diff --git a/controller/app/src/main/java/org/appdevforall/k2go/contentexchange/data/ConexCoreDataSource.java b/controller/app/src/main/java/org/appdevforall/k2go/contentexchange/data/ConexCoreDataSource.java index 332a3cf42..eb8891339 100644 --- a/controller/app/src/main/java/org/appdevforall/k2go/contentexchange/data/ConexCoreDataSource.java +++ b/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 diff --git a/controller/app/src/main/java/org/appdevforall/k2go/contentexchange/data/ConexCoreImportSource.java b/controller/app/src/main/java/org/appdevforall/k2go/contentexchange/data/ConexCoreImportSource.java index 13fab8cf4..d8fed15e4 100644 --- a/controller/app/src/main/java/org/appdevforall/k2go/contentexchange/data/ConexCoreImportSource.java +++ b/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 validate(List 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 diff --git a/controller/app/src/main/java/org/appdevforall/k2go/contentexchange/domain/ImportSource.java b/controller/app/src/main/java/org/appdevforall/k2go/contentexchange/domain/ImportSource.java index 16c865491..2a08d6525 100644 --- a/controller/app/src/main/java/org/appdevforall/k2go/contentexchange/domain/ImportSource.java +++ b/controller/app/src/main/java/org/appdevforall/k2go/contentexchange/domain/ImportSource.java @@ -19,11 +19,22 @@ public interface ImportSource { List validate(List 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 diff --git a/controller/app/src/main/java/org/appdevforall/k2go/contentexchange/domain/ImportStatus.java b/controller/app/src/main/java/org/appdevforall/k2go/contentexchange/domain/ImportStatus.java new file mode 100644 index 000000000..bf10054c0 --- /dev/null +++ b/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). + * + *

{@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; + } +} diff --git a/controller/app/src/main/java/org/appdevforall/k2go/contentexchange/presentation/ContentImportService.java b/controller/app/src/main/java/org/appdevforall/k2go/contentexchange/presentation/ContentImportService.java index 94204884f..760c012cd 100644 --- a/controller/app/src/main/java/org/appdevforall/k2go/contentexchange/presentation/ContentImportService.java +++ b/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) { diff --git a/controller/app/src/main/java/org/appdevforall/k2go/contentexchange/presentation/ImportProgressFragment.java b/controller/app/src/main/java/org/appdevforall/k2go/contentexchange/presentation/ImportProgressFragment.java index d40b4071d..81b4bdac0 100644 --- a/controller/app/src/main/java/org/appdevforall/k2go/contentexchange/presentation/ImportProgressFragment.java +++ b/controller/app/src/main/java/org/appdevforall/k2go/contentexchange/presentation/ImportProgressFragment.java @@ -54,6 +54,14 @@ public class ImportProgressFragment extends Fragment { /** One-shot guard so a re-entrant render() does not schedule the back-pop twice. */ private boolean leaving; + /** Seconds before a fully successful run opens the Library (same pause as the Clone exit). */ + private static final int REDIRECT_SECS = 3; + private TextView redirect; + private TextView redirectCancel; + private int redirectSecs = -1; // -1 = no countdown running + private boolean redirectCancelled; // view-scoped: Cancel holds for this screen only + private final Runnable redirectTick = this::onRedirectTick; + @Nullable @Override public View onCreateView(@NonNull LayoutInflater inflater, @Nullable ViewGroup container, @Nullable Bundle s) { @@ -69,6 +77,13 @@ public View onCreateView(@NonNull LayoutInflater inflater, @Nullable ViewGroup c cancel = root.findViewById(R.id.k2go_cxi_cancel); retry = root.findViewById(R.id.k2go_cxi_retry); finish = root.findViewById(R.id.k2go_cxi_finish); + redirect = root.findViewById(R.id.k2go_cxi_redirect); + redirectCancel = root.findViewById(R.id.k2go_cxi_redirect_cancel); + redirectCancel.setOnClickListener(v -> { + redirectCancelled = true; + stopRedirect(); + render(); + }); // Two-phones transfer animation, shared with Clone (R.raw.k2go_clone_loop). Reduce motion holds // a single frame instead of looping; the screen never depends on movement to be understood. @@ -108,6 +123,7 @@ public void onResume() { public void onPause() { super.onPause(); session.setListener(null); + stopRedirect(); // render() on resume restarts it unless the user cancelled if (anim != null && anim.isAnimating()) { anim.pauseAnimation(); } @@ -175,6 +191,7 @@ private void renderCurrentItem(boolean running) { private void renderSettled(boolean running, String error) { if (running) { message.setVisibility(View.GONE); + renderRedirect(false); return; } int done = countDone(); @@ -187,8 +204,62 @@ private void renderSettled(boolean running, String error) { } else { text = getString(R.string.k2go_cx_import_partial_fmt, done, total); } + boolean success = error == null && session.allDone(); message.setText(text); + message.setTextColor(androidx.core.content.ContextCompat.getColor(requireContext(), + success ? R.color.k2go_leaf : R.color.k2go_ink)); message.setVisibility(View.VISIBLE); + renderRedirect(success && !redirectCancelled); + } + + /** On full success, count down to the Library; Cancel (or any non-success state) hides it. */ + private void renderRedirect(boolean active) { + if (!active) { + stopRedirect(); + redirect.setVisibility(View.GONE); + redirectCancel.setVisibility(View.GONE); + return; + } + if (redirectSecs < 0) { + redirectSecs = REDIRECT_SECS; + onRedirectTick(); + } + redirectCancel.setVisibility(View.VISIBLE); + } + + private void onRedirectTick() { + View v = getView(); + if (v == null || !isAdded() || redirectSecs < 0) { + return; + } + if (redirectSecs == 0) { + redirectSecs = -1; + openLibrary(); + return; + } + redirect.setText(getString(R.string.k2go_clone_exit_redirecting, redirectSecs)); + redirect.setVisibility(View.VISIBLE); + redirectSecs--; + v.postDelayed(redirectTick, 1000L); + } + + private void stopRedirect() { + View v = getView(); + if (v != null) { + v.removeCallbacks(redirectTick); + } + redirectSecs = -1; + } + + /** Close the run and land on the Library tab, clearing the Content Exchange screens above it. */ + private void openLibrary() { + ContentImportService.finishSession(requireContext()); + startActivity(new android.content.Intent(requireContext(), + org.appdevforall.k2go.redesign.LibraryActivity.class) + .addFlags(android.content.Intent.FLAG_ACTIVITY_CLEAR_TOP + | android.content.Intent.FLAG_ACTIVITY_SINGLE_TOP) + .putExtra(org.appdevforall.k2go.redesign.LibraryActivity.EXTRA_TAB, R.id.nav_library)); + requireActivity().finish(); } private int countDone() { @@ -227,6 +298,7 @@ public void onDestroyView() { // The scheduled pop runs on the view's handler; if the view is gone it will not run, so clear // the guard here so it does not stay true across a view recreation. leaving = false; + stopRedirect(); super.onDestroyView(); } diff --git a/controller/app/src/main/res/layout/fragment_k2go_content_import.xml b/controller/app/src/main/res/layout/fragment_k2go_content_import.xml index 78f7adce3..0348c6d9b 100644 --- a/controller/app/src/main/res/layout/fragment_k2go_content_import.xml +++ b/controller/app/src/main/res/layout/fragment_k2go_content_import.xml @@ -135,6 +135,29 @@ android:textAppearance="?attr/textAppearanceBodyLarge" android:textColor="@color/k2go_ink" android:visibility="gone" /> + + + + + diff --git a/controller/docs/ADR-long-running-operation-contract.md b/controller/docs/ADR-long-running-operation-contract.md index ee9963977..ebd00807b 100644 --- a/controller/docs/ADR-long-running-operation-contract.md +++ b/controller/docs/ADR-long-running-operation-contract.md @@ -68,3 +68,33 @@ shape (K2GO-436, K2GO-449). forced unique, and the holder says why. - This ADR closes the K2GO-458 / K2GO-461 / K2GO-462 family. The first landed piece is the identifier registry (K2GO-463). + +## Liveness, not deadlines (K2GO-376) + +A long operation that runs a child process (a Kolibri export or import CLI) is +supervised by LIVENESS, not by a fixed timeout. A fixed timeout is wrong for it: +a large channel is slow but healthy, and a small one that hangs must still be +caught. A fixed timeout also leaks across layers. The import 504 proved it: the +app had a 20 s read timeout, so we raised it; then the dash-node / nginx relay +cut at its own 60 s; raising that would only move the wall again. Each fix added +a second place that encoded "how long is too long" : a duplicate truth. + +The rule, enforced by one shared watchdog module used by both the import and +export ops: + +- Bytes are moving (the child's `/proc//io`, or the destination directory + grows) -> the child is alive. There is NO time limit. +- CPU advances with no new bytes -> it may be a database phase or a loop. Allow + a LONGER window before giving up. +- Nothing moves (no bytes, no CPU) -> it is stalled: terminate it and report. + +The stall window is RELATIVE TO THE TASK, not a wall clock: `max(floor, k * the +longest gap between byte-progress events seen so far in this run)`. A task that +naturally pauses between bursts widens its own tolerance. An exit by signal 9 is +reported as "killed (likely out of memory)", so an OOM kill is not opaque. + +The operation becomes a start/poll/cancel long op: POST starts it and returns an +opId, GET returns status and progress, DELETE cancels. Each poll is a SHORT +request, so a relay or proxy read timeout never trips, and the client keeps a +short per-request timeout, never a long one. The "how long is too long" decision +has ONE owner: the server-side watchdog. The client does not encode a deadline. diff --git a/static/conex-core/CHANGELOG.md b/static/conex-core/CHANGELOG.md index 58cef91a6..f04477b28 100644 --- a/static/conex-core/CHANGELOG.md +++ b/static/conex-core/CHANGELOG.md @@ -5,6 +5,7 @@ One line per version, newest first. Every REST-facing change bumps `conex_core.V role), so this file is the human record of what each bump enables. Keep entries short: `version - change (TICKET)`. +- **0.3.4** - import becomes a long operation like export: POST /v1/import/item now STARTS the ingest and returns an opId, the client polls GET /v1/import/status/{op}, and DELETE /v1/import/{op} cancels, so a 35-141 s Kolibri import no longer times out at the dash-node/nginx relay. A shared liveness watchdog supervises the Kolibri import and export CLIs by byte and CPU progress (no fixed timeout; the window is relative to the task) and names a signal-9 exit as a likely out-of-memory kill. (K2GO-376) - **0.3.3** - observability: log every error response and uncaught handler exception to stdout (the supervisor captures it to /var/log/conex-core.log), and line-buffer stdout/stderr in __main__ so output reaches the log while the server runs forever, so a failed import or share leaves a box-side trace; an uncaught error now returns a 500 JSON instead of dropping the connection. (K2GO-376) - **0.3.2** - /v1/import/item returns 507 (not 422) when a stage runs out of disk (adapters tag it code "no_space"), so the front stops the run with a space message; /v1/export/prepare drops the proot-based free-space preflight too (same false-507 as import), failing cleanly during staging instead. (K2GO-376) - **0.3.1** - /v1/import/item drops the free-space preflight (proot under-reports free space, so the 507 false-rejected valid imports); a real ENOSPC during ingest now fails cleanly as 422. (K2GO-376) diff --git a/static/conex-core/conex_core/__init__.py b/static/conex-core/conex_core/__init__.py index c2b1750b1..20019912a 100644 --- a/static/conex-core/conex_core/__init__.py +++ b/static/conex-core/conex_core/__init__.py @@ -7,5 +7,5 @@ # The core build. Bumped per core change so a box can report which core it runs (GET # /v1/version), mirroring dash-node's package.json version and /system/version. The # conex_core Ansible role writes this into the [conex_core] section of iiab.ini. -VERSION = "0.3.3" +VERSION = "0.3.4" __version__ = VERSION diff --git a/static/conex-core/conex_core/adapters/calibre.py b/static/conex-core/conex_core/adapters/calibre.py index bfff8ed01..47598a1af 100644 --- a/static/conex-core/conex_core/adapters/calibre.py +++ b/static/conex-core/conex_core/adapters/calibre.py @@ -125,7 +125,10 @@ def check(self, dedup: dict) -> str: return SAME_VERSION return ABSENT - def import_item(self, entry: dict, payload_path: str) -> dict: + def import_item(self, entry: dict, payload_path: str, should_cancel=None, + on_progress=None) -> dict: + # The import long op passes should_cancel / on_progress for a supervised CLI; Calibre's + # import is a short in-process upload, so it ignores both and runs to completion. # The payload is the tar of the book's own directory. Extract it safely, then add each # content-format file through Calibre-Web's upload (calibredb is not installed). # Soft identity: a re-import can create a duplicate record, so the front checks first. diff --git a/static/conex-core/conex_core/adapters/kiwix.py b/static/conex-core/conex_core/adapters/kiwix.py index 9aa3c7230..b5a352b74 100644 --- a/static/conex-core/conex_core/adapters/kiwix.py +++ b/static/conex-core/conex_core/adapters/kiwix.py @@ -209,7 +209,10 @@ def check(self, dedup: dict) -> str: return SAME_VERSION return OLDER_HERE if local_version < version else NEWER_HERE - def import_item(self, entry: dict, payload_path: str) -> dict: + def import_item(self, entry: dict, payload_path: str, should_cancel=None, + on_progress=None) -> dict: + # The import long op passes should_cancel / on_progress for a supervised CLI; the Kiwix + # import is a file move plus a re-index, so it ignores both and runs to completion. # The payload is the .zim itself: place it in the zims dir under its own name, then # re-run the indexer so the library entry appears. Overwriting the same name is # idempotent (same content, same UUID). diff --git a/static/conex-core/conex_core/adapters/kolibri.py b/static/conex-core/conex_core/adapters/kolibri.py index 109a1b843..64a6f1571 100644 --- a/static/conex-core/conex_core/adapters/kolibri.py +++ b/static/conex-core/conex_core/adapters/kolibri.py @@ -13,6 +13,7 @@ from __future__ import annotations import errno +import logging import os import re import shutil @@ -24,6 +25,7 @@ from ..export import free_bytes, tar_stream from ..importing import safe_extract_tar, single_child_dir, volume_nearly_full +from ..watchdog import CANCELLED, OK, SuperviseResult, run_supervised from ..model import ( ABSENT, AUTHORITATIVE, @@ -38,6 +40,8 @@ _CHANNEL_ID_RE = re.compile(r"^[0-9a-fA-F]{32}$") +_LOG = logging.getLogger("conex_core") + def export_commands(channel_id: str, dest: str, kolibri_bin: str) -> List[List[str]]: """The two Kolibri management commands that stage a channel export into dest: the @@ -76,27 +80,6 @@ def delete_command(channel_id: str, kolibri_bin: str) -> List[str]: return [kolibri_bin, "manage", "deletechannel", channel_id] -def _run_cancellable(argv: List[str], env: dict, should_cancel=None) -> bool: - """Run one child process to completion, polling ``should_cancel`` about once a second. - When it returns True the process is terminated (then killed if it does not exit) and this - returns False. Otherwise returns True on exit code 0. This is how an aborted export stops - the long-running Kolibri CLI instead of leaving it writing into a removed dir.""" - proc = subprocess.Popen(argv, env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) - while True: - try: - proc.wait(timeout=1.0) - return proc.returncode == 0 - except subprocess.TimeoutExpired: - if should_cancel and should_cancel(): - proc.terminate() - try: - proc.wait(timeout=5) - except subprocess.TimeoutExpired: - proc.kill() - proc.wait() - return False - - def _to_int(*values) -> Optional[int]: """First value that parses as an int, else None. The version may arrive as an int or a string over JSON, and the build mirrors it.""" @@ -214,10 +197,7 @@ def stage_export(self, local_ref: str, dest: str, should_cancel=None) -> bool: # the export CLI on abort instead of leaving it running into a deleted dir. if not _CHANNEL_ID_RE.match(local_ref or ""): return False - try: - return self._stage_export(local_ref, dest, should_cancel) - except OSError: - return False + return self._stage_export(local_ref, dest, should_cancel) def open_fetch(self, local_ref: str, staged_dir: Optional[str] = None) -> Optional[ExportStream]: # Tar-stream the staged tree. The op layer sets on_close (drop the op + remove temp). @@ -230,13 +210,17 @@ def open_fetch(self, local_ref: str, staged_dir: Optional[str] = None) -> Option ) def _stage_export(self, channel_id: str, dest: str, should_cancel=None) -> bool: - """Run the Kolibri export CLI into dest. A seam, so a test can stage a fake tree - without a live Kolibri. Each command runs as a child process polled against - ``should_cancel``, so an abort terminates the export instead of letting it finish into a - deleted dir. Returns True on success, False on CLI failure or cancellation.""" + """Run the Kolibri export CLI into dest under the liveness watchdog. A seam, so a test can + stage a fake tree without a live Kolibri. The watchdog ends a child that stalls or runs + away and honors should_cancel, so an abort stops the export instead of letting it finish + into a deleted dir. Returns True on success, False on cancellation or a CLI failure; the + failure detail is logged box-side, and the export op reports a generic staging error.""" env = dict(os.environ, KOLIBRI_HOME=self._home) for argv in export_commands(channel_id, dest, self._bin): - if not _run_cancellable(argv, env, should_cancel): + result = run_supervised(argv, env, dest_dir=dest, should_cancel=should_cancel, log=_LOG) + if not result.ok: + if result.state != CANCELLED: + _LOG.warning("kolibri export %s: %s", channel_id, result.detail) return False return True @@ -252,9 +236,11 @@ def check(self, dedup: dict) -> str: return SAME_VERSION return OLDER_HERE if local_version < incoming else NEWER_HERE - def import_item(self, entry: dict, payload_path: str) -> dict: - # The payload is the export tar. Extract it safely, then delegate the disk import to - # Kolibri. Idempotent: Kolibri's import re-registers and dedups, never double-counts. + def import_item(self, entry: dict, payload_path: str, should_cancel=None, + on_progress=None) -> dict: + # The payload is the export tar. Extract it safely, then delegate the disk import to Kolibri + # under the liveness watchdog (import is a long op). Idempotent: Kolibri's import + # re-registers and dedups, never double-counts. channel_id = (entry.get("identity") or {}).get("localRef", "") if not _CHANNEL_ID_RE.match(channel_id or ""): return {"ok": False, "reason": "invalid channel id"} @@ -277,29 +263,36 @@ def import_item(self, entry: dict, payload_path: str) -> dict: failure["code"] = "no_space" return failure source = single_child_dir(work) or work - if not self._stage_import(channel_id, source): - # The Kolibri CLI hides its errno. When it fails with the content volume essentially - # full, the cause is out-of-space (the channel copy into KOLIBRI_HOME is the third - # full copy on one volume), so report it structurally; otherwise it is a real import - # failure. + outcome = self._stage_import(channel_id, source, should_cancel, on_progress) + if outcome.state == CANCELLED: + # The op was aborted, so the CLI was stopped. The op is already being dropped; the + # reason is only for completeness. + return {"ok": False, "reason": "import cancelled"} + if not outcome.ok: + # The Kolibri CLI stopped. When the content volume is essentially full, the cause is + # out-of-space (the channel copy into KOLIBRI_HOME is a third full copy on one + # volume), so report it structurally; otherwise report the watchdog detail (a stall, + # an OOM kill, or a non-zero exit) so a failed import is not opaque. if volume_nearly_full(self._home): return {"ok": False, "code": "no_space", "reason": "kolibri import failed: no space left on the content volume"} - return {"ok": False, "reason": "kolibri import failed"} + return {"ok": False, "reason": f"kolibri import {outcome.detail}"} return {"ok": True, "app": "kolibri", "localRef": channel_id} finally: shutil.rmtree(work, ignore_errors=True) - def _stage_import(self, channel_id: str, source: str) -> bool: - """Run the Kolibri disk-import CLI from source. A seam, like _stage_export.""" + def _stage_import(self, channel_id: str, source: str, should_cancel=None, + on_progress=None) -> SuperviseResult: + """Run the Kolibri disk-import CLI from source under the liveness watchdog. A seam, like + _stage_export. Returns the watchdog result: ok, or the first command's failure with its + detail. KOLIBRI_HOME is the dest-growth fallback when /proc//io is unreadable.""" env = dict(os.environ, KOLIBRI_HOME=self._home) for argv in import_commands(channel_id, source, self._bin): - result = subprocess.run( - argv, env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False - ) - if result.returncode != 0: - return False - return True + result = run_supervised(argv, env, dest_dir=self._home, should_cancel=should_cancel, + on_progress=on_progress, log=_LOG) + if not result.ok: + return result + return SuperviseResult(OK, 0, "ok") def delete(self, local_ref: str) -> dict: # Delegate to Kolibri's own delete (it refcounts shared files). Measure the reclaim as a diff --git a/static/conex-core/conex_core/export_ops.py b/static/conex-core/conex_core/export_ops.py index 3ce0dd041..9e0a0160d 100644 --- a/static/conex-core/conex_core/export_ops.py +++ b/static/conex-core/conex_core/export_ops.py @@ -1,144 +1,91 @@ """Export long-operation registry (K2GO-507). -Exporting a large item (a multi-gigabyte Kolibri channel) must stage the whole channel -before the first byte can stream, which takes minutes. A single blocking request times out -on the client. So export becomes a long operation: prepare starts the work and returns an -opId at once, the client polls status for liveness and progress, then fetches the bytes. -The front owns the sequence (ADR D1c); this registry owns the in-flight set and the temp -lifecycle, uniform for every app. The adapters only report a plan, stage into a given -directory and open the stream: the orchestration is here, not in an adapter. - -Single source of truth: the op state and the staged bytes are owned by the core. The op -registry is in-memory, so a conex-core restart loses the ops; this is acceptable because the -front recovers by RE-PREPARE and the startup sweep reclaims any orphan temp left by a dead -process. The filesystem is the source for staged bytes; the registry is the source for the -in-flight set. No durable op state, no second owner. - -Every op temp has exactly one cleanup owner across the four end states: - 1. normal completion : the fetch stream's on_close drops the op and removes the temp. - 2. client abort : DELETE drops the op and removes the temp. - 3. abandonment : a lazy TTL sweep (run at the head of each op request) removes ops - whose last_seen is older than the timeout, so an abandoned prepare - never orphans a multi-gigabyte temp while the service keeps running. - 4. process death : a startup sweep removes every op temp left on disk, since a fresh - process holds no ops and anything matching the temp prefix is an - orphan. -The TTL sweep is lazy (no background timer to own): it needs no thread lifecycle, and the -startup sweep is the backstop for a temp left with no further requests coming. +Exporting a large item (a multi-gigabyte Kolibri channel) must stage the whole channel before the +first byte can stream, which takes minutes. A single blocking request times out on the client. So +export is a long operation: prepare starts the work and returns an opId at once, the client polls +status for liveness and progress, then fetches the bytes. The front owns the sequence (ADR D1c). + +The in-flight set, the lazy TTL sweep, the cancel signal and the one cleanup primitive are shared +with import in OpRegistry (ops_base). This module adds only what export needs: a staged temp per op, +the staged-bytes heartbeat, and the fetch stream. + +Every op temp has exactly one cleanup owner across four end states: normal completion (the fetch +stream's on_close drops the op and removes the temp), client abort (DELETE), abandonment (the lazy +TTL sweep), and process death (the startup sweep reclaims every export op temp on disk, since a +fresh process holds no ops). """ from __future__ import annotations import os -import secrets import shutil import tempfile import threading import time -from dataclasses import dataclass, field -from typing import Any, Dict, Optional, Tuple +from dataclasses import dataclass +from typing import Optional -Response = Tuple[int, Any] +from .ops_base import ERROR, READY, STAGING, Op, OpRegistry, Response +from .watchdog import dir_size -# The op temp prefix. Op-owned and app-agnostic (the op layer creates the temp, not the -# adapter), and distinct from the import flow's "cximport-" temps, so the startup sweep -# reclaims only export op temps and never touches an in-flight import. +# The op temp prefix. Op-owned and app-agnostic, and distinct from the import flow's "cximport-" +# temps, so the export startup sweep reclaims only export op temps. TEMP_PREFIX = "cxexport-" -# Prefixes the startup sweep also reclaims for one release: before this change the Kolibri -# adapter staged into "cxkolibri-" temps, so an orphan from an older build (or a device test) -# must still be swept. Not used for the import flow's "cximport-" temps, which self-clean. +# Prefixes the startup sweep also reclaims for one release: before this change the Kolibri adapter +# staged into "cxkolibri-" temps, so an orphan from an older build (or a device test) must still be +# swept. The import flow's "cximport-" temps are reclaimed by the import registry's own sweep. _LEGACY_SWEEP_PREFIXES = ("cxkolibri-",) -# How long an op may go without a status or fetch before the lazy sweep reclaims it. The -# client polls status throughout a staging export, so a live op keeps bumping last_seen; -# the timeout only catches an abandoned op. -DEFAULT_TTL_SECONDS = 600 - -# Op states. -STAGING = "staging" -READY = "ready" -ERROR = "error" - @dataclass -class _Op: - app: str - local_ref: str - temp: Optional[str] # the staging dir for a staging op, None for a direct app - state: str - total_bytes: int - error: Optional[str] = None - last_seen: float = field(default_factory=time.monotonic) - # True while a fetch stream is open on this op. The TTL sweep never reclaims an in-use op, - # so a multi-gigabyte fetch that streams longer than the TTL (and bumps nothing meanwhile) - # is not swept out from under tar_stream. Set by fetch() when the stream opens, cleared by - # dropping the op in the stream's on_close. Lost on process death, but then the op itself - # is gone and the startup sweep reclaims the orphan temp. - in_use: bool = False +class _ExportOp(Op): + local_ref: str = "" + temp: Optional[str] = None # the staging dir for a staging op, None for a direct app -def _dir_size(path: str) -> int: - """Sum of the file sizes under path (a staged tree), or 0 when it is gone. This is the - staging heartbeat: it grows as the export CLI writes, so the client sees progress.""" - total = 0 - for dirpath, _dirnames, filenames in os.walk(path): - for name in filenames: - try: - total += os.path.getsize(os.path.join(dirpath, name)) - except OSError: - pass - return total +class ExportOps(OpRegistry): + """Registry of in-flight export operations. One instance lives for the server's lifetime.""" - -class ExportOps: - """Thread-safe registry of in-flight export operations. One instance lives for the - server's lifetime; the handler threads call into it concurrently.""" - - def __init__(self, registry, staging_dir: str, ttl_seconds: int = DEFAULT_TTL_SECONDS): + def __init__(self, registry, staging_dir: str, ttl_seconds: int = 600): + super().__init__(ttl_seconds) self._registry = registry self._staging_dir = staging_dir - self._ttl = ttl_seconds - self._lock = threading.Lock() - self._ops: Dict[str, _Op] = {} # ---- REST-facing operations (return (status, payload)) ---- def prepare(self, app: str, local_ref: str) -> Response: - """Resolve the item, preflight the staging disk, and either start a background - staging thread (apps that stage) or mark the op ready at once (direct apps).""" + """Resolve the item, then either start a background staging thread (apps that stage) or mark + the op ready at once (direct apps).""" self._sweep_expired() if not self._registry.has(app): return 404, {"error": f"unknown app: {app}"} plan = self._registry.get(app).prepare_export(local_ref) if plan is None: return 404, {"error": "item not found or not exportable"} - # No free-space preflight. The box runs under proot, whose statvfs under-reports the free - # space on the content volume, so this check false-rejected valid exports on devices with - # ample room (the import path dropped the same check for the same reason). A genuine ENOSPC + # No free-space preflight: the box runs under proot, whose statvfs under-reports the free + # space on the content volume, so a check here false-rejected valid exports. A genuine ENOSPC # during staging surfaces as an op error through _run_stage, and the temp is swept. - op_id = secrets.token_hex(16) + op_id = self._new_id() if plan.stages: try: os.makedirs(self._staging_dir, exist_ok=True) temp = tempfile.mkdtemp(prefix=TEMP_PREFIX, dir=self._staging_dir) except OSError: return 507, {"error": "cannot create staging dir"} - op = _Op(app, local_ref, temp, STAGING, plan.total_bytes) - with self._lock: - self._ops[op_id] = op + self._put(op_id, _ExportOp(app=app, state=STAGING, total_bytes=plan.total_bytes, + local_ref=local_ref, temp=temp)) worker = threading.Thread( target=self._run_stage, args=(op_id, app, local_ref, temp), daemon=True) worker.start() return 200, {"opId": op_id, "state": STAGING, "totalBytes": plan.total_bytes} - op = _Op(app, local_ref, None, READY, plan.total_bytes) - with self._lock: - self._ops[op_id] = op + self._put(op_id, _ExportOp(app=app, state=READY, total_bytes=plan.total_bytes, + local_ref=local_ref, temp=None)) return 200, {"opId": op_id, "state": READY, "totalBytes": plan.total_bytes} def status(self, op_id: str) -> Response: - """Report the op state plus the staged-bytes heartbeat. A 404 means the op is - unknown (swept or never existed), so the client re-prepares.""" + """Report the op state plus the staged-bytes heartbeat. A 404 means the op is unknown + (swept or never existed), so the client re-prepares.""" self._sweep_expired() with self._lock: op = self._ops.get(op_id) @@ -149,7 +96,7 @@ def status(self, op_id: str) -> Response: if state == READY: staged = total elif temp: - staged = _dir_size(temp) # du outside the lock: the walk may be slow + staged = dir_size(temp) # du outside the lock: the walk may be slow else: staged = 0 payload = {"state": state, "stagedBytes": staged, "totalBytes": total} @@ -158,9 +105,8 @@ def status(self, op_id: str) -> Response: return 200, payload def fetch(self, op_id: str): - """Open the byte stream for a ready op. Returns (200, ExportStream) for the server - to stream; the stream's on_close drops the op and removes the temp. 409 when the op - is not ready, 404 when it is unknown.""" + """Open the byte stream for a ready op. Returns (200, ExportStream); the stream's on_close + drops the op and removes the temp. 409 when not ready, 404 when unknown.""" self._sweep_expired() with self._lock: op = self._ops.get(op_id) @@ -170,35 +116,26 @@ def fetch(self, op_id: str): if op.state != READY: return 409, {"error": f"op not ready: {op.state}"} app, local_ref, temp = op.app, op.local_ref, op.temp - # Open the stream OUTSIDE the lock. The op is not pinned yet: on_open pins it once the - # server actually begins streaming, so a failure here (or before streaming starts) - # leaves the op merely fresh and reclaimable by the normal sweep, never stuck in_use. + # Open the stream OUTSIDE the lock. The op is not pinned yet: on_open pins it once the server + # actually begins streaming, so a failure here leaves the op merely fresh and reclaimable by + # the normal sweep, never stuck in_use. try: stream = self._registry.get(app).open_fetch(local_ref, temp) except Exception: self._finish(op_id) return 500, {"error": "failed to open export"} if stream is None: - # The item vanished between prepare and fetch: clean up and report absent. self._finish(op_id) return 404, {"error": "item not found or not exportable"} stream.on_open = lambda: self._pin(op_id) stream.on_close = lambda: self._finish(op_id) return 200, stream - def delete(self, op_id: str) -> Response: - """Client abort: drop the op and remove its temp. Idempotent, so a repeat or an - unknown op is still 200 {ok: True}.""" - self._sweep_expired() - self._finish(op_id) - return 200, {"ok": True} - # ---- lifecycle ---- def sweep_startup(self) -> None: - """Remove every export op temp on disk. The server calls this once on start: a fresh - process holds no ops, so anything under the staging dir with the op prefix is an - orphan left by a dead process.""" + """Remove every export op temp on disk. A fresh process holds no ops, so anything under the + staging dir with an export op prefix is an orphan left by a dead process.""" try: names = os.listdir(self._staging_dir) except OSError: @@ -209,66 +146,31 @@ def sweep_startup(self) -> None: shutil.rmtree(os.path.join(self._staging_dir, name), ignore_errors=True) def _run_stage(self, op_id: str, app: str, local_ref: str, temp: str) -> None: - """Stage the item into temp off the request thread, then flip the op to ready or - error. If the op was dropped meanwhile (DELETE or sweep), the flip is a no-op and the - temp was already removed by whoever dropped it. The adapter polls ``should_cancel`` - and stops its staging work (terminates the subprocess) once the op is gone, so an - abort actually cancels the server-side work instead of letting it run on into a - deleted dir.""" - def should_cancel() -> bool: - with self._lock: - return op_id not in self._ops - + """Stage the item into temp off the request thread, then flip the op to ready or error. If + the op was dropped meanwhile (DELETE or sweep), the flip is a no-op and the temp was already + removed by whoever dropped it. The adapter polls should_cancel and stops its staging work + once the op is gone, so an abort actually cancels the server-side work.""" + should_cancel = self.should_cancel(op_id) try: ok = self._registry.get(app).stage_export(local_ref, temp, should_cancel) - except Exception as exc: # any adapter / CLI failure surfaces as an op error - self._set_error(op_id, f"stage failed: {exc}") + except Exception as exc: # any adapter / CLI failure surfaces as an op error with its detail + self._transition(op_id, STAGING, ERROR, error=f"stage failed: {exc}") return if ok: - self._set_state(op_id, STAGING, READY) + self._transition(op_id, STAGING, READY) else: - self._set_error(op_id, "staging failed") - - def _set_state(self, op_id: str, expect: str, new: str) -> None: - with self._lock: - op = self._ops.get(op_id) - if op is not None and op.state == expect: - op.state = new - op.last_seen = time.monotonic() - - def _set_error(self, op_id: str, message: str) -> None: - with self._lock: - op = self._ops.get(op_id) - if op is not None and op.state == STAGING: - op.state = ERROR - op.error = message - op.last_seen = time.monotonic() + self._transition(op_id, STAGING, ERROR, error="staging failed") def _pin(self, op_id: str) -> None: - """Mark the op in use for the life of an open fetch stream. The server calls this at the - top of streaming; _finish (the stream's on_close) drops the op and clears the pin with - it. Keeping the pin here, not in fetch(), means a pre-stream failure never leaves it - set.""" + """Mark the op in use for the life of an open fetch stream. The server calls this at the top + of streaming; _finish (the stream's on_close) drops the op and clears the pin with it.""" with self._lock: op = self._ops.get(op_id) if op is not None: op.in_use = True - def _finish(self, op_id: str) -> None: - """Drop the op and remove its temp. The one cleanup primitive, idempotent.""" - with self._lock: - op = self._ops.pop(op_id, None) - if op is not None and op.temp: - shutil.rmtree(op.temp, ignore_errors=True) - - def _sweep_expired(self) -> None: - now = time.monotonic() - stale = [] - with self._lock: - for op_id, op in list(self._ops.items()): - # Never reclaim an op with an open fetch stream: its temp is in use. - if not op.in_use and now - op.last_seen > self._ttl: - stale.append(self._ops.pop(op_id)) - for op in stale: - if op.temp: - shutil.rmtree(op.temp, ignore_errors=True) + def _cleanup(self, op: Op) -> None: + """Remove the op's staging temp, if it owns one.""" + temp = getattr(op, "temp", None) + if temp: + shutil.rmtree(temp, ignore_errors=True) diff --git a/static/conex-core/conex_core/import_ops.py b/static/conex-core/conex_core/import_ops.py new file mode 100644 index 000000000..f3a038747 --- /dev/null +++ b/static/conex-core/conex_core/import_ops.py @@ -0,0 +1,133 @@ +"""Import long-operation registry (K2GO-376). + +Importing a channel runs the Kolibri CLI, which takes 35 to 141 s for a real channel. A single +blocking POST times out at the dash-node / nginx relay (its default read timeout is 60 s). So +import is a long operation, like export: start returns an opId at once, the client polls status for +liveness and progress, and a DELETE cancels. Each poll is a short request, so the relay read +timeout never trips. + +This reuses the shared OpRegistry (ops_base): the in-flight set, the lazy TTL sweep, the cancel +signal and cleanup. Import adds only its own fields (the ingest result) and the background worker +that runs the adapter's import under the liveness watchdog. Import does not own a staging temp: the +adapter's own work temp ("cximport-") self-cleans on completion, and the startup sweep reclaims any +such orphan a process that died mid-import left behind. +""" + +from __future__ import annotations + +import os +import shutil +import threading +import time +from dataclasses import dataclass +from typing import Any, Optional + +from .importing import is_within_dir +from .ops_base import DONE, ERROR, RUNNING, Op, OpRegistry, Response + +# The import work-temp prefix, created by the adapters' import_item. The startup sweep reclaims an +# orphan left by a process that died mid-import (at startup nothing is in-flight). +TEMP_PREFIX = "cximport-" + + +@dataclass +class _ImportOp(Op): + moved_bytes: int = 0 + result: Optional[dict] = None + code: Optional[str] = None # a structural error code, e.g. "no_space" + + +def _entry_bytes(entry: Any) -> int: + """The item's declared size, the progress denominator. 0 when absent or unparsable.""" + try: + return max(0, int(entry.get("bytes", 0))) + except (AttributeError, TypeError, ValueError): + return 0 + + +class ImportOps(OpRegistry): + """Registry of in-flight import operations. One instance lives for the server's lifetime; the + handler threads call in concurrently.""" + + def __init__(self, registry, import_root: str, ttl_seconds: int = 600): + super().__init__(ttl_seconds) + self._registry = registry + self._import_root = import_root + + # ---- REST-facing operations (return (status, payload)) ---- + + def start(self, app: str, entry: Any, payload_path: str) -> Response: + """Validate the request, start the ingest off the request thread, and return the opId at + once. The same containment rule as the old synchronous import: the payload must live inside + the import area, never an arbitrary path.""" + self._sweep_expired() + if not self._registry.has(app): + return 404, {"error": f"unknown app: {app}"} + if not isinstance(entry, dict) or not isinstance(payload_path, str) or not payload_path: + return 400, {"error": "entry and payloadPath are required"} + if not is_within_dir(self._import_root, payload_path): + return 400, {"error": "payloadPath is outside the import area"} + total = _entry_bytes(entry) + op_id = self._new_id() + self._put(op_id, _ImportOp(app=app, state=RUNNING, total_bytes=total)) + worker = threading.Thread( + target=self._run_import, args=(op_id, app, entry, payload_path, total), daemon=True) + worker.start() + return 200, {"opId": op_id, "state": RUNNING, "totalBytes": total} + + def status(self, op_id: str) -> Response: + """Report the op state, the moved-bytes heartbeat, and the result once done. A 404 means the + op is unknown (swept or never existed), so the client re-starts.""" + self._sweep_expired() + with self._lock: + op = self._ops.get(op_id) + if op is None: + return 404, {"error": "unknown op"} + op.last_seen = time.monotonic() + payload = {"state": op.state, "movedBytes": op.moved_bytes, "totalBytes": op.total_bytes} + if op.error: + payload["error"] = op.error + if op.code: + payload["code"] = op.code + if op.state == DONE and op.result: + payload["result"] = op.result + return 200, payload + + # ---- lifecycle ---- + + def sweep_startup(self) -> None: + """Remove every import work-temp on disk. A fresh process holds no import ops, so any + cximport- dir under the import area is an orphan left by a process that died mid-import.""" + try: + names = os.listdir(self._import_root) + except OSError: + return + for name in names: + if name.startswith(TEMP_PREFIX): + shutil.rmtree(os.path.join(self._import_root, name), ignore_errors=True) + + def _run_import(self, op_id: str, app: str, entry: Any, payload_path: str, total: int) -> None: + """Ingest the item off the request thread, under the liveness watchdog, then flip the op to + done (with the result) or error (with the detail, and a structural code when the adapter set + one, e.g. no_space). If the op was dropped meanwhile (DELETE or sweep), the flip is a no-op; + the adapter saw should_cancel and stopped its CLI.""" + should_cancel = self.should_cancel(op_id) + + def on_progress(moved: int) -> None: + # The watchdog reports cumulative syscall bytes, which for an import exceed the payload + # (read the tar, then write the content), so clamp to the known total: a monotonic bar + # that fills and holds, never past 100%. + self._touch(op_id, moved_bytes=min(moved, total) if total else moved) + + try: + result = self._registry.get(app).import_item( + entry, payload_path, should_cancel=should_cancel, on_progress=on_progress) + except Exception as exc: # any adapter failure surfaces as an op error with its detail + self._transition(op_id, RUNNING, ERROR, error=f"import failed: {exc}") + return + if isinstance(result, dict) and result.get("ok"): + self._transition(op_id, RUNNING, DONE, result=result, moved_bytes=total) + else: + reason = result.get("reason", "import failed") if isinstance(result, dict) else "import failed" + code = result.get("code") if isinstance(result, dict) else None + self._transition(op_id, RUNNING, ERROR, error=reason, code=code) diff --git a/static/conex-core/conex_core/ops_base.py b/static/conex-core/conex_core/ops_base.py new file mode 100644 index 000000000..696ae85ac --- /dev/null +++ b/static/conex-core/conex_core/ops_base.py @@ -0,0 +1,125 @@ +"""Shared registry for in-flight long operations (K2GO-376, generalizing K2GO-507). + +Export and import are both long operations: a request starts one and returns an opId, the client +polls status for liveness and progress, and a DELETE cancels. The in-flight set, the lazy TTL +abandonment sweep, the cancel signal and the one cleanup primitive are identical for both, so they +live here once. Export adds a staged temp and a fetch stream; import adds a result. Neither +re-implements the registry. + +One source of truth: op state is owned by the core and is in-memory, so a restart drops the ops and +the client recovers (export re-prepares, import re-starts); each kind's startup sweep reclaims any +temp an earlier process left on disk. No durable op state, no second owner. + +Every op is reclaimed by exactly one of four paths: normal completion (the subclass drops it), +client abort (DELETE), abandonment (the lazy TTL sweep at the head of each request), and process +death (the startup sweep). The TTL sweep is lazy, so there is no background timer to own. +""" + +from __future__ import annotations + +import secrets +import threading +import time +from dataclasses import dataclass, field +from typing import Any, Dict, Optional, Tuple + +Response = Tuple[int, Any] + +# Op states. Export moves staging -> ready; import moves running -> done; both can reach error. +STAGING = "staging" +READY = "ready" +RUNNING = "running" +DONE = "done" +ERROR = "error" + +# How long an op may go without a status poll before the lazy sweep reclaims it. A live op bumps +# last_seen on every poll, so the timeout only catches an abandoned op. +DEFAULT_TTL_SECONDS = 600 + + +@dataclass +class Op: + """Base op record. Subclasses add their kind's own fields (export: a temp; import: a result).""" + app: str + state: str + total_bytes: int = 0 + error: Optional[str] = None + last_seen: float = field(default_factory=time.monotonic) + # True while a fetch stream is open (export only): the TTL sweep never reclaims an in-use op, so + # a multi-gigabyte fetch that outlasts the TTL is not swept out from under the stream. + in_use: bool = False + + +class OpRegistry: + """Thread-safe registry of in-flight operations with a lazy TTL sweep. One instance per kind; + handler threads call in concurrently. A subclass overrides _cleanup to release an op's on-disk + resource, and builds its own Op subclass.""" + + def __init__(self, ttl_seconds: int = DEFAULT_TTL_SECONDS): + self._ttl = ttl_seconds + self._lock = threading.Lock() + self._ops: Dict[str, Op] = {} + + def _new_id(self) -> str: + return secrets.token_hex(16) + + def _put(self, op_id: str, op: Op) -> None: + with self._lock: + self._ops[op_id] = op + + def should_cancel(self, op_id: str): + """A closure the background worker polls: True once the op is gone (a DELETE or a sweep), + so the worker stops its child instead of running on into a dropped op.""" + def _gone() -> bool: + with self._lock: + return op_id not in self._ops + return _gone + + def _transition(self, op_id: str, expect: str, new_state: str, **fields: Any) -> None: + """Move an op from ``expect`` to ``new_state`` and set any extra fields, under the lock. A + no-op if the op is gone or already moved on (its id is unique, so this only guards against a + late worker flipping an op that a DELETE already dropped).""" + with self._lock: + op = self._ops.get(op_id) + if op is not None and op.state == expect: + op.state = new_state + for name, value in fields.items(): + setattr(op, name, value) + op.last_seen = time.monotonic() + + def _touch(self, op_id: str, **fields: Any) -> None: + """Set fields on an op without changing its state (e.g. a progress update), under the lock.""" + with self._lock: + op = self._ops.get(op_id) + if op is not None: + for name, value in fields.items(): + setattr(op, name, value) + + def delete(self, op_id: str) -> Response: + """Client abort: drop the op (its worker sees should_cancel and stops) and clean up. A + repeat or an unknown op is still 200 {ok: True}.""" + self._sweep_expired() + self._finish(op_id) + return 200, {"ok": True} + + def _finish(self, op_id: str) -> None: + """The one cleanup primitive, idempotent: drop the op and release its resource.""" + with self._lock: + op = self._ops.pop(op_id, None) + if op is not None: + self._cleanup(op) + + def _cleanup(self, op: Op) -> None: + """Release an op's on-disk resource. The base op owns none; a subclass overrides.""" + + def _sweep_expired(self) -> None: + now = time.monotonic() + stale = [] + with self._lock: + for op_id in list(self._ops): + op = self._ops[op_id] + # Never reclaim an in-use op (an open fetch stream): its temp is being read. + if not op.in_use and now - op.last_seen > self._ttl: + stale.append(self._ops.pop(op_id)) + for op in stale: + self._cleanup(op) diff --git a/static/conex-core/conex_core/ports.py b/static/conex-core/conex_core/ports.py index 166b58a70..d2d8fc323 100644 --- a/static/conex-core/conex_core/ports.py +++ b/static/conex-core/conex_core/ports.py @@ -107,10 +107,16 @@ def check(self, dedup: dict) -> str: A soft idKind (Calibre) makes the verdict advisory, never an automatic skip.""" return ABSENT - def import_item(self, entry: dict, payload_path: str) -> dict: + def import_item(self, entry: dict, payload_path: str, + should_cancel: Optional[Callable[[], bool]] = None, + on_progress: Optional[Callable[[int], None]] = None) -> dict: """Ingest one already-local item archive into the app index, delegating to the app's own import. Idempotent: a re-import must not double-count or corrupt. ``payload_path`` - is a box-visible local file the front placed. Default: not supported.""" + is a box-visible local file the front placed. Runs off the request thread as the import + long operation, so it may be slow. ``should_cancel``, when supplied, is polled during the + long CLI work so an aborted op stops the ingest; ``on_progress(moved_bytes)`` reports the + bytes moved so far for the progress heartbeat. An app whose import is quick and in-process + may ignore both. Default: not supported.""" return {"ok": False, "reason": "import not supported"} def delete(self, local_ref: str) -> dict: diff --git a/static/conex-core/conex_core/router.py b/static/conex-core/conex_core/router.py index 8b5b9fef7..e3cbfc071 100644 --- a/static/conex-core/conex_core/router.py +++ b/static/conex-core/conex_core/router.py @@ -13,7 +13,7 @@ from . import VERSION from .export import disk_total_free from .export_ops import ExportOps -from .importing import is_within_dir +from .import_ops import ImportOps from .registry import Registry from .share_ops import ShareOps @@ -93,33 +93,6 @@ def _import_validate(registry: Registry, body: Any) -> Response: return 200, {"results": results} -def _import_item(registry: Registry, body: Any) -> Response: - """Ingest one already-local item archive. The front placed payloadPath on a box-visible - path inside the import area and drives the per-item sequence.""" - if not isinstance(body, dict): - return 400, {"error": "body must be {app, entry, payloadPath}"} - app = body.get("app") - entry = body.get("entry") - payload_path = body.get("payloadPath") - if not registry.has(app): - return 404, {"error": f"unknown app: {app}"} - if not isinstance(entry, dict) or not isinstance(payload_path, str) or not payload_path: - return 400, {"error": "entry and payloadPath are required"} - # Containment: the payload must live inside the import area, never an arbitrary path. - if not is_within_dir(registry.import_root, payload_path): - return 400, {"error": "payloadPath is outside the import area"} - # No free-space preflight here. The box runs under proot, whose statvfs under-reports the - # free space on the content volume, so a disk check here false-rejected imports on devices - # with ample room. The phone app is the one source of truth for free space (Android StatFs) - # and gates before this call; a real ENOSPC during ingest still fails cleanly. A space - # shortfall returns 507 (the adapter tags it code "no_space") so the front stops the run with - # a space message; any other failure stays 422 (a per-item retry). - result = registry.get(app).import_item(entry, payload_path) - if result.get("ok"): - return 200, result - return (507, result) if result.get("code") == "no_space" else (422, result) - - def _delete_item(registry: Registry, app: str, local_ref: str) -> Response: """Delete one item, delegating to the app's own delete. 404 when the item is absent, so a repeated delete is idempotent-safe. The destructive confirm is the front's (D1c); the core @@ -170,13 +143,14 @@ def _export_prepare(ops: Optional[ExportOps], body: Any) -> Response: def dispatch(registry: Registry, method: str, path: str, query: dict, body: Any = None, ops: Optional[ExportOps] = None, share: Optional[ShareOps] = None, - token: Optional[str] = None) -> Response: + token: Optional[str] = None, import_ops: Optional[ImportOps] = None) -> Response: """Route one request. ``query`` is a dict of first values; ``body`` is the parsed JSON - for a POST; ``ops`` is the export long-operation registry and ``share`` is the share-gate - registry (the server supplies both; the routes that need them guard against None). ``token`` - is the share session token: the server reads it from the ``X-Share-Token`` request header - (never the URL, so it stays out of access logs) and passes it here, so the router stays - header-agnostic and unit-testable without a socket.""" + for a POST; ``ops`` is the export long-operation registry, ``import_ops`` the import + long-operation registry, and ``share`` the share-gate registry (the server supplies all; the + routes that need them guard against None). ``token`` is the share session token: the server + reads it from the ``X-Share-Token`` request header (never the URL, so it stays out of access + logs) and passes it here, so the router stays header-agnostic and unit-testable without a + socket.""" trimmed = path.rstrip("/") or "/" segments = [unquote(s) for s in trimmed.split("/") if s != ""] @@ -204,9 +178,15 @@ def dispatch(registry: Registry, method: str, path: str, query: dict, body: Any # POST /v1/import/validate : what will import and what is skipped, nothing moved. if segments == ["v1", "import", "validate"]: return _import_validate(registry, body) - # POST /v1/import/item : ingest one already-local item archive. + # POST /v1/import/item : start the import long operation, return its opId. Import is a long + # op (K2GO-376): the ingest runs minutes on a real channel, so a single blocking POST times + # out at the relay. The op starts off the request thread; the client polls import status. if segments == ["v1", "import", "item"]: - return _import_item(registry, body) + if import_ops is None: + return 500, {"error": "import ops unavailable"} + if not isinstance(body, dict): + return 400, {"error": "body must be {app, entry, payloadPath}"} + return import_ops.start(body.get("app"), body.get("entry"), body.get("payloadPath")) return 404, {"error": "not found"} if method == "DELETE": @@ -215,6 +195,11 @@ def dispatch(registry: Registry, method: str, path: str, query: dict, body: Any if ops is None: return 500, {"error": "export ops unavailable"} return ops.delete(segments[2]) + # DELETE /v1/import/{opId} : abort an import long operation. + if len(segments) == 3 and segments[0] == "v1" and segments[1] == "import": + if import_ops is None: + return 500, {"error": "import ops unavailable"} + return import_ops.delete(segments[2]) # DELETE /v1/share/{sessionId} : cancel a share session (abort, by host or receiver). if len(segments) == 3 and segments[0] == "v1" and segments[1] == "share": if share is None: @@ -242,6 +227,12 @@ def dispatch(registry: Registry, method: str, path: str, query: dict, body: Any return 500, {"error": "export ops unavailable"} return ops.status(segments[3]) + # GET /v1/import/status/{opId} : the import op state + the moved-bytes heartbeat + the result. + if len(segments) == 4 and segments[0] == "v1" and segments[1] == "import" and segments[2] == "status": + if import_ops is None: + return 500, {"error": "import ops unavailable"} + return import_ops.status(segments[3]) + # GET /v1/export/fetch/{opId} : stream the prepared item (payload is an ExportStream the # server streams). 409 until the op is ready; the stream's on_close drops the op. if len(segments) == 4 and segments[0] == "v1" and segments[1] == "export" and segments[2] == "fetch": diff --git a/static/conex-core/conex_core/server.py b/static/conex-core/conex_core/server.py index e8f52285d..4dd74d876 100644 --- a/static/conex-core/conex_core/server.py +++ b/static/conex-core/conex_core/server.py @@ -14,6 +14,7 @@ from . import VERSION from .config import Config from .export_ops import ExportOps +from .import_ops import ImportOps from .ports import ExportStream from .registry import Registry from .router import dispatch @@ -30,7 +31,8 @@ log = logging.getLogger("conex_core") -def make_handler(registry: Registry, ops: ExportOps, share: ShareOps = None): +def make_handler(registry: Registry, ops: ExportOps, share: ShareOps = None, + import_ops: ImportOps = None): class Handler(BaseHTTPRequestHandler): server_version = f"conex-core/{VERSION}" # HTTP/1.1 so the chunked export stream is valid (HTTP/1.0 would hand the client the @@ -47,7 +49,7 @@ def _dispatch(self, method: str, path: str, query: dict, body=None): # with a JSON body, never a dropped connection the front reads as a network error. try: return dispatch(registry, method, path, query, body, ops=ops, share=share, - token=self._share_token()) + token=self._share_token(), import_ops=import_ops) except Exception: log.exception("unhandled error: %s %s", method, path) return 500, {"error": "internal error"} @@ -161,14 +163,17 @@ def serve(config: Config) -> None: format="%(asctime)s conex-core %(levelname)s %(message)s") registry = Registry.load(config) ops = ExportOps(registry, config.staging_dir) + import_ops = ImportOps(registry, registry.import_root) share = ShareOps(registry, ops) - # Reclaim export op temps left by a dead process before serving (K2GO-507 process-death - # cleanup owner): a fresh process holds no ops, so any op temp on disk is an orphan. The - # share sweep then drops any in-memory sessions (K2GO-376 closer 2); a share session's own - # on-disk footprint is its child export temps, already reclaimed by the export sweep above. + # Reclaim op temps left by a dead process before serving (the process-death cleanup owner): a + # fresh process holds no ops, so any op temp on disk is an orphan. Export reclaims its staging + # temps, import its work temps; the share sweep then drops any in-memory sessions (K2GO-376 + # closer 2), whose only on-disk footprint is child export temps already reclaimed above. ops.sweep_startup() + import_ops.sweep_startup() share.sweep_startup() - httpd = ThreadingHTTPServer((config.host, config.port), make_handler(registry, ops, share)) + httpd = ThreadingHTTPServer( + (config.host, config.port), make_handler(registry, ops, share, import_ops)) log.info("conex-core %s listening on http://%s:%d (apps: %s)", VERSION, config.host, config.port, ", ".join(registry.apps()) or "none") httpd.serve_forever() diff --git a/static/conex-core/conex_core/watchdog.py b/static/conex-core/conex_core/watchdog.py new file mode 100644 index 000000000..639063e34 --- /dev/null +++ b/static/conex-core/conex_core/watchdog.py @@ -0,0 +1,234 @@ +"""Liveness watchdog for a supervised child process (K2GO-376). + +A content import or export runs a Kolibri CLI child that can take minutes. A fixed timeout is +wrong for it: a large channel is slow but healthy, and a small one that hangs must still be +caught. So this module supervises the child by LIVENESS, not by a deadline: + + - Bytes are moving -> the child is alive. There is no time limit. + - CPU only, no bytes -> it may be a legitimate database phase, or a loop. Allow a longer + window before giving up. + - Nothing moves -> it is stalled. Terminate it and report. + +The stall window is RELATIVE TO THE TASK, not a wall clock. It is max(floor, k * the longest gap +between byte-progress events seen so far in this run), so a task that naturally pauses between +bursts widens its own tolerance. + +LivenessMonitor is pure: it takes samples and returns a verdict, so it is unit-tested without a +subprocess. run_supervised wires a real child, the /proc samplers, cancellation and exit +classification around it. The byte signal is /proc//io (rchar+wchar), which is readable under +proot for a same-uid child; directory growth is the fallback when /proc is not readable. +""" + +from __future__ import annotations + +import os +import signal +import subprocess +import time +from dataclasses import dataclass +from typing import Callable, Optional + +# The end states run_supervised reports. +OK = "ok" +CANCELLED = "cancelled" +STALLED = "stalled" +FAILED = "failed" +KILLED = "killed" + + +@dataclass +class SuperviseResult: + state: str # one of OK / CANCELLED / STALLED / FAILED / KILLED + returncode: Optional[int] + detail: str + + @property + def ok(self) -> bool: + return self.state == OK + + +def dir_size(path: Optional[str]) -> int: + """Sum of the file sizes under path, or 0 when it is missing or None. Shared with export_ops + as the staging heartbeat, and used here as the byte-progress fallback when /proc//io + cannot be read.""" + if not path: + return 0 + total = 0 + for dirpath, _dirnames, filenames in os.walk(path): + for name in filenames: + try: + total += os.path.getsize(os.path.join(dirpath, name)) + except OSError: + pass + return total + + +def proc_io_bytes(pid: int) -> Optional[int]: + """Bytes the process has moved through read()/write() syscalls (rchar+wchar from + /proc//io), or None when the file cannot be read. Verified readable under proot for a + same-uid child. rchar+wchar is preferred over read_bytes/write_bytes: it counts a page-cached + write too, so it reflects a copy in progress before the data reaches the disk.""" + try: + total = 0 + with open(f"/proc/{pid}/io", "r", encoding="ascii", errors="replace") as handle: + for line in handle: + if line.startswith(("rchar:", "wchar:")): + total += int(line.split(":", 1)[1]) + return total + except (OSError, ValueError): + return None + + +def proc_cpu_ticks(pid: int) -> Optional[int]: + """utime+stime in clock ticks from /proc//stat, or None. Field 2 (comm) is in parentheses + and may hold spaces or parentheses, so read the fields AFTER the last ')': utime and stime are + then at offsets 11 and 12.""" + try: + with open(f"/proc/{pid}/stat", "r", encoding="ascii", errors="replace") as handle: + data = handle.read() + fields = data.rpartition(")")[2].split() + return int(fields[11]) + int(fields[12]) + except (OSError, ValueError, IndexError): + return None + + +class LivenessMonitor: + """Pure stall detector. Feed it (now, activity_bytes, cpu_ticks) samples; it reports whether + the child still looks alive. Bytes moving resets the clock (no limit). CPU-only extends the + window. Nothing moving trips the stall after a task-relative window.""" + + def __init__(self, floor_seconds: float = 120.0, gap_multiplier: float = 4.0, + cpu_grace: float = 3.0): + self._floor = floor_seconds + self._gap_multiplier = gap_multiplier + self._cpu_grace = cpu_grace + self._last_activity = 0 + self._last_cpu: Optional[int] = None + self._progress_at = 0.0 + self._longest_gap = 0.0 + self._started = False + + def start(self, now: float, activity: int, cpu: Optional[int]) -> None: + self._last_activity = activity + self._last_cpu = cpu + self._progress_at = now + self._longest_gap = 0.0 + self._started = True + + def update(self, now: float, activity: int, cpu: Optional[int]) -> Optional[str]: + """Return None while the child looks alive, or a human-readable reason once it should be + killed. The first call seeds the baselines and always returns None.""" + if not self._started: + self.start(now, activity, cpu) + return None + if activity > self._last_activity: + # Bytes moved: the child is alive. Record this gap so the window grows to fit the + # task's own rhythm. + self._longest_gap = max(self._longest_gap, now - self._progress_at) + self._progress_at = now + self._last_activity = activity + self._last_cpu = cpu + return None + idle = now - self._progress_at + window = max(self._floor, self._gap_multiplier * self._longest_gap) + if cpu is not None and self._last_cpu is not None and cpu > self._last_cpu: + # CPU advances with no new bytes: a database phase, or a loop. Give it a longer window; + # if even that passes, treat it as a loop rather than progress. + self._last_cpu = cpu + if idle > self._cpu_grace * window: + return f"no byte progress for {idle:.0f}s while CPU-bound (likely a loop)" + return None + if idle > window: + return f"no progress for {idle:.0f}s (no bytes, no CPU)" + return None + + +def _kill(proc: subprocess.Popen) -> None: + """Terminate, then kill, giving the child a short grace to exit on SIGTERM.""" + try: + proc.terminate() + except OSError: + pass + try: + proc.wait(timeout=5) + return + except subprocess.TimeoutExpired: + pass + try: + proc.kill() + proc.wait(timeout=5) + except (OSError, subprocess.TimeoutExpired): + pass + + +def classify_exit(returncode: int) -> SuperviseResult: + """Map a child return code to a result. A negative code is death by signal; signal 9 (SIGKILL) + on a child that was not cancelled is almost always the OOM killer, so it is named as such.""" + if returncode == 0: + return SuperviseResult(OK, 0, "ok") + if returncode < 0: + sig = -returncode + # SIGKILL is 9 on Linux (the only platform the core serves on); getattr keeps this file + # importable on a host whose signal module has no SIGKILL, so the tests run anywhere. + if sig == getattr(signal, "SIGKILL", 9): + return SuperviseResult(KILLED, returncode, "killed (likely out of memory)") + return SuperviseResult(KILLED, returncode, f"killed by signal {sig}") + return SuperviseResult(FAILED, returncode, f"exited with code {returncode}") + + +def run_supervised(argv, env, *, dest_dir: Optional[str] = None, + should_cancel: Optional[Callable[[], bool]] = None, + poll_seconds: float = 1.0, floor_seconds: float = 120.0, + gap_multiplier: float = 4.0, cpu_grace: float = 3.0, + on_progress: Optional[Callable[[int], None]] = None, + log=None) -> SuperviseResult: + """Run argv as a child and supervise it by liveness. The child's stdout/stderr go to DEVNULL + (the caller owns user output). dest_dir is the directory whose growth is the byte-progress + fallback when /proc//io is unreadable. should_cancel is polled each interval; True + terminates the child and returns CANCELLED. on_progress(activity_bytes) is called each poll so + the caller can surface progress.""" + try: + proc = subprocess.Popen(argv, env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + except OSError as exc: + # The child could not start at all (a missing binary, a bad interpreter). Own it here as a + # clean failure rather than letting the raw OSError escape the supervisor. + return SuperviseResult(FAILED, None, f"could not start: {exc}") + pid = proc.pid + # Decide the byte source once: the box either exposes /proc//io or it does not. + use_io = proc_io_bytes(pid) is not None + + def activity() -> int: + if use_io: + value = proc_io_bytes(pid) + if value is not None: + return value + return dir_size(dest_dir) + + monitor = LivenessMonitor(floor_seconds, gap_multiplier, cpu_grace) + monitor.start(time.monotonic(), activity(), proc_cpu_ticks(pid)) + try: + while True: + returncode = proc.poll() + if returncode is not None: + return classify_exit(returncode) + if should_cancel is not None and should_cancel(): + _kill(proc) + return SuperviseResult(CANCELLED, None, "cancelled") + time.sleep(poll_seconds) + current = activity() + if on_progress is not None: + try: + on_progress(current) + except Exception: # a progress sink must never break the supervision loop + pass + reason = monitor.update(time.monotonic(), current, proc_cpu_ticks(pid)) + if reason is not None: + _kill(proc) + if log is not None: + log.warning("supervised child stalled: %s (%s)", reason, argv[0]) + return SuperviseResult(STALLED, None, reason) + finally: + # Never leak the child. A normal exit already reaped it, so _kill on a dead process is a + # harmless no-op; this only matters if the loop left with the child still running. + if proc.poll() is None: + _kill(proc) diff --git a/static/conex-core/tests/test_import_ops.py b/static/conex-core/tests/test_import_ops.py new file mode 100644 index 000000000..4c892ebc6 --- /dev/null +++ b/static/conex-core/tests/test_import_ops.py @@ -0,0 +1,192 @@ +import os +import tempfile +import threading +import time +import unittest + +from conex_core.import_ops import ImportOps +from conex_core.ports import ContentAdapter +from conex_core.registry import Registry + + +class _ImportFake(ContentAdapter): + """A fake import adapter: import_item returns a configured result, optionally firing progress + first and optionally waiting on a gate so a test can observe the running state.""" + + app = "kiwix" + + def __init__(self, result=None, gate=None, progress=None): + self._result = result if result is not None else {"ok": True, "app": "kiwix", "localRef": "r"} + self._gate = gate + self._progress = progress or [] + self.saw_cancel = False + + def is_installed(self): + return True + + def list(self): + return [] + + def import_item(self, entry, payload_path, should_cancel=None, on_progress=None): + for value in self._progress: + if on_progress: + on_progress(value) + if self._gate is not None: + self._gate.wait(5) + if should_cancel and should_cancel(): + self.saw_cancel = True + return self._result + + +def _ops(adapter, root, ttl=60): + reg = Registry([adapter], import_root=root) + return ImportOps(reg, import_root=root, ttl_seconds=ttl) + + +def _payload(root, name="a.zim"): + path = os.path.join(root, name) + with open(path, "wb") as f: + f.write(b"x") + return path + + +def _wait_state(ops, op_id, want, timeout=5): + end = time.time() + timeout + body = {} + while time.time() < end: + _, body = ops.status(op_id) + if body.get("state") == want: + return body + time.sleep(0.01) + return body + + +class StartTest(unittest.TestCase): + def test_start_runs_and_completes_to_done_with_result(self): + with tempfile.TemporaryDirectory() as d: + payload = _payload(d) + ops = _ops(_ImportFake(result={"ok": True, "app": "kiwix", "localRef": "r"}), d) + status, body = ops.start("kiwix", {"identity": {"localRef": "a.zim"}, "bytes": 10}, payload) + self.assertEqual(status, 200) + self.assertEqual(body["state"], "running") + self.assertIn("opId", body) + done = _wait_state(ops, body["opId"], "done") + self.assertEqual(done["state"], "done") + self.assertEqual(done["result"]["localRef"], "r") + self.assertEqual(done["movedBytes"], 10) + self.assertEqual(done["totalBytes"], 10) + + def test_start_unknown_app_404(self): + with tempfile.TemporaryDirectory() as d: + ops = _ops(_ImportFake(), d) + status, _ = ops.start("nope", {"identity": {"localRef": "a"}}, os.path.join(d, "p")) + self.assertEqual(status, 404) + + def test_start_bad_body_400(self): + with tempfile.TemporaryDirectory() as d: + ops = _ops(_ImportFake(), d) + self.assertEqual(ops.start("kiwix", "not-a-dict", os.path.join(d, "p"))[0], 400) + self.assertEqual(ops.start("kiwix", {"x": 1}, "")[0], 400) + + def test_start_rejects_payload_outside_area(self): + with tempfile.TemporaryDirectory() as d: + ops = _ops(_ImportFake(), d) + status, _ = ops.start("kiwix", {"identity": {"localRef": "a"}}, "/etc/passwd") + self.assertEqual(status, 400) + + +class StatusTest(unittest.TestCase): + def test_unknown_op_404(self): + with tempfile.TemporaryDirectory() as d: + self.assertEqual(_ops(_ImportFake(), d).status("deadbeef")[0], 404) + + def test_failing_adapter_becomes_error_without_code(self): + with tempfile.TemporaryDirectory() as d: + payload = _payload(d) + ops = _ops(_ImportFake(result={"ok": False, "reason": "bad archive"}), d) + _, body = ops.start("kiwix", {"identity": {"localRef": "a.zim"}, "bytes": 5}, payload) + err = _wait_state(ops, body["opId"], "error") + self.assertEqual(err["error"], "bad archive") + self.assertNotIn("code", err) + + def test_no_space_result_carries_structural_code(self): + with tempfile.TemporaryDirectory() as d: + payload = _payload(d) + ops = _ops(_ImportFake(result={"ok": False, "code": "no_space", "reason": "full"}), d) + _, body = ops.start("kiwix", {"identity": {"localRef": "a.zim"}, "bytes": 5}, payload) + err = _wait_state(ops, body["opId"], "error") + self.assertEqual(err["code"], "no_space") + + def test_progress_is_reported_and_clamped_to_total(self): + with tempfile.TemporaryDirectory() as d: + payload = _payload(d) + gate = threading.Event() + # The watchdog reports cumulative syscall bytes that exceed the payload; the op clamps + # moved_bytes to the known total (10), so the bar never passes 100%. + ops = _ops(_ImportFake(gate=gate, progress=[100]), d) + _, body = ops.start("kiwix", {"identity": {"localRef": "a.zim"}, "bytes": 10}, payload) + op_id = body["opId"] + end = time.time() + 5 + moved = 0 + while time.time() < end: + _, st = ops.status(op_id) + if st["state"] == "running" and st["movedBytes"] > 0: + moved = st["movedBytes"] + break + time.sleep(0.01) + self.assertEqual(moved, 10) + gate.set() + _wait_state(ops, op_id, "done") + + +class CancelAndSweepTest(unittest.TestCase): + def test_delete_drops_op_and_cancels_the_worker_idempotent(self): + with tempfile.TemporaryDirectory() as d: + payload = _payload(d) + gate = threading.Event() + fake = _ImportFake(gate=gate) + ops = _ops(fake, d) + _, body = ops.start("kiwix", {"identity": {"localRef": "a.zim"}, "bytes": 1}, payload) + op_id = body["opId"] + self.assertEqual(ops.delete(op_id), (200, {"ok": True})) + self.assertEqual(ops.status(op_id)[0], 404) + gate.set() # release the worker; it sees should_cancel True (the op is gone) + end = time.time() + 2 + while time.time() < end and not fake.saw_cancel: + time.sleep(0.01) + self.assertTrue(fake.saw_cancel) + # idempotent: a repeat and an unknown op are both 200 ok. + self.assertEqual(ops.delete(op_id), (200, {"ok": True})) + self.assertEqual(ops.delete("never-existed"), (200, {"ok": True})) + + def test_ttl_sweep_reclaims_an_abandoned_op(self): + with tempfile.TemporaryDirectory() as d: + payload = _payload(d) + ops = _ops(_ImportFake(), d, ttl=60) + _, body = ops.start("kiwix", {"identity": {"localRef": "a.zim"}, "bytes": 1}, payload) + op_id = body["opId"] + _wait_state(ops, op_id, "done") + ops._ops[op_id].last_seen -= 120 + ops._sweep_expired() + self.assertNotIn(op_id, ops._ops) + + def test_startup_sweep_reclaims_import_temps_only(self): + with tempfile.TemporaryDirectory() as d: + mine = os.path.join(d, "cximport-abc") + export = os.path.join(d, "cxexport-xyz") # an export temp: import's sweep leaves it + unrelated = os.path.join(d, "kolibri") + for p in (mine, export, unrelated): + os.makedirs(p) + ops = _ops(_ImportFake(), d) + ops.sweep_startup() + self.assertFalse(os.path.exists(mine)) + self.assertTrue(os.path.exists(export)) + self.assertTrue(os.path.exists(unrelated)) + + def test_startup_sweep_tolerates_missing_dir(self): + ops = _ops(_ImportFake(), "/no/such/import/dir") + ops.sweep_startup() # must not raise + + +if __name__ == "__main__": + unittest.main() diff --git a/static/conex-core/tests/test_kolibri.py b/static/conex-core/tests/test_kolibri.py index e6f06e5d7..86998b507 100644 --- a/static/conex-core/tests/test_kolibri.py +++ b/static/conex-core/tests/test_kolibri.py @@ -1,8 +1,6 @@ import os import sqlite3 -import sys import tempfile -import time import unittest import io @@ -11,13 +9,13 @@ import conex_core.adapters.kolibri as kolibri_mod from conex_core.adapters.kolibri import ( KolibriAdapter, - _run_cancellable, build_channel_item, delete_command, export_commands, import_commands, ) from conex_core.model import AUTHORITATIVE +from conex_core.watchdog import FAILED, OK, SuperviseResult VALID_ID = "c1f2b7e6ac9f56a2bb44fa7a48b66dce" @@ -131,17 +129,17 @@ def test_absent_when_no_such_channel(self): class _ImportStubKolibri(KolibriAdapter): - def _stage_import(self, channel_id, source): + def _stage_import(self, channel_id, source, should_cancel=None, on_progress=None): self.import_channel = channel_id self.import_source = source - return True + return SuperviseResult(OK, 0, "ok") class _ImportFailStubKolibri(KolibriAdapter): """The CLI import seam fails (non-zero exit), so import_item takes its failure branch.""" - def _stage_import(self, channel_id, source): - return False + def _stage_import(self, channel_id, source, should_cancel=None, on_progress=None): + return SuperviseResult(FAILED, 1, "exited with code 1") def _bundle_tar(tar_path, channel_id): @@ -335,26 +333,6 @@ def test_open_fetch_without_staged_dir_is_none(self): self.assertIsNone(adapter.open_fetch(VALID_ID, os.path.join(d, "missing"))) -class RunCancellableTest(unittest.TestCase): - def test_completed_command_returns_true(self): - self.assertTrue(_run_cancellable([sys.executable, "-c", "pass"], dict(os.environ))) - - def test_failed_command_returns_false(self): - self.assertFalse( - _run_cancellable([sys.executable, "-c", "import sys; sys.exit(3)"], dict(os.environ))) - - def test_cancel_terminates_a_long_command(self): - start = time.time() - # A 30 s child that is cancelled at once: it must be terminated promptly, not waited out. - ok = _run_cancellable( - [sys.executable, "-c", "import time; time.sleep(30)"], - dict(os.environ), - should_cancel=lambda: True, - ) - self.assertFalse(ok) - self.assertLess(time.time() - start, 10) - - class DeleteCommandTest(unittest.TestCase): def test_builds_the_delegated_delete_command(self): self.assertEqual( diff --git a/static/conex-core/tests/test_router.py b/static/conex-core/tests/test_router.py index 839cc6165..02564d43f 100644 --- a/static/conex-core/tests/test_router.py +++ b/static/conex-core/tests/test_router.py @@ -42,7 +42,7 @@ def list(self): def check(self, dedup): return self._verdict - def import_item(self, entry, payload_path): + def import_item(self, entry, payload_path, should_cancel=None, on_progress=None): self.imported = (entry, payload_path) return self._import_result @@ -190,61 +190,47 @@ def test_import_validate_partial(self): self.assertEqual(out["results"][0]["verdict"], "ABSENT") self.assertEqual(out["results"][1]["status"], "NO_APP") - def test_import_item_happy_path(self): - with tempfile.TemporaryDirectory() as root: - fake = FakeAdapter("kiwix", [], import_result={"ok": True, "app": "kiwix"}) - reg = Registry([fake], import_root=root) - payload = os.path.join(root, "a.zim") - body = {"app": "kiwix", "entry": {"identity": {"localRef": "a.zim"}, "bytes": 10}, "payloadPath": payload} - status, out = dispatch(reg, "POST", "/v1/import/item", {}, body) - self.assertEqual(status, 200) - self.assertTrue(out["ok"]) - self.assertEqual(fake.imported[1], payload) - - def test_import_item_no_free_space_preflight(self): - # The box runs under proot, whose statvfs under-reports free space, so the router must - # not preflight the disk: even an implausibly large declared size still reaches the - # adapter. The phone app is the one source of truth for free space (Android StatFs); a - # real ENOSPC fails cleanly in the adapter (422). + # Import is a long op (K2GO-376): POST /v1/import/item STARTS the op and returns its opId; the + # ingest result is polled via GET /v1/import/status. These check the routing and the missing-ops + # guard; the op mechanics (the ingest result, no_space, cancel) live in test_import_ops. + + def test_import_start_returns_opid(self): + from conex_core.import_ops import ImportOps with tempfile.TemporaryDirectory() as root: fake = FakeAdapter("kiwix", [], import_result={"ok": True, "app": "kiwix"}) reg = Registry([fake], import_root=root) + import_ops = ImportOps(reg, import_root=root, ttl_seconds=60) payload = os.path.join(root, "a.zim") - body = {"app": "kiwix", - "entry": {"identity": {"localRef": "a.zim"}, "bytes": 10 ** 18}, + body = {"app": "kiwix", "entry": {"identity": {"localRef": "a.zim"}, "bytes": 10}, "payloadPath": payload} - status, out = dispatch(reg, "POST", "/v1/import/item", {}, body) + status, out = dispatch(reg, "POST", "/v1/import/item", {}, body, import_ops=import_ops) self.assertEqual(status, 200) - self.assertTrue(out["ok"]) + self.assertEqual(out["state"], "running") + self.assertIn("opId", out) - def test_import_item_rejects_payload_outside_area(self): + def test_import_start_without_import_ops_500(self): with tempfile.TemporaryDirectory() as root: reg = Registry([FakeAdapter("kiwix", [])], import_root=root) - body = {"app": "kiwix", "entry": {"identity": {"localRef": "a.zim"}}, "payloadPath": "/etc/passwd"} - status, _ = dispatch(reg, "POST", "/v1/import/item", {}, body) - self.assertEqual(status, 400) + body = {"app": "kiwix", "entry": {"identity": {"localRef": "a.zim"}}, + "payloadPath": os.path.join(root, "x")} + status, _ = dispatch(reg, "POST", "/v1/import/item", {}, body, import_ops=None) + self.assertEqual(status, 500) - def test_import_item_adapter_failure_422(self): + def test_import_start_rejects_payload_outside_area(self): + from conex_core.import_ops import ImportOps with tempfile.TemporaryDirectory() as root: - fake = FakeAdapter("kiwix", [], import_result={"ok": False, "reason": "bad archive"}) - reg = Registry([fake], import_root=root) - body = {"app": "kiwix", "entry": {"identity": {"localRef": "a.zim"}}, "payloadPath": os.path.join(root, "x")} - status, out = dispatch(reg, "POST", "/v1/import/item", {}, body) - self.assertEqual(status, 422) - self.assertFalse(out["ok"]) - - def test_import_item_no_space_507(self): - # A space shortfall (the adapter tags code "no_space") maps to 507, not 422, so the front - # stops the run with a space message instead of a per-item retry. - with tempfile.TemporaryDirectory() as root: - fake = FakeAdapter("kiwix", [], - import_result={"ok": False, "code": "no_space", "reason": "extract failed"}) - reg = Registry([fake], import_root=root) - body = {"app": "kiwix", "entry": {"identity": {"localRef": "a.zim"}}, "payloadPath": os.path.join(root, "x")} - status, out = dispatch(reg, "POST", "/v1/import/item", {}, body) - self.assertEqual(status, 507) - self.assertFalse(out["ok"]) - self.assertEqual(out["code"], "no_space") + reg = Registry([FakeAdapter("kiwix", [])], import_root=root) + import_ops = ImportOps(reg, import_root=root, ttl_seconds=60) + body = {"app": "kiwix", "entry": {"identity": {"localRef": "a.zim"}}, + "payloadPath": "/etc/passwd"} + status, _ = dispatch(reg, "POST", "/v1/import/item", {}, body, import_ops=import_ops) + self.assertEqual(status, 400) + + def test_import_status_and_delete_without_import_ops_500(self): + reg = _registry() + for method, path in [("GET", "/v1/import/status/abc"), ("DELETE", "/v1/import/abc")]: + status, _ = dispatch(reg, method, path, {}, import_ops=None) + self.assertEqual(status, 500) class DeleteRouteTest(unittest.TestCase): diff --git a/static/conex-core/tests/test_watchdog.py b/static/conex-core/tests/test_watchdog.py new file mode 100644 index 000000000..d32dfa4f8 --- /dev/null +++ b/static/conex-core/tests/test_watchdog.py @@ -0,0 +1,122 @@ +import sys +import threading +import unittest + +from conex_core.watchdog import ( + CANCELLED, + FAILED, + KILLED, + OK, + STALLED, + LivenessMonitor, + classify_exit, + run_supervised, +) + + +class LivenessMonitorTest(unittest.TestCase): + def _monitor(self): + # Small floor so a test trips the stall in a few simulated seconds. + return LivenessMonitor(floor_seconds=10.0, gap_multiplier=4.0, cpu_grace=3.0) + + def test_first_sample_seeds_and_is_alive(self): + monitor = self._monitor() + self.assertIsNone(monitor.update(0.0, 0, 0)) + + def test_bytes_moving_is_always_alive(self): + monitor = self._monitor() + monitor.start(0.0, 0, 0) + # Each sample moves bytes well past the floor; the child is never declared stalled. + for t in range(1, 200, 20): + self.assertIsNone(monitor.update(float(t), t * 1000, 0)) + + def test_nothing_moving_stalls_after_the_floor_window(self): + monitor = self._monitor() + monitor.start(0.0, 0, 0) + self.assertIsNone(monitor.update(5.0, 0, 0)) # idle 5 < floor 10 + reason = monitor.update(11.0, 0, 0) # idle 11 > floor 10 + self.assertIsNotNone(reason) + self.assertIn("no progress", reason) + + def test_cpu_only_extends_the_window(self): + monitor = self._monitor() + monitor.start(0.0, 0, 0) + # CPU keeps advancing with no new bytes: tolerated up to cpu_grace * window (3 * 10 = 30 s). + self.assertIsNone(monitor.update(11.0, 0, 5)) # past the floor, but CPU moved + self.assertIsNone(monitor.update(25.0, 0, 9)) # still inside the CPU grace window + reason = monitor.update(31.0, 0, 12) # past 30 s of CPU-only + self.assertIsNotNone(reason) + self.assertIn("CPU-bound", reason) + + def test_window_is_relative_to_the_longest_gap(self): + monitor = self._monitor() + monitor.start(0.0, 0, 0) + # A healthy 30 s gap between byte bursts raises the window to 4 * 30 = 120 s. + self.assertIsNone(monitor.update(30.0, 1000, 0)) + self.assertIsNone(monitor.update(140.0, 1000, 0)) # idle 110 < 120: still alive + reason = monitor.update(260.0, 1000, 0) # idle 230 > 120: stalled + self.assertIsNotNone(reason) + + +class ClassifyExitTest(unittest.TestCase): + def test_zero_is_ok(self): + result = classify_exit(0) + self.assertEqual(result.state, OK) + self.assertTrue(result.ok) + + def test_nonzero_is_failed_with_code(self): + result = classify_exit(1) + self.assertEqual(result.state, FAILED) + self.assertIn("code 1", result.detail) + + def test_sigkill_is_named_as_oom(self): + result = classify_exit(-9) + self.assertEqual(result.state, KILLED) + self.assertIn("out of memory", result.detail) + + def test_other_signal_is_named(self): + result = classify_exit(-15) + self.assertEqual(result.state, KILLED) + self.assertIn("signal 15", result.detail) + + +class RunSupervisedTest(unittest.TestCase): + """Drives a real short-lived child. The stall logic is covered purely by LivenessMonitorTest; + here we only confirm the exit, cancel and progress wiring around a live process.""" + + def _run(self, code, **kwargs): + return run_supervised([sys.executable, "-c", code], None, poll_seconds=0.02, **kwargs) + + def test_clean_exit_is_ok(self): + result = self._run("import sys; sys.exit(0)") + self.assertEqual(result.state, OK) + + def test_nonzero_exit_is_failed(self): + result = self._run("import sys; sys.exit(3)") + self.assertEqual(result.state, FAILED) + self.assertEqual(result.returncode, 3) + + def test_cancel_terminates_the_child(self): + flag = threading.Event() + + def should_cancel(): + flag.set() # ask to cancel on the first poll + return True + + result = run_supervised( + [sys.executable, "-c", "import time; time.sleep(30)"], None, + poll_seconds=0.02, should_cancel=should_cancel) + self.assertEqual(result.state, CANCELLED) + self.assertTrue(flag.is_set()) + + def test_on_progress_is_called(self): + seen = [] + result = run_supervised( + [sys.executable, "-c", "import time; time.sleep(0.2)"], None, + poll_seconds=0.02, on_progress=seen.append) + self.assertEqual(result.state, OK) + self.assertTrue(seen) # at least one progress sample while the child ran + + +if __name__ == "__main__": + unittest.main() diff --git a/static/dashboard/CHANGELOG.md b/static/dashboard/CHANGELOG.md index d760d82f9..2aa6c731f 100644 --- a/static/dashboard/CHANGELOG.md +++ b/static/dashboard/CHANGELOG.md @@ -4,6 +4,7 @@ One line per version, newest first. Every REST-facing change bumps the version i (the app surfaces it via `/system/dashboard/update-check` and the "Update available" pill), so this file is the human record of what each bump enables. Keep entries short: `version - change (TICKET)`. +- **1.4.8** - The /k2go-api nginx location waits up to 3600 s for a reply (the default was 60 s). A Kolibri content import through the conex-core relay takes minutes and nginx cut it with a 504. Localhost-only. (K2GO-376) - **1.4.7** - FQR map downloads pause and resume on the durable job engine (K2GO-317). Pause now sends the extract a graceful SIGTERM, so the custom go-pmtiles build flushes its on-disk resume journal (`.resume`); resume re-runs tile-extract, which continues from that journal instead of re-extracting each layer from zero. The maps runner keeps the partial and the journal on pause, purges both on cancel or any other failure, reads the json progress line by line (a split stdout chunk no longer drops a second of progress), and reports whole-job percent with speed and eta. Resume from the journal needs the box rootfs to carry the m16 pmtiles binary and the patched tile-extract.py. Localhost-only. (K2GO-317) - **1.4.6** - Rename the content-exchange sigla cx to conex: the relay file is now conex-proxy and the core-port env var is CONEX_PORT (naming only; same /v1 localhost relay to the core on 127.0.0.1:9500). (chore) - **1.4.5** - Relay to the content-exchange core (conex-core). dash-node now forwards `/v1/*` (reached externally at `/k2go-api/v1/*`) to the in-box conex-core REST service on `127.0.0.1:9500`: GET and POST, including the streamed `GET /v1/export/item`. dash-node only relays; the core owns the content state (one owner per fact). No proxy library (Node core `http`); the core's port is env-overridable via `CONEX_PORT`. Localhost-only. (K2GO-427) diff --git a/static/dashboard/dash-node-nginx.conf b/static/dashboard/dash-node-nginx.conf index 2c798309f..49b7b142a 100644 --- a/static/dashboard/dash-node-nginx.conf +++ b/static/dashboard/dash-node-nginx.conf @@ -15,6 +15,9 @@ location ^~ /k2go-api/ { error_page 403 =404 /404.html; proxy_pass http://127.0.0.1:4000/api/; + # Default 60 s cut a long call (a Kolibri content import takes minutes) with a 504 (K2GO-376). + proxy_read_timeout 3600s; + proxy_send_timeout 3600s; proxy_set_header Host $http_host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; diff --git a/static/dashboard/package.json b/static/dashboard/package.json index 2edfb8cd9..c79539f20 100644 --- a/static/dashboard/package.json +++ b/static/dashboard/package.json @@ -1,6 +1,6 @@ { "name": "dashboard-console", - "version": "1.4.7", + "version": "1.4.8", "description": "", "main": "index.js", "scripts": {