K2GO-507 feat(cx-core): export as a prepare/poll/fetch long operation

static/cx-core/README.md

@@ -6,7 +6,9 @@ Calibre. It is client-agnostic: dash-node (K2Go) or the IIAB Console reach it ov
 REST API. See `controller/docs/ADR-content-exchange.md` for the frozen Pillar 0
 contracts (identity, manifest, adapter port, core REST API).
 
-This directory is the first slice: the READ side of Pillar 1, with the Kiwix adapter.
+The read, export, import and delete sides are built for all three adapters. Export runs
+as a liveness-paced long operation (prepare / poll / fetch), so a multi-gigabyte item does
+not time out on the client (K2GO-507).
 
 ## What is here
 
@@ -33,16 +35,39 @@ Environment (defaults match the box layout):
 - `CX_HOST` (default `127.0.0.1`), `CX_PORT` (default `9500`).
 - `CX_ZIMS_DIR` (default `/library/zims/content/`): where the ZIM files live.
 
-## Endpoints (read side)
+## Endpoints
 
-- `GET /v1/apps`: installed content apps that have a loaded adapter.
+Read:
+
+- `GET /v1/version`: the core build (mirrors dash-node `/system/version`).
+- `GET /v1/apps`: installed content apps that have a loaded adapter, plus `version`.
 - `GET /v1/content[?app=<app>]`: every item this box has, with identity + size. Filterable.
 - `GET /v1/content/{app}/{localRef}`: one item detail.
+- `GET /v1/storage`: content-volume disk summary + per-app content totals.
 
 Each item carries two identity handles: `localRef` (the app's own handle for
 export / delete) and `dedup` (the cross-box "already have this?" key). For Kiwix both
 are read, not computed: `dedup.build` is the ZIM header UUID.
 
+Export (a long operation, K2GO-507): a large item (a Kolibri channel) must stage before
+its bytes can stream, which takes minutes, so export is prepare / poll / fetch instead of
+one blocking request. The front owns the sequence; the core owns the op and the temp.
+
+- `POST /v1/export/prepare {app, localRef}` -> `{opId, state, totalBytes}`. `state` is
+  `staging` (a background thread is staging) or `ready` (nothing to stage). `507` when the
+  staging disk cannot hold the item.
+- `GET /v1/export/status/{opId}` -> `{state, stagedBytes, totalBytes, error?}`. `stagedBytes`
+  is the staging heartbeat (it grows as the export is assembled). `404` when the op is
+  unknown (swept), so the client re-prepares.
+- `GET /v1/export/fetch/{opId}` -> the chunked octet stream. `409` until the op is ready;
+  the stream's close drops the op and removes its temp.
+- `DELETE /v1/export/{opId}` -> `{ok: true}`. Client abort; idempotent.
+
+Import / delete:
+
+- `POST /v1/export/plan`, `POST /v1/dedup/check`, `POST /v1/import/validate`,
+  `POST /v1/import/item`, `DELETE /v1/content/{app}/{localRef}`.
+
 ## Test
 
 ```
@@ -51,17 +76,15 @@ python -m unittest discover -s tests -t .
 ```
 
 The tests cover the pure logic with no device: ZIM header UUID parsing, filename
-decomposition, soft-key normalization, the manifest JSON shape, and route dispatch.
-
-## Not in this slice
+decomposition, soft-key normalization, the manifest JSON shape, route dispatch, the export
+long-operation registry (prepare / status / fetch / delete, the TTL and startup sweeps),
+and the streaming over a real socket.
 
-Export, import, delete, the dedup-check endpoint, the Kolibri and Calibre adapters, the
-offline-disk and network carriers, strict-mode hashing, and the manager UI. The adapter
-boundary is kept clean so Kolibri and Calibre slot in next.
+## Not built yet
 
-Deployment (D1d: an Ansible-role baseline plus a dash-node-style live self-update) is
-not built here; this slice only has to be runnable. Live on-device verification needs a
-box and is a follow-up.
+The network and QR carriers, strict-mode content hashing, the cx-core live self-update
+(D1d; the Ansible role baseline is in `tools/upstream-patches/overlays/roles/cx_core`), and
+the dash-node <-> core listing consolidation (D1b).
 
 The registry resolves installed apps once at startup (a snapshot). Reacting to a module
 being installed or removed at runtime is a deferred design item: it must consume the

static/cx-core/cx_core/__init__.py

@@ -1,7 +1,11 @@
 """Content-exchange core: an in-box REST service for granular content exchange.
 
-See controller/docs/ADR-content-exchange.md for the frozen Pillar 0 contracts.
-This package holds the read side of Pillar 1 (list + identity + size).
+See controller/docs/ADR-content-exchange.md for the frozen Pillar 0 contracts and the
+K2GO-507 export long-operation design.
 """
 
-__version__ = "0.1.0"
+# 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
+# cx_core Ansible role writes this into the [cx_core] section of iiab.ini.
+VERSION = "0.2.0"
+__version__ = VERSION

static/cx-core/cx_core/adapters/calibre.py

@@ -14,7 +14,6 @@
 
 from __future__ import annotations
 
-import os
 import os
 import shutil
 import sqlite3
@@ -25,7 +24,7 @@
 from ..export import resolve_within, tar_stream
 from ..importing import safe_extract_tar, single_child_dir
 from ..model import ABSENT, SAME_VERSION, SOFT, Dedup, Identity, ManifestItem, normalize_text
-from ..ports import ContentAdapter, ExportStream
+from ..ports import ContentAdapter, ExportPlan, ExportStream
 from .calibreweb import CalibreWebClient, resolve_calibre_credentials
 
 # One row per book, with its authors, formats, total byte-size and one external id if any.
@@ -218,22 +217,43 @@ def _book_path(self, uuid: str) -> Optional[str]:
             con.close()
         return row[0] if row and row[0] else None
 
-    def open_export(self, local_ref: str) -> Optional[ExportStream]:
-        # Export the book's own directory (its format files, cover and opf), the layout
-        # Calibre itself reads back. books.path comes from the DB, so a containment guard
-        # keeps a crafted path from escaping the library root.
-        rel = self._book_path(local_ref)
-        if rel is None:
+    def prepare_export(self, local_ref: str) -> Optional[ExportPlan]:
+        # A book's directory streams directly as a tar, so nothing is staged. total_bytes is
+        # the item's summed data size (the tar also carries the cover / opf sidecars, so the
+        # stream is a little larger; total_bytes is the progress denominator, not exact).
+        book_dir = self._export_dir(local_ref)
+        if book_dir is None:
             return None
-        book_dir = resolve_within(self._lib_root, rel)
-        if book_dir is None or not os.path.isdir(book_dir):
+        item = self.get(local_ref)
+        return ExportPlan(
+            stages=False,
+            total_bytes=item.bytes if item else 0,
+            filename=f"{local_ref}.tar",
+            content_type="application/x-tar",
+        )
+
+    def open_fetch(self, local_ref: str, staged_dir: Optional[str] = None) -> Optional[ExportStream]:
+        # Direct app: staged_dir is unused. Re-resolve at fetch time.
+        book_dir = self._export_dir(local_ref)
+        if book_dir is None:
             return None
         return ExportStream(
             filename=f"{local_ref}.tar",
             content_type="application/x-tar",
             chunks=tar_stream(book_dir, arcname=local_ref),
         )
 
+    def _export_dir(self, local_ref: str) -> Optional[str]:
+        # books.path comes from the DB, so a containment guard keeps a crafted path from
+        # escaping the library root.
+        rel = self._book_path(local_ref)
+        if rel is None:
+            return None
+        book_dir = resolve_within(self._lib_root, rel)
+        if book_dir is None or not os.path.isdir(book_dir):
+            return None
+        return book_dir
+
     def _query(self, only_uuid: Optional[str]) -> List[ManifestItem]:
         if not os.path.exists(self._db):
             return []

static/cx-core/cx_core/adapters/kiwix.py

@@ -32,7 +32,7 @@
     Identity,
     ManifestItem,
 )
-from ..ports import ContentAdapter, ExportStream
+from ..ports import ContentAdapter, ExportPlan, ExportStream
 
 # openzim header: magic number is uint32 little-endian 0x044D495A at offset 0, and the
 # 16-byte UUID is at offset 8. We only need the first 24 bytes.
@@ -160,20 +160,38 @@ def get(self, local_ref: str) -> Optional[ManifestItem]:
             return None
         return build_item(local_ref, size, read_zim_uuid_from_file(path))
 
-    def open_export(self, local_ref: str) -> Optional[ExportStream]:
-        # The item IS the .zim file: stream it as-is (no archive wrapper). Same containment
+    def prepare_export(self, local_ref: str) -> Optional[ExportPlan]:
+        # A ZIM streams directly (it IS the file), so nothing is staged. Same containment
         # guard as get(), so a traversal ref never reaches the filesystem.
-        if not is_safe_local_ref(local_ref) or not local_ref.lower().endswith(".zim"):
+        path = self._export_path(local_ref)
+        if path is None:
             return None
-        path = os.path.join(self._dir, local_ref)
-        if not os.path.isfile(path):
+        return ExportPlan(
+            stages=False,
+            total_bytes=os.path.getsize(path),
+            filename=local_ref,
+            content_type="application/octet-stream",
+        )
+
+    def open_fetch(self, local_ref: str, staged_dir: Optional[str] = None) -> Optional[ExportStream]:
+        # Direct app: staged_dir is unused. Re-validate at fetch time (the file may have
+        # vanished between prepare and fetch).
+        path = self._export_path(local_ref)
+        if path is None:
             return None
         return ExportStream(
             filename=local_ref,
             content_type="application/octet-stream",
             chunks=file_stream(path),
         )
 
+    def _export_path(self, local_ref: str) -> Optional[str]:
+        """The on-disk path of an exportable ZIM, or None. Containment-guarded, like get()."""
+        if not is_safe_local_ref(local_ref) or not local_ref.lower().endswith(".zim"):
+            return None
+        path = os.path.join(self._dir, local_ref)
+        return path if os.path.isfile(path) else None
+
     def check(self, dedup: dict) -> str:
         # Same (creator, lang, flavour) line is the comparable set; the UUID is the exact
         # build and the date is the version axis. Authoritative, so the verdict is a hard fact.

static/cx-core/cx_core/adapters/kolibri.py

@@ -21,7 +21,7 @@
 import tempfile
 from typing import List, Optional
 
-from ..export import free_bytes, has_room, tar_stream
+from ..export import free_bytes, tar_stream
 from ..importing import safe_extract_tar, single_child_dir
 from ..model import (
     ABSENT,
@@ -33,7 +33,7 @@
     Identity,
     ManifestItem,
 )
-from ..ports import ContentAdapter, ExportStream
+from ..ports import ContentAdapter, ExportPlan, ExportStream
 
 _CHANNEL_ID_RE = re.compile(r"^[0-9a-fA-F]{32}$")
 
@@ -75,6 +75,27 @@ 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."""
@@ -169,48 +190,52 @@ def get(self, local_ref: str) -> Optional[ManifestItem]:
         items = self._query(local_ref)
         return items[0] if items else None
 
-    def open_export(self, local_ref: str) -> Optional[ExportStream]:
-        # Delegate to Kolibri's own export: it stages the channel DB + the channel's
-        # content-addressed files into a staging tree, which we then tar-stream. Read-only on
-        # the source (export copies out). The tree is removed after streaming.
+    def prepare_export(self, local_ref: str) -> Optional[ExportPlan]:
+        # A channel must be staged (the Kolibri export CLI writes the DB + content tree)
+        # before it can tar-stream, so stages=True. The op layer preflights the staging disk,
+        # creates the temp and runs the staging thread; this only validates and reports.
         if not _CHANNEL_ID_RE.match(local_ref or ""):
             return None
         channel = self._query(local_ref)
         if not channel:  # not an installed channel
             return None
-        # Preflight the staging disk against the channel size, so we fail cleanly here
-        # instead of hitting ENOSPC part way through the export (the box owns its own disk
-        # check, mirroring the front's phone-disk preflight).
-        try:
-            os.makedirs(self._staging_dir, exist_ok=True)
-        except OSError:
-            return None
-        if not has_room(self._staging_dir, channel[0].bytes):
-            return None
-        dest = tempfile.mkdtemp(prefix="cxkolibri-", dir=self._staging_dir)
+        return ExportPlan(
+            stages=True,
+            total_bytes=channel[0].bytes,
+            filename=f"{local_ref}.tar",
+            content_type="application/x-tar",
+        )
+
+    def stage_export(self, local_ref: str, dest: str, should_cancel=None) -> bool:
+        # Stage the channel into dest (the op layer's temp). Delegate to Kolibri's own export:
+        # it writes the channel DB + the content-addressed files that import dedups for free.
+        # Read-only on the source (export copies out). ``should_cancel`` lets the op layer stop
+        # 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:
-            staged = self._stage_export(local_ref, dest)
+            return self._stage_export(local_ref, dest, should_cancel)
         except OSError:
-            staged = False
-        if not staged:
-            shutil.rmtree(dest, ignore_errors=True)
+            return False
+
+    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).
+        if not staged_dir or not os.path.isdir(staged_dir):
             return None
         return ExportStream(
             filename=f"{local_ref}.tar",
             content_type="application/x-tar",
-            chunks=tar_stream(dest, arcname=local_ref),
-            on_close=lambda: shutil.rmtree(dest, ignore_errors=True),
+            chunks=tar_stream(staged_dir, arcname=local_ref),
         )
 
-    def _stage_export(self, channel_id: str, dest: str) -> bool:
+    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. Returns True on success."""
+        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."""
         env = dict(os.environ, KOLIBRI_HOME=self._home)
         for argv in export_commands(channel_id, dest, self._bin):
-            result = subprocess.run(
-                argv, env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False
-            )
-            if result.returncode != 0:
+            if not _run_cancellable(argv, env, should_cancel):
                 return False
         return True