Bulk import boxes: imported-box tasks only work on their own import's boxes

src/main/java/org/ecocean/Annotation.java

@@ -409,6 +409,11 @@ public String importedBoxImportTaskId() {
         return (marker == null) ? null : marker.optString("importTaskId", null);
     }
 
+    // true only for a box that the given bulk import created
+    public boolean isImportedBoxOf(String importTaskId) {
+        return Util.stringExists(importTaskId) && importTaskId.equals(importedBoxImportTaskId());
+    }
+
     public void setFeatures(ArrayList<Feature> f) {
         features = f;
         this.setVersion();

src/main/java/org/ecocean/ia/IA.java

@@ -1025,11 +1025,11 @@ private static Task errorChildHolding(Task parent, List<Annotation> anns, String
      * instead of committing a holder that omits an existing box.
      */
     private static Task errorChildReloading(Shepherd myShepherd, Task parent, List<String> annIds,
-        String code, String message) {
+        String importId, String code, String message) {
         List<Annotation> anns = new ArrayList<Annotation>();
         for (String id : annIds) {
             Annotation a = strictAnnotation(myShepherd, id);
-            if (a != null) anns.add(a);
+            if ((a != null) && a.isImportedBoxOf(importId)) anns.add(a);
         }
         return errorChildHolding(parent, anns, code, message);
     }
@@ -1047,7 +1047,12 @@ static int intakeImportedBoxes(Shepherd myShepherd, List<String> inputIds, Task
         String context, String baseUrl) {
         if ((inputIds == null) || inputIds.isEmpty() || (root == null)) return 0;
         boolean skipIdent = IBEISIA.taskParametersSkipIdent(root);
+        // extraction mode is only for the boxes of the import this root was created for: any
+        // other annotation in the envelope is left exactly as it was (no task, no status change)
+        String importId = (root.getParameters() == null) ? null
+            : root.getParameters().optString("importTaskId", null);
         List<StagedBox> staged = new ArrayList<StagedBox>();
+        int foreign = 0;   // named, existing, but not a box of this root's import
         Map<String, Task> speciesTasks = new LinkedHashMap<String, Task>();
         Map<String, List<Annotation>> wbiaSpecies = new LinkedHashMap<String, List<Annotation>>();
         boolean committed = false;
@@ -1068,6 +1073,12 @@ static int intakeImportedBoxes(Shepherd myShepherd, List<String> inputIds, Task
                         " no longer exists; skipped");
                     continue;
                 }
+                if (!ann.isImportedBoxOf(importId)) {
+                    System.out.println("WARN: IA.intakeImportedBoxes() annotation " + id +
+                        " is not a box of import " + importId + "; skipped");
+                    foreign++;
+                    continue;
+                }
                 MediaAsset ma = ann.getMediaAsset();
                 Taxonomy taxy = ann.getTaxonomy(myShepherd);
                 if ((ma == null) || (taxy == null) || !Util.stringExists(taxy.getScientificName())) {
@@ -1080,6 +1091,24 @@ static int intakeImportedBoxes(Shepherd myShepherd, List<String> inputIds, Task
                 if (!bySpecies.containsKey(key)) bySpecies.put(key, new ArrayList<Annotation>());
                 bySpecies.get(key).add(ann);
             }
+            if (bySpecies.isEmpty() && noTaxonomy.isEmpty()) {
+                // nothing was taken, so no task below will ever end the root: end it here, so a
+                // client polling it does not wait forever. Boxes of another import (or a root with
+                // no import) make the envelope invalid; boxes that were all deleted leave nothing to do.
+                if (foreign > 0) {
+                    markTaskError(root, "INVALID_ENVELOPE", foreign +
+                        " named annotation(s) are not boxes of import " + importId);
+                } else {
+                    root.setStatus("completed");
+                    root.setCompletionDateInMilliseconds();
+                }
+                myShepherd.getPM().makePersistent(root);
+                // a failed commit goes to the staging recovery below, which ends the root in error
+                if (!myShepherd.commitDBTransactionWithStatus())
+                    throw new IllegalStateException("could not commit the end of a root that took no box");
+                myShepherd.beginDBTransaction();
+                return 0;
+            }
             if (!noTaxonomy.isEmpty()) {
                 Task t = errorChildHolding(root, noTaxonomy, "NO_TAXONOMY",
                     "imported box has no media asset or no taxonomy");
@@ -1144,8 +1173,8 @@ static int intakeImportedBoxes(Shepherd myShepherd, List<String> inputIds, Task
                     throw new IllegalStateException("transaction not active after restart");
                 // reload every box: a lookup failure here fails the recovery (below) rather than
                 // committing a holder that omits an existing box
-                Task t = errorChildReloading(myShepherd, root, inputIds, "EXTRACTION_PERSIST",
-                    stagingFailure);
+                Task t = errorChildReloading(myShepherd, root, inputIds, importId,
+                    "EXTRACTION_PERSIST", stagingFailure);
                 markTaskError(root, "EXTRACTION_PERSIST", stagingFailure);
                 myShepherd.getPM().makePersistent(t);
                 myShepherd.getPM().makePersistent(root);

src/main/java/org/ecocean/ia/ImportedBoxReplay.java

@@ -20,32 +20,55 @@
  */
 public final class ImportedBoxReplay {
     public static final int MAX_PER_RUN = 500;
+    // bound on candidate rows examined per run, so rows the text match catches but that are not
+    // replayable (no resume message, or the marker is not the top-level flag) cannot stall startup
+    static final int MAX_SCANNED = 20 * MAX_PER_RUN;
     static final String MARKER = "\"" + Task.PARAM_IMPORTED_BOX + "\":true";
 
     private ImportedBoxReplay() {}
 
-    /** The resume messages of non-terminal imported-box children modified before the cutoff, oldest first. */
+    /**
+     * The resume messages of non-terminal imported-box children modified before the cutoff, oldest
+     * first, at most {@code max}. The text match is only a prefilter, so candidates are read in
+     * pages and only real imported-box children with a resume message count toward {@code max}.
+     */
     static List<String> fetchStaleResumeMessages(Shepherd shep, long modifiedCutoff, int max) {
+        return fetchStaleResumeMessages(shep, modifiedCutoff, max, MAX_SCANNED);
+    }
+
+    static List<String> fetchStaleResumeMessages(Shepherd shep, long modifiedCutoff, int max,
+        int maxScanned) {
         List<String> messages = new ArrayList<String>();
+        int limit = Math.max(1, max);
         // the marker is a JSON fragment; DataNucleus supports String.indexOf in JDOQL (see MetricsBot)
         String jdoql = "SELECT FROM org.ecocean.ia.Task WHERE " +
             "(status == null || (status != 'completed' && status != 'error' && status != 'dropped-stale'))" +
             " && modified < " + modifiedCutoff +
             " && parameters.indexOf('" + MARKER + "') > -1";
-        Query q = shep.getPM().newQuery(jdoql);
-        try {
-            q.setOrdering("modified ascending");
-            q.setRange(0, Math.max(1, max));
-            @SuppressWarnings("unchecked")
-            List<Task> stale = (List<Task>)q.execute();
-            if (stale != null) {
-                for (Task t : stale) {
-                    String msg = t.getQueueResumeMessage();
-                    if (msg != null) messages.add(msg);
+        long offset = 0;
+        while ((messages.size() < limit) && (offset < maxScanned)) {
+            int pageSize = (int)Math.min(limit, maxScanned - offset);   // never past the scan bound
+            Query q = shep.getPM().newQuery(jdoql);
+            int rows = 0;
+            try {
+                q.setOrdering("modified ascending, id ascending");   // stable across pages
+                q.setRange(offset, offset + pageSize);
+                @SuppressWarnings("unchecked")
+                List<Task> page = (List<Task>)q.execute();
+                if (page != null) {
+                    for (Task t : page) {
+                        rows++;
+                        // the marker must be the real top-level flag, not text found anywhere
+                        if (!Task.isImportedBoxTask(t)) continue;
+                        String msg = t.getQueueResumeMessage();
+                        if ((msg != null) && (messages.size() < limit)) messages.add(msg);
+                    }
                 }
+            } finally {
+                q.closeAll();
             }
-        } finally {
-            q.closeAll();
+            if (rows < pageSize) break;   // no further candidates
+            offset += rows;
         }
         return messages;
     }

src/main/java/org/ecocean/ia/MlServiceProcessor.java

@@ -80,12 +80,29 @@ public MlServiceJobOutcome process(JSONObject jobData) {
         if (jobData == null) {
             return MlServiceJobOutcome.validationError("INVALID_PAYLOAD", "payload is null");
         }
-        if (jobData.optBoolean("deferredMatch", false)) {
+        String taskId = jobData.optString("taskId", null);
+        boolean deferred = jobData.optBoolean("deferredMatch", false);
+        List<String> named = deferred
+            ? jsonArrayToStringList(jobData.optJSONArray("annotationIds")) : new ArrayList<String>();
+        // a payload that would be dispatched as detection (mediaAssetId wins below) names no box
+        if (!deferred && jobData.has("annotationId") && !jobData.has("mediaAssetId"))
+            named.add(jobData.optString("annotationId", null));
+        String mismatch;
+        try {
+            mismatch = importedBoxJobMismatch(taskId, named);
+        } catch (Exception ex) {
+            return MlServiceJobOutcome.persistError("PERSIST", "imported-box check failed: " + ex);
+        }
+        if (mismatch != null) {
+            System.out.println("WARN: MlServiceProcessor.process() dropped a job for task " + taskId +
+                ": " + mismatch);
+            return MlServiceJobOutcome.stale(mismatch);
+        }
+        if (deferred) {
             return runDeferredMatch(jobData);
         }
 
         String taxonomyString = jobData.optString("taxonomyString", null);
-        String taskId = jobData.optString("taskId", null);
         String encounterId = jobData.optString("encounterId", null);
 
         if (jobData.has("mediaAssetId")) {
@@ -100,6 +117,69 @@ public MlServiceJobOutcome process(JSONObject jobData) {
             "neither mediaAssetId nor annotationId in payload");
     }
 
+    /**
+     * Bulk-import boxes (spec 4.4): an imported-box child works on exactly one box, the one named
+     * in the job the server stored on it when it created the child, and that box must carry the
+     * stamp of the child's import. Returns why a job must be dropped without any write, or null
+     * when it may run (always null for a task that is not an imported-box child). Reads only.
+     */
+    String importedBoxJobMismatch(String taskId, List<String> annIds) {
+        if (taskId == null) return null;
+        Shepherd shep = openShepherd("importedBoxJobMismatch");
+        try {
+            shep.beginDBTransaction();
+            // strict lookups: only a genuinely absent row reads as absent; a datastore failure
+            // propagates, so an outage can never look like "not an imported-box task"
+            Task task;
+            try {
+                task = (Task)shep.getPM().getObjectById(
+                    shep.getPM().newObjectIdInstance(Task.class, taskId), true);
+            } catch (javax.jdo.JDOObjectNotFoundException notFound) {
+                return null;
+            }
+            if (task.hasUnreadableParameters()) {
+                // fail closed when anything persisted says this could be an imported-box child;
+                // any other task with unreadable parameters is not one, and its job runs as before
+                if (mayBeImportedBoxChild(task))
+                    throw new IllegalStateException("task " + taskId + " has unreadable parameters");
+                return null;
+            }
+            if (!Task.isImportedBoxTask(task)) return null;
+            String boxId = null;
+            String stored = task.getQueueResumeMessage();
+            if (stored != null) boxId = new JSONObject(stored).optString("annotationId", null);
+            if (boxId == null) return "imported-box task has no stored box";
+            if ((annIds == null) || (annIds.size() != 1) || !boxId.equals(annIds.get(0)))
+                return "the job does not name this task's box";
+            Annotation ann = IA.strictAnnotation(shep, boxId);   // null: deleted, handled downstream
+            if ((ann != null) &&
+                !ann.isImportedBoxOf(task.getParameters().optString("importTaskId", null)))
+                return "the annotation is not a box of this task's import";
+            return null;
+        } finally {
+            shep.rollbackAndClose();
+        }
+    }
+
+    /**
+     * For a task whose parameters cannot be read: true when other persisted evidence could make it
+     * an imported-box child, namely the marker text, a stored job that names an annotation (only
+     * extraction children store one), or a held annotation that is an imported box.
+     */
+    static boolean mayBeImportedBoxChild(Task task) {
+        if (task.rawParametersMentionImportedBox()) return true;
+        String stored = task.getQueueResumeMessage();
+        if (stored != null) {
+            JSONObject job = Util.stringToJSONObject(stored);
+            if ((job == null) || job.has("annotationId")) return true;   // unreadable job: fail closed
+        }
+        List<Annotation> held = task.getObjectAnnotations();
+        if (held != null) {
+            for (Annotation ann : held) if ((ann != null) && ann.isImportedBox()) return true;
+        }
+        return false;
+    }
+
     private MlServiceJobOutcome processDetection(JSONObject jobData, String taxonomyString,
         String taskId, String encounterId, String maId) {
         DetectionContext det = null;

src/main/java/org/ecocean/ia/Task.java

@@ -456,6 +456,38 @@ public JSONObject getParameters() { // only return as JSONObject!
         return Util.stringToJSONObject(parameters);
     }
 
+    // stored parameters that do not parse: getParameters() answers null for these, exactly as
+    // it does for a task with no parameters, so a caller that must tell the two apart asks here
+    public boolean hasUnreadableParameters() {
+        return (parameters != null) && !parameters.trim().isEmpty() && (getParameters() == null);
+    }
+
+    // the stored parameter text mentions the imported-box marker key at all, readable or not
+    public boolean rawParametersMentionImportedBox() {
+        return (parameters != null) && parameters.contains(PARAM_IMPORTED_BOX);
+    }
+
+    /**
+     * Removes the imported-box marker key at every depth of client-supplied parameters: only the
+     * server sets it, on the children it creates (spec 4.4). Returns its argument, for chaining.
+     */
+    public static JSONObject stripImportedBoxMarker(JSONObject params) {
+        stripImportedBoxMarkerFrom(params);
+        return params;
+    }
+
+    private static void stripImportedBoxMarkerFrom(Object value) {
+        if (value instanceof JSONObject) {
+            JSONObject obj = (JSONObject)value;
+            obj.remove(PARAM_IMPORTED_BOX);
+            for (String key : new java.util.ArrayList<String>(obj.keySet()))
+                stripImportedBoxMarkerFrom(obj.opt(key));
+        } else if (value instanceof org.json.JSONArray) {
+            org.json.JSONArray arr = (org.json.JSONArray)value;
+            for (int i = 0; i < arr.length(); i++) stripImportedBoxMarkerFrom(arr.opt(i));
+        }
+    }
+
     // see comment above: should this even be public?  (or exist)
     public void setParameters(String s) { // best be json, yo
         parameters = s;

src/main/java/org/ecocean/servlet/IAGateway.java

@@ -478,7 +478,9 @@ static Task resolveGatewayRoot(Task loaded, String taskId, JSONObject j, JSONObj
         // a legacy identify request ({identify:{annotationIds}}) names its own identification
         // task, which holds the query annotation: reuse it unless it belongs to a run
         boolean identifyJob = (j != null) && (j.optJSONObject("identify") != null);
+        // an imported-box child is server-owned: a request never runs under it or rewrites it
         boolean replace = (loaded != null) && ((Task.runIdOf(loaded) != null) ||
+            Task.isImportedBoxTask(loaded) ||
             (!identifyJob && org.ecocean.ia.IA.mustReplaceRequestRoot(loaded)));
         if (replace) {
             Task fresh = new Task();
@@ -510,6 +512,8 @@ static JSONObject gatewayTaskParams(JSONObject j, User tuser) {
             tparams.put("userId", tuser.getUUID());
             tparams.put("username", tuser.getUsername());
         }
+        // the imported-box marker is only ever set by the server, on the children it creates
+        Task.stripImportedBoxMarker(tparams);
         return org.ecocean.ia.IA.scrubRunParams(tparams);
     }
 
@@ -1345,8 +1349,7 @@ static List<String> importedBoxAnnotationIds(ImportTask itask) {
         for (Encounter enc : itask.getEncounters()) {
             if ((enc == null) || (enc.getAnnotations() == null)) continue;
             for (Annotation ann : enc.getAnnotations()) {
-                if ((ann == null) || !ann.isImportedBox()) continue;
-                if ((importId == null) || !importId.equals(ann.importedBoxImportTaskId())) continue;
+                if ((ann == null) || !ann.isImportedBoxOf(importId)) continue;
                 if (!ids.contains(ann.getId())) ids.add(ann.getId());
             }
         }
@@ -1430,6 +1433,8 @@ public static JSONObject handleBulkImport(JSONObject jin, JSONObject res, Shephe
         JSONObject taskParameters = jin.optJSONObject("taskParameters");
         String importTaskId = null;
         if (taskParameters != null) importTaskId = taskParameters.optString("importTaskId", null);
+        // the imported-box marker is only ever set by the server, on the children it creates
+        if (taskParameters != null) Task.stripImportedBoxMarker(taskParameters);
         ImportTask itask = null;
         if (importTaskId != null) itask = myShepherd.getImportTask(importTaskId);
         if (itask == null) {

src/test/java/org/ecocean/api/submission/SubmissionProcessingTest.java

@@ -173,7 +173,7 @@ static ImportTask boxedImportTask() {
         ImportTask task = mock(ImportTask.class);
         when(task.getId()).thenReturn("job"); when(task.annotationsImported()).thenReturn(true);
         org.ecocean.Annotation ann = mock(org.ecocean.Annotation.class);
-        when(ann.isImportedBox()).thenReturn(true); when(ann.importedBoxImportTaskId()).thenReturn("job"); when(ann.getId()).thenReturn("ann-1");
+        when(ann.isImportedBox()).thenReturn(true); when(ann.importedBoxImportTaskId()).thenReturn("job"); when(ann.isImportedBoxOf("job")).thenReturn(true); when(ann.getId()).thenReturn("ann-1");
         org.ecocean.Encounter enc = mock(org.ecocean.Encounter.class);
         when(enc.getAnnotations()).thenReturn(new java.util.ArrayList<>(java.util.List.of(ann)));
         when(task.getEncounters()).thenReturn(new java.util.ArrayList<>(java.util.List.of(enc)));