K2GO-376 fix(content-exchange): longer import read timeout for slow Kolibri

controller/app/src/main/java/org/appdevforall/k2go/contentexchange/data/ConexCoreDataSource.java

@@ -75,11 +75,30 @@ public JSONObject postImportValidate(JSONObject body) throws Exception {
         return postJson(V1 + "/import/validate", body);
     }
 
-    /** {@code POST /v1/import/item} with {@code {app, entry, payloadPath}} -> the ingest result. */
-    public JSONObject postImportItem(JSONObject body) throws Exception {
+    /**
+     * {@code POST /v1/import/item} with {@code {app, entry, payloadPath}} -> {@code {opId, state,
+     * totalBytes}}. Starts the import long operation (K2GO-376) and returns fast, so a 35-141 s
+     * Kolibri import does not block one request past the dash-node/nginx relay read timeout.
+     */
+    public JSONObject postImportStart(JSONObject body) throws Exception {
         return postJson(V1 + "/import/item", body);
     }
 
+    /**
+     * {@code GET /v1/import/status/{opId}} -> {@code {state, movedBytes, totalBytes, error?, code?,
+     * result?}}. A 404 means the op was swept; it surfaces as {@link ConexHttpException} with
+     * {@code code() == 404} so the caller re-starts rather than poll a dead id. Each poll is a short
+     * request on the generic read timeout. K2GO-376.
+     */
+    public JSONObject getImportStatus(String opId) throws Exception {
+        return getJson(V1 + "/import/status/" + encodeSegment(opId));
+    }
+
+    /** {@code DELETE /v1/import/{opId}} -> {@code {ok}}. Client abort of an in-flight import op. K2GO-376. */
+    public JSONObject deleteImportOp(String opId) throws Exception {
+        return deleteJson(V1 + "/import/" + encodeSegment(opId));
+    }
+
     /**
      * {@code DELETE /v1/content/{app}/{localRef}} -> {@code {ok, freedBytes}} on success or
      * {@code {ok:false, reason}} on an adapter failure. Path segments are percent-encoded (the core

controller/app/src/main/java/org/appdevforall/k2go/contentexchange/data/ConexCoreImportSource.java

@@ -12,8 +12,8 @@
 package org.appdevforall.k2go.contentexchange.data;
 
 import org.appdevforall.k2go.contentexchange.domain.ImportEntry;
-import org.appdevforall.k2go.contentexchange.domain.ImportNoSpaceException;
 import org.appdevforall.k2go.contentexchange.domain.ImportSource;
+import org.appdevforall.k2go.contentexchange.domain.ImportStatus;
 import org.appdevforall.k2go.contentexchange.domain.ImportVerdict;
 import org.appdevforall.k2go.contentexchange.domain.ItemRef;
 import org.json.JSONArray;
@@ -54,24 +54,60 @@ public List<ImportVerdict> validate(List<ImportEntry> entries) throws Exception
     }
 
     @Override
-    public void importItem(String app, String entryJson, String payloadBoxPath) throws Exception {
+    public String startImport(String app, String entryJson, String payloadBoxPath) throws Exception {
         JSONObject body = new JSONObject()
                 .put("app", app)
                 .put("entry", new JSONObject(entryJson))
                 .put("payloadPath", payloadBoxPath);
+        JSONObject r = dataSource.postImportStart(body);
+        String opId = r.optString("opId", "");
+        if (opId.isEmpty()) {
+            throw new Exception("import start returned no opId");
+        }
+        return opId;
+    }
+
+    @Override
+    public ImportStatus importStatus(String opId) throws Exception {
+        JSONObject r;
         try {
-            JSONObject result = dataSource.postImportItem(body);
-            if (!result.optBoolean("ok", false)) {
-                throw new Exception(result.optString("reason", result.optString("error", "import failed")));
-            }
+            r = dataSource.getImportStatus(opId);
         } catch (ConexCoreDataSource.ConexHttpException e) {
-            // 507 = a stage ran out of disk (the core tags the result code "no_space"). Map it to a
-            // typed signal so the service stops the run with a space message, not a per-item retry.
-            if (e.code() == 507) {
-                throw new ImportNoSpaceException(e.getMessage());
+            if (e.code() == 404) {
+                return ImportStatus.gone();   // the op was swept: the caller re-starts
             }
             throw e;
         }
+        ImportStatus.State state;
+        switch (r.optString("state", "")) {
+            case "done":
+                state = ImportStatus.State.DONE;
+                break;
+            case "error":
+                state = ImportStatus.State.ERROR;
+                break;
+            default:
+                state = ImportStatus.State.RUNNING;
+                break;
+        }
+        // The core tags an out-of-space failure with code "no_space" so the run stops with a space
+        // message, not a per-item retry.
+        String error = r.isNull("error") ? null : r.optString("error", null);
+        boolean noSpace = "no_space".equals(r.optString("code", ""));
+        return new ImportStatus(state, r.optLong("movedBytes", 0L), r.optLong("totalBytes", 0L),
+                error, noSpace);
+    }
+
+    @Override
+    public void abortImport(String opId) {
+        if (opId == null || opId.isEmpty()) {
+            return;
+        }
+        try {
+            dataSource.deleteImportOp(opId);
+        } catch (Exception ignored) {
+            // Best-effort: a swept or already-dropped op 404s; the box's TTL sweep is the backstop.
+        }
     }
 
     @Override

controller/app/src/main/java/org/appdevforall/k2go/contentexchange/domain/ImportSource.java

@@ -19,11 +19,22 @@ public interface ImportSource {
     List<ImportVerdict> validate(List<ImportEntry> entries) throws Exception;
 
     /**
-     * Ingest one item whose archive already sits at {@code payloadBoxPath}: a box-visible path inside
-     * the core's import area. The core delegates to the app's own import. Throws when the core rejects
-     * the item (bad path, no space, adapter failure).
+     * Start the import long operation for one item whose archive already sits at
+     * {@code payloadBoxPath}: a box-visible path inside the core's import area. Returns the opId to
+     * poll (K2GO-376). A Kolibri import takes 35 to 141 s, so it cannot be one blocking request; the
+     * caller starts it here and polls {@link #importStatus}. Throws when the core rejects the start
+     * (bad path, unknown app).
      */
-    void importItem(String app, String entryJson, String payloadBoxPath) throws Exception;
+    String startImport(String app, String entryJson, String payloadBoxPath) throws Exception;
+
+    /**
+     * Poll one import operation's status. {@link ImportStatus.State#GONE} means the op was swept, so
+     * the caller re-starts once. Throws on a transport failure (the caller tolerates a few blips).
+     */
+    ImportStatus importStatus(String opId) throws Exception;
+
+    /** Abort one import operation (best-effort): the box stops the ingest and drops the op. */
+    void abortImport(String opId);
 
     /**
      * Delete one item from the box, delegating to the app's own delete. Available for the replace-older

controller/app/src/main/java/org/appdevforall/k2go/contentexchange/domain/ImportStatus.java

@@ -0,0 +1,68 @@
+/*
+ * ============================================================================
+ * Name        : ImportStatus.java
+ * Author      : AppDevForAll
+ * Copyright   : Copyright (c) 2026 AppDevForAll
+ * Description : K2GO-376. One poll of an import long operation's progress on the box. Pure value,
+ *               no Android or JSON: the Data layer maps the REST status to this.
+ * ============================================================================
+ */
+package org.appdevforall.k2go.contentexchange.domain;
+
+/**
+ * A single status reading for an in-flight import operation. The import ingests an already-staged
+ * item on the box (a Kolibri channel import takes 35 to 141 s), so the app polls this instead of
+ * blocking one long request (which times out at the relay).
+ *
+ * <p>{@code movedBytes} is the server-side progress and the liveness heartbeat. {@link State#GONE}
+ * marks an operation the box no longer knows (its registry entry was swept), which tells the caller
+ * to re-start rather than keep polling a dead id. {@link #noSpace()} is set when the box reported a
+ * structural out-of-space code, so the caller stops the run with a space message.
+ */
+public final class ImportStatus {
+
+    public enum State { RUNNING, DONE, ERROR, GONE }
+
+    private final State state;
+    private final long movedBytes;
+    private final long totalBytes;
+    private final String error;
+    private final boolean noSpace;
+
+    public ImportStatus(State state, long movedBytes, long totalBytes, String error, boolean noSpace) {
+        this.state = state != null ? state : State.ERROR;
+        this.movedBytes = Math.max(0L, movedBytes);
+        this.totalBytes = Math.max(0L, totalBytes);
+        this.error = error;
+        this.noSpace = noSpace;
+    }
+
+    /** The operation was swept and is unknown to the box: the caller re-starts. */
+    public static ImportStatus gone() {
+        return new ImportStatus(State.GONE, 0L, 0L, null, false);
+    }
+
+    public State state() {
+        return state;
+    }
+
+    /** Bytes moved so far on the box (server-side progress and the liveness signal). */
+    public long movedBytes() {
+        return movedBytes;
+    }
+
+    /** The item's total size in bytes (0 when unknown). */
+    public long totalBytes() {
+        return totalBytes;
+    }
+
+    /** The failure detail when {@link #state} is {@link State#ERROR}, else null. */
+    public String error() {
+        return error;
+    }
+
+    /** True when the failure is a structural out-of-space: the run stops with a space message. */
+    public boolean noSpace() {
+        return noSpace;
+    }
+}

controller/app/src/main/java/org/appdevforall/k2go/contentexchange/presentation/ContentImportService.java

@@ -41,6 +41,7 @@
 import org.appdevforall.k2go.contentexchange.domain.ImportCarrier;
 import org.appdevforall.k2go.contentexchange.domain.ImportNoSpaceException;
 import org.appdevforall.k2go.contentexchange.domain.ImportSource;
+import org.appdevforall.k2go.contentexchange.domain.ImportStatus;
 import org.appdevforall.k2go.contentexchange.domain.ItemRef;
 import org.appdevforall.k2go.download.domain.DownloadEta;
 import org.appdevforall.k2go.notifications.NotificationIds;
@@ -81,6 +82,11 @@ public final class ContentImportService extends Service {
     /** Minimum width of the trailing window used to sample the transfer rate (delta bytes / delta
      *  time), so the speed shown is the recent rate, not the cumulative average from the run start. */
     private static final long RATE_WINDOW_MS = 500L;
+    /** K2GO-376: gap between import-ingest status polls (mirrors ContentExportService). */
+    private static final long POLL_INTERVAL_MS = 1000L;
+    /** K2GO-376: consecutive status-poll failures tolerated before an item fails. The ingest runs
+     *  minutes, so one relay blip or a busy core must not discard it (mirrors ContentExportService). */
+    private static final int MAX_POLL_ERRORS = 10;
 
     /** The single run state the progress fragment observes. */
     private static final ImportSession SESSION = new ImportSession();
@@ -305,7 +311,7 @@ private void drain() {
             String stored = BoxImportStaging.safeName(SESSION.payloadName(i));
             try {
                 copyToStaging(carrier, staging, SESSION.payloadName(i), stored, i, job);
-                source.importItem(ref.app(), SESSION.entryJson(i), BoxImportStaging.boxPath(stored));
+                ingest(source, ref, i, stored);
                 job.setDone(i);
             } catch (Exception e) {
                 // The staging copy or the ingest failed: mark the item and keep its slot for a retry.
@@ -402,6 +408,82 @@ private void copyToStaging(ImportCarrier carrier, BoxImportStaging staging, Stri
         }
     }
 
+    /**
+     * Ingest one already-staged item on the box as a long operation (K2GO-376): start it, then poll
+     * its status until it is done. Each poll is a SHORT request, so a slow Kolibri import (35-141 s)
+     * never times out at the dash-node/nginx relay. The poll IS the liveness signal: there is no
+     * wall-clock deadline here (the box's own watchdog ends a stalled ingest). A swept op (status
+     * GONE) re-starts exactly once; a second loss fails the item. A box error throws (the
+     * out-of-space case as {@link ImportNoSpaceException}, so the run stops with a space message, not
+     * a per-item retry). The item stays active while this runs; its bar sits at the copied bytes
+     * (the copy is done), as the old blocking ingest showed.
+     */
+    private void ingest(ImportSource source, ItemRef ref, int i, String storedName) throws Exception {
+        String boxPath = BoxImportStaging.boxPath(storedName);
+        String opId = source.startImport(ref.app(), SESSION.entryJson(i), boxPath);
+        boolean restarted = false;
+        int pollErrors = 0;
+        while (true) {
+            if (SESSION.isCancelled()) {
+                source.abortImport(opId);
+                throw new IOException("cancelled");
+            }
+            sleepPoll();
+            ImportStatus st;
+            try {
+                st = source.importStatus(opId);
+            } catch (Exception e) {
+                // A swept op returns GONE (handled below), not an exception; this is a transient poll
+                // failure (relay blip, busy core). Keep the same op and retry, up to the blip budget,
+                // so an ingest of minutes is not lost to one hiccup.
+                if (++pollErrors > MAX_POLL_ERRORS) {
+                    source.abortImport(opId);
+                    throw e;
+                }
+                continue;
+            }
+            pollErrors = 0;
+            switch (st.state()) {
+                case DONE:
+                    // Release the consumed op now (an idempotent drop on the box), so a finished
+                    // import does not linger until the TTL sweep. This is import's normal-completion
+                    // cleanup, since it has no fetch stream whose close drops the op as export does.
+                    source.abortImport(opId);
+                    return;
+                case ERROR:
+                    if (st.noSpace()) {
+                        throw new ImportNoSpaceException(st.error() != null && !st.error().isEmpty()
+                                ? st.error() : "no space left on the device");
+                    }
+                    throw new Exception(st.error() != null && !st.error().isEmpty()
+                            ? st.error() : "the box could not import this item");
+                case GONE:
+                    // The op was swept (e.g. a core restart). Re-start once; a second loss fails it.
+                    if (restarted) {
+                        throw new Exception("the import operation was lost");
+                    }
+                    restarted = true;
+                    opId = source.startImport(ref.app(), SESSION.entryJson(i), boxPath);
+                    break;
+                case RUNNING:
+                default:
+                    // Keep the session fresh so the UI shows the item is still being imported.
+                    SESSION.publish();
+                    maybeNotify(SystemClock.elapsedRealtime());
+                    break;
+            }
+        }
+    }
+
+    private static void sleepPoll() throws IOException {
+        try {
+            Thread.sleep(POLL_INTERVAL_MS);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            throw new IOException("interrupted");   // worker shutdown: end the item
+        }
+    }
+
     // ---- helpers -------------------------------------------------------------------------------
 
     private void sampleRate(ExportJob job, long dtMs, long dBytes) {