Make the OpenSearch max_result_window raise take effect after the first time

src/main/java/org/ecocean/OpenSearch.java

@@ -1196,16 +1196,57 @@ public Map<String, Long> getAllVersions(String indexName)
                 res = queryRawScroll(query);
             }
         }
-        // this is a little hacky, but allows us to page thru results enough to cover what we have
-        if (versions.size() > 10000) {
-            putSettings(indexName,
-                new JSONObject("{\"index.max_result_window\": " +
-                Math.round(1.2 * versions.size()) + "}"));
-            MAX_RESULT_WINDOW_CACHE.remove(indexName);
-        }
+        // keep deep paging (from + size) able to reach everything we have indexed
+        ensureMaxResultWindow(indexName, versions.size());
         return versions;
     }
 
+    /**
+     * The index.max_result_window a sync pass should store for an index holding indexedDocCount
+     * top-level documents: 1.2x the count (headroom for growth between passes), or -1 when the
+     * index fits inside the default window and the setting should be left alone.
+     */
+    static int targetMaxResultWindow(long indexedDocCount) {
+        if (indexedDocCount <= DEFAULT_MAX_RESULT_WINDOW) return -1;
+        return (int)Math.min(Integer.MAX_VALUE, Math.round(1.2d * indexedDocCount));
+    }
+
+    /**
+     * Grow-only: raise index.max_result_window to targetMaxResultWindow() when that exceeds the
+     * value stored on the index. The stored value is read fresh and strictly (never the 60 s cache
+     * or the tolerant default), and any read failure skips the update, so a transient error can
+     * never lower a window, e.g. one an operator raised by hand. Grow-only assumes serialized
+     * settings writers: the GET and the PUT are not atomic, so a concurrent manual increase can
+     * be overwritten and is not automatically restored.
+     */
+    void ensureMaxResultWindow(String indexName, long indexedDocCount)
+    throws IOException {
+        int target = targetMaxResultWindow(indexedDocCount);
+        if (target < 0) return;
+        int current;
+        try {
+            current = readMaxResultWindow(indexName);
+        } catch (Exception ex) {
+            System.out.println("OpenSearch.ensureMaxResultWindow(" + indexName +
+                ") skipped: could not read current setting: " + ex);
+            return;
+        }
+        if (target <= current) return;
+        putSettings(indexName, new JSONObject().put("index.max_result_window", target));
+        MAX_RESULT_WINDOW_CACHE.remove(indexName);
+    }
+
+    // strict read for update decisions: throws on transport or response-shape errors and on a
+    // value that is not an integer; only a settings response that omits the key means the default
+    private int readMaxResultWindow(String indexName)
+    throws IOException {
+        JSONObject index = new JSONObject(getRestResponse(new Request("GET",
+            indexName + "/_settings"))).getJSONObject(indexName).getJSONObject("settings")
+            .getJSONObject("index");
+        if (!index.has("max_result_window")) return DEFAULT_MAX_RESULT_WINDOW;
+        return Integer.parseInt(index.get("max_result_window").toString());
+    }
+
     public JSONObject getSettings(final String indexName)
     throws IOException {
         Request settingsRequest = new Request("GET", indexName + "/_settings");
@@ -1226,8 +1267,11 @@ public JSONObject getSettings(final String indexName)
     public void putSettings(final String indexName, final JSONObject settings)
     throws IOException {
         if (settings == null) throw new IOException("null data passed");
-        Request settingsRequest = new Request("PUT",
-            indexName + "/_settings?preserve_existing=true");
+        // no preserve_existing: that flag leaves an already-stored setting untouched, which made
+        // every max_result_window raise after the first one a silent no-op. The other caller is
+        // createIndex(), whose wildbook_keyword_normalizer definition is authoritative, so an
+        // index template that pre-populated that key is overwritten by design.
+        Request settingsRequest = new Request("PUT", indexName + "/_settings");
         settingsRequest.setJsonEntity(settings.toString());
         String rtn = getRestResponse(settingsRequest);
         System.out.println("OpenSearch.putSettings() on " + indexName + ": " + settings + " => " +

src/test/java/org/ecocean/OpenSearchPaginationTest.java

@@ -1,9 +1,17 @@
 package org.ecocean;
 
 import static org.junit.jupiter.api.Assertions.*;
-
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.spy;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.http.util.EntityUtils;
 import org.json.JSONObject;
 import org.junit.jupiter.api.Test;
+import org.opensearch.client.Request;
 
 class OpenSearchPaginationTest {
     // ===== parseMaxResultWindow: reads the live index setting, defaults to 10000 =====
@@ -72,4 +80,145 @@ class OpenSearchPaginationTest {
     @Test void staleContext_nullMessageIsNotRetryable() {
         assertFalse(OpenSearch.isStaleSearchContextError(null));
     }
+
+    // ===== max_result_window bump (background sync): grow-only, strict read, no preserve_existing =====
+
+    private static String settingsJson(String indexName, Object window) {
+        JSONObject index = new JSONObject();
+        if (window != null) index.put("max_result_window", window);
+        return new JSONObject().put(indexName, new JSONObject().put("settings",
+            new JSONObject().put("index", index))).toString();
+    }
+
+    /** spy whose GET _settings answers with getBody (or throws getFailure) and which records every PUT */
+    private static OpenSearch stubbed(String getBody, IOException getFailure, List<Request> puts)
+    throws Exception {
+        OpenSearch os = spy(new OpenSearch());
+        doAnswer(inv -> {
+            Request request = inv.getArgument(0);
+            if ("PUT".equals(request.getMethod())) {
+                puts.add(request);
+                return "{\"acknowledged\":true}";
+            }
+            if (getFailure != null) throw getFailure;
+            return getBody;
+        }).when(os).getRestResponse(any(Request.class));
+        return os;
+    }
+
+    private static int windowIn(Request put) throws Exception {
+        return new JSONObject(EntityUtils.toString(put.getEntity())).getInt("index.max_result_window");
+    }
+
+    @Test void targetMaxResultWindow_isOnePointTwoTimesTheCount() {
+        assertEquals(14624, OpenSearch.targetMaxResultWindow(12187)); // the stale value seen in the field
+        assertEquals(653682, OpenSearch.targetMaxResultWindow(544735));
+        assertEquals(12001, OpenSearch.targetMaxResultWindow(10001));
+    }
+
+    @Test void targetMaxResultWindow_leavesIndexesAtOrBelowTheDefaultAlone() {
+        assertEquals(-1, OpenSearch.targetMaxResultWindow(10000));
+        assertEquals(-1, OpenSearch.targetMaxResultWindow(0));
+        assertEquals(-1, OpenSearch.targetMaxResultWindow(-5));
+    }
+
+    @Test void targetMaxResultWindow_clampsToIntegerRange() {
+        assertEquals(Integer.MAX_VALUE, OpenSearch.targetMaxResultWindow(Integer.MAX_VALUE));
+        assertEquals(Integer.MAX_VALUE, OpenSearch.targetMaxResultWindow(Long.MAX_VALUE));
+    }
+
+    @Test void ensureMaxResultWindow_raisesStoredWindowWithoutPreserveExisting() throws Exception {
+        List<Request> puts = new ArrayList<Request>();
+        OpenSearch os = stubbed(settingsJson("encounter", "14875"), null, puts);
+        os.ensureMaxResultWindow("encounter", 544735);
+        assertEquals(1, puts.size());
+        assertEquals("encounter/_settings", puts.get(0).getEndpoint());
+        assertTrue(puts.get(0).getParameters().isEmpty(),
+            "preserve_existing=true made every update after the first a no-op");
+        assertEquals(653682, windowIn(puts.get(0)));
+    }
+
+    @Test void ensureMaxResultWindow_neverLowersALargerStoredWindow() throws Exception {
+        List<Request> puts = new ArrayList<Request>();
+        OpenSearch os = stubbed(settingsJson("encounter", "660000"), null, puts);
+        os.ensureMaxResultWindow("encounter", 544735);
+        assertTrue(puts.isEmpty(), "a hand-raised window must survive the sync");
+    }
+
+    @Test void ensureMaxResultWindow_skipsWhenSettingsReadFails() throws Exception {
+        List<Request> puts = new ArrayList<Request>();
+        OpenSearch os = stubbed(null, new IOException("connection refused"), puts);
+        os.ensureMaxResultWindow("encounter", 544735);
+        assertTrue(puts.isEmpty(), "a failed read must not be mistaken for the default window");
+    }
+
+    @Test void ensureMaxResultWindow_skipsWhenSettingsResponseIsMalformed() throws Exception {
+        List<Request> puts = new ArrayList<Request>();
+        OpenSearch os = stubbed("{}", null, puts);
+        os.ensureMaxResultWindow("encounter", 544735);
+        assertTrue(puts.isEmpty());
+    }
+
+    @Test void ensureMaxResultWindow_skipsWhenStoredValueIsUnparseable() throws Exception {
+        List<Request> puts = new ArrayList<Request>();
+        OpenSearch os = stubbed(settingsJson("encounter", "lots"), null, puts);
+        os.ensureMaxResultWindow("encounter", 544735);
+        assertTrue(puts.isEmpty());
+    }
+
+    @Test void ensureMaxResultWindow_treatsAnAbsentSettingAsTheDefault() throws Exception {
+        List<Request> puts = new ArrayList<Request>();
+        OpenSearch os = stubbed(settingsJson("encounter", null), null, puts);
+        os.ensureMaxResultWindow("encounter", 12187);
+        assertEquals(1, puts.size());
+        assertEquals(14624, windowIn(puts.get(0)));
+    }
+
+    @Test void ensureMaxResultWindow_doesNotTouchOpenSearchForSmallIndexes() throws Exception {
+        List<Request> all = new ArrayList<Request>();
+        OpenSearch os = spy(new OpenSearch());
+        doAnswer(inv -> {
+            all.add(inv.getArgument(0));
+            return settingsJson("encounter", "10000");
+        }).when(os).getRestResponse(any(Request.class));
+        os.ensureMaxResultWindow("encounter", 10000);
+        assertTrue(all.isEmpty(), "no GET and no PUT below the default window");
+    }
+
+    @Test void ensureMaxResultWindow_decidesFromTheLiveSettingNotTheCache() throws Exception {
+        List<Request> puts = new ArrayList<Request>();
+        String[] stored = { "14875" };
+        OpenSearch os = spy(new OpenSearch());
+        doAnswer(inv -> {
+            Request request = inv.getArgument(0);
+            if ("PUT".equals(request.getMethod())) {
+                puts.add(request);
+                return "{\"acknowledged\":true}";
+            }
+            return settingsJson("individual", stored[0]);
+        }).when(os).getRestResponse(any(Request.class));
+        assertEquals(14875, os.getMaxResultWindow("individual")); // primes the 60 s cache
+        stored[0] = "660000"; // an operator raised it by hand since
+        os.ensureMaxResultWindow("individual", 544735);
+        assertTrue(puts.isEmpty(), "a stale cached value must not justify lowering the live setting");
+    }
+
+    @Test void ensureMaxResultWindow_refreshesTheCachedWindowRightAway() throws Exception {
+        String[] stored = { "14875" };
+        OpenSearch os = spy(new OpenSearch());
+        doAnswer(inv -> {
+            Request request = inv.getArgument(0);
+            if ("PUT".equals(request.getMethod())) {
+                assertEquals("annotation/_settings", request.getEndpoint());
+                assertTrue(request.getParameters().isEmpty());
+                stored[0] = Integer.toString(windowIn(request));
+                return "{\"acknowledged\":true}";
+            }
+            return settingsJson("annotation", stored[0]);
+        }).when(os).getRestResponse(any(Request.class));
+        assertEquals(14875, os.getMaxResultWindow("annotation")); // primes the 60 s cache
+        os.ensureMaxResultWindow("annotation", 786983);
+        assertEquals(944380, os.getMaxResultWindow("annotation"),
+            "the API gate must see the new window without waiting for the cache to expire");
+    }
 }