From c9e2e4b44fc89004174da36fddc399596ed84912 Mon Sep 17 00:00:00 2001 From: "Luis Guzman (AppDevForAll)" Date: Sat, 24 Oct 2026 01:29:19 -0600 Subject: [PATCH 1/4] K2GO-376 feat(cx-core): add the /v1/share gate reusing the export long-op Token-gated share session over a scoped item set (POST /v1/share, GET manifest, GET fetch per item reusing the export op, DELETE). Closes on the first of five closers; loopback only (network carrier + pairing are later app slices). Core 0.3.0. --- static/cx-core/README.md | 41 +- static/cx-core/cx_core/__init__.py | 2 +- static/cx-core/cx_core/router.py | 33 +- static/cx-core/cx_core/server.py | 17 +- static/cx-core/cx_core/share_ops.py | 357 ++++++++++++++++ static/cx-core/tests/test_share_ops.py | 537 +++++++++++++++++++++++++ 6 files changed, 971 insertions(+), 16 deletions(-) create mode 100644 static/cx-core/cx_core/share_ops.py create mode 100644 static/cx-core/tests/test_share_ops.py diff --git a/static/cx-core/README.md b/static/cx-core/README.md index baa76be4d..b1cfcf4de 100644 --- a/static/cx-core/README.md +++ b/static/cx-core/README.md @@ -8,7 +8,9 @@ contracts (identity, manifest, adapter port, core REST API). 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). +not time out on the client (K2GO-507). Share (K2GO-376) is the network twin: a token-gated +session over a scoped item set that reuses the export op per item (the gate contract only; +pairing and the network carrier are later app slices). ## What is here @@ -16,6 +18,8 @@ not time out on the client (K2GO-507). - `cx_core/ports.py`: the inner contract, the app-adapter port (read methods). - `cx_core/adapters/kiwix.py`: the first adapter, read side (ZIM header UUID + listing). - `cx_core/registry.py`: loads an adapter only when its app is installed. +- `cx_core/export_ops.py`: the export long-operation registry (op lifecycle + temp sweeps). +- `cx_core/share_ops.py`: the share gate (token + scoped item set + the five closers), K2GO-376. - `cx_core/router.py`: pure `/v1` route dispatch (unit-testable without a socket). - `cx_core/server.py` + `__main__.py`: a thin stdlib `http.server` wrapper, localhost only. @@ -68,6 +72,29 @@ Import / delete: - `POST /v1/export/plan`, `POST /v1/dedup/check`, `POST /v1/import/validate`, `POST /v1/import/item`, `DELETE /v1/content/{app}/{localRef}`. +Share (a gate over the export op, K2GO-376): Share is the network twin of Export. The +receiver pulls selected items from the sender's live cx-core, so the transfer can close +itself (the sender machine is on, unlike Clone). A share session is a token plus a scoped +item set plus one shared lifecycle; each item is a lazily created export fetch, so the +staging, the in_use pin and the on_close cleanup are the export op's. The gate closes on the +first of five closers (TTL / idle, startup sweep, DELETE, implicit all-delivered, and the +sender service end, which is app-side). This slice is loopback only: pairing (QR, hotspot, +WLAN) is a later app slice. The session token travels as the `token` query parameter. + +- `POST /v1/share {items: [{app, localRef}]}` -> `{sessionId, token, manifestUrl}`. Opens a + session scoped to the item set. +- `GET /v1/share/{sessionId}/manifest?token=` -> the bundle manifest over the scope + (per-item identity, size, upstream hash when present, each with its `itemRef`), plus + `missing` and `totalBytes`. +- `GET /v1/share/{sessionId}/fetch/{itemRef}?token=` -> the chunked octet stream for + one item. `409 {state: staging}` while a staging item (Kolibri) stages, so the receiver + retries; `200` streams once ready. +- `DELETE /v1/share/{sessionId}?token=` -> `{ok: true}`. Closes the session now and + revokes the token and every child temp. `404` on an unknown or already-closed session. + +Token checks return `401` when no token is sent, `403` on a mismatch, `404` on an unknown or +expired session. + ## Test ``` @@ -77,14 +104,16 @@ 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, route dispatch, the export -long-operation registry (prepare / status / fetch / delete, the TTL and startup sweeps), -and the streaming over a real socket. +long-operation registry (prepare / status / fetch / delete, the TTL and startup sweeps), the +share gate (create / manifest / fetch / delete, token enforcement, the five closers), and the +streaming over a real socket. ## Not built yet -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 Share pairing and network carrier (QR, hotspot, WLAN: the app slices that build on the +`/v1/share` gate above), 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 diff --git a/static/cx-core/cx_core/__init__.py b/static/cx-core/cx_core/__init__.py index fe8a0fa7e..e0213fe9c 100644 --- a/static/cx-core/cx_core/__init__.py +++ b/static/cx-core/cx_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 # cx_core Ansible role writes this into the [cx_core] section of iiab.ini. -VERSION = "0.2.0" +VERSION = "0.3.0" __version__ = VERSION diff --git a/static/cx-core/cx_core/router.py b/static/cx-core/cx_core/router.py index 23a289031..cdcade110 100644 --- a/static/cx-core/cx_core/router.py +++ b/static/cx-core/cx_core/router.py @@ -15,6 +15,7 @@ from .export_ops import ExportOps from .importing import is_within_dir from .registry import Registry +from .share_ops import ShareOps Response = Tuple[int, Any] @@ -165,10 +166,12 @@ 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) -> Response: + ops: Optional[ExportOps] = None, share: Optional[ShareOps] = 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 (the server supplies it, the - export-op routes need it).""" + 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). The + share routes read the session token from ``query["token"]``, 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 != ""] @@ -179,6 +182,12 @@ def dispatch(registry: Registry, method: str, path: str, query: dict, body: Any # POST /v1/export/plan : plan an export (sizes only, no payload moved). if segments == ["v1", "export", "plan"]: return _export_plan(registry, body) + # POST /v1/share : open a share session, return its sessionId + token + manifest pointer. + if segments == ["v1", "share"]: + if share is None: + return 500, {"error": "share unavailable"} + items = body.get("items") if isinstance(body, dict) else None + return share.create(items) # POST /v1/dedup/check : receiver dedup verdicts, nothing moved. if segments == ["v1", "dedup", "check"]: return _dedup_check(registry, body) @@ -196,6 +205,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/share/{sessionId} : close a share session (the explicit "done" closer). + if len(segments) == 3 and segments[0] == "v1" and segments[1] == "share": + if share is None: + return 500, {"error": "share unavailable"} + return share.delete(segments[2], query.get("token")) # DELETE /v1/content/{app}/{localRef} : reclaim space by removing one item. if len(segments) == 4 and segments[0] == "v1" and segments[1] == "content": return _delete_item(registry, segments[2], segments[3]) @@ -225,6 +239,19 @@ def dispatch(registry: Registry, method: str, path: str, query: dict, body: Any return 500, {"error": "export ops unavailable"} return ops.fetch(segments[3]) + # GET /v1/share/{sessionId}/manifest : the bundle manifest over the session's item set. + if len(segments) == 4 and segments[0] == "v1" and segments[1] == "share" and segments[3] == "manifest": + if share is None: + return 500, {"error": "share unavailable"} + return share.manifest(segments[2], query.get("token")) + + # GET /v1/share/{sessionId}/fetch/{itemRef} : stream one shared item (payload is an + # ExportStream the server streams). 409 while the item stages; the receiver retries. + if len(segments) == 5 and segments[0] == "v1" and segments[1] == "share" and segments[3] == "fetch": + if share is None: + return 500, {"error": "share unavailable"} + return share.fetch(segments[2], segments[4], query.get("token")) + # GET /v1/storage : box content-volume disk summary + per-app content totals. if segments == ["v1", "storage"]: return _storage(registry) diff --git a/static/cx-core/cx_core/server.py b/static/cx-core/cx_core/server.py index 704db9891..77bb1b9ac 100644 --- a/static/cx-core/cx_core/server.py +++ b/static/cx-core/cx_core/server.py @@ -15,13 +15,14 @@ from .ports import ExportStream from .registry import Registry from .router import dispatch +from .share_ops import ShareOps _MAX_BODY = 1 * 1024 * 1024 # a plan request is small; cap the read defensively _BAD_JSON = object() # sentinel: body present but not valid JSON _TOO_LARGE = object() # sentinel: declared body exceeds _MAX_BODY -def make_handler(registry: Registry, ops: ExportOps): +def make_handler(registry: Registry, ops: ExportOps, share: ShareOps = None): class Handler(BaseHTTPRequestHandler): server_version = f"cx-core/{VERSION}" # HTTP/1.1 so the chunked export stream is valid (HTTP/1.0 would hand the client the @@ -31,7 +32,7 @@ class Handler(BaseHTTPRequestHandler): def do_GET(self) -> None: # noqa: N802 (http.server naming) parts = urlsplit(self.path) query = {k: v[0] for k, v in parse_qs(parts.query).items()} - status, payload = dispatch(registry, "GET", parts.path, query, ops=ops) + status, payload = dispatch(registry, "GET", parts.path, query, ops=ops, share=share) if isinstance(payload, ExportStream): self._stream(status, payload) else: @@ -48,13 +49,13 @@ def do_POST(self) -> None: # noqa: N802 (http.server naming) if body is _BAD_JSON: self._send(400, {"error": "invalid JSON body"}) return - status, payload = dispatch(registry, "POST", parts.path, {}, body, ops=ops) + status, payload = dispatch(registry, "POST", parts.path, {}, body, ops=ops, share=share) self._send(status, payload) def do_DELETE(self) -> None: # noqa: N802 (http.server naming) parts = urlsplit(self.path) query = {k: v[0] for k, v in parse_qs(parts.query).items()} - status, payload = dispatch(registry, "DELETE", parts.path, query, ops=ops) + status, payload = dispatch(registry, "DELETE", parts.path, query, ops=ops, share=share) self._send(status, payload) def _read_json_body(self): @@ -118,10 +119,14 @@ def log_message(self, *_args) -> None: def serve(config: Config) -> None: registry = Registry.load(config) ops = ExportOps(registry, config.staging_dir) + 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. + # 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. ops.sweep_startup() - httpd = ThreadingHTTPServer((config.host, config.port), make_handler(registry, ops)) + share.sweep_startup() + httpd = ThreadingHTTPServer((config.host, config.port), make_handler(registry, ops, share)) print(f"cx-core {VERSION} listening on http://{config.host}:{config.port} " f"(apps: {', '.join(registry.apps()) or 'none'})") httpd.serve_forever() diff --git a/static/cx-core/cx_core/share_ops.py b/static/cx-core/cx_core/share_ops.py new file mode 100644 index 000000000..0b293c604 --- /dev/null +++ b/static/cx-core/cx_core/share_ops.py @@ -0,0 +1,357 @@ +"""Share gate: the shared contract for granular wireless Share (K2GO-376). + +Share is the NETWORK TWIN of Export. The bundle, the manifest and the import path are the +same; only the carrier changes. The receiver PULLS from the sender's live cx-core (the +machine is ON), so the transfer can close itself: this is the whole point of Share over +Clone, whose standalone rsync daemon is mute on the host side and can only be stopped by +the user. See controller/docs/ADR-content-exchange.md, section "Pillar Share". + +A share session is a TOKEN plus a SCOPED set of items plus ONE shared lifecycle. It does not +reinvent the transfer: each shared item is a lazily created export op (export_ops.py), so the +staging, the in_use pin and the on_close cleanup are the ones already hardened and device +validated for K2GO-507. The session is the GATE over those child ops: it holds the token and +the scope, it ties the child ops together, and it can revoke them all at once. + +Ownership (one source per fact, so this is not a second owner of anything the export op +owns): + : the session owns the token, the scoped item set and the session liveness. + : the child export op owns the staged temp, the stream and the per-item liveness. +Both track an "is a stream open" pin, but at different scopes and for different cleanup: the +session pin keeps the gate open for a long single-item transfer; the export op pin keeps that +item's temp from being swept. They are nested scopes, not a duplicated truth. + +The gate closes on the FIRST of five independent closers (ADR "Gate / token lifecycle"): + 1. TTL + idle : last_seen bumps on each manifest / fetch, so an active transfer stays + alive and an abandoned one expires. This is the jaw Clone lacks. + 2. startup sweep: a fresh process holds no sessions (they live only in memory); the child + temps are cxexport-* reclaimed by the export op's own startup sweep. + 3. DELETE : the explicit "done" closer (the receiver finished, or tapped Finish). + 4. implicit : when every scoped item has been delivered, the gate auto-closes. + 5. sender end : APP-SIDE. The sender's foreground share service calls DELETE in its + onDestroy to revoke the gate. The core exposes DELETE; it cannot observe + the Android service lifecycle, so this closer is documented here, not + implemented in the core. + +Default = closed: staying open requires active fetches. The token is bound to the session +(revoking the session revokes the token) and scoped to the item set, so there is no separate +credential lifecycle to leak. The only surface a token holder reaches is /v1/share/* scoped +to this session's items; the wider /v1 API is never widened by a share session. + +This slice is loopback only: pairing (QR, hotspot, WLAN) is a later presentation slice and is +outside the core, the way plugging in the Ethernet cable is not part of content exchange. +""" + +from __future__ import annotations + +import secrets +import threading +import time +from dataclasses import dataclass, field +from typing import Any, Dict, List, Tuple, Union + +from .export_ops import DEFAULT_TTL_SECONDS, ExportOps +from .registry import Registry + +Response = Tuple[int, Any] + + +@dataclass +class _Session: + token: str + # The scoped item set, as (app, local_ref) pairs. The index is the item's itemRef in the + # manifest and the fetch URL: an opaque per-session handle, so a Calibre local_ref that + # contains slashes never lands in a URL path segment. + items: List[Tuple[str, str]] + # itemRef -> the child export opId, created lazily on first fetch of that item. + item_ops: Dict[str, str] = field(default_factory=dict) + # itemRefs whose stream has closed, for the implicit auto-close (closer 4). + delivered: set = field(default_factory=set) + last_seen: float = field(default_factory=time.monotonic) + # Count of open item streams. The TTL sweep never reclaims a session with a stream open, + # so a single multi-gigabyte item that streams longer than the TTL (and bumps nothing + # meanwhile) does not expire the gate under its own transfer. + in_use: int = 0 + + +class ShareOps: + """Thread-safe registry of share sessions. One instance lives for the server's lifetime; + the handler threads call into it concurrently. It wraps the one ExportOps instance and + delegates every per-item transfer to it.""" + + def __init__(self, registry: Registry, export_ops: ExportOps, + ttl_seconds: int = DEFAULT_TTL_SECONDS): + self._registry = registry + self._export = export_ops + self._ttl = ttl_seconds + self._lock = threading.Lock() + self._sessions: Dict[str, _Session] = {} + + # ---- REST-facing operations (return (status, payload), or (200, ExportStream)) ---- + + def create(self, items: Any) -> Response: + """Open a share session over a scoped item set. Returns the sessionId, an opaque + token bound to the session, and the manifest pointer. Partial is first-class: the set + is recorded as requested and each item is resolved at manifest / fetch time, so an + item that is absent is reported, not fatal.""" + if not isinstance(items, list) or not items: + return 400, {"error": "body must be {items: [{app, localRef}]} with at least one item"} + scoped: List[Tuple[str, str]] = [] + seen = set() + for req in items: + app = req.get("app") if isinstance(req, dict) else None + local_ref = req.get("localRef") if isinstance(req, dict) else None + if not isinstance(app, str) or not isinstance(local_ref, str) or not local_ref: + return 400, {"error": "each item needs a string app and a non-empty localRef"} + key = (app, local_ref) + if key not in seen: + seen.add(key) + scoped.append(key) + session_id = secrets.token_hex(16) + token = secrets.token_urlsafe(32) + with self._lock: + self._sessions[session_id] = _Session(token=token, items=scoped) + # The pointer carries no token: the QR carries the address, the token and this pointer + # as separate fields (ADR), and the receiver combines them. + return 200, { + "sessionId": session_id, + "token": token, + "manifestUrl": f"/v1/share/{session_id}/manifest", + } + + def manifest(self, session_id: str, token: Any) -> Response: + """The bundle manifest over the session's scoped set: per-item identity, size and + upstream hash when present, each tagged with its itemRef. Reuses the read-side item + JSON, so it is the same manifest shape the disk export builds.""" + session = self._resolve(session_id, token) + if isinstance(session, tuple): + return session + found: List[dict] = [] + missing: List[dict] = [] + total = 0 + for index, (app, local_ref) in enumerate(session.items): + item_ref = str(index) + if not self._registry.has(app): + missing.append({"itemRef": item_ref, "app": app, "localRef": local_ref, + "reason": "unknown app"}) + continue + item = self._registry.get(app).get(local_ref) + if item is None: + missing.append({"itemRef": item_ref, "app": app, "localRef": local_ref, + "reason": "not found"}) + continue + entry = item.to_json() + entry["itemRef"] = item_ref + found.append(entry) + total += item.bytes + return 200, {"sessionId": session_id, "items": found, "missing": missing, + "totalBytes": total} + + def fetch(self, session_id: str, item_ref: str, token: Any): + """Stream one shared item. Lazily creates the child export op the first time an item is + fetched, then drives it: 409 {state: staging} while a staging app (Kolibri) stages, so + the receiver retries this same URL until it streams; (200, ExportStream) once ready. + The stream is wrapped so the session pin and the delivered set track with the export + op's own pin and temp cleanup.""" + session = self._resolve(session_id, token) + if isinstance(session, tuple): + return session + target = self._item(session, item_ref) + if target is None: + return 404, {"error": "unknown item"} + app, local_ref = target + # A child op that was swept or already finished returns 404; re-prepare and retry once, + # so a resumed fetch re-stages instead of dead-ending on a stale opId. + for attempt in (1, 2): + op_id = self._ensure_op(session_id, item_ref, app, local_ref) + if isinstance(op_id, tuple): + return op_id # prepare failed: 404 absent, 507 no space + status, result = self._export.fetch(op_id) + if status == 404 and attempt == 1: + self._drop_op(session_id, item_ref, op_id) + continue + if status == 409: + return self._staging_response(op_id, item_ref) + if status != 200: + return status, result + self._wrap_stream(session_id, item_ref, result) + return 200, result + return 404, {"error": "item not available"} + + def _staging_response(self, op_id: str, item_ref: str) -> Response: + """The item is not ready: report it through the one fetch endpoint, since the gate has + no separate status route (do not widen the API). 500 on a staging error so the receiver + stops retrying; 409 with the staging heartbeat otherwise so the receiver retries and can + show progress.""" + st_status, st = self._export.status(op_id) + if st_status == 200 and st.get("state") == "error": + return 500, {"state": "error", "itemRef": item_ref, "error": st.get("error")} + progress = {"state": "staging", "itemRef": item_ref} + if st_status == 200: + progress["stagedBytes"] = st.get("stagedBytes", 0) + progress["totalBytes"] = st.get("totalBytes", 0) + return 409, progress + + def delete(self, session_id: str, token: Any) -> Response: + """Close the session now (closer 3, the explicit "done"; also the app-side closer 5). + Revokes the token with the session and drops every child op + its temp. 404 on an + unknown or already-closed session, so a double close (receiver Finish plus sender + onDestroy) is harmless: the second caller sees the gate already closed.""" + self._sweep_expired() + with self._lock: + session = self._sessions.get(session_id) + if session is None: + return 404, {"error": "unknown session"} + denied = self._check_token(session, token) + if denied is not None: + return denied + self._sessions.pop(session_id, None) + child_ops = list(session.item_ops.values()) + for op_id in child_ops: + self._export.delete(op_id) # drops each child op + removes its staged temp + return 200, {"ok": True} + + # ---- lifecycle ---- + + def sweep_startup(self) -> None: + """Drop any in-memory sessions on boot (closer 2). A fresh process holds none, so this + is defensive; the on-disk footprint of a session is its child export temps (cxexport-*), + reclaimed by the export op's own startup sweep, which the server runs on boot. Reusing + that single sweep avoids a second temp owner.""" + with self._lock: + self._sessions.clear() + + # ---- token + item resolution ---- + + def _resolve(self, session_id: str, token: Any) -> Union[_Session, Response]: + """Return the live session, bumping last_seen (closer 1), or an error response: 404 on + an unknown or expired session, 401 when no token is sent, 403 on a token mismatch.""" + self._sweep_expired() + with self._lock: + session = self._sessions.get(session_id) + if session is None: + return 404, {"error": "unknown session"} + denied = self._check_token(session, token) + if denied is not None: + return denied + session.last_seen = time.monotonic() + return session + + @staticmethod + def _check_token(session: _Session, token: Any): + """None when the token is valid, else the error response. compare_digest is + constant-time, so a wrong token leaks no timing signal.""" + if not isinstance(token, str) or token == "": + return 401, {"error": "token required"} + if not secrets.compare_digest(token, session.token): + return 403, {"error": "token mismatch"} + return None + + @staticmethod + def _item(session: _Session, item_ref: str): + """Resolve an itemRef (its index in the scoped set) to (app, local_ref), or None.""" + try: + index = int(item_ref) + except (TypeError, ValueError): + return None + if 0 <= index < len(session.items): + return session.items[index] + return None + + def _ensure_op(self, session_id: str, item_ref: str, app: str, local_ref: str): + """Return the child export opId for this item, preparing one if none exists. Returns an + error response tuple when prepare fails (absent item or no staging space).""" + with self._lock: + session = self._sessions.get(session_id) + if session is None: + return 404, {"error": "unknown session"} + op_id = session.item_ops.get(item_ref) + if op_id is not None: + return op_id + status, payload = self._export.prepare(app, local_ref) + if status != 200: + return status, payload + op_id = payload["opId"] + with self._lock: + session = self._sessions.get(session_id) + if session is None: + # The session closed while we prepared: drop the orphan op we just made. + self._export.delete(op_id) + return 404, {"error": "unknown session"} + # Another concurrent fetch may have prepared first; keep the first and drop ours. + existing = session.item_ops.get(item_ref) + if existing is not None: + self._export.delete(op_id) + return existing + session.item_ops[item_ref] = op_id + return op_id + + def _drop_op(self, session_id: str, item_ref: str, op_id: str) -> None: + """Forget a child op mapping whose op has vanished, so the next fetch re-prepares.""" + with self._lock: + session = self._sessions.get(session_id) + if session is not None and session.item_ops.get(item_ref) == op_id: + session.item_ops.pop(item_ref, None) + + def _wrap_stream(self, session_id: str, item_ref: str, stream) -> None: + """Layer the session pin + delivered bookkeeping onto the export op's own on_open / + on_close, so the two lifecycles stay a matched pair. The server fires on_open at the + top of streaming and on_close in the same try's finally, so both pins are always + released, even on a client disconnect.""" + inner_open = stream.on_open + inner_close = stream.on_close + + def on_open() -> None: + self._pin(session_id) + if inner_open: + inner_open() + + def on_close() -> None: + # The export op's close runs first (drops the child op and removes its temp), then + # the session bookkeeping records the item delivered and may auto-close the gate. + if inner_close: + inner_close() + self._unpin(session_id, item_ref) + + stream.on_open = on_open + stream.on_close = on_close + + def _pin(self, session_id: str) -> None: + with self._lock: + session = self._sessions.get(session_id) + if session is not None: + session.in_use += 1 + + def _unpin(self, session_id: str, item_ref: str) -> None: + """Release one stream pin, record the item delivered, and auto-close the gate once + every scoped item is delivered and no stream is open (closer 4). A mid-stream drop also + marks the item delivered, so the implicit close is a convenience; DELETE (closer 3) and + the TTL (closer 1) are the robust closers, and a receiver that still needs an item + re-pairs (the gate is cheap to re-create).""" + child_ops: List[str] = [] + with self._lock: + session = self._sessions.get(session_id) + if session is None: + return + if session.in_use > 0: + session.in_use -= 1 + session.item_ops.pop(item_ref, None) # the child op finished; a re-fetch re-prepares + session.delivered.add(item_ref) + session.last_seen = time.monotonic() + all_refs = {str(i) for i in range(len(session.items))} + if session.in_use == 0 and session.delivered.issuperset(all_refs): + self._sessions.pop(session_id, None) + child_ops = list(session.item_ops.values()) + for op_id in child_ops: + self._export.delete(op_id) + + def _sweep_expired(self) -> None: + now = time.monotonic() + expired: List[_Session] = [] + with self._lock: + for session_id, session in list(self._sessions.items()): + # Never reclaim a session with an open stream: its transfer is live. + if session.in_use == 0 and now - session.last_seen > self._ttl: + expired.append(self._sessions.pop(session_id)) + for session in expired: + for op_id in list(session.item_ops.values()): + self._export.delete(op_id) diff --git a/static/cx-core/tests/test_share_ops.py b/static/cx-core/tests/test_share_ops.py new file mode 100644 index 000000000..6985869ca --- /dev/null +++ b/static/cx-core/tests/test_share_ops.py @@ -0,0 +1,537 @@ +"""Share gate tests (K2GO-376): the shared contract for granular wireless Share. + +Everything runs on loopback, like the export op tests: a share session wraps the export op, +so these cover the gate (token, scope, the five closers) and the per-item transfer reuse. The +app forks (sender / receiver) build against this contract. +""" + +import http.client +import json +import os +import tempfile +import threading +import time +import unittest +from http.server import ThreadingHTTPServer + +from cx_core.export_ops import ExportOps +from cx_core.model import AUTHORITATIVE, Dedup, Identity, ManifestItem +from cx_core.ports import ContentAdapter, ExportPlan, ExportStream +from cx_core.registry import Registry +from cx_core.router import dispatch +from cx_core.server import make_handler +from cx_core.share_ops import ShareOps + + +def _mitem(app, local_ref, total): + return ManifestItem( + app=app, kind="zim", title=local_ref, bytes=total, + identity=Identity(local_ref, Dedup(line={}, build=None, version="", id_kind=AUTHORITATIVE))) + + +class _DirectFake(ContentAdapter): + """A direct app (Kiwix / Calibre): nothing to stage, ready at once. ``items`` maps a + local_ref to its byte size.""" + + app = "kiwix" + + def __init__(self, items): + self._items = dict(items) + + def is_installed(self): + return True + + def list(self): + return [_mitem(self.app, ref, total) for ref, total in self._items.items()] + + def prepare_export(self, local_ref): + if local_ref not in self._items: + return None + return ExportPlan(False, self._items[local_ref], local_ref, "application/octet-stream") + + def open_fetch(self, local_ref, staged_dir=None): + if local_ref not in self._items: + return None + return ExportStream(local_ref, "application/octet-stream", [b"x" * self._items[local_ref]]) + + +class _StagingFake(ContentAdapter): + """A staging app (Kolibri): stage_export writes into dest, optionally gated so a test can + observe the staging state before the op flips to ready.""" + + app = "kolibri" + + def __init__(self, items, gate=None, fail=False): + self._items = dict(items) + self._gate = gate + self._fail = fail + + def is_installed(self): + return True + + def list(self): + return [_mitem(self.app, ref, total) for ref, total in self._items.items()] + + def prepare_export(self, local_ref): + if local_ref not in self._items: + return None + return ExportPlan(True, self._items[local_ref], f"{local_ref}.tar", "application/x-tar") + + def stage_export(self, local_ref, dest, should_cancel=None): + if self._gate is not None: + self._gate.wait(5) + if self._fail: + return False + with open(os.path.join(dest, "part.bin"), "wb") as f: + f.write(b"x" * self._items[local_ref]) + return True + + def open_fetch(self, local_ref, staged_dir=None): + path = os.path.join(staged_dir or "", "part.bin") + data = b"" + if os.path.isfile(path): + with open(path, "rb") as f: + data = f.read() + return ExportStream(f"{local_ref}.tar", "application/x-tar", [data]) + + +def _share(adapters, staging, ttl=60): + reg = Registry(adapters, import_root=staging, content_volume=staging) + ops = ExportOps(reg, staging_dir=staging, ttl_seconds=ttl) + share = ShareOps(reg, ops, ttl_seconds=ttl) + return share, ops, reg + + +def _create(share, items): + status, body = share.create(items) + assert status == 200, (status, body) + return body["sessionId"], body["token"] + + +def _drain(stream): + """Drive a stream as the server would: on_open, consume the chunks, on_close.""" + if stream.on_open: + stream.on_open() + try: + return b"".join(stream.chunks) + finally: + if stream.on_close: + stream.on_close() + + +def _fetch_ready(share, sid, item_ref, token, timeout=5): + end = time.time() + timeout + while time.time() < end: + status, result = share.fetch(sid, item_ref, token) + if status == 200: + return result + if status != 409: + raise AssertionError(f"unexpected fetch status {status}: {result}") + time.sleep(0.01) + raise AssertionError("item never became ready") + + +class CreateTest(unittest.TestCase): + def test_create_returns_session_token_and_pointer(self): + with tempfile.TemporaryDirectory() as d: + share, _ops, _reg = _share([_DirectFake({"a.zim": 10})], d) + status, body = share.create([{"app": "kiwix", "localRef": "a.zim"}]) + self.assertEqual(status, 200) + self.assertIn("sessionId", body) + self.assertTrue(body["token"]) + self.assertEqual(body["manifestUrl"], f"/v1/share/{body['sessionId']}/manifest") + # The pointer carries no token: the token travels as a separate field. + self.assertNotIn(body["token"], body["manifestUrl"]) + + def test_distinct_sessions_get_distinct_tokens(self): + with tempfile.TemporaryDirectory() as d: + share, _ops, _reg = _share([_DirectFake({"a.zim": 10})], d) + a = share.create([{"app": "kiwix", "localRef": "a.zim"}])[1] + b = share.create([{"app": "kiwix", "localRef": "a.zim"}])[1] + self.assertNotEqual(a["sessionId"], b["sessionId"]) + self.assertNotEqual(a["token"], b["token"]) + + def test_empty_or_bad_body_400(self): + with tempfile.TemporaryDirectory() as d: + share, _ops, _reg = _share([_DirectFake({"a.zim": 10})], d) + self.assertEqual(share.create([])[0], 400) + self.assertEqual(share.create("nope")[0], 400) + self.assertEqual(share.create([{"app": "kiwix"}])[0], 400) # no localRef + self.assertEqual(share.create([{"localRef": "a.zim"}])[0], 400) # no app + self.assertEqual(share.create([{"app": "kiwix", "localRef": ""}])[0], 400) + + def test_duplicate_items_are_deduped(self): + with tempfile.TemporaryDirectory() as d: + share, _ops, _reg = _share([_DirectFake({"a.zim": 10})], d) + sid, token = _create(share, [ + {"app": "kiwix", "localRef": "a.zim"}, + {"app": "kiwix", "localRef": "a.zim"}, + ]) + _, body = share.manifest(sid, token) + self.assertEqual(len(body["items"]), 1) + + +class ManifestTest(unittest.TestCase): + def test_manifest_lists_scoped_items_with_itemref_and_total(self): + with tempfile.TemporaryDirectory() as d: + share, _ops, _reg = _share([_DirectFake({"a.zim": 10, "b.zim": 25})], d) + sid, token = _create(share, [ + {"app": "kiwix", "localRef": "a.zim"}, + {"app": "kiwix", "localRef": "b.zim"}, + ]) + status, body = share.manifest(sid, token) + self.assertEqual(status, 200) + self.assertEqual([i["itemRef"] for i in body["items"]], ["0", "1"]) + self.assertEqual([i["identity"]["localRef"] for i in body["items"]], ["a.zim", "b.zim"]) + self.assertEqual(body["totalBytes"], 35) + self.assertEqual(body["missing"], []) + + def test_manifest_reports_missing_items_partial(self): + with tempfile.TemporaryDirectory() as d: + share, _ops, _reg = _share([_DirectFake({"a.zim": 10})], d) + sid, token = _create(share, [ + {"app": "kiwix", "localRef": "a.zim"}, + {"app": "kiwix", "localRef": "gone.zim"}, + {"app": "nope", "localRef": "x"}, + ]) + _, body = share.manifest(sid, token) + self.assertEqual([i["identity"]["localRef"] for i in body["items"]], ["a.zim"]) + reasons = {m["reason"] for m in body["missing"]} + self.assertEqual(reasons, {"not found", "unknown app"}) + self.assertEqual(body["totalBytes"], 10) + + def test_manifest_token_enforcement(self): + with tempfile.TemporaryDirectory() as d: + share, _ops, _reg = _share([_DirectFake({"a.zim": 10})], d) + sid, token = _create(share, [{"app": "kiwix", "localRef": "a.zim"}]) + self.assertEqual(share.manifest(sid, None)[0], 401) # no token + self.assertEqual(share.manifest(sid, "")[0], 401) # empty token + self.assertEqual(share.manifest(sid, "wrong")[0], 403) # mismatch + self.assertEqual(share.manifest("deadbeef", token)[0], 404) # unknown session + self.assertEqual(share.manifest(sid, token)[0], 200) # correct + + +class FetchTest(unittest.TestCase): + def test_fetch_direct_item_streams_bytes(self): + with tempfile.TemporaryDirectory() as d: + share, _ops, _reg = _share([_DirectFake({"a.zim": 10, "b.zim": 7})], d) + sid, token = _create(share, [ + {"app": "kiwix", "localRef": "a.zim"}, + {"app": "kiwix", "localRef": "b.zim"}, + ]) + status, stream = share.fetch(sid, "0", token) + self.assertEqual(status, 200) + self.assertEqual(len(_drain(stream)), 10) + + def test_fetch_token_enforcement(self): + with tempfile.TemporaryDirectory() as d: + share, _ops, _reg = _share([_DirectFake({"a.zim": 10})], d) + sid, token = _create(share, [{"app": "kiwix", "localRef": "a.zim"}]) + self.assertEqual(share.fetch(sid, "0", None)[0], 401) + self.assertEqual(share.fetch(sid, "0", "wrong")[0], 403) + self.assertEqual(share.fetch("deadbeef", "0", token)[0], 404) + + def test_fetch_unknown_itemref_404(self): + with tempfile.TemporaryDirectory() as d: + share, _ops, _reg = _share([_DirectFake({"a.zim": 10})], d) + sid, token = _create(share, [{"app": "kiwix", "localRef": "a.zim"}]) + self.assertEqual(share.fetch(sid, "9", token)[0], 404) # out of range + self.assertEqual(share.fetch(sid, "x", token)[0], 404) # not an index + + def test_fetch_absent_item_404(self): + # The item was in scope at create but is not exportable now: prepare returns None. + with tempfile.TemporaryDirectory() as d: + share, _ops, _reg = _share([_DirectFake({"a.zim": 10})], d) + sid, token = _create(share, [{"app": "kiwix", "localRef": "gone.zim"}]) + self.assertEqual(share.fetch(sid, "0", token)[0], 404) + + def test_fetch_staging_item_409_then_streams(self): + with tempfile.TemporaryDirectory() as d: + gate = threading.Event() # hold staging open + share, _ops, _reg = _share([_StagingFake({"chan": 120}, gate=gate)], d) + sid, token = _create(share, [{"app": "kolibri", "localRef": "chan"}]) + status, body = share.fetch(sid, "0", token) + self.assertEqual(status, 409) + self.assertEqual(body["state"], "staging") + gate.set() + stream = _fetch_ready(share, sid, "0", token) + self.assertEqual(len(_drain(stream)), 120) + + def test_fetch_staging_error_surfaces_500(self): + with tempfile.TemporaryDirectory() as d: + gate = threading.Event() + gate.set() # let staging run, but it fails + share, _ops, _reg = _share([_StagingFake({"chan": 50}, gate=gate, fail=True)], d) + sid, token = _create(share, [{"app": "kolibri", "localRef": "chan"}]) + end = time.time() + 5 + seen = None + while time.time() < end: + status, body = share.fetch(sid, "0", token) + if status == 500: + seen = body + break + time.sleep(0.01) + self.assertIsNotNone(seen, "staging error never surfaced") + self.assertEqual(seen["state"], "error") + + +class AutoCloseTest(unittest.TestCase): + def test_delivering_every_item_auto_closes_the_gate(self): + # Closer 4 (implicit progress): once every scoped item has been delivered, the gate + # closes itself, so a later manifest is 404. + with tempfile.TemporaryDirectory() as d: + share, _ops, _reg = _share([_DirectFake({"a.zim": 10, "b.zim": 7})], d) + sid, token = _create(share, [ + {"app": "kiwix", "localRef": "a.zim"}, + {"app": "kiwix", "localRef": "b.zim"}, + ]) + _drain(share.fetch(sid, "0", token)[1]) + self.assertEqual(share.manifest(sid, token)[0], 200) # one item left, still open + _drain(share.fetch(sid, "1", token)[1]) + self.assertEqual(share.manifest(sid, token)[0], 404) # all delivered: auto-closed + + +class DeleteTest(unittest.TestCase): + def test_delete_closes_and_revokes_child_temp(self): + # Closer 3 (explicit done). Fetch one of two items so the gate does not auto-close, + # leave a staged child op, then DELETE and confirm the gate and the temp are gone. + with tempfile.TemporaryDirectory() as d: + gate = threading.Event() + gate.set() + share, ops, _reg = _share([_StagingFake({"c0": 20, "c1": 30}, gate=gate)], d) + sid, token = _create(share, [ + {"app": "kolibri", "localRef": "c0"}, + {"app": "kolibri", "localRef": "c1"}, + ]) + # Create a ready-but-unconsumed child op for item 1 (so a temp exists on disk). + share.fetch(sid, "1", token) # starts staging + op_id = share._sessions[sid].item_ops["1"] + end = time.time() + 5 + while ops._ops[op_id].state != "ready" and time.time() < end: + time.sleep(0.01) + temp = ops._ops[op_id].temp + self.assertTrue(os.path.isdir(temp)) + status, body = share.delete(sid, token) + self.assertEqual(status, 200) + self.assertTrue(body["ok"]) + self.assertNotIn(sid, share._sessions) + self.assertFalse(os.path.exists(temp)) # child temp revoked + self.assertEqual(share.manifest(sid, token)[0], 404) + + def test_delete_token_enforcement(self): + with tempfile.TemporaryDirectory() as d: + share, _ops, _reg = _share([_DirectFake({"a.zim": 10})], d) + sid, token = _create(share, [{"app": "kiwix", "localRef": "a.zim"}]) + self.assertEqual(share.delete(sid, None)[0], 401) + self.assertEqual(share.delete(sid, "wrong")[0], 403) + self.assertIn(sid, share._sessions) # still open after the rejected closes + self.assertEqual(share.delete(sid, token)[0], 200) + + def test_double_delete_is_404_already_closed(self): + # The receiver Finish and the sender onDestroy may both DELETE; the second sees 404. + with tempfile.TemporaryDirectory() as d: + share, _ops, _reg = _share([_DirectFake({"a.zim": 10})], d) + sid, token = _create(share, [{"app": "kiwix", "localRef": "a.zim"}]) + self.assertEqual(share.delete(sid, token)[0], 200) + self.assertEqual(share.delete(sid, token)[0], 404) + + +class SweepTest(unittest.TestCase): + def test_ttl_sweep_reclaims_abandoned_session_and_temp(self): + # Closer 1 (TTL + idle): an abandoned session expires and takes its child temp with it. + with tempfile.TemporaryDirectory() as d: + gate = threading.Event() + gate.set() + share, ops, _reg = _share([_StagingFake({"chan": 40}, gate=gate)], d, ttl=60) + sid, token = _create(share, [{"app": "kolibri", "localRef": "chan"}]) + share.fetch(sid, "0", token) # starts staging, leaves a child op + temp + op_id = share._sessions[sid].item_ops["0"] + end = time.time() + 5 + while ops._ops[op_id].state != "ready" and time.time() < end: + time.sleep(0.01) + temp = ops._ops[op_id].temp + share._sessions[sid].last_seen -= 120 # older than the TTL + share._sweep_expired() + self.assertNotIn(sid, share._sessions) + self.assertFalse(os.path.exists(temp)) + + def test_ttl_sweep_never_reclaims_a_session_with_an_open_stream(self): + # A single large item can stream longer than the TTL without a new request to bump + # last_seen; the gate must survive under its own transfer (in_use). + with tempfile.TemporaryDirectory() as d: + share, _ops, _reg = _share([_DirectFake({"a.zim": 10, "b.zim": 7})], d, ttl=60) + sid, token = _create(share, [ + {"app": "kiwix", "localRef": "a.zim"}, + {"app": "kiwix", "localRef": "b.zim"}, + ]) + status, stream = share.fetch(sid, "0", token) + self.assertEqual(status, 200) + stream.on_open() # the server pins at top of streaming + share._sessions[sid].last_seen -= 120 + share._sweep_expired() + self.assertIn(sid, share._sessions) # pinned: survives the sweep + b"".join(stream.chunks) + stream.on_close() # unpins; item 1 undelivered, stays open + self.assertIn(sid, share._sessions) + self.assertEqual(share._sessions[sid].in_use, 0) + + def test_startup_sweep_clears_in_memory_sessions(self): + with tempfile.TemporaryDirectory() as d: + share, _ops, _reg = _share([_DirectFake({"a.zim": 10})], d) + sid, _token = _create(share, [{"app": "kiwix", "localRef": "a.zim"}]) + self.assertIn(sid, share._sessions) + share.sweep_startup() + self.assertEqual(share._sessions, {}) + + def test_boot_reclaims_a_leftover_share_session_and_its_temp(self): + # Closer 2 (process death): sessions live only in memory, so a fresh process holds + # none; the session's on-disk footprint is its child export temps, reclaimed by the + # export op's startup sweep. This mirrors exactly what serve() runs on boot. + with tempfile.TemporaryDirectory() as d: + gate = threading.Event() + gate.set() + share, ops, _reg = _share([_StagingFake({"chan": 40}, gate=gate)], d, ttl=60) + sid, token = _create(share, [{"app": "kolibri", "localRef": "chan"}]) + share.fetch(sid, "0", token) # leaves a staged cxexport-* temp on disk + op_id = share._sessions[sid].item_ops["0"] + end = time.time() + 5 + while ops._ops[op_id].state != "ready" and time.time() < end: + time.sleep(0.01) + temp = ops._ops[op_id].temp + self.assertTrue(os.path.isdir(temp)) + # Simulate a restart: a new process, new registries on the same staging dir. + new_share, new_ops, _ = _share([_StagingFake({"chan": 40})], d, ttl=60) + new_ops.sweep_startup() # reclaims the orphan temp + new_share.sweep_startup() # holds no sessions + self.assertFalse(os.path.exists(temp)) + self.assertEqual(new_share._sessions, {}) + + +class ShareRouteTest(unittest.TestCase): + """The share routes forward to the ShareOps registry; the gate mechanics live above. Here + we check the routing, the token-via-query wiring, and the missing-share guard.""" + + def setUp(self): + self._d = tempfile.TemporaryDirectory() + self.share, self.ops, self.reg = _share([_DirectFake({"a.zim": 10, "b.zim": 7})], self._d.name) + + def tearDown(self): + self._d.cleanup() + + def test_post_creates_and_get_manifest_and_fetch_and_delete(self): + status, body = dispatch(self.reg, "POST", "/v1/share", {}, + {"items": [{"app": "kiwix", "localRef": "a.zim"}, + {"app": "kiwix", "localRef": "b.zim"}]}, + ops=self.ops, share=self.share) + self.assertEqual(status, 200) + sid, token = body["sessionId"], body["token"] + + status, man = dispatch(self.reg, "GET", f"/v1/share/{sid}/manifest", + {"token": token}, ops=self.ops, share=self.share) + self.assertEqual(status, 200) + self.assertEqual(len(man["items"]), 2) + + status, stream = dispatch(self.reg, "GET", f"/v1/share/{sid}/fetch/0", + {"token": token}, ops=self.ops, share=self.share) + self.assertEqual(status, 200) + self.assertIsInstance(stream, ExportStream) + self.assertEqual(len(_drain(stream)), 10) + + status, out = dispatch(self.reg, "DELETE", f"/v1/share/{sid}", + {"token": token}, ops=self.ops, share=self.share) + self.assertEqual(status, 200) + self.assertTrue(out["ok"]) + + def test_manifest_without_token_401(self): + sid, _token = _create(self.share, [{"app": "kiwix", "localRef": "a.zim"}]) + status, _ = dispatch(self.reg, "GET", f"/v1/share/{sid}/manifest", {}, + ops=self.ops, share=self.share) + self.assertEqual(status, 401) + + def test_share_routes_without_share_500(self): + for method, path, body in [ + ("POST", "/v1/share", {"items": [{"app": "kiwix", "localRef": "a.zim"}]}), + ("GET", "/v1/share/abc/manifest", None), + ("GET", "/v1/share/abc/fetch/0", None), + ("DELETE", "/v1/share/abc", None), + ]: + status, _ = dispatch(self.reg, method, path, {}, body, ops=self.ops, share=None) + self.assertEqual(status, 500, path) + + +class ShareHttpTest(unittest.TestCase): + """End-to-end over a real socket: POST create, GET manifest, GET fetch (chunked stream + decoded by a real client), DELETE, all with the token as a query parameter.""" + + def setUp(self): + self._d = tempfile.TemporaryDirectory() + self.payload = b"x" * 4096 + share, ops, reg = _share([_DirectFake({"a.zim": len(self.payload), "b.zim": 7})], self._d.name) + self.httpd = ThreadingHTTPServer(("127.0.0.1", 0), make_handler(reg, ops, share)) + self.port = self.httpd.server_address[1] + self._t = threading.Thread(target=self.httpd.serve_forever, daemon=True) + self._t.start() + + def tearDown(self): + self.httpd.shutdown() + self.httpd.server_close() + self._d.cleanup() + + def _conn(self): + return http.client.HTTPConnection("127.0.0.1", self.port, timeout=10) + + # Connection: close on each request so the server does not keep a worker thread blocked on + # the next request line, which otherwise logs a reset when the suite tears the server down. + _CLOSE = {"Connection": "close"} + + def _json(self, method, path, body=None): + c = self._conn() + headers = dict(self._CLOSE) + if body is not None: + headers["Content-Type"] = "application/json" + c.request(method, path, body=json.dumps(body) if body is not None else None, headers=headers) + resp = c.getresponse() + status = resp.status + out = resp.read() + c.close() + return status, (json.loads(out) if out else None) + + def test_round_trip_over_http(self): + status, created = self._json( + "POST", "/v1/share", + {"items": [{"app": "kiwix", "localRef": "a.zim"}, {"app": "kiwix", "localRef": "b.zim"}]}) + self.assertEqual(status, 200) + sid, token = created["sessionId"], created["token"] + self.assertEqual(created["manifestUrl"], f"/v1/share/{sid}/manifest") + + status, man = self._json("GET", f"/v1/share/{sid}/manifest?token={token}") + self.assertEqual(status, 200) + self.assertEqual(man["totalBytes"], len(self.payload) + 7) + self.assertEqual([i["itemRef"] for i in man["items"]], ["0", "1"]) + + c = self._conn() + c.request("GET", f"/v1/share/{sid}/fetch/0?token={token}", headers=self._CLOSE) + resp = c.getresponse() + self.assertEqual(resp.status, 200) + got = resp.read() + c.close() + self.assertEqual(got, self.payload) + + # One of two items fetched: the gate is still open, so an explicit DELETE closes it. + status, out = self._json("DELETE", f"/v1/share/{sid}?token={token}") + self.assertEqual(status, 200) + self.assertTrue(out["ok"]) + + status, _ = self._json("GET", f"/v1/share/{sid}/manifest?token={token}") + self.assertEqual(status, 404) # closed + + def test_manifest_wrong_token_over_http_403(self): + _, created = self._json( + "POST", "/v1/share", {"items": [{"app": "kiwix", "localRef": "a.zim"}]}) + sid = created["sessionId"] + status, _ = self._json("GET", f"/v1/share/{sid}/manifest?token=wrong") + self.assertEqual(status, 403) + + +if __name__ == "__main__": + unittest.main() From d377d31280b52a25d4550c9a03ae53c8c3d1661e Mon Sep 17 00:00:00 2001 From: "Luis Guzman (AppDevForAll)" Date: Sat, 24 Oct 2026 16:06:50 -0600 Subject: [PATCH 2/4] K2GO-376 refactor(cx-core): split the share gate into connection and session lifetimes The durable session no longer expires on the short connection idle; it closes on COMPLETE, CANCEL, or a long safety cap (the per-item connection TTL stays the export op's). Move the token to the X-Share-Token header so it never reaches access logs. Single-flight per item. Count an item delivered only on a clean close (ExportStream.completed); add POST /v1/share/{id}/complete; drop the implicit drop-based auto-close. --- static/cx-core/README.md | 43 ++-- static/cx-core/cx_core/ports.py | 8 +- static/cx-core/cx_core/router.py | 21 +- static/cx-core/cx_core/server.py | 18 +- static/cx-core/cx_core/share_ops.py | 337 ++++++++++++++----------- static/cx-core/tests/test_share_ops.py | 267 +++++++++++++------- 6 files changed, 428 insertions(+), 266 deletions(-) diff --git a/static/cx-core/README.md b/static/cx-core/README.md index b1cfcf4de..4900f3b3f 100644 --- a/static/cx-core/README.md +++ b/static/cx-core/README.md @@ -19,7 +19,7 @@ pairing and the network carrier are later app slices). - `cx_core/adapters/kiwix.py`: the first adapter, read side (ZIM header UUID + listing). - `cx_core/registry.py`: loads an adapter only when its app is installed. - `cx_core/export_ops.py`: the export long-operation registry (op lifecycle + temp sweeps). -- `cx_core/share_ops.py`: the share gate (token + scoped item set + the five closers), K2GO-376. +- `cx_core/share_ops.py`: the share gate (token + scoped item set + two lifetimes), K2GO-376. - `cx_core/router.py`: pure `/v1` route dispatch (unit-testable without a socket). - `cx_core/server.py` + `__main__.py`: a thin stdlib `http.server` wrapper, localhost only. @@ -74,23 +74,32 @@ Import / delete: Share (a gate over the export op, K2GO-376): Share is the network twin of Export. The receiver pulls selected items from the sender's live cx-core, so the transfer can close -itself (the sender machine is on, unlike Clone). A share session is a token plus a scoped -item set plus one shared lifecycle; each item is a lazily created export fetch, so the -staging, the in_use pin and the on_close cleanup are the export op's. The gate closes on the -first of five closers (TTL / idle, startup sweep, DELETE, implicit all-delivered, and the -sender service end, which is app-side). This slice is loopback only: pairing (QR, hotspot, -WLAN) is a later app slice. The session token travels as the `token` query parameter. +itself (the sender machine is on, unlike Clone). The gate has two lifetimes: the CONNECTION +(one open stream) reuses the export op's idle TTL + in_use pin per item, and the durable +SESSION (a token + a scoped item set + per-item delivered state) does not expire on the short +connection idle. The session closes on COMPLETE (the receiver has every item), CANCEL +(DELETE, by host or receiver; the sender's foreground service calls it in onDestroy), a long +SAFETY CAP (hours of no activity), or process death (in-memory + the export startup sweep). +Each item is a lazily created export fetch, so staging, the in_use pin and the on_close +cleanup are the export op's. An item counts as delivered only on a CLEAN close (the server +marks a full body), so a mid-stream drop stays pending and a resume re-fetches it. This slice +is loopback only: pairing (QR, hotspot, WLAN) is a later app slice. + +The session token travels in the `X-Share-Token` request header, never the URL (it is the +only auth, so it must not reach access logs). - `POST /v1/share {items: [{app, localRef}]}` -> `{sessionId, token, manifestUrl}`. Opens a session scoped to the item set. -- `GET /v1/share/{sessionId}/manifest?token=` -> the bundle manifest over the scope - (per-item identity, size, upstream hash when present, each with its `itemRef`), plus - `missing` and `totalBytes`. -- `GET /v1/share/{sessionId}/fetch/{itemRef}?token=` -> the chunked octet stream for - one item. `409 {state: staging}` while a staging item (Kolibri) stages, so the receiver - retries; `200` streams once ready. -- `DELETE /v1/share/{sessionId}?token=` -> `{ok: true}`. Closes the session now and - revokes the token and every child temp. `404` on an unknown or already-closed session. +- `GET /v1/share/{sessionId}/manifest` -> the bundle manifest over the scope (per-item + identity, size, upstream hash when present, each with its `itemRef` and whether it is already + `delivered`), plus `missing` and `totalBytes`. +- `GET /v1/share/{sessionId}/fetch/{itemRef}` -> the chunked octet stream for one item. + `409 {state: staging}` while a staging item (Kolibri) stages (the receiver retries), + `409 {state: busy}` when the item already has an open stream (single-flight), `200` streams + once ready. +- `POST /v1/share/{sessionId}/complete` -> `{ok: true}`. The receiver's clean-success close. +- `DELETE /v1/share/{sessionId}` -> `{ok: true}`. Cancel (abort) by host or receiver. Both + closers revoke the token and every child temp; `404` on an unknown or already-closed session. Token checks return `401` when no token is sent, `403` on a mismatch, `404` on an unknown or expired session. @@ -105,8 +114,8 @@ 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, route dispatch, the export long-operation registry (prepare / status / fetch / delete, the TTL and startup sweeps), the -share gate (create / manifest / fetch / delete, token enforcement, the five closers), and the -streaming over a real socket. +share gate (create / manifest / fetch / complete / delete, header-token enforcement, +single-flight, clean-close delivery, the safety cap), and the streaming over a real socket. ## Not built yet diff --git a/static/cx-core/cx_core/ports.py b/static/cx-core/cx_core/ports.py index 25a49a9f6..166b58a70 100644 --- a/static/cx-core/cx_core/ports.py +++ b/static/cx-core/cx_core/ports.py @@ -24,13 +24,19 @@ class ExportStream: error); the EXPORT OP LAYER sets both: on_open pins the op for the stream's life, on_close unpins and drops it. The server fires on_open at the top of the same try whose finally runs on_close, so the pin and its release are a matched pair even when the client disconnects. - The adapter only opens the stream; the op layer owns the lifecycle (K2GO-507).""" + The adapter only opens the stream; the op layer owns the lifecycle (K2GO-507). + + ``completed`` is set True by the server after the FULL body has been written (the clean + terminator), and stays False on a mid-stream client drop. The export op ignores it (it drops + the temp either way), but the share gate reads it so a share item counts as delivered only on + a clean close, never on a drop (K2GO-376).""" filename: str content_type: str chunks: Iterable[bytes] on_close: Optional[Callable[[], None]] = None on_open: Optional[Callable[[], None]] = None + completed: bool = False @dataclass diff --git a/static/cx-core/cx_core/router.py b/static/cx-core/cx_core/router.py index cdcade110..a44d65232 100644 --- a/static/cx-core/cx_core/router.py +++ b/static/cx-core/cx_core/router.py @@ -166,11 +166,13 @@ 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) -> Response: + ops: Optional[ExportOps] = None, share: Optional[ShareOps] = None, + token: Optional[str] = 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). The - share routes read the session token from ``query["token"]``, so the router stays + 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.""" trimmed = path.rstrip("/") or "/" segments = [unquote(s) for s in trimmed.split("/") if s != ""] @@ -188,6 +190,11 @@ def dispatch(registry: Registry, method: str, path: str, query: dict, body: Any return 500, {"error": "share unavailable"} items = body.get("items") if isinstance(body, dict) else None return share.create(items) + # POST /v1/share/{sessionId}/complete : the receiver signals a clean, done transfer. + if len(segments) == 4 and segments[0] == "v1" and segments[1] == "share" and segments[3] == "complete": + if share is None: + return 500, {"error": "share unavailable"} + return share.complete(segments[2], token) # POST /v1/dedup/check : receiver dedup verdicts, nothing moved. if segments == ["v1", "dedup", "check"]: return _dedup_check(registry, body) @@ -205,11 +212,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/share/{sessionId} : close a share session (the explicit "done" closer). + # 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: return 500, {"error": "share unavailable"} - return share.delete(segments[2], query.get("token")) + return share.delete(segments[2], token) # DELETE /v1/content/{app}/{localRef} : reclaim space by removing one item. if len(segments) == 4 and segments[0] == "v1" and segments[1] == "content": return _delete_item(registry, segments[2], segments[3]) @@ -243,14 +250,14 @@ def dispatch(registry: Registry, method: str, path: str, query: dict, body: Any if len(segments) == 4 and segments[0] == "v1" and segments[1] == "share" and segments[3] == "manifest": if share is None: return 500, {"error": "share unavailable"} - return share.manifest(segments[2], query.get("token")) + return share.manifest(segments[2], token) # GET /v1/share/{sessionId}/fetch/{itemRef} : stream one shared item (payload is an # ExportStream the server streams). 409 while the item stages; the receiver retries. if len(segments) == 5 and segments[0] == "v1" and segments[1] == "share" and segments[3] == "fetch": if share is None: return 500, {"error": "share unavailable"} - return share.fetch(segments[2], segments[4], query.get("token")) + return share.fetch(segments[2], segments[4], token) # GET /v1/storage : box content-volume disk summary + per-app content totals. if segments == ["v1", "storage"]: diff --git a/static/cx-core/cx_core/server.py b/static/cx-core/cx_core/server.py index 77bb1b9ac..9a16b6c74 100644 --- a/static/cx-core/cx_core/server.py +++ b/static/cx-core/cx_core/server.py @@ -29,10 +29,16 @@ class Handler(BaseHTTPRequestHandler): # chunk-size lines as body). The JSON path sends Content-Length, so keep-alive holds. protocol_version = "HTTP/1.1" + def _share_token(self): + # The share session token travels in a header, never the URL query, so it does not + # land in access logs (it is the only auth for the gate). + return self.headers.get("X-Share-Token") + def do_GET(self) -> None: # noqa: N802 (http.server naming) parts = urlsplit(self.path) query = {k: v[0] for k, v in parse_qs(parts.query).items()} - status, payload = dispatch(registry, "GET", parts.path, query, ops=ops, share=share) + status, payload = dispatch(registry, "GET", parts.path, query, ops=ops, share=share, + token=self._share_token()) if isinstance(payload, ExportStream): self._stream(status, payload) else: @@ -49,13 +55,15 @@ def do_POST(self) -> None: # noqa: N802 (http.server naming) if body is _BAD_JSON: self._send(400, {"error": "invalid JSON body"}) return - status, payload = dispatch(registry, "POST", parts.path, {}, body, ops=ops, share=share) + status, payload = dispatch(registry, "POST", parts.path, {}, body, ops=ops, share=share, + token=self._share_token()) self._send(status, payload) def do_DELETE(self) -> None: # noqa: N802 (http.server naming) parts = urlsplit(self.path) query = {k: v[0] for k, v in parse_qs(parts.query).items()} - status, payload = dispatch(registry, "DELETE", parts.path, query, ops=ops, share=share) + status, payload = dispatch(registry, "DELETE", parts.path, query, ops=ops, share=share, + token=self._share_token()) self._send(status, payload) def _read_json_body(self): @@ -105,6 +113,10 @@ def _stream(self, status: int, stream: ExportStream) -> None: self.wfile.write(block) self.wfile.write(b"\r\n") self.wfile.write(b"0\r\n\r\n") + # The full body (and its terminator) is written: a clean close. A mid-stream + # client drop raises above and leaves this False, so the share gate counts the + # item delivered only on a clean close (K2GO-376). + stream.completed = True finally: if stream.on_close: stream.on_close() diff --git a/static/cx-core/cx_core/share_ops.py b/static/cx-core/cx_core/share_ops.py index 0b293c604..cd2c79bf0 100644 --- a/static/cx-core/cx_core/share_ops.py +++ b/static/cx-core/cx_core/share_ops.py @@ -6,36 +6,50 @@ Clone, whose standalone rsync daemon is mute on the host side and can only be stopped by the user. See controller/docs/ADR-content-exchange.md, section "Pillar Share". -A share session is a TOKEN plus a SCOPED set of items plus ONE shared lifecycle. It does not -reinvent the transfer: each shared item is a lazily created export op (export_ops.py), so the -staging, the in_use pin and the on_close cleanup are the ones already hardened and device -validated for K2GO-507. The session is the GATE over those child ops: it holds the token and -the scope, it ties the child ops together, and it can revoke them all at once. - -Ownership (one source per fact, so this is not a second owner of anything the export op -owns): - : the session owns the token, the scoped item set and the session liveness. - : the child export op owns the staged temp, the stream and the per-item liveness. -Both track an "is a stream open" pin, but at different scopes and for different cleanup: the -session pin keeps the gate open for a long single-item transfer; the export op pin keeps that -item's temp from being swept. They are nested scopes, not a duplicated truth. - -The gate closes on the FIRST of five independent closers (ADR "Gate / token lifecycle"): - 1. TTL + idle : last_seen bumps on each manifest / fetch, so an active transfer stays - alive and an abandoned one expires. This is the jaw Clone lacks. - 2. startup sweep: a fresh process holds no sessions (they live only in memory); the child - temps are cxexport-* reclaimed by the export op's own startup sweep. - 3. DELETE : the explicit "done" closer (the receiver finished, or tapped Finish). - 4. implicit : when every scoped item has been delivered, the gate auto-closes. - 5. sender end : APP-SIDE. The sender's foreground share service calls DELETE in its - onDestroy to revoke the gate. The core exposes DELETE; it cannot observe - the Android service lifecycle, so this closer is documented here, not - implemented in the core. - -Default = closed: staying open requires active fetches. The token is bound to the session -(revoking the session revokes the token) and scoped to the item set, so there is no separate -credential lifecycle to leak. The only surface a token holder reaches is /v1/share/* scoped -to this session's items; the wider /v1 API is never widened by a share session. +Two lifetimes (the refinement that makes large, resumable transfers work; collapsing them +into one short idle TTL was the first-cut error: a multi-hour transfer that pauses would be +reclaimed mid-way): + + CONNECTION (ephemeral, Layer 1): ONE open stream. Its idle TTL and in_use pin belong to the + CHILD EXPORT OP (export_ops.py), reused per item: a dead socket's op is reclaimed after the + export idle TTL (~10 min), a live stream is pinned so its temp is never swept under it, and a + per-item resume is just a re-fetch (the export op re-stages). This gate adds no connection + TTL of its own; it reuses the export op's. + + SESSION (durable, Layer 2): a TOKEN plus a SCOPED item set plus the per-item delivered state. + It does NOT expire on the short connection idle. It lives until ONE of: + : COMPLETE : the receiver signals it has every item (clean success) -> release. + : CANCEL (DELETE, by host or receiver) : abort; the sender foreground service calls DELETE + in onDestroy (the app-side closer). + : a long SAFETY CAP (hours of no activity), so a truly abandoned session is reclaimed; this + is NOT the ~10-minute connection idle. An active transfer keeps the session fresh. + : process death : the session is in-memory (v1 decision), so a restart loses it and the + export op's startup sweep reclaims the child temps on boot. + +The session does not reinvent the transfer: each shared item is a lazily created export op, so +staging + the in_use pin + on_close cleanup are the hardened, device-validated K2GO-507 ones. +The session is the GATE over those child ops: it holds the token and the scope, ties the child +ops together, and revokes them all at once on close. + +Ownership (one source per fact, so this is not a second owner of anything the export op owns): + : the session owns the token, the scoped item set, the delivered state and the session's own + liveness (the safety cap). + : the child export op owns the staged temp, the stream and the connection liveness (idle TTL + + in_use pin). +The ``streaming`` set is the session's single truth for "which items have an open stream": it +both single-flights a per-item fetch (so two streams never share one child op's temp) and keeps +the safety cap from reclaiming a session mid-transfer. That is a nested scope of the export +op's own per-op pin, not a duplicated truth. + +delivered is set ONLY on a CLEAN item close (the server marks ExportStream.completed after the +full body), never on a mid-stream drop, so a dropped item stays pending and a resume re-fetches +it. The manifest surfaces per-item delivered, so a resuming receiver skips what it already has. +The clean close is signalled; it is never guessed. + +The token travels in the ``X-Share-Token`` request header, never the URL, so it stays out of +access logs (it is the only auth). It is opaque, bound to the session (revoking the session +revokes the token) and scoped to the item set. The only surface a token holder reaches is +/v1/share/* scoped to this session's items; the wider /v1 API is never widened. This slice is loopback only: pairing (QR, hotspot, WLAN) is a later presentation slice and is outside the core, the way plugging in the Ethernet cable is not part of content exchange. @@ -49,11 +63,16 @@ from dataclasses import dataclass, field from typing import Any, Dict, List, Tuple, Union -from .export_ops import DEFAULT_TTL_SECONDS, ExportOps from .registry import Registry Response = Tuple[int, Any] +# How long a session may go with NO activity before the safety cap reclaims it. Hours, not the +# export op's ~10-minute connection idle: a paused transfer that resumes within the window +# survives, and a truly abandoned session is eventually reclaimed. The cap is a backstop; the +# normal closes are COMPLETE and CANCEL. +DEFAULT_SAFETY_CAP_SECONDS = 6 * 3600 + @dataclass class _Session: @@ -64,35 +83,34 @@ class _Session: items: List[Tuple[str, str]] # itemRef -> the child export opId, created lazily on first fetch of that item. item_ops: Dict[str, str] = field(default_factory=dict) - # itemRefs whose stream has closed, for the implicit auto-close (closer 4). + # itemRefs delivered on a CLEAN stream close (full body). Surfaced in the manifest for resume. delivered: set = field(default_factory=set) + # itemRefs with an open (or opening) stream: single-flight + "do not cap mid-transfer". + streaming: set = field(default_factory=set) + # Last activity, for the safety cap only (bumped on manifest / fetch / complete / close). last_seen: float = field(default_factory=time.monotonic) - # Count of open item streams. The TTL sweep never reclaims a session with a stream open, - # so a single multi-gigabyte item that streams longer than the TTL (and bumps nothing - # meanwhile) does not expire the gate under its own transfer. - in_use: int = 0 class ShareOps: """Thread-safe registry of share sessions. One instance lives for the server's lifetime; the handler threads call into it concurrently. It wraps the one ExportOps instance and - delegates every per-item transfer to it.""" + delegates every per-item transfer (the connection lifetime) to it.""" - def __init__(self, registry: Registry, export_ops: ExportOps, - ttl_seconds: int = DEFAULT_TTL_SECONDS): + def __init__(self, registry: Registry, export_ops, + safety_cap_seconds: int = DEFAULT_SAFETY_CAP_SECONDS): self._registry = registry self._export = export_ops - self._ttl = ttl_seconds + self._cap = safety_cap_seconds self._lock = threading.Lock() self._sessions: Dict[str, _Session] = {} # ---- REST-facing operations (return (status, payload), or (200, ExportStream)) ---- def create(self, items: Any) -> Response: - """Open a share session over a scoped item set. Returns the sessionId, an opaque - token bound to the session, and the manifest pointer. Partial is first-class: the set - is recorded as requested and each item is resolved at manifest / fetch time, so an - item that is absent is reported, not fatal.""" + """Open a share session over a scoped item set. Returns the sessionId, an opaque token + bound to the session, and the manifest pointer. Partial is first-class: the set is + recorded as requested and each item is resolved at manifest / fetch time, so an absent + item is reported, not fatal.""" if not isinstance(items, list) or not items: return 400, {"error": "body must be {items: [{app, localRef}]} with at least one item"} scoped: List[Tuple[str, str]] = [] @@ -111,7 +129,7 @@ def create(self, items: Any) -> Response: with self._lock: self._sessions[session_id] = _Session(token=token, items=scoped) # The pointer carries no token: the QR carries the address, the token and this pointer - # as separate fields (ADR), and the receiver combines them. + # as separate fields (ADR), and the receiver sends the token as the X-Share-Token header. return 200, { "sessionId": session_id, "token": token, @@ -120,8 +138,9 @@ def create(self, items: Any) -> Response: def manifest(self, session_id: str, token: Any) -> Response: """The bundle manifest over the session's scoped set: per-item identity, size and - upstream hash when present, each tagged with its itemRef. Reuses the read-side item - JSON, so it is the same manifest shape the disk export builds.""" + upstream hash when present, each tagged with its itemRef and whether it is already + delivered (so a resuming receiver skips it). Reuses the read-side item JSON, so it is the + same manifest shape the disk export builds.""" session = self._resolve(session_id, token) if isinstance(session, tuple): return session @@ -141,6 +160,7 @@ def manifest(self, session_id: str, token: Any) -> Response: continue entry = item.to_json() entry["itemRef"] = item_ref + entry["delivered"] = item_ref in session.delivered found.append(entry) total += item.bytes return 200, {"sessionId": session_id, "items": found, "missing": missing, @@ -148,10 +168,10 @@ def manifest(self, session_id: str, token: Any) -> Response: def fetch(self, session_id: str, item_ref: str, token: Any): """Stream one shared item. Lazily creates the child export op the first time an item is - fetched, then drives it: 409 {state: staging} while a staging app (Kolibri) stages, so - the receiver retries this same URL until it streams; (200, ExportStream) once ready. - The stream is wrapped so the session pin and the delivered set track with the export - op's own pin and temp cleanup.""" + fetched, then drives it: 409 {state: staging} while a staging app (Kolibri) stages (the + receiver retries this same URL), 409 {state: busy} when the item already has an open + stream (single-flight), (200, ExportStream) once ready. The stream's on_close releases + the single-flight slot and, on a clean close, records the item delivered.""" session = self._resolve(session_id, token) if isinstance(session, tuple): return session @@ -159,62 +179,60 @@ def fetch(self, session_id: str, item_ref: str, token: Any): if target is None: return 404, {"error": "unknown item"} app, local_ref = target - # A child op that was swept or already finished returns 404; re-prepare and retry once, - # so a resumed fetch re-stages instead of dead-ending on a stale opId. - for attempt in (1, 2): - op_id = self._ensure_op(session_id, item_ref, app, local_ref) - if isinstance(op_id, tuple): - return op_id # prepare failed: 404 absent, 507 no space - status, result = self._export.fetch(op_id) - if status == 404 and attempt == 1: - self._drop_op(session_id, item_ref, op_id) - continue - if status == 409: - return self._staging_response(op_id, item_ref) - if status != 200: - return status, result - self._wrap_stream(session_id, item_ref, result) - return 200, result - return 404, {"error": "item not available"} - - def _staging_response(self, op_id: str, item_ref: str) -> Response: - """The item is not ready: report it through the one fetch endpoint, since the gate has - no separate status route (do not widen the API). 500 on a staging error so the receiver - stops retrying; 409 with the staging heartbeat otherwise so the receiver retries and can - show progress.""" - st_status, st = self._export.status(op_id) - if st_status == 200 and st.get("state") == "error": - return 500, {"state": "error", "itemRef": item_ref, "error": st.get("error")} - progress = {"state": "staging", "itemRef": item_ref} - if st_status == 200: - progress["stagedBytes"] = st.get("stagedBytes", 0) - progress["totalBytes"] = st.get("totalBytes", 0) - return 409, progress - - def delete(self, session_id: str, token: Any) -> Response: - """Close the session now (closer 3, the explicit "done"; also the app-side closer 5). - Revokes the token with the session and drops every child op + its temp. 404 on an - unknown or already-closed session, so a double close (receiver Finish plus sender - onDestroy) is harmless: the second caller sees the gate already closed.""" - self._sweep_expired() + # Single-flight per item: one open stream at a time, so two fetches never stream the same + # child op's temp (the first close removes it under the second). A second concurrent + # fetch of the same item gets 409 busy and retries. Claim the slot before touching the + # export op; the stream's on_close releases it on the 200 path, the finally on every + # other path. with self._lock: - session = self._sessions.get(session_id) - if session is None: + live = self._sessions.get(session_id) + if live is None: return 404, {"error": "unknown session"} - denied = self._check_token(session, token) - if denied is not None: - return denied - self._sessions.pop(session_id, None) - child_ops = list(session.item_ops.values()) - for op_id in child_ops: - self._export.delete(op_id) # drops each child op + removes its staged temp - return 200, {"ok": True} + if item_ref in live.streaming: + return 409, {"state": "busy", "itemRef": item_ref} + live.streaming.add(item_ref) + keep = False + try: + # A child op swept or already finished returns 404; re-prepare and retry once, so a + # resumed fetch re-stages instead of dead-ending on a stale opId. + for attempt in (1, 2): + op_id = self._ensure_op(session_id, item_ref, app, local_ref) + if isinstance(op_id, tuple): + return op_id # prepare failed: 404 absent, 507 no space + status, result = self._export.fetch(op_id) + if status == 404 and attempt == 1: + self._drop_op(session_id, item_ref, op_id) + continue + if status == 409: + return self._staging_response(op_id, item_ref) + if status != 200: + return status, result + self._wrap_close(session_id, item_ref, result) + keep = True # the stream now owns the slot; its on_close releases it + return 200, result + return 404, {"error": "item not available"} + finally: + if not keep: + self._release_streaming(session_id, item_ref) + + def complete(self, session_id: str, token: Any) -> Response: + """Clean-success close: the receiver has every item it wanted and releases the session. + This is the receiver-driven close (supersedes any drop-based auto-close). Revokes the + token and every child op + temp. 404 on an unknown or already-closed session.""" + return self._close(session_id, token) + + def delete(self, session_id: str, token: Any) -> Response: + """Cancel (abort) close, by the host or the receiver; also the app-side closer (the + sender's foreground service calls this in onDestroy). Revokes the token and every child + op + temp. 404 on an unknown or already-closed session, so a double close (receiver and + sender) is harmless: the second caller sees the gate already closed.""" + return self._close(session_id, token) # ---- lifecycle ---- def sweep_startup(self) -> None: - """Drop any in-memory sessions on boot (closer 2). A fresh process holds none, so this - is defensive; the on-disk footprint of a session is its child export temps (cxexport-*), + """Drop any in-memory sessions on boot. A fresh process holds none, so this is + defensive; the on-disk footprint of a session is its child export temps (cxexport-*), reclaimed by the export op's own startup sweep, which the server runs on boot. Reusing that single sweep avoids a second temp owner.""" with self._lock: @@ -223,9 +241,10 @@ def sweep_startup(self) -> None: # ---- token + item resolution ---- def _resolve(self, session_id: str, token: Any) -> Union[_Session, Response]: - """Return the live session, bumping last_seen (closer 1), or an error response: 404 on - an unknown or expired session, 401 when no token is sent, 403 on a token mismatch.""" - self._sweep_expired() + """Return the live session, bumping last_seen (keeps an active session off the safety + cap), or an error response: 404 on an unknown or expired session, 401 when no token is + sent, 403 on a token mismatch.""" + self._sweep_capped() with self._lock: session = self._sessions.get(session_id) if session is None: @@ -236,6 +255,23 @@ def _resolve(self, session_id: str, token: Any) -> Union[_Session, Response]: session.last_seen = time.monotonic() return session + def _close(self, session_id: str, token: Any) -> Response: + """Resolve + token-check + revoke. The single close primitive behind COMPLETE and + CANCEL: both revoke the token and the child ops; they differ only in the app's intent.""" + self._sweep_capped() + with self._lock: + session = self._sessions.get(session_id) + if session is None: + return 404, {"error": "unknown session"} + denied = self._check_token(session, token) + if denied is not None: + return denied + self._sessions.pop(session_id, None) + child_ops = list(session.item_ops.values()) + for op_id in child_ops: + self._export.delete(op_id) # drops each child op + removes its staged temp + return 200, {"ok": True} + @staticmethod def _check_token(session: _Session, token: Any): """None when the token is valid, else the error response. compare_digest is @@ -274,14 +310,12 @@ def _ensure_op(self, session_id: str, item_ref: str, app: str, local_ref: str): with self._lock: session = self._sessions.get(session_id) if session is None: - # The session closed while we prepared: drop the orphan op we just made. + # The session closed (CANCEL / cap) while we prepared: drop the orphan op we + # just made so its temp is not leaked. self._export.delete(op_id) return 404, {"error": "unknown session"} - # Another concurrent fetch may have prepared first; keep the first and drop ours. - existing = session.item_ops.get(item_ref) - if existing is not None: - self._export.delete(op_id) - return existing + # Single-flight guarantees no concurrent fetch of this same item, so item_ops has no + # entry for it here; store ours as the one child op for the item. session.item_ops[item_ref] = op_id return op_id @@ -292,66 +326,65 @@ def _drop_op(self, session_id: str, item_ref: str, op_id: str) -> None: if session is not None and session.item_ops.get(item_ref) == op_id: session.item_ops.pop(item_ref, None) - def _wrap_stream(self, session_id: str, item_ref: str, stream) -> None: - """Layer the session pin + delivered bookkeeping onto the export op's own on_open / - on_close, so the two lifecycles stay a matched pair. The server fires on_open at the - top of streaming and on_close in the same try's finally, so both pins are always - released, even on a client disconnect.""" - inner_open = stream.on_open - inner_close = stream.on_close + def _staging_response(self, op_id: str, item_ref: str) -> Response: + """The item is not ready: report it through the one fetch endpoint, since the gate has + no separate status route (do not widen the API). 500 on a staging error so the receiver + stops retrying; 409 {state: staging} with the heartbeat otherwise so it retries and can + show progress.""" + st_status, st = self._export.status(op_id) + if st_status == 200 and st.get("state") == "error": + return 500, {"state": "error", "itemRef": item_ref, "error": st.get("error")} + progress = {"state": "staging", "itemRef": item_ref} + if st_status == 200: + progress["stagedBytes"] = st.get("stagedBytes", 0) + progress["totalBytes"] = st.get("totalBytes", 0) + return 409, progress - def on_open() -> None: - self._pin(session_id) - if inner_open: - inner_open() + def _wrap_close(self, session_id: str, item_ref: str, stream) -> None: + """Layer the single-flight release + delivered bookkeeping onto the export op's own + on_close. The server fires on_close in the same try's finally that opens the stream, so + it always runs, even on a client disconnect. on_open stays the export op's (its per-op + pin); only on_close needs the session bookkeeping.""" + inner_close = stream.on_close def on_close() -> None: # The export op's close runs first (drops the child op and removes its temp), then - # the session bookkeeping records the item delivered and may auto-close the gate. + # the session records the outcome. A clean close (full body) marks the item + # delivered; a mid-stream drop leaves it pending so a resume re-fetches it. if inner_close: inner_close() - self._unpin(session_id, item_ref) + self._on_stream_close(session_id, item_ref, bool(getattr(stream, "completed", False))) - stream.on_open = on_open stream.on_close = on_close - def _pin(self, session_id: str) -> None: - with self._lock: - session = self._sessions.get(session_id) - if session is not None: - session.in_use += 1 - - def _unpin(self, session_id: str, item_ref: str) -> None: - """Release one stream pin, record the item delivered, and auto-close the gate once - every scoped item is delivered and no stream is open (closer 4). A mid-stream drop also - marks the item delivered, so the implicit close is a convenience; DELETE (closer 3) and - the TTL (closer 1) are the robust closers, and a receiver that still needs an item - re-pairs (the gate is cheap to re-create).""" - child_ops: List[str] = [] + def _on_stream_close(self, session_id: str, item_ref: str, completed: bool) -> None: with self._lock: session = self._sessions.get(session_id) if session is None: return - if session.in_use > 0: - session.in_use -= 1 + session.streaming.discard(item_ref) session.item_ops.pop(item_ref, None) # the child op finished; a re-fetch re-prepares - session.delivered.add(item_ref) session.last_seen = time.monotonic() - all_refs = {str(i) for i in range(len(session.items))} - if session.in_use == 0 and session.delivered.issuperset(all_refs): - self._sessions.pop(session_id, None) - child_ops = list(session.item_ops.values()) - for op_id in child_ops: - self._export.delete(op_id) + if completed: + session.delivered.add(item_ref) + + def _release_streaming(self, session_id: str, item_ref: str) -> None: + """Release a single-flight slot claimed by a fetch that did not open a stream (staging, + error, or absent).""" + with self._lock: + session = self._sessions.get(session_id) + if session is not None: + session.streaming.discard(item_ref) - def _sweep_expired(self) -> None: + def _sweep_capped(self) -> None: now = time.monotonic() - expired: List[_Session] = [] + capped: List[_Session] = [] with self._lock: for session_id, session in list(self._sessions.items()): - # Never reclaim a session with an open stream: its transfer is live. - if session.in_use == 0 and now - session.last_seen > self._ttl: - expired.append(self._sessions.pop(session_id)) - for session in expired: + # Never reclaim a session with an open stream: its transfer is live. Otherwise an + # abandoned session (no activity past the safety cap) is reclaimed. + if not session.streaming and now - session.last_seen > self._cap: + capped.append(self._sessions.pop(session_id)) + for session in capped: for op_id in list(session.item_ops.values()): self._export.delete(op_id) diff --git a/static/cx-core/tests/test_share_ops.py b/static/cx-core/tests/test_share_ops.py index 6985869ca..934eb1319 100644 --- a/static/cx-core/tests/test_share_ops.py +++ b/static/cx-core/tests/test_share_ops.py @@ -1,8 +1,8 @@ """Share gate tests (K2GO-376): the shared contract for granular wireless Share. Everything runs on loopback, like the export op tests: a share session wraps the export op, -so these cover the gate (token, scope, the five closers) and the per-item transfer reuse. The -app forks (sender / receiver) build against this contract. +so these cover the gate (token in a header, the scope, the two lifetimes, single-flight, +clean-close delivery) and the per-item transfer reuse. The app forks build against this. """ import http.client @@ -95,10 +95,12 @@ def open_fetch(self, local_ref, staged_dir=None): return ExportStream(f"{local_ref}.tar", "application/x-tar", [data]) -def _share(adapters, staging, ttl=60): +def _share(adapters, staging, cap=3600, conn_ttl=600): + """Build a share gate over an export op. ``cap`` is the Layer-2 session safety cap; the + ``conn_ttl`` is the Layer-1 export-op (connection) idle TTL: two separate timeouts.""" reg = Registry(adapters, import_root=staging, content_volume=staging) - ops = ExportOps(reg, staging_dir=staging, ttl_seconds=ttl) - share = ShareOps(reg, ops, ttl_seconds=ttl) + ops = ExportOps(reg, staging_dir=staging, ttl_seconds=conn_ttl) + share = ShareOps(reg, ops, safety_cap_seconds=cap) return share, ops, reg @@ -109,16 +111,28 @@ def _create(share, items): def _drain(stream): - """Drive a stream as the server would: on_open, consume the chunks, on_close.""" + """Drive a stream as the server would on a CLEAN close: on_open, consume the chunks, mark + completed (the server sets this after the full body), on_close.""" if stream.on_open: stream.on_open() try: - return b"".join(stream.chunks) + data = b"".join(stream.chunks) + stream.completed = True + return data finally: if stream.on_close: stream.on_close() +def _drain_drop(stream): + """Drive a stream as a MID-STREAM client drop: open, then close without a full read, so + ``completed`` stays False and the item is not counted delivered.""" + if stream.on_open: + stream.on_open() + if stream.on_close: + stream.on_close() + + def _fetch_ready(share, sid, item_ref, token, timeout=5): end = time.time() + timeout while time.time() < end: @@ -131,6 +145,12 @@ def _fetch_ready(share, sid, item_ref, token, timeout=5): raise AssertionError("item never became ready") +def _wait_ready(ops, op_id, timeout=5): + end = time.time() + timeout + while ops._ops[op_id].state != "ready" and time.time() < end: + time.sleep(0.01) + + class CreateTest(unittest.TestCase): def test_create_returns_session_token_and_pointer(self): with tempfile.TemporaryDirectory() as d: @@ -140,7 +160,7 @@ def test_create_returns_session_token_and_pointer(self): self.assertIn("sessionId", body) self.assertTrue(body["token"]) self.assertEqual(body["manifestUrl"], f"/v1/share/{body['sessionId']}/manifest") - # The pointer carries no token: the token travels as a separate field. + # The pointer carries no token: the token travels as the X-Share-Token header. self.assertNotIn(body["token"], body["manifestUrl"]) def test_distinct_sessions_get_distinct_tokens(self): @@ -156,8 +176,8 @@ def test_empty_or_bad_body_400(self): share, _ops, _reg = _share([_DirectFake({"a.zim": 10})], d) self.assertEqual(share.create([])[0], 400) self.assertEqual(share.create("nope")[0], 400) - self.assertEqual(share.create([{"app": "kiwix"}])[0], 400) # no localRef - self.assertEqual(share.create([{"localRef": "a.zim"}])[0], 400) # no app + self.assertEqual(share.create([{"app": "kiwix"}])[0], 400) + self.assertEqual(share.create([{"localRef": "a.zim"}])[0], 400) self.assertEqual(share.create([{"app": "kiwix", "localRef": ""}])[0], 400) def test_duplicate_items_are_deduped(self): @@ -172,7 +192,7 @@ def test_duplicate_items_are_deduped(self): class ManifestTest(unittest.TestCase): - def test_manifest_lists_scoped_items_with_itemref_and_total(self): + def test_manifest_lists_scoped_items_with_itemref_total_and_delivered(self): with tempfile.TemporaryDirectory() as d: share, _ops, _reg = _share([_DirectFake({"a.zim": 10, "b.zim": 25})], d) sid, token = _create(share, [ @@ -183,6 +203,7 @@ def test_manifest_lists_scoped_items_with_itemref_and_total(self): self.assertEqual(status, 200) self.assertEqual([i["itemRef"] for i in body["items"]], ["0", "1"]) self.assertEqual([i["identity"]["localRef"] for i in body["items"]], ["a.zim", "b.zim"]) + self.assertEqual([i["delivered"] for i in body["items"]], [False, False]) self.assertEqual(body["totalBytes"], 35) self.assertEqual(body["missing"], []) @@ -239,7 +260,6 @@ def test_fetch_unknown_itemref_404(self): self.assertEqual(share.fetch(sid, "x", token)[0], 404) # not an index def test_fetch_absent_item_404(self): - # The item was in scope at create but is not exportable now: prepare returns None. with tempfile.TemporaryDirectory() as d: share, _ops, _reg = _share([_DirectFake({"a.zim": 10})], d) sid, token = _create(share, [{"app": "kiwix", "localRef": "gone.zim"}]) @@ -274,11 +294,27 @@ def test_fetch_staging_error_surfaces_500(self): self.assertIsNotNone(seen, "staging error never surfaced") self.assertEqual(seen["state"], "error") - -class AutoCloseTest(unittest.TestCase): - def test_delivering_every_item_auto_closes_the_gate(self): - # Closer 4 (implicit progress): once every scoped item has been delivered, the gate - # closes itself, so a later manifest is 404. + def test_single_flight_rejects_a_second_concurrent_fetch_of_one_item(self): + # Two open streams on one item would share the child op's temp, and the first close would + # remove it under the second. The gate single-flights: a second fetch of a busy item gets + # 409 busy until the first stream closes. + with tempfile.TemporaryDirectory() as d: + share, _ops, _reg = _share([_DirectFake({"a.zim": 10})], d) + sid, token = _create(share, [{"app": "kiwix", "localRef": "a.zim"}]) + status1, stream1 = share.fetch(sid, "0", token) # claims the slot + self.assertEqual(status1, 200) + status2, body2 = share.fetch(sid, "0", token) # same item still open + self.assertEqual(status2, 409) + self.assertEqual(body2["state"], "busy") + _drain(stream1) # closes the first stream: releases + status3, stream3 = share.fetch(sid, "0", token) # free again + self.assertEqual(status3, 200) + _drain(stream3) + + +class CompleteTest(unittest.TestCase): + def test_complete_closes_cleanly(self): + # Closer: COMPLETE (the receiver has every item). Supersedes any drop-based auto-close. with tempfile.TemporaryDirectory() as d: share, _ops, _reg = _share([_DirectFake({"a.zim": 10, "b.zim": 7})], d) sid, token = _create(share, [ @@ -286,15 +322,52 @@ def test_delivering_every_item_auto_closes_the_gate(self): {"app": "kiwix", "localRef": "b.zim"}, ]) _drain(share.fetch(sid, "0", token)[1]) - self.assertEqual(share.manifest(sid, token)[0], 200) # one item left, still open _drain(share.fetch(sid, "1", token)[1]) - self.assertEqual(share.manifest(sid, token)[0], 404) # all delivered: auto-closed + status, body = share.complete(sid, token) + self.assertEqual(status, 200) + self.assertTrue(body["ok"]) + self.assertEqual(share.manifest(sid, token)[0], 404) # released + + def test_complete_token_enforcement_and_double_complete(self): + with tempfile.TemporaryDirectory() as d: + share, _ops, _reg = _share([_DirectFake({"a.zim": 10})], d) + sid, token = _create(share, [{"app": "kiwix", "localRef": "a.zim"}]) + self.assertEqual(share.complete(sid, None)[0], 401) + self.assertEqual(share.complete(sid, "wrong")[0], 403) + self.assertEqual(share.complete("deadbeef", token)[0], 404) + self.assertEqual(share.complete(sid, token)[0], 200) + self.assertEqual(share.complete(sid, token)[0], 404) # already closed -class DeleteTest(unittest.TestCase): - def test_delete_closes_and_revokes_child_temp(self): - # Closer 3 (explicit done). Fetch one of two items so the gate does not auto-close, - # leave a staged child op, then DELETE and confirm the gate and the temp are gone. +class DeliveredTest(unittest.TestCase): + def test_delivered_only_on_clean_close_dropped_item_resumes(self): + # delivered is set on a clean close only. A dropped item stays pending and re-fetches + # (the resume model): the gate does not close on a drop. + with tempfile.TemporaryDirectory() as d: + share, _ops, _reg = _share([_DirectFake({"a.zim": 10, "b.zim": 7})], d) + sid, token = _create(share, [ + {"app": "kiwix", "localRef": "a.zim"}, + {"app": "kiwix", "localRef": "b.zim"}, + ]) + _drain(share.fetch(sid, "0", token)[1]) # clean -> delivered + _drain_drop(share.fetch(sid, "1", token)[1]) # drop -> not delivered + _, man = share.manifest(sid, token) + delivered = {i["itemRef"]: i["delivered"] for i in man["items"]} + self.assertTrue(delivered["0"]) + self.assertFalse(delivered["1"]) + self.assertEqual(share.manifest(sid, token)[0], 200) # a dropped item did not close it + # Resume: re-fetch the dropped item, this time cleanly. + status, stream = share.fetch(sid, "1", token) + self.assertEqual(status, 200) + _drain(stream) + _, man2 = share.manifest(sid, token) + self.assertTrue({i["itemRef"]: i["delivered"] for i in man2["items"]}["1"]) + + +class CancelTest(unittest.TestCase): + def test_delete_cancels_and_revokes_child_temp(self): + # Closer: CANCEL (DELETE). Fetch starts a staging child op (leaves a temp on disk); the + # cancel revokes the session, the token and the child temp. with tempfile.TemporaryDirectory() as d: gate = threading.Event() gate.set() @@ -303,12 +376,9 @@ def test_delete_closes_and_revokes_child_temp(self): {"app": "kolibri", "localRef": "c0"}, {"app": "kolibri", "localRef": "c1"}, ]) - # Create a ready-but-unconsumed child op for item 1 (so a temp exists on disk). - share.fetch(sid, "1", token) # starts staging + share.fetch(sid, "1", token) # 409 staging, but creates the child op + temp op_id = share._sessions[sid].item_ops["1"] - end = time.time() + 5 - while ops._ops[op_id].state != "ready" and time.time() < end: - time.sleep(0.01) + _wait_ready(ops, op_id) temp = ops._ops[op_id].temp self.assertTrue(os.path.isdir(temp)) status, body = share.delete(sid, token) @@ -328,7 +398,7 @@ def test_delete_token_enforcement(self): self.assertEqual(share.delete(sid, token)[0], 200) def test_double_delete_is_404_already_closed(self): - # The receiver Finish and the sender onDestroy may both DELETE; the second sees 404. + # The receiver and the sender onDestroy may both DELETE; the second sees 404. with tempfile.TemporaryDirectory() as d: share, _ops, _reg = _share([_DirectFake({"a.zim": 10})], d) sid, token = _create(share, [{"app": "kiwix", "localRef": "a.zim"}]) @@ -336,44 +406,49 @@ def test_double_delete_is_404_already_closed(self): self.assertEqual(share.delete(sid, token)[0], 404) -class SweepTest(unittest.TestCase): - def test_ttl_sweep_reclaims_abandoned_session_and_temp(self): - # Closer 1 (TTL + idle): an abandoned session expires and takes its child temp with it. +class SafetyCapTest(unittest.TestCase): + def test_session_survives_past_the_connection_idle(self): + # The point of the two lifetimes: a paused transfer older than the connection idle TTL + # (10 min) but younger than the session safety cap (hours) must NOT be reclaimed. + with tempfile.TemporaryDirectory() as d: + share, _ops, _reg = _share([_DirectFake({"a.zim": 10})], d, cap=3600, conn_ttl=60) + sid, token = _create(share, [{"app": "kiwix", "localRef": "a.zim"}]) + share._sessions[sid].last_seen -= 120 # older than conn idle 60, younger than cap 3600 + share._sweep_capped() + self.assertIn(sid, share._sessions) + self.assertEqual(share.manifest(sid, token)[0], 200) + + def test_safety_cap_reclaims_abandoned_session_and_temp(self): with tempfile.TemporaryDirectory() as d: gate = threading.Event() gate.set() - share, ops, _reg = _share([_StagingFake({"chan": 40}, gate=gate)], d, ttl=60) + share, ops, _reg = _share([_StagingFake({"chan": 40}, gate=gate)], d, cap=3600) sid, token = _create(share, [{"app": "kolibri", "localRef": "chan"}]) - share.fetch(sid, "0", token) # starts staging, leaves a child op + temp + share.fetch(sid, "0", token) # 409 staging, leaves a child op + temp op_id = share._sessions[sid].item_ops["0"] - end = time.time() + 5 - while ops._ops[op_id].state != "ready" and time.time() < end: - time.sleep(0.01) + _wait_ready(ops, op_id) temp = ops._ops[op_id].temp - share._sessions[sid].last_seen -= 120 # older than the TTL - share._sweep_expired() + share._sessions[sid].last_seen -= 4000 # abandoned past the cap + share._sweep_capped() self.assertNotIn(sid, share._sessions) self.assertFalse(os.path.exists(temp)) - def test_ttl_sweep_never_reclaims_a_session_with_an_open_stream(self): - # A single large item can stream longer than the TTL without a new request to bump - # last_seen; the gate must survive under its own transfer (in_use). + def test_safety_cap_never_reclaims_a_session_with_an_open_stream(self): with tempfile.TemporaryDirectory() as d: - share, _ops, _reg = _share([_DirectFake({"a.zim": 10, "b.zim": 7})], d, ttl=60) + share, _ops, _reg = _share([_DirectFake({"a.zim": 10, "b.zim": 7})], d, cap=3600) sid, token = _create(share, [ {"app": "kiwix", "localRef": "a.zim"}, {"app": "kiwix", "localRef": "b.zim"}, ]) - status, stream = share.fetch(sid, "0", token) + status, stream = share.fetch(sid, "0", token) # item 0 now streaming self.assertEqual(status, 200) - stream.on_open() # the server pins at top of streaming - share._sessions[sid].last_seen -= 120 - share._sweep_expired() - self.assertIn(sid, share._sessions) # pinned: survives the sweep - b"".join(stream.chunks) - stream.on_close() # unpins; item 1 undelivered, stays open - self.assertIn(sid, share._sessions) - self.assertEqual(share._sessions[sid].in_use, 0) + share._sessions[sid].last_seen -= 4000 # past the cap + share._sweep_capped() + self.assertIn(sid, share._sessions) # streaming: survives + _drain(stream) # closes: clears streaming + share._sessions[sid].last_seen -= 4000 + share._sweep_capped() + self.assertNotIn(sid, share._sessions) # now abandoned: reclaimed def test_startup_sweep_clears_in_memory_sessions(self): with tempfile.TemporaryDirectory() as d: @@ -384,23 +459,21 @@ def test_startup_sweep_clears_in_memory_sessions(self): self.assertEqual(share._sessions, {}) def test_boot_reclaims_a_leftover_share_session_and_its_temp(self): - # Closer 2 (process death): sessions live only in memory, so a fresh process holds - # none; the session's on-disk footprint is its child export temps, reclaimed by the - # export op's startup sweep. This mirrors exactly what serve() runs on boot. + # Process death: sessions live only in memory, so a fresh process holds none; the + # session's on-disk footprint is its child export temps, reclaimed by the export op's + # startup sweep. This mirrors exactly what serve() runs on boot. with tempfile.TemporaryDirectory() as d: gate = threading.Event() gate.set() - share, ops, _reg = _share([_StagingFake({"chan": 40}, gate=gate)], d, ttl=60) + share, ops, _reg = _share([_StagingFake({"chan": 40}, gate=gate)], d) sid, token = _create(share, [{"app": "kolibri", "localRef": "chan"}]) share.fetch(sid, "0", token) # leaves a staged cxexport-* temp on disk op_id = share._sessions[sid].item_ops["0"] - end = time.time() + 5 - while ops._ops[op_id].state != "ready" and time.time() < end: - time.sleep(0.01) + _wait_ready(ops, op_id) temp = ops._ops[op_id].temp self.assertTrue(os.path.isdir(temp)) # Simulate a restart: a new process, new registries on the same staging dir. - new_share, new_ops, _ = _share([_StagingFake({"chan": 40})], d, ttl=60) + new_share, new_ops, _ = _share([_StagingFake({"chan": 40})], d) new_ops.sweep_startup() # reclaims the orphan temp new_share.sweep_startup() # holds no sessions self.assertFalse(os.path.exists(temp)) @@ -409,7 +482,8 @@ def test_boot_reclaims_a_leftover_share_session_and_its_temp(self): class ShareRouteTest(unittest.TestCase): """The share routes forward to the ShareOps registry; the gate mechanics live above. Here - we check the routing, the token-via-query wiring, and the missing-share guard.""" + we check the routing, the token-via-param wiring (the server reads the X-Share-Token header + and passes it as dispatch's token), and the missing-share guard.""" def setUp(self): self._d = tempfile.TemporaryDirectory() @@ -418,7 +492,7 @@ def setUp(self): def tearDown(self): self._d.cleanup() - def test_post_creates_and_get_manifest_and_fetch_and_delete(self): + def test_post_create_manifest_fetch_complete(self): status, body = dispatch(self.reg, "POST", "/v1/share", {}, {"items": [{"app": "kiwix", "localRef": "a.zim"}, {"app": "kiwix", "localRef": "b.zim"}]}, @@ -426,42 +500,51 @@ def test_post_creates_and_get_manifest_and_fetch_and_delete(self): self.assertEqual(status, 200) sid, token = body["sessionId"], body["token"] - status, man = dispatch(self.reg, "GET", f"/v1/share/{sid}/manifest", - {"token": token}, ops=self.ops, share=self.share) + status, man = dispatch(self.reg, "GET", f"/v1/share/{sid}/manifest", {}, + ops=self.ops, share=self.share, token=token) self.assertEqual(status, 200) self.assertEqual(len(man["items"]), 2) - status, stream = dispatch(self.reg, "GET", f"/v1/share/{sid}/fetch/0", - {"token": token}, ops=self.ops, share=self.share) + status, stream = dispatch(self.reg, "GET", f"/v1/share/{sid}/fetch/0", {}, + ops=self.ops, share=self.share, token=token) self.assertEqual(status, 200) self.assertIsInstance(stream, ExportStream) self.assertEqual(len(_drain(stream)), 10) - status, out = dispatch(self.reg, "DELETE", f"/v1/share/{sid}", - {"token": token}, ops=self.ops, share=self.share) + status, out = dispatch(self.reg, "POST", f"/v1/share/{sid}/complete", {}, None, + ops=self.ops, share=self.share, token=token) + self.assertEqual(status, 200) + self.assertTrue(out["ok"]) + + def test_delete_route_cancels(self): + sid, token = _create(self.share, [{"app": "kiwix", "localRef": "a.zim"}]) + status, out = dispatch(self.reg, "DELETE", f"/v1/share/{sid}", {}, + ops=self.ops, share=self.share, token=token) self.assertEqual(status, 200) self.assertTrue(out["ok"]) def test_manifest_without_token_401(self): sid, _token = _create(self.share, [{"app": "kiwix", "localRef": "a.zim"}]) status, _ = dispatch(self.reg, "GET", f"/v1/share/{sid}/manifest", {}, - ops=self.ops, share=self.share) + ops=self.ops, share=self.share, token=None) self.assertEqual(status, 401) def test_share_routes_without_share_500(self): for method, path, body in [ ("POST", "/v1/share", {"items": [{"app": "kiwix", "localRef": "a.zim"}]}), + ("POST", "/v1/share/abc/complete", None), ("GET", "/v1/share/abc/manifest", None), ("GET", "/v1/share/abc/fetch/0", None), ("DELETE", "/v1/share/abc", None), ]: - status, _ = dispatch(self.reg, method, path, {}, body, ops=self.ops, share=None) + status, _ = dispatch(self.reg, method, path, {}, body, ops=self.ops, share=None, + token="t") self.assertEqual(status, 500, path) class ShareHttpTest(unittest.TestCase): """End-to-end over a real socket: POST create, GET manifest, GET fetch (chunked stream - decoded by a real client), DELETE, all with the token as a query parameter.""" + decoded by a real client), POST complete, all with the token in the X-Share-Token header.""" def setUp(self): self._d = tempfile.TemporaryDirectory() @@ -482,13 +565,13 @@ def _conn(self): # Connection: close on each request so the server does not keep a worker thread blocked on # the next request line, which otherwise logs a reset when the suite tears the server down. - _CLOSE = {"Connection": "close"} - - def _json(self, method, path, body=None): + def _req(self, method, path, body=None, token=None): c = self._conn() - headers = dict(self._CLOSE) + headers = {"Connection": "close"} if body is not None: headers["Content-Type"] = "application/json" + if token is not None: + headers["X-Share-Token"] = token c.request(method, path, body=json.dumps(body) if body is not None else None, headers=headers) resp = c.getresponse() status = resp.status @@ -497,39 +580,51 @@ def _json(self, method, path, body=None): return status, (json.loads(out) if out else None) def test_round_trip_over_http(self): - status, created = self._json( + status, created = self._req( "POST", "/v1/share", {"items": [{"app": "kiwix", "localRef": "a.zim"}, {"app": "kiwix", "localRef": "b.zim"}]}) self.assertEqual(status, 200) sid, token = created["sessionId"], created["token"] self.assertEqual(created["manifestUrl"], f"/v1/share/{sid}/manifest") - status, man = self._json("GET", f"/v1/share/{sid}/manifest?token={token}") + status, man = self._req("GET", f"/v1/share/{sid}/manifest", token=token) self.assertEqual(status, 200) self.assertEqual(man["totalBytes"], len(self.payload) + 7) self.assertEqual([i["itemRef"] for i in man["items"]], ["0", "1"]) c = self._conn() - c.request("GET", f"/v1/share/{sid}/fetch/0?token={token}", headers=self._CLOSE) + c.request("GET", f"/v1/share/{sid}/fetch/0", headers={"Connection": "close", "X-Share-Token": token}) resp = c.getresponse() self.assertEqual(resp.status, 200) got = resp.read() c.close() self.assertEqual(got, self.payload) - # One of two items fetched: the gate is still open, so an explicit DELETE closes it. - status, out = self._json("DELETE", f"/v1/share/{sid}?token={token}") + # A clean fetch marks the item delivered in the manifest (resume skips it). + status, man2 = self._req("GET", f"/v1/share/{sid}/manifest", token=token) + delivered = {i["itemRef"]: i["delivered"] for i in man2["items"]} + self.assertTrue(delivered["0"]) + self.assertFalse(delivered["1"]) + + status, out = self._req("POST", f"/v1/share/{sid}/complete", token=token) self.assertEqual(status, 200) self.assertTrue(out["ok"]) - status, _ = self._json("GET", f"/v1/share/{sid}/manifest?token={token}") - self.assertEqual(status, 404) # closed + status, _ = self._req("GET", f"/v1/share/{sid}/manifest", token=token) + self.assertEqual(status, 404) # released + + def test_token_in_query_is_not_accepted(self): + # FIX: the token must NOT be read from the URL (it would land in access logs). A query + # token with no header is unauthenticated. + _, created = self._req("POST", "/v1/share", {"items": [{"app": "kiwix", "localRef": "a.zim"}]}) + sid, token = created["sessionId"], created["token"] + status, _ = self._req("GET", f"/v1/share/{sid}/manifest?token={token}") # no header + self.assertEqual(status, 401) - def test_manifest_wrong_token_over_http_403(self): - _, created = self._json( - "POST", "/v1/share", {"items": [{"app": "kiwix", "localRef": "a.zim"}]}) + def test_manifest_wrong_token_header_403(self): + _, created = self._req("POST", "/v1/share", {"items": [{"app": "kiwix", "localRef": "a.zim"}]}) sid = created["sessionId"] - status, _ = self._json("GET", f"/v1/share/{sid}/manifest?token=wrong") + status, _ = self._req("GET", f"/v1/share/{sid}/manifest", token="wrong") self.assertEqual(status, 403) From fbf3ec9031f5095dd1e27161a54d0c8f5c78bb6f Mon Sep 17 00:00:00 2001 From: "Luis Guzman (AppDevForAll)" Date: Sat, 24 Oct 2026 16:27:48 -0600 Subject: [PATCH 3/4] K2GO-376 fix(cx-core): guarantee stream on_close on a pre-header client drop Move the response header writes inside the try whose finally runs on_close, so a client disconnect during the headers still releases the export op pin and the share single-flight slot. A stranded slot blocked the safety cap and leaked the durable session (content exposed past the cap). Also note the manifest delivered flag is an advisory sender-side hint, not receiver-persisted state. --- static/cx-core/cx_core/server.py | 19 +++++++------ static/cx-core/cx_core/share_ops.py | 10 +++++-- static/cx-core/tests/test_share_ops.py | 38 ++++++++++++++++++++++++++ 3 files changed, 56 insertions(+), 11 deletions(-) diff --git a/static/cx-core/cx_core/server.py b/static/cx-core/cx_core/server.py index 9a16b6c74..ab908faec 100644 --- a/static/cx-core/cx_core/server.py +++ b/static/cx-core/cx_core/server.py @@ -95,15 +95,18 @@ def _stream(self, status: int, stream: ExportStream) -> None: # Close after the stream: the body length is unknown up front, so a chunked # response plus an explicit close keeps the connection state simple. self.close_connection = True - self.send_response(status) - self.send_header("Content-Type", stream.content_type) - self.send_header("Content-Disposition", f'attachment; filename="{stream.filename}"') - self.send_header("Transfer-Encoding", "chunked") - self.send_header("Connection", "close") - self.end_headers() try: - # Pin the op for the stream's life inside the same try whose finally runs - # on_close, so the pin is always released (even on a client disconnect). + # Write the headers INSIDE the same try whose finally runs on_close, so on_close + # is guaranteed on every exit: a client that disconnects during the header write + # must still release the op pin and the share single-flight slot (K2GO-376), + # never strand them. on_open follows the headers, so a pre-header failure runs + # on_close without a matching on_open: the op cleanup (_finish) is safe either way. + self.send_response(status) + self.send_header("Content-Type", stream.content_type) + self.send_header("Content-Disposition", f'attachment; filename="{stream.filename}"') + self.send_header("Transfer-Encoding", "chunked") + self.send_header("Connection", "close") + self.end_headers() if stream.on_open: stream.on_open() for block in stream.chunks: diff --git a/static/cx-core/cx_core/share_ops.py b/static/cx-core/cx_core/share_ops.py index cd2c79bf0..c2ba4bd90 100644 --- a/static/cx-core/cx_core/share_ops.py +++ b/static/cx-core/cx_core/share_ops.py @@ -138,9 +138,13 @@ def create(self, items: Any) -> Response: def manifest(self, session_id: str, token: Any) -> Response: """The bundle manifest over the session's scoped set: per-item identity, size and - upstream hash when present, each tagged with its itemRef and whether it is already - delivered (so a resuming receiver skips it). Reuses the read-side item JSON, so it is the - same manifest shape the disk export builds.""" + upstream hash when present, each tagged with its itemRef and a ``delivered`` flag. + Reuses the read-side item JSON, so it is the same manifest shape the disk export builds. + + ``delivered`` is ADVISORY: it means the SENDER streamed this item to completion once (a + clean stream close), NOT that the receiver persisted it. It is a progress / resume HINT. + The receiver's resume authority must be its OWN persisted state, so it never skips an item + it received but failed to save.""" session = self._resolve(session_id, token) if isinstance(session, tuple): return session diff --git a/static/cx-core/tests/test_share_ops.py b/static/cx-core/tests/test_share_ops.py index 934eb1319..9a5fb881f 100644 --- a/static/cx-core/tests/test_share_ops.py +++ b/static/cx-core/tests/test_share_ops.py @@ -542,6 +542,44 @@ def test_share_routes_without_share_500(self): self.assertEqual(status, 500, path) +class PreHeaderDropTest(unittest.TestCase): + """A client that disconnects DURING the response header write must still fire the stream's + on_close, so the single-flight slot is released and the safety cap can reclaim the session. + The server writes the headers inside the same try whose finally runs on_close; this checks + that guarantee through the real handler (white-box, because a real-socket drop timed exactly + at the header write is not deterministic).""" + + def test_pre_header_disconnect_releases_slot_so_cap_reclaims(self): + with tempfile.TemporaryDirectory() as d: + share, ops, reg = _share([_DirectFake({"a.zim": 10})], d, cap=3600) + sid, token = _create(share, [{"app": "kiwix", "localRef": "a.zim"}]) + status, stream = share.fetch(sid, "0", token) # claims the single-flight slot + self.assertEqual(status, 200) + self.assertIn("0", share._sessions[sid].streaming) + + # A handler whose header write fails, as if the client dropped during the headers. + handler_cls = make_handler(reg, ops, share) + + class _FailHeaders(handler_cls): + def send_response(self, *_a, **_k): + raise BrokenPipeError("client gone during headers") + + handler = _FailHeaders.__new__(_FailHeaders) # skip the socket __init__ + handler.close_connection = False + with self.assertRaises(BrokenPipeError): + handler._stream(200, stream) + + # on_close ran in the finally despite the header failure: the slot is released and the + # item is NOT delivered (the body never streamed). + self.assertNotIn("0", share._sessions[sid].streaming) + _, man = share.manifest(sid, token) + self.assertFalse(man["items"][0]["delivered"]) + # With the slot released, the cap reclaims the otherwise-abandoned session. + share._sessions[sid].last_seen -= 4000 + share._sweep_capped() + self.assertNotIn(sid, share._sessions) + + class ShareHttpTest(unittest.TestCase): """End-to-end over a real socket: POST create, GET manifest, GET fetch (chunked stream decoded by a real client), POST complete, all with the token in the X-Share-Token header.""" From 96a1116ac5a94d5789eae3406322fa0234a5691e Mon Sep 17 00:00:00 2001 From: "Luis Guzman (AppDevForAll)" Date: Sat, 24 Oct 2026 17:21:21 -0600 Subject: [PATCH 4/4] K2GO-376 docs(cx-core): add CHANGELOG (0.1.0/0.2.0/0.3.0) --- static/cx-core/CHANGELOG.md | 10 ++++++++++ 1 file changed, 10 insertions(+) create mode 100644 static/cx-core/CHANGELOG.md diff --git a/static/cx-core/CHANGELOG.md b/static/cx-core/CHANGELOG.md new file mode 100644 index 000000000..bd32ad458 --- /dev/null +++ b/static/cx-core/CHANGELOG.md @@ -0,0 +1,10 @@ +# cx-core (content-exchange core) - changelog + +One line per version, newest first. Every REST-facing change bumps `cx_core.VERSION` (reported by +`GET /v1/version`, and written into the `[cx_core]` section of iiab.ini by the cx_core Ansible +role), so this file is the human record of what each bump enables. Keep entries short: +`version - change (TICKET)`. + +- **0.3.0** - /v1/share gate: a token-scoped, two-lifetime share session (ephemeral connection vs durable session) over a scoped item set, reusing the export op per item, for granular wireless Share. (K2GO-376) +- **0.2.0** - export as a prepare/poll/fetch long operation, so a large Kolibri channel stages without timing out the client. (K2GO-507) +- **0.1.0** - initial content-exchange core: read (list + identity), export, import and delete over /v1 across Kiwix, Kolibri and Calibre. (K2GO-427)