diff --git a/claude-notes/instructions/hub-storage-hygiene.md b/claude-notes/instructions/hub-storage-hygiene.md new file mode 100644 index 000000000..b37ddbc6d --- /dev/null +++ b/claude-notes/instructions/hub-storage-hygiene.md @@ -0,0 +1,80 @@ +# Hub storage hygiene (`hub admin …`) + +Maintainer tools for quarto-hub sync-server deployments +(bd-eiku4ymo; design + rationale in +`claude-notes/plans/2026-07-24-capture-meta-and-hub-admin-tools.md`). + +## Why + +Every engine re-execution writes a **new** capture binary doc and +repoints the index sidecar; the old doc is orphaned in samod storage +forever. Since #410 captures also embed figure bytes, so orphans are +big. These tools inventory a storage location, identify orphaned +captures, and reclaim them — through a pipeline designed so a mistake +must get past four independent gates before any byte is gone. + +## The pipeline + +```bash +# 1. SCAN — read-only. Safe on a live server (snapshot semantics; no +# consistency is guaranteed anywhere in this system anyway). +hub admin scan --data-dir /srv/hub/data --output manifest.json +# → prints an inventory by doc kind + the removable candidates, +# each with evidence (createdAt, sourcePath, engines, WHY). +# --json prints the manifest to stdout (versioned, camelCase — +# stable for tooling). --older-than-days N gates candidates +# (default 30; negative disables the gate). --include-unstamped +# opts in captures recorded before the audit envelope existed. + +# 2. COLLECT — quarantine, never delete. Dry-run by default. Refuses: +# while a server holds hub.lock; if the manifest's dataDir isn't +# this data dir; per-doc, if re-verification fails (doc became +# referenced since the scan, kind changed, age gate no longer met). +hub admin collect --data-dir /srv/hub/data --manifest manifest.json # plan +hub admin collect --data-dir /srv/hub/data --manifest manifest.json --execute # quarantine +# → moves each doc's chunks to data/trash//docs/, +# writes batch.json (embedded manifest + per-chunk SHA-256s). + +# 3. RESTORE — undo a batch (or named docs), hash-verified. +hub admin restore --data-dir /srv/hub/data --batch /srv/hub/data/trash/ [doc-id ...] + +# 4. PURGE — the ONLY operation that deletes bytes. Whole batches, +# past the retention window (default 30d), dry-run by default. +hub admin purge --data-dir /srv/hub/data # list +hub admin purge --data-dir /srv/hub/data --execute # delete eligible +``` + +`quarto hub admin …` is the same tool (the `quarto` binary embeds hub). + +## Safety properties to rely on + +- **Allowlist**: only binary docs with the capture MIME + (`application/x-engine-capture+gzip`) are ever collectible. Project + indexes, project sets, file docs, and unknown-shaped docs (i.e. + anything a newer schema added) are structurally protected — + `scan` reports them, nothing can touch them. +- **Locking**: `collect`/`restore`/`purge` *acquire* the server's own + `hub.lock` (exclusive flock) for their duration. A running server + makes them refuse; symmetrically, a server cannot start + mid-operation. +- **Quarantine lives inside the data dir** (`trash/`): renames on one + filesystem, covered by your existing backup regime, can't be + orphaned from its server. +- **Unstamped captures** (recorded before the meta envelope) have no + age evidence and are excluded by default; batches without a + readable `batch.json` similarly can never be purged. + +## Caveats + +- **Peers**: liveness is computed from THIS storage location only. + Run against the authoritative store. A doc collected here can + re-sync from a peer that still holds it — which is the safe + direction (if it's referenced, that's exactly recovery; if not, + it becomes a future scan candidate again). +- **`q2 preview` data dirs are ephemeral** (temp dirs deleted on + shutdown) — these tools are for persistent hub deployments + (`hub --project` / standalone `--data-dir`), or for snapshots of a + store copied out for maintenance. +- Unreferenced **non-capture** docs (e.g. indexes of abandoned + projects) are reported but not collectible in v1; that's a + follow-up strand once real inventories are seen. diff --git a/claude-notes/plans/2026-07-24-capture-meta-and-hub-admin-tools.md b/claude-notes/plans/2026-07-24-capture-meta-and-hub-admin-tools.md new file mode 100644 index 000000000..f8c4e1590 --- /dev/null +++ b/claude-notes/plans/2026-07-24-capture-meta-and-hub-admin-tools.md @@ -0,0 +1,359 @@ +# Capture-doc metadata envelope + sync-server maintainer tools + +**Braid strand:** bd-eiku4ymo +**Status:** implemented on branch +`braid/bd-eiku4ymo-capture-docs-uncompressed-auditgc`; binary E2E +verified (record below) + +## Overview + +Two deliverables, one storyline: + +1. **Part A — audit/GC metadata envelope** (the strand as filed): stamp + engine-capture binary docs with a small **uncompressed** `meta` map + so sync-server audits can read provenance without gunzipping + payloads. +2. **Part B — minimal sync-server maintainer tools** (added on review, + 2026-07-24): `hub admin` subcommands that (a) inventory a samod + storage location and identify orphaned capture docs, producing a + manifest of safely-removable document ids, and (b) act on such a + manifest — with a design that minimizes the blast radius of any + accidental deletion. + +### Why captures orphan (recap) + +Every re-execution creates a **new** capture binary doc and repoints +the index sidecar's `CaptureRef` (`re_execute.rs::perform_re_execute`, +`capture_driver.rs`, hub-provider `execute.rs`); nothing deletes the +old doc. On hub deployments these accumulate forever, and bd-qbhp2cvv +(#410) made each one substantially bigger (embedded figure bytes). + +## Architecture facts the design rests on (verified in-tree) + +- **Binary-doc schema** (`quarto-hub/src/resource.rs:107`): automerge + ROOT with `content: Bytes`, `mimeType: String`, `hash: String`. + Capture docs are identified by + `mimeType == "application/x-engine-capture+gzip"` — already + readable without decompression. +- **Doc kinds distinguishable by ROOT shape**: + - project index: `files` map (+ V2 `captures` sidecar map) — + `quarto-hub/src/index.rs` + - project set (collections home, #394): `projects` map keyed by + indexDocId — `ts-packages/quarto-automerge-schema` + - text file: `text` + - binary file: `content` + `mimeType` + `hash` + - anything else: unknown +- **Reference graph**: index doc → file docs (`files: path → docId`) + and capture docs (`captures: path → CaptureRef{captureDocId}`); + project-set doc → index docs. That is the complete in-storage + reference graph today. +- **Storage**: samod `Storage` trait (`load`, `load_range(prefix)`, + `put`, `delete`) with `TokioFilesystemStorage` rooted at + `/automerge/` (`context.rs:283`). Admin tools reuse this + adapter — no hand-rolled directory walking, and the on-disk chunk + layout (splayed doc-id dirs with snapshot/incremental chunks) stays + samod's implementation detail. +- **Concurrency guard**: the server holds `/hub.lock`. +- **CLI**: the `hub` binary is currently a flat clap `Args` (no + subcommands); `quarto hub` forwards to the same entry, so new + subcommands surface in both. +- **Capture writers** (all native): three call sites already share + `quarto_core::engine::capture_files::gzip_captures` (Phase 4 of + #410), then each does `create_binary_document` + `repo.create`. + +--- + +## Part A — capture-doc metadata envelope + +New in `quarto-hub/src/resource.rs`: + +```text +ROOT +├── content: Bytes +├── mimeType: String +├── hash: String +└── meta: Map // NEW — uncompressed, audit-readable + ├── kind: "engine-capture" + ├── schemaVersion: 1 (int) + ├── createdAt: String // RFC 3339, UTC, from the writer's clock + ├── sourcePath: String // project-relative qmd path, fwd slashes + └── engines: List // e.g. ["knitr"] +``` + +- `pub struct CaptureDocMeta { source_path: String, engines: Vec }` + and `pub fn create_capture_document(gzipped: &[u8], meta: &CaptureDocMeta) -> Result` + — builds content/mimeType/hash exactly as today (MIME stays the + discriminator; `meta.kind` is belt-and-suspenders) plus the `meta` + map. `createdAt` stamped inside the helper. +- The three writers switch from `create_binary_document(..., + CAPTURE_MIME_TYPE)` to `create_capture_document(...)`, passing the + rel path + `captures.iter().map(|c| c.engine_name)`. +- Readers are untouched: the SPA/WASM reader only reads `content`; + docs without `meta` (all existing captures) stay valid forever. + `meta` is written once at creation and never mutated (no CRDT merge + concerns). + +### Part A tests (TDD) + +- [ ] Unit (`resource.rs`): `create_capture_document` produces + content/mimeType/hash byte-identical semantics to + `create_binary_document` + a `meta` map with all five fields; + `createdAt` parses as RFC 3339. +- [ ] Unit: `detect_document_type` still classifies the doc as Binary. +- [ ] Integration (per writer, cheapest at `capture_driver`): a + recorded capture doc read back from the repo carries `meta` + with the right `sourcePath`/`engines`. + +--- + +## Part B — sync-server maintainer tools (`hub admin …`) + +### Design principles (damage minimization first) + +1. **Read and write are separate tools.** `scan` never writes; + `collect` never decides. `collect` accepts **only a scan manifest** + — never ad-hoc doc ids from the command line. +2. **Nothing is ever unlinked by `collect`.** Collection = **quarantine**: + chunks move to a trash area inside the data dir, preserving the + storage-key layout, restorable byte-for-byte. Actual unlinking is a + third, separately-gated step (`purge`) with a retention window. +3. **Allowlist, not blocklist.** v1 may only collect docs that are + (a) binary docs with the capture MIME type and (b) unreferenced. + Index docs, project sets, file docs, and **unknown-shaped docs are + never collectible** — a future schema the scanner doesn't know + reads as "unknown" and is automatically protected. +4. **Age gate.** Only captures whose `meta.createdAt` is older than + `--older-than ` (default 30) are candidates. Pre-envelope + captures have no timestamp → **excluded by default**; an explicit + `--include-unstamped` opts them in (still quarantined, never + purged before retention). +5. **Re-verify at collect time.** The manifest is a photograph; + between scan and collect, a doc may become referenced (a client + synced an older index state back, a re-execute repointed, a project + was re-added). `collect` recomputes liveness for the candidate set + against *current* storage and skips (and reports) anything that no + longer qualifies. +6. **Never operate under a live server** (default). `collect`, + `restore`, and `purge` refuse when `hub.lock` is held; `scan` may + run live (it is read-only; staleness is handled by principle 5). + samod holds loaded docs in memory and may rewrite chunks — moving + files under it risks resurrection or wasted work, so the escape + hatch (`--allow-live`) exists only for `scan`-equivalent safety + analysis, not for collect. +7. **Everything is auditable.** The manifest and every quarantine + batch carry: tool version, scan timestamp, storage path, full + inventory counts, and per-doc evidence (kind, MIME, size, meta + fields, and *why* it was classified removable). The trash batch + directory embeds the manifest that produced it. + +### Subcommands + +```bash +hub admin scan --data-dir [--older-than 30d] [--include-unstamped] + [--output manifest.json] [--json] +hub admin collect --data-dir --manifest manifest.json [--execute] +hub admin restore --data-dir --batch [doc-id ...] +hub admin purge --data-dir [--retention 30d] [--execute] +``` + +**`scan`** (read-only): +1. Enumerate doc ids via the storage adapter (`load_range` from the + root key; skip `storage-adapter-id`). +2. Load each doc (snapshot + incremental chunks → `Automerge`), + classify by ROOT shape (see facts above). +3. Root set = every project-index doc + every project-set doc. + (Project sets are roots because their only inbound pointers live in + client IndexedDB, invisible to us. Index docs are roots for the + same reason — share URLs.) +4. Live set = roots ∪ every doc id referenced from any index `files` + map, any index `captures` sidecar, any project-set `projects` map. +5. Candidates = capture-MIME binary docs ∉ live set, past the age + gate. +6. Emit the manifest + a human summary (counts and bytes by kind, + candidate count and reclaimable bytes). The inventory alone makes + `scan` a useful audit tool with zero risk — it also reports + *non-capture* unreferenced docs (informational only, explicitly + marked `not collectible in v1`). + +Manifest shape (JSON, versioned): + +```json +{ + "manifestVersion": 1, + "tool": "hub admin scan ", + "scannedAt": "2026-07-24T…Z", + "dataDir": "/srv/hub/data", + "inventory": { "project-index": 3, "project-set": 1, "text-file": 41, + "binary-file": 12, "engine-capture": 9, "unknown": 0, + "bytesByKind": { "…": 0 } }, + "candidates": [ + { "docId": "4EBfN…", "kind": "engine-capture", + "sizeBytes": 812345, + "meta": { "createdAt": "…", "sourcePath": "posts/a.qmd", + "engines": ["knitr"] }, + "reason": "capture MIME; not referenced by any index/captures/project-set; age 41d > 30d" } + ], + "notCollectible": [ { "docId": "…", "kind": "unknown", "reason": "unknown shape" } ] +} +``` + +**`collect`** (quarantine; dry-run by default): +1. Refuse if `hub.lock` held. Refuse if manifest's `dataDir` doesn't + match (paranoia against pointing the wrong manifest at the wrong + server). +2. Re-verify each candidate (principle 5): still exists, still + capture-MIME, still unreferenced, still past age gate. +3. Dry-run prints the verified plan. With `--execute`: for each doc, + move its chunks to + `/trash/-//` + and write `batch.json` (the manifest + per-doc verification results + + chunk inventory with hashes) into the batch dir. + Move is per-doc atomic-ish (rename within the same filesystem); + a failure mid-batch leaves already-moved docs quarantined and + reports precisely which. +4. `trash/` lives inside the data dir on purpose: same filesystem + (rename not copy), covered by whatever backup regime the operator + already has, and impossible to orphan the trash from its server. + +**`restore`**: moves a batch's chunks (or a named subset) back into +place, verifying the target keys are absent first (a doc re-created +under the same id since collection → refuse for that doc, report). +Chunk hashes from `batch.json` are verified on the way back. + +**`purge`** (the only unlink): deletes trash **batches**, never +individual live-store docs; only batches older than `--retention` +(default 30d); dry-run by default; `--execute` required; prints what +it deletes. Accidental-deletion story end-to-end: a mistake must +survive `scan` (evidence-based), `collect --execute` (re-verified, +quarantine only), a retention window, and `purge --execute` before +bytes are actually gone — with `restore` available at every point +until purge. + +### What the tools deliberately do NOT do (v1) + +- Collect anything other than capture-MIME docs (index/file/set/ + unknown docs are inventory-only). +- Run against live servers for mutating operations. +- Reason about **peers**: on a replicated/peered deployment, this + storage location's reference graph is the only truth consulted. + The runbook must say: run against the authoritative store; a doc + collected here can resurface via a peer that still holds it (samod + sync would re-fetch it if referenced — which is exactly the safe + direction; an *unreferenced* doc re-synced from a peer just becomes + a future scan candidate again). +- Compact/GC automerge history inside live docs (different problem). + +### Placement + +- `crates/quarto-hub/src/admin/` — `mod.rs`, `classify.rs` + (shape classification + reference extraction, pure functions over + `Automerge`, unit-testable without storage), `scan.rs`, + `manifest.rs` (serde types, versioned), `collect.rs` (quarantine + + restore + purge). +- `main.rs` grows `#[command(subcommand)]` with the existing flat + serve behavior as the default command, so `hub --port …` keeps + working; `hub admin ` dispatches to the new module. (`quarto + hub admin …` comes along for free.) +- Operator runbook: `claude-notes/instructions/hub-storage-hygiene.md` + (promote to `dev-docs/` if/when deployments want it public). + +### Part B tests (TDD) + +- [ ] Unit (`classify.rs`): each doc kind fixture classifies + correctly; unknown shapes classify as unknown; reference + extraction returns exact id sets for index and project-set docs. +- [ ] Integration (build a real repo via `HubContext` in a temp dir: + project with files + a capture, then re-execute so an orphan + exists): + - `scan` inventory counts match; exactly the orphaned capture is + a candidate; the live capture and all file docs are not. + - unstamped (legacy) capture excluded without + `--include-unstamped`, included with it. + - `collect` dry-run changes nothing on disk (tree hash equal). + - `collect --execute` quarantines only the candidate; reopening + the project via `HubContext` still serves every file and the + live capture. + - re-verification: re-reference the candidate between scan and + collect (write it back into a `captures` sidecar) → skipped. + - `restore` brings chunks back byte-identical (hash check); + subsequent `repo.find` loads the doc. + - `purge` refuses young batches; deletes old ones with + `--execute`. + - lock guard: collect refuses while a `HubContext` holds the + data dir. +- [ ] E2E: on a locally-run `hub` with a real project (create, + execute, re-execute), stop the server, run scan→collect→restore + →purge through the actual binary; record invocations + output in + this plan. + +## Design decisions (reviewed with Carlos, 2026-07-24) + +1. **`hub admin` subcommands** on the existing binary (no separate + `hub-admin` artifact); `quarto hub admin` comes along for free. +2. **30d defaults** for both `--older-than` and `--retention`, both + CLI-configurable (they already are in the design; keeping them + flags, no config-file plumbing in v1). +3. **Live-server `scan` allowed** — read-only, snapshot semantics; no + consistency is guaranteed anywhere in this system anyway (even an + offline store can be behind a peer). Mutating subcommands still + require the lock to be free. +4. **Unreferenced non-capture docs**: report-only in v1; collection of + e.g. abandoned project indexes is a follow-up strand once real + inventories are seen. +5. **`--json` output is in** for `scan`; the manifest schema is + versioned so a future hub-client admin page can consume it. + +## Work items + +- [x] Phase A: metadata envelope (commit 38af41d3) — includes + `read_capture_meta` and the `create_capture_document_at` test + seam for the age gate; `CAPTURE_MIME_TYPE` consolidated into + quarto-hub. +- [x] Phase B1: `classify.rs` + unit tests +- [x] Phase B2: `scan` + manifest + integration tests (commit + e1ed2c3b). Implementation note: whole-store enumeration cannot + use `Storage::load_range([])` — the filesystem adapter splays + doc ids across two path components and load_range returns raw + components; enumeration is an explicit backend-specific step + (`list_doc_ids_filesystem`), per-doc chunk loads stay on the + adapter. Caught by the real-store integration test. +- [x] Phase B3: `collect` (quarantine) + `restore` + `purge` + lock + guard + integration tests (commit 4ffc1f8f). Refinement: the + guard doesn't just *check* hub.lock — mutating subcommands + ACQUIRE the server's exclusive flock for their duration, so a + server also can't start mid-operation. +- [x] Phase B4: CLI wiring (`hub admin`), binary E2E, runbook + (`claude-notes/instructions/hub-storage-hygiene.md`) +- [x] Phase C: `cargo xtask verify --skip-hub-build` passes (Rust-only + change; hub-client/WASM do not depend on quarto-hub) + +## End-to-end verification record (2026-07-24) + +Store built entirely by shipped binaries: `q2 preview` on the knitr +repro records a (meta-stamped) capture; a real +`POST /api/preview/re-execute` supersedes it, orphaning the old doc. +Then the `hub` binary drives the whole pipeline +(`scratchpad/e2e-admin/run.sh`; output inspected): + +1. **Live scan** (server still running, read-only): inventory + `engine-capture 2 / project-index 1 / text-file 1`; exactly the + superseded capture flagged — `22821 bytes hello.qmd` (sourcePath + from the Phase A envelope). `--older-than-days=-1` exercised the + age-gate-disable escape hatch (fresh captures). +2. Store copied out live (preview temp data dirs are deleted on + shutdown — operating on a snapshot mirrors backup-based + maintenance), `hub admin scan --output manifest.json` offline. +3. `collect` dry-run: `WOULD COLLECT`, tree untouched. +4. `collect --execute`: `QUARANTINED` into + `trash/20260724T162747Z-scan081c78a2`, restore command printed. + Post-collect scan: 0 candidates. +5. `restore`: `RESTORED` (hash-verified); scan shows the candidate + again. +6. `purge`: young batch `KEPT` at default retention; + `--retention-days=-1 --execute` → `PURGED`. + +**Bonus validation:** the first E2E attempt ran a stale `q2` binary +(pre-Phase A) whose captures were genuinely unstamped — scan reported +the orphan as `unstamped (pre-envelope), excluded by default` and +collected nothing. The legacy-capture protection gate, validated with +authentic legacy data. diff --git a/crates/quarto-hub-provider/src/execute.rs b/crates/quarto-hub-provider/src/execute.rs index 366b6df45..f1bfdd414 100644 --- a/crates/quarto-hub-provider/src/execute.rs +++ b/crates/quarto-hub-provider/src/execute.rs @@ -41,7 +41,6 @@ use quarto_core::engine::EngineRegistry; use quarto_core::engine::preview_record::{compute_input_qmd, record_capture}; use quarto_core::project::ProjectContext; use quarto_hub::index::{CaptureRef, CaptureState, IndexDocument}; -use quarto_hub::resource::create_binary_document; use quarto_system_runtime::{NativeRuntime, SystemRuntime}; use quarto_trace::EngineCapture; use samod::Repo; @@ -50,13 +49,14 @@ use crate::ProviderError; use crate::consent::ConsentGate; use crate::exec_channel::{BEACON_INTERVAL, ExecMessage, parse_exec_message}; -/// MIME type stamped on capture binary docs. Must stay byte-identical to -/// `quarto_preview::capture_driver::CAPTURE_MIME_TYPE` and the literal the TS -/// consumers use (`ts-packages/quarto-sync-client`, -/// `hub-client/.../ReactPreview.capture.integration.test.tsx`). Duplicated here -/// rather than depending on the heavy `quarto-preview` crate (which pulls in an -/// axum server); the value is a self-describing label, not validated on read. -pub const CAPTURE_MIME_TYPE: &str = "application/x-engine-capture+gzip"; +/// MIME type stamped on capture binary docs. Re-exported from +/// quarto-hub, the single source of truth (bd-eiku4ymo) — the former +/// duplication between here and `quarto-preview` is gone now that +/// quarto-hub (a dependency of both) owns the constant. The TS +/// consumers (`ts-packages/quarto-sync-client`, +/// `hub-client/.../ReactPreview.capture.integration.test.tsx`) hold +/// the same literal. +pub use quarto_hub::resource::CAPTURE_MIME_TYPE; /// Result of a single execution attempt. #[derive(Debug, Clone, PartialEq, Eq)] @@ -302,7 +302,7 @@ impl Provider { return Err("engine produced no capture (no code cells?)".to_string()); } - let new_doc_id = write_capture_doc(&self.repo, &captures) + let new_doc_id = write_capture_doc(&self.repo, rel_path, &captures) .await .map_err(|e| format!("failed to store capture binary doc: {e}"))?; @@ -433,11 +433,16 @@ fn is_safe_relative(rel_path: &str) -> bool { /// `gzip_captures` — serialize + gzip + 10MB size warning, bd-qbhp2cvv). async fn write_capture_doc( repo: &Repo, + rel_path: &str, captures: &[EngineCapture], ) -> Result { let gzipped = quarto_core::engine::capture_files::gzip_captures(captures) .map_err(|e| ProviderError::Protocol(format!("serialize/gzip captures: {e}")))?; - let doc = create_binary_document(&gzipped, CAPTURE_MIME_TYPE) + let meta = quarto_hub::resource::CaptureDocMeta { + source_path: rel_path.to_string(), + engines: captures.iter().map(|c| c.engine_name.clone()).collect(), + }; + let doc = quarto_hub::resource::create_capture_document(&gzipped, &meta) .map_err(|e| ProviderError::Protocol(format!("binary doc: {e}")))?; let handle = repo .create(doc) diff --git a/crates/quarto-hub/Cargo.toml b/crates/quarto-hub/Cargo.toml index 64ea31796..e24b32548 100644 --- a/crates/quarto-hub/Cargo.toml +++ b/crates/quarto-hub/Cargo.toml @@ -94,13 +94,15 @@ hex = "0.4" # MIME type detection from file content infer = "0.19" +# RFC 3339 createdAt stamps on capture-doc meta (bd-eiku4ymo). +chrono = "0.4" + [dev-dependencies] tempfile = "3" # Phase 2 Bearer-auth integration tests: spin up a mock OIDC provider, # mint signed JWTs, exercise the hub router against random ports. rsa = { version = "0.9", features = ["pem", "getrandom"] } base64 = "0.22" -chrono = "0.4" parking_lot = "0.12" [lints] diff --git a/crates/quarto-hub/src/admin/classify.rs b/crates/quarto-hub/src/admin/classify.rs new file mode 100644 index 000000000..c06283bb1 --- /dev/null +++ b/crates/quarto-hub/src/admin/classify.rs @@ -0,0 +1,298 @@ +//! Document classification and reference extraction for `hub admin` +//! (bd-eiku4ymo). +//! +//! Pure functions over loaded [`Automerge`] documents — no storage, +//! no repo — so the safety-critical logic (what a doc *is*, and what +//! it *references*) is unit-testable in isolation. +//! +//! Classification is by ROOT shape, mirroring the schemas in +//! `crate::index` (project index), `ts-packages/quarto-automerge-schema` +//! (project set), and `crate::resource` (text/binary docs): +//! +//! | kind | ROOT signature | +//! |-----------------|-----------------------------------------------| +//! | `ProjectIndex` | `files` map (V2 adds a `captures` sidecar) | +//! | `ProjectSet` | `projects` map | +//! | `TextFile` | `text` | +//! | `EngineCapture` | `content` + `mimeType == CAPTURE_MIME_TYPE` | +//! | `BinaryFile` | `content` (any other MIME) | +//! | `Unknown` | anything else | +//! +//! **`Unknown` is load-bearing for safety**: a doc written by a +//! future schema this scanner doesn't know classifies as `Unknown` +//! and is therefore never collectible (the collector's allowlist +//! admits only `EngineCapture`). Never "improve" classification by +//! guessing. + +use automerge::{Automerge, ObjType, ROOT, ReadDoc}; + +use crate::resource::CAPTURE_MIME_TYPE; + +/// What a stored automerge document is, by ROOT shape. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum DocKind { + /// A project index (`crate::index::IndexDocument`). Always a + /// liveness root. + ProjectIndex, + /// A collections/projects-home set document. Always a liveness + /// root (its only inbound pointers live in client IndexedDB). + ProjectSet, + /// A text file document. + TextFile, + /// An engine-capture binary doc — the only collectible kind. + EngineCapture, + /// Any other binary doc (images, PDFs, fonts, …). + BinaryFile, + /// Unrecognized shape. Never collectible; reported for audit. + Unknown, +} + +impl DocKind { + /// Stable string for manifests and human summaries. + pub fn as_str(&self) -> &'static str { + match self { + DocKind::ProjectIndex => "project-index", + DocKind::ProjectSet => "project-set", + DocKind::TextFile => "text-file", + DocKind::EngineCapture => "engine-capture", + DocKind::BinaryFile => "binary-file", + DocKind::Unknown => "unknown", + } + } + + /// Liveness roots: docs whose inbound pointers live outside the + /// store (share URLs, client IndexedDB), so they must be treated + /// as always-reachable. + pub fn is_root(&self) -> bool { + matches!(self, DocKind::ProjectIndex | DocKind::ProjectSet) + } +} + +/// Classify a document by its ROOT shape. Total: every doc gets a +/// kind, unrecognized shapes get [`DocKind::Unknown`]. +pub fn classify(doc: &Automerge) -> DocKind { + let has = |key: &str| doc.get(ROOT, key).ok().flatten().is_some(); + + if has("files") { + return DocKind::ProjectIndex; + } + if has("projects") { + return DocKind::ProjectSet; + } + if has("text") { + return DocKind::TextFile; + } + if has("content") { + let mime = doc + .get(ROOT, "mimeType") + .ok() + .flatten() + .and_then(|(value, _)| value.to_str().map(str::to_string)); + return if mime.as_deref() == Some(CAPTURE_MIME_TYPE) { + DocKind::EngineCapture + } else { + DocKind::BinaryFile + }; + } + DocKind::Unknown +} + +/// Extract every document id this document references — the edges of +/// the liveness graph. +/// +/// - `ProjectIndex`: values of the `files` map (path → docId) and the +/// `captureDocId` of every `captures` sidecar entry. +/// - `ProjectSet`: keys of the `projects` map (indexDocIds), plus each +/// entry's `indexDocId` field defensively (schema says they match). +/// - Everything else references nothing. +/// +/// Ids are normalized by stripping an `automerge:` prefix when +/// present (the TS schema stores keys without it, but defensiveness +/// is cheap and a missed reference means an unsafe collection). +pub fn referenced_doc_ids(doc: &Automerge) -> Vec { + let mut out = Vec::new(); + match classify(doc) { + DocKind::ProjectIndex => { + if let Some((_, files_obj)) = doc.get(ROOT, "files").ok().flatten() { + let keys: Vec = doc.keys(&files_obj).collect(); + for key in keys { + if let Some(id) = read_str(doc, &files_obj, &key) { + out.push(normalize_id(&id)); + } + } + } + if let Some((_, caps_obj)) = doc.get(ROOT, "captures").ok().flatten() { + let keys: Vec = doc.keys(&caps_obj).collect(); + for key in keys { + if let Some((value, entry_obj)) = doc.get(&caps_obj, &key).ok().flatten() + && matches!(value, automerge::Value::Object(ObjType::Map)) + && let Some(id) = read_str(doc, &entry_obj, "captureDocId") + { + out.push(normalize_id(&id)); + } + } + } + } + DocKind::ProjectSet => { + if let Some((_, projects_obj)) = doc.get(ROOT, "projects").ok().flatten() { + let keys: Vec = doc.keys(&projects_obj).collect(); + for key in keys { + out.push(normalize_id(&key)); + if let Some((value, entry_obj)) = doc.get(&projects_obj, &key).ok().flatten() + && matches!(value, automerge::Value::Object(ObjType::Map)) + && let Some(id) = read_str(doc, &entry_obj, "indexDocId") + { + out.push(normalize_id(&id)); + } + } + } + } + _ => {} + } + out.sort(); + out.dedup(); + out +} + +fn read_str(doc: &Automerge, obj: &automerge::ObjId, key: &str) -> Option { + let (value, _) = doc.get(obj, key).ok().flatten()?; + value.to_str().map(str::to_string) +} + +fn normalize_id(id: &str) -> String { + id.strip_prefix("automerge:").unwrap_or(id).to_string() +} + +#[cfg(test)] +mod tests { + use super::*; + use automerge::transaction::Transactable; + + use crate::resource::{CaptureDocMeta, create_binary_document, create_capture_document}; + + fn index_doc(files: &[(&str, &str)], captures: &[(&str, &str)]) -> Automerge { + let mut doc = Automerge::new(); + doc.transact::<_, _, automerge::AutomergeError>(|tx| { + let files_obj = tx.put_object(ROOT, "files", ObjType::Map)?; + for (path, id) in files { + tx.put(&files_obj, *path, *id)?; + } + if !captures.is_empty() { + let caps_obj = tx.put_object(ROOT, "captures", ObjType::Map)?; + for (path, id) in captures { + let entry = tx.put_object(&caps_obj, *path, ObjType::Map)?; + tx.put(&entry, "captureDocId", *id)?; + tx.put(&entry, "staleness", false)?; + } + } + Ok(()) + }) + .unwrap(); + doc + } + + fn project_set_doc(entries: &[(&str, &str)]) -> Automerge { + let mut doc = Automerge::new(); + doc.transact::<_, _, automerge::AutomergeError>(|tx| { + let projects_obj = tx.put_object(ROOT, "projects", ObjType::Map)?; + for (key, index_id) in entries { + let entry = tx.put_object(&projects_obj, *key, ObjType::Map)?; + tx.put(&entry, "indexDocId", *index_id)?; + tx.put(&entry, "name", "A project")?; + } + Ok(()) + }) + .unwrap(); + doc + } + + fn text_doc() -> Automerge { + let mut doc = Automerge::new(); + doc.transact::<_, _, automerge::AutomergeError>(|tx| { + let text_obj = tx.put_object(ROOT, "text", ObjType::Text)?; + tx.update_text(&text_obj, "hello")?; + Ok(()) + }) + .unwrap(); + doc + } + + fn capture_doc() -> Automerge { + create_capture_document( + b"gz", + &CaptureDocMeta { + source_path: "a.qmd".into(), + engines: vec!["knitr".into()], + }, + ) + .unwrap() + } + + #[test] + fn classifies_every_known_kind() { + assert_eq!(classify(&index_doc(&[], &[])), DocKind::ProjectIndex); + assert_eq!(classify(&project_set_doc(&[])), DocKind::ProjectSet); + assert_eq!(classify(&text_doc()), DocKind::TextFile); + assert_eq!(classify(&capture_doc()), DocKind::EngineCapture); + assert_eq!( + classify(&create_binary_document(b"png", "image/png").unwrap()), + DocKind::BinaryFile + ); + assert_eq!(classify(&Automerge::new()), DocKind::Unknown); + } + + #[test] + fn legacy_capture_without_meta_still_classifies_by_mime() { + // Pre-envelope capture docs carry only the MIME discriminator. + let doc = create_binary_document(b"gz", CAPTURE_MIME_TYPE).unwrap(); + assert_eq!(classify(&doc), DocKind::EngineCapture); + } + + #[test] + fn unknown_shape_is_never_a_root_and_never_collectible_kind() { + // A future schema (here: a doc with only a `widgets` map) must + // classify as Unknown — the collector's allowlist then protects + // it automatically. + let mut doc = Automerge::new(); + doc.transact::<_, _, automerge::AutomergeError>(|tx| { + tx.put_object(ROOT, "widgets", ObjType::Map)?; + Ok(()) + }) + .unwrap(); + assert_eq!(classify(&doc), DocKind::Unknown); + assert!(!classify(&doc).is_root()); + } + + #[test] + fn roots_are_index_and_project_set() { + assert!(DocKind::ProjectIndex.is_root()); + assert!(DocKind::ProjectSet.is_root()); + assert!(!DocKind::TextFile.is_root()); + assert!(!DocKind::EngineCapture.is_root()); + assert!(!DocKind::BinaryFile.is_root()); + } + + #[test] + fn index_references_files_and_capture_sidecar() { + let doc = index_doc( + &[("a.qmd", "docA"), ("img.png", "docB")], + &[("a.qmd", "capA")], + ); + assert_eq!(referenced_doc_ids(&doc), vec!["capA", "docA", "docB"]); + } + + #[test] + fn project_set_references_index_ids_from_keys_and_entries() { + // Entry value disagreeing with its key: both are collected + // (defensive — a missed reference is an unsafe collection). + let doc = project_set_doc(&[("idx1", "idx1"), ("idx2", "automerge:idx2b")]); + assert_eq!(referenced_doc_ids(&doc), vec!["idx1", "idx2", "idx2b"]); + } + + #[test] + fn leaf_docs_reference_nothing() { + assert!(referenced_doc_ids(&text_doc()).is_empty()); + assert!(referenced_doc_ids(&capture_doc()).is_empty()); + assert!(referenced_doc_ids(&Automerge::new()).is_empty()); + } +} diff --git a/crates/quarto-hub/src/admin/collect.rs b/crates/quarto-hub/src/admin/collect.rs new file mode 100644 index 000000000..190143924 --- /dev/null +++ b/crates/quarto-hub/src/admin/collect.rs @@ -0,0 +1,467 @@ +//! `hub admin collect` / `restore` / `purge` — the mutating maintainer +//! tools (bd-eiku4ymo). +//! +//! Safety model (see `admin` module docs + the plan): `collect` +//! consumes **only** a scan manifest, re-verifies every candidate +//! against *current* storage, and **quarantines** — it renames each +//! doc's chunk directory into `/trash//docs/` +//! and never unlinks anything. `restore` moves a batch back, +//! hash-verified. `purge` is the only unlink in the whole tool set: +//! whole trash batches, past a retention window, dry-run by default. +//! +//! All three acquire the server's own `hub.lock` (exclusive flock) +//! for their duration — a running server makes them refuse, and they +//! symmetrically prevent a server from starting mid-operation. +//! +//! The trash area lives inside the data dir on purpose: renames stay +//! on one filesystem (atomic-ish, no copy), the operator's existing +//! backup regime covers it, and a batch can never be orphaned from +//! its server. + +use std::fs::File; +use std::path::{Path, PathBuf}; + +use fs2::FileExt as _; +use serde::{Deserialize, Serialize}; +use sha2::{Digest as _, Sha256}; + +use super::classify::DocKind; +use super::manifest::{CandidateDoc, MANIFEST_VERSION, ScanManifest}; +use super::scan::{ScanOptions, list_doc_ids_filesystem, live_doc_ids, load_all_docs}; +use crate::resource::read_capture_meta; + +/// Version of the `batch.json` schema written into quarantine batches. +pub const BATCH_VERSION: u32 = 1; + +/// Result of a collect run (dry or executed). +#[derive(Debug)] +pub struct CollectOutcome { + /// Candidates that passed re-verification (quarantined when + /// `execute`; the plan when dry-run). + pub verified: Vec, + /// Candidates skipped at re-verification, with reasons. + pub skipped: Vec<(String, String)>, + /// The batch directory (only when `execute` and something moved). + pub batch_dir: Option, +} + +/// `batch.json`: the audit record embedded in every quarantine batch. +#[derive(Debug, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct BatchRecord { + pub batch_version: u32, + pub tool: String, + /// RFC 3339 UTC creation time — `purge`'s retention gate reads this. + pub created_at: String, + pub data_dir: String, + /// The scan manifest that authorized this batch, embedded whole. + pub manifest: ScanManifest, + /// Per-doc quarantine records. + pub docs: Vec, + /// Candidates skipped at re-verification (kept for the audit trail). + pub skipped: Vec, +} + +#[derive(Debug, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct BatchDoc { + pub doc_id: String, + /// Chunk files moved, relative to the doc's directory, with + /// SHA-256 hashes — `restore` verifies these on the way back. + pub chunks: Vec, +} + +#[derive(Debug, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct BatchChunk { + pub rel_path: String, + pub sha256: String, + pub bytes: u64, +} + +#[derive(Debug, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct BatchSkip { + pub doc_id: String, + pub reason: String, +} + +/// Hold the server's own exclusive lock for the duration of a +/// mutating operation. Refuses (with a clear message) when a server — +/// or another admin operation — holds it. +pub struct AdminLock { + _file: File, +} + +impl AdminLock { + pub fn acquire(data_dir: &Path) -> Result { + let lock_path = data_dir.join("hub.lock"); + let file = File::options() + .create(true) + .truncate(false) + .write(true) + .open(&lock_path) + .map_err(|e| format!("cannot open {}: {e}", lock_path.display()))?; + file.try_lock_exclusive().map_err(|_| { + format!( + "{} is locked — a hub server (or another admin command) is running \ + against this data dir. Stop it first; mutating admin operations \ + never run under a live server.", + lock_path.display() + ) + })?; + Ok(Self { _file: file }) + } +} + +/// The splayed on-disk directory of a doc's chunks: +/// `//` (samod's `key_to_path`). +fn doc_dir(automerge_dir: &Path, doc_id: &str) -> PathBuf { + let first_two: String = doc_id.chars().take(2).collect(); + let rest: String = doc_id.chars().skip(2).collect(); + automerge_dir.join(first_two).join(rest) +} + +/// Re-verify + quarantine. Dry-run unless `execute`. +/// +/// `data_dir` is the hub data dir (contains `automerge/`, `hub.lock`, +/// and — after this — `trash/`). +pub async fn collect( + data_dir: &Path, + manifest: &ScanManifest, + execute: bool, +) -> Result { + if manifest.manifest_version != MANIFEST_VERSION { + return Err(format!( + "manifest version {} is not supported by this tool (expected {})", + manifest.manifest_version, MANIFEST_VERSION + )); + } + // Paranoia gate: a manifest scanned elsewhere must not authorize + // collection here. + let canonical = data_dir + .canonicalize() + .map_err(|e| format!("cannot canonicalize {}: {e}", data_dir.display()))?; + let manifest_dir = Path::new(&manifest.data_dir) + .canonicalize() + .unwrap_or_else(|_| PathBuf::from(&manifest.data_dir)); + if manifest_dir != canonical { + return Err(format!( + "manifest was scanned from {:?} but --data-dir is {:?}; refusing \ + (re-scan this data dir to produce a matching manifest)", + manifest.data_dir, canonical + )); + } + + let _lock = AdminLock::acquire(&canonical)?; + let automerge_dir = canonical.join("automerge"); + + // Re-verification (principle 5): recompute kind + liveness + age + // against CURRENT storage, with the manifest's own gate options. + let storage = samod::storage::TokioFilesystemStorage::new(&automerge_dir); + let doc_ids = list_doc_ids_filesystem(&automerge_dir); + let docs = load_all_docs(&storage, &doc_ids).await; + let live = live_doc_ids(&docs); + let opts = ScanOptions { + older_than_days: manifest.older_than_days, + include_unstamped: manifest.include_unstamped, + }; + let now = chrono::Utc::now(); + + let mut verified = Vec::new(); + let mut skipped = Vec::new(); + for candidate in &manifest.candidates { + let reason = verify_candidate(candidate, &docs, &live, &opts, now); + match reason { + Ok(()) => verified.push(candidate.clone()), + Err(why) => skipped.push((candidate.doc_id.clone(), why)), + } + } + + if !execute || verified.is_empty() { + return Ok(CollectOutcome { + verified, + skipped, + batch_dir: None, + }); + } + + // Quarantine: one rename per doc directory, hashes recorded first. + let batch_name = format!( + "{}-scan{}", + now.format("%Y%m%dT%H%M%SZ"), + short_hash(&manifest.scanned_at) + ); + let batch_dir = canonical.join("trash").join(&batch_name); + let batch_docs_dir = batch_dir.join("docs"); + std::fs::create_dir_all(&batch_docs_dir) + .map_err(|e| format!("cannot create {}: {e}", batch_docs_dir.display()))?; + + let mut batch_docs = Vec::new(); + for candidate in &verified { + let src = doc_dir(&automerge_dir, &candidate.doc_id); + let chunks = + hash_dir_files(&src).map_err(|e| format!("hashing {} failed: {e}", src.display()))?; + let dst = batch_docs_dir.join(&candidate.doc_id); + std::fs::rename(&src, &dst).map_err(|e| { + format!( + "quarantine rename {} -> {} failed: {e} (already-moved docs remain \ + quarantined in {})", + src.display(), + dst.display(), + batch_dir.display() + ) + })?; + // Best-effort: drop the now-possibly-empty 2-char splay dir. + if let Some(parent) = src.parent() { + let _ = std::fs::remove_dir(parent); // fails (kept) unless empty + } + batch_docs.push(BatchDoc { + doc_id: candidate.doc_id.clone(), + chunks, + }); + } + + let record = BatchRecord { + batch_version: BATCH_VERSION, + tool: format!("hub admin collect {}", env!("CARGO_PKG_VERSION")), + created_at: now.to_rfc3339(), + data_dir: canonical.to_string_lossy().into_owned(), + manifest: manifest.clone(), + docs: batch_docs, + skipped: skipped + .iter() + .map(|(doc_id, reason)| BatchSkip { + doc_id: doc_id.clone(), + reason: reason.clone(), + }) + .collect(), + }; + let record_json = + serde_json::to_vec_pretty(&record).map_err(|e| format!("serializing batch.json: {e}"))?; + std::fs::write(batch_dir.join("batch.json"), record_json) + .map_err(|e| format!("writing batch.json: {e}"))?; + + Ok(CollectOutcome { + verified, + skipped, + batch_dir: Some(batch_dir), + }) +} + +/// One candidate's re-verification. `Ok(())` = still safely removable. +fn verify_candidate( + candidate: &CandidateDoc, + docs: &[super::scan::LoadedDoc], + live: &std::collections::HashSet, + opts: &ScanOptions, + now: chrono::DateTime, +) -> Result<(), String> { + let Some(current) = docs.iter().find(|d| d.doc_id == candidate.doc_id) else { + return Err("no longer exists in storage".to_string()); + }; + if current.kind != DocKind::EngineCapture { + return Err(format!( + "kind is now {:?}, not engine-capture", + current.kind.as_str() + )); + } + if live.contains(&candidate.doc_id) { + return Err("became referenced since the scan".to_string()); + } + let created_at = read_capture_meta(¤t.doc).and_then(|m| m.created_at); + match created_at { + Some(ts) => { + let t = chrono::DateTime::parse_from_rfc3339(&ts) + .map_err(|e| format!("unparseable createdAt {ts:?}: {e}"))?; + let age_days = (now - t.with_timezone(&chrono::Utc)).num_days(); + if age_days <= opts.older_than_days { + return Err(format!( + "age {age_days}d no longer passes the {}d gate", + opts.older_than_days + )); + } + } + None => { + if !opts.include_unstamped { + return Err("unstamped and manifest did not include unstamped".to_string()); + } + } + } + Ok(()) +} + +/// Restore a quarantine batch (or a subset of its docs) back into the +/// automerge store. Hash-verified; refuses per-doc when the target +/// directory already exists (a doc re-created under the same id). +pub fn restore( + data_dir: &Path, + batch_dir: &Path, + only_doc_ids: &[String], +) -> Result)>, String> { + let canonical = data_dir + .canonicalize() + .map_err(|e| format!("cannot canonicalize {}: {e}", data_dir.display()))?; + let _lock = AdminLock::acquire(&canonical)?; + let automerge_dir = canonical.join("automerge"); + + let record: BatchRecord = serde_json::from_slice( + &std::fs::read(batch_dir.join("batch.json")) + .map_err(|e| format!("reading batch.json: {e}"))?, + ) + .map_err(|e| format!("parsing batch.json: {e}"))?; + + let mut results = Vec::new(); + for doc in &record.docs { + if !only_doc_ids.is_empty() && !only_doc_ids.contains(&doc.doc_id) { + continue; + } + results.push(( + doc.doc_id.clone(), + restore_one(&automerge_dir, batch_dir, doc), + )); + } + Ok(results) +} + +fn restore_one(automerge_dir: &Path, batch_dir: &Path, doc: &BatchDoc) -> Result<(), String> { + let src = batch_dir.join("docs").join(&doc.doc_id); + if !src.is_dir() { + return Err("not present in this batch (already restored?)".to_string()); + } + // Verify hashes BEFORE moving anything back. + let current = hash_dir_files(&src).map_err(|e| format!("hashing quarantined chunks: {e}"))?; + let expected: std::collections::BTreeMap<&str, &str> = doc + .chunks + .iter() + .map(|c| (c.rel_path.as_str(), c.sha256.as_str())) + .collect(); + for chunk in ¤t { + match expected.get(chunk.rel_path.as_str()) { + Some(hash) if *hash == chunk.sha256 => {} + Some(_) => { + return Err(format!( + "chunk {} hash mismatch — quarantined bytes were modified; refusing", + chunk.rel_path + )); + } + None => { + return Err(format!( + "chunk {} not in batch record — refusing", + chunk.rel_path + )); + } + } + } + let dst = doc_dir(automerge_dir, &doc.doc_id); + if dst.exists() { + return Err(format!( + "{} already exists (doc re-created since collection); refusing to overwrite", + dst.display() + )); + } + if let Some(parent) = dst.parent() { + std::fs::create_dir_all(parent).map_err(|e| format!("mkdir {}: {e}", parent.display()))?; + } + std::fs::rename(&src, &dst).map_err(|e| format!("restore rename failed: {e}")) +} + +/// A trash batch eligible (or not) for purging. +#[derive(Debug)] +pub struct PurgeCandidate { + pub batch_dir: PathBuf, + pub created_at: Option, + pub age_days: Option, + pub eligible: bool, +} + +/// List trash batches and, with `execute`, delete those older than +/// `retention_days`. THE ONLY UNLINK IN THE TOOL SET. Batch-level +/// only — never individual docs, never anything outside `trash/`. +pub fn purge( + data_dir: &Path, + retention_days: i64, + execute: bool, +) -> Result, String> { + let canonical = data_dir + .canonicalize() + .map_err(|e| format!("cannot canonicalize {}: {e}", data_dir.display()))?; + let _lock = AdminLock::acquire(&canonical)?; + let trash = canonical.join("trash"); + let now = chrono::Utc::now(); + + let mut out = Vec::new(); + let Ok(entries) = std::fs::read_dir(&trash) else { + return Ok(out); // no trash dir: nothing to purge + }; + for entry in entries.flatten() { + let batch_dir = entry.path(); + if !batch_dir.is_dir() { + continue; + } + let created_at: Option = std::fs::read(batch_dir.join("batch.json")) + .ok() + .and_then(|bytes| serde_json::from_slice::(&bytes).ok()) + .map(|r| r.created_at); + let age_days = created_at.as_deref().and_then(|ts| { + chrono::DateTime::parse_from_rfc3339(ts) + .ok() + .map(|t| (now - t.with_timezone(&chrono::Utc)).num_days()) + }); + // A batch with no readable batch.json has no age evidence: + // never eligible (same protective stance as unstamped captures). + let eligible = age_days.is_some_and(|d| d > retention_days); + if eligible && execute { + std::fs::remove_dir_all(&batch_dir) + .map_err(|e| format!("purging {}: {e}", batch_dir.display()))?; + } + out.push(PurgeCandidate { + batch_dir, + created_at, + age_days, + eligible, + }); + } + Ok(out) +} + +fn short_hash(input: &str) -> String { + let mut hasher = Sha256::new(); + hasher.update(input.as_bytes()); + hex::encode(hasher.finalize())[..8].to_string() +} + +/// Hash every regular file under `dir` (recursively), keyed by +/// forward-slash relative path. +fn hash_dir_files(dir: &Path) -> std::io::Result> { + fn walk(root: &Path, dir: &Path, out: &mut Vec) -> std::io::Result<()> { + for entry in std::fs::read_dir(dir)? { + let entry = entry?; + let path = entry.path(); + if path.is_dir() { + walk(root, &path, out)?; + } else if path.is_file() { + let bytes = std::fs::read(&path)?; + let mut hasher = Sha256::new(); + hasher.update(&bytes); + let rel = path + .strip_prefix(root) + .expect("walk stays under root") + .components() + .map(|c| c.as_os_str().to_string_lossy().into_owned()) + .collect::>() + .join("/"); + out.push(BatchChunk { + rel_path: rel, + sha256: hex::encode(hasher.finalize()), + bytes: bytes.len() as u64, + }); + } + } + Ok(()) + } + let mut out = Vec::new(); + walk(dir, dir, &mut out)?; + out.sort_by(|a, b| a.rel_path.cmp(&b.rel_path)); + Ok(out) +} diff --git a/crates/quarto-hub/src/admin/manifest.rs b/crates/quarto-hub/src/admin/manifest.rs new file mode 100644 index 000000000..3ab001130 --- /dev/null +++ b/crates/quarto-hub/src/admin/manifest.rs @@ -0,0 +1,93 @@ +//! Versioned scan-manifest types (`hub admin scan --output`, +//! bd-eiku4ymo). +//! +//! The manifest is the **only** input `hub admin collect` accepts — +//! never ad-hoc doc ids — so its shape is a contract: versioned, +//! self-describing (tool, timestamp, data dir), and carrying enough +//! per-doc evidence that a human can audit *why* each candidate was +//! deemed removable. Field names are camelCase on the wire so a +//! future hub-client admin page can consume the same JSON. + +use serde::{Deserialize, Serialize}; +use std::collections::BTreeMap; + +/// Current manifest schema version. Bump on any incompatible change; +/// `collect` refuses manifests with a version it doesn't know. +pub const MANIFEST_VERSION: u32 = 1; + +/// Top-level scan output. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct ScanManifest { + /// Schema version of this manifest ([`MANIFEST_VERSION`]). + pub manifest_version: u32, + /// Tool + crate version that produced this manifest. + pub tool: String, + /// RFC 3339 UTC timestamp of the scan. + pub scanned_at: String, + /// The `--data-dir` the scan ran against (canonicalized). + /// `collect` refuses a manifest whose dataDir doesn't match its + /// own — pointing yesterday's manifest at a different server must + /// fail loudly. + pub data_dir: String, + /// Age gate that was in effect, in days (candidates are older). + pub older_than_days: i64, + /// Whether unstamped (pre-envelope) captures were eligible. + pub include_unstamped: bool, + /// Counts of every doc kind seen, keyed by [`DocKind::as_str`] + /// (BTreeMap for stable JSON output). + pub inventory: BTreeMap, + /// Docs the scan determined are safely removable. + pub candidates: Vec, + /// Unreferenced docs that are NOT collectible in v1 (non-capture + /// kinds, unknown shapes, too-young or unstamped captures). + /// Informational: these are the follow-up-strand material. + pub not_collectible: Vec, +} + +/// Per-kind inventory counters. +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct KindStats { + pub count: usize, + pub bytes: u64, +} + +/// A doc the scan deems safely removable. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CandidateDoc { + pub doc_id: String, + /// Always `"engine-capture"` in v1 (the collector re-checks). + pub kind: String, + /// Total bytes of this doc's storage chunks. + pub size_bytes: u64, + /// The uncompressed audit meta (absent on legacy captures — which + /// are only candidates under `--include-unstamped`). + #[serde(skip_serializing_if = "Option::is_none")] + pub meta: Option, + /// Human-readable evidence for why this doc is removable. + pub reason: String, +} + +/// The capture-doc `meta` envelope as carried in the manifest. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CandidateMeta { + #[serde(skip_serializing_if = "Option::is_none")] + pub created_at: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub source_path: Option, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub engines: Vec, +} + +/// An unreferenced-but-protected doc, reported for visibility. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct ReportedDoc { + pub doc_id: String, + pub kind: String, + pub size_bytes: u64, + pub reason: String, +} diff --git a/crates/quarto-hub/src/admin/mod.rs b/crates/quarto-hub/src/admin/mod.rs new file mode 100644 index 000000000..1b2719e10 --- /dev/null +++ b/crates/quarto-hub/src/admin/mod.rs @@ -0,0 +1,21 @@ +//! Sync-server maintainer tools (`hub admin …`, bd-eiku4ymo). +//! +//! Design: `claude-notes/plans/2026-07-24-capture-meta-and-hub-admin-tools.md`. +//! +//! The safety model is a pipeline of independent gates — a mistake +//! must survive all of them before bytes are gone: +//! +//! 1. [`scan`] is read-only and evidence-based; it emits a versioned +//! manifest of orphaned engine-capture docs (allowlist: nothing +//! else is ever collectible) plus a full inventory. +//! 2. `collect` consumes only a manifest, re-verifies every candidate +//! against current storage, and **quarantines** (renames into +//! `/trash//`) — it never unlinks. +//! 3. `restore` moves a batch back, hash-verified. +//! 4. `purge` is the only unlink: whole trash batches, past a +//! retention window, dry-run by default. + +pub mod classify; +pub mod collect; +pub mod manifest; +pub mod scan; diff --git a/crates/quarto-hub/src/admin/scan.rs b/crates/quarto-hub/src/admin/scan.rs new file mode 100644 index 000000000..e09ad227e --- /dev/null +++ b/crates/quarto-hub/src/admin/scan.rs @@ -0,0 +1,574 @@ +//! `hub admin scan` — read-only orphan analysis of a samod storage +//! location (bd-eiku4ymo). +//! +//! Enumerates every stored document through the same [`Storage`] +//! adapter the server uses, classifies each by shape +//! ([`classify`](super::classify)), computes liveness as the closure +//! of the reference graph from the roots (project indexes + project +//! sets), and reports orphaned engine-capture docs past the age gate +//! as removable candidates. +//! +//! Safe to run against a live server: strictly read-only, snapshot +//! semantics. (No stronger consistency exists anywhere in this system +//! — an offline store can equally be behind a peer.) The collector +//! re-verifies against current storage before acting, so scan-time +//! staleness cannot cause an unsafe collection. + +use std::collections::{BTreeMap, HashMap, HashSet}; + +use automerge::Automerge; +use samod::storage::{Storage, StorageKey}; + +use super::classify::{DocKind, classify, referenced_doc_ids}; +use super::manifest::{ + CandidateDoc, CandidateMeta, KindStats, MANIFEST_VERSION, ReportedDoc, ScanManifest, +}; +use crate::resource::read_capture_meta; + +/// Options for a scan run. +#[derive(Debug, Clone)] +pub struct ScanOptions { + /// Minimum age (days since `meta.createdAt`) for a capture to be + /// collectible. Decision (2026-07-24): default 30. + pub older_than_days: i64, + /// Whether captures without a `meta` envelope (pre-bd-eiku4ymo) + /// are eligible. Default false: no timestamp means no age + /// evidence, so they are protected unless the operator opts in. + pub include_unstamped: bool, +} + +impl Default for ScanOptions { + fn default() -> Self { + Self { + older_than_days: 30, + include_unstamped: false, + } + } +} + +/// One stored document, loaded and classified. +#[derive(Debug)] +pub struct LoadedDoc { + pub doc_id: String, + pub kind: DocKind, + pub doc: Automerge, + /// Sum of this doc's chunk sizes on storage. + pub size_bytes: u64, + /// Number of chunks that failed to parse (fail-soft, mirroring + /// samod's own loader). A doc whose every chunk fails loads empty + /// and classifies as Unknown — i.e. protected. + pub bad_chunks: usize, +} + +/// Enumerate document ids from a **filesystem** samod store. +/// +/// Enumeration is deliberately NOT done through +/// `Storage::load_range([])`: samod's filesystem adapter splays a +/// key's first component across two path segments +/// (`key_to_path`), and its `load_range` rebuilds keys from raw path +/// components — so an empty-prefix listing returns the doc id split +/// in half. samod itself only ever uses per-doc prefixes (where the +/// mapping round-trips); whole-store enumeration is simply outside +/// the `Storage` contract. So we enumerate by mirroring the splay: +/// doc id = `` + ``. Per-doc +/// chunk loading then goes through the adapter (`load_range([id])`), +/// which is correct on every backend. +/// +/// The adapter's own identity record (`st/orage-adapter-id`) is a +/// *file* at level 2, not a directory, so the `is_dir` requirement +/// skips it structurally. +pub fn list_doc_ids_filesystem(automerge_dir: &std::path::Path) -> Vec { + let mut out = Vec::new(); + let Ok(level1) = std::fs::read_dir(automerge_dir) else { + return out; + }; + for l1 in level1.flatten() { + if !l1.path().is_dir() { + continue; + } + let Some(prefix) = l1.file_name().to_str().map(str::to_string) else { + continue; + }; + let Ok(level2) = std::fs::read_dir(l1.path()) else { + continue; + }; + for l2 in level2.flatten() { + if !l2.path().is_dir() { + continue; + } + if let Some(rest) = l2.file_name().to_str() { + out.push(format!("{prefix}{rest}")); + } + } + } + out.sort(); + out +} + +/// Load and classify the given documents through the storage adapter. +/// Shared by scan and by the collector's re-verification pass. +/// +/// Enumeration is the caller's job (backend-specific — see +/// [`list_doc_ids_filesystem`]); chunk loading uses per-doc +/// `load_range`, which round-trips correctly on every adapter. +pub async fn load_all_docs(storage: &S, doc_ids: &[String]) -> Vec { + let mut out = Vec::new(); + for doc_id in doc_ids { + let Ok(prefix) = StorageKey::from_parts([doc_id.as_str()]) else { + tracing::warn!(doc_id = %doc_id, "scan: id not a valid storage key; skipping"); + continue; + }; + let chunk_map = storage.load_range(prefix).await; + let mut chunks: Vec<(Vec, Vec)> = chunk_map + .into_iter() + .map(|(key, bytes)| ((&key).into_iter().cloned().collect(), bytes)) + .collect(); + if chunks.is_empty() { + continue; + } + // Snapshots load before incrementals (samod's own order); + // then key order, for deterministic runs. + chunks.sort_by_key(|(parts, _)| { + (parts.get(1).is_none_or(|s| s != "snapshot"), parts.clone()) + }); + let mut doc = Automerge::new(); + let mut size_bytes = 0u64; + let mut bad_chunks = 0usize; + for (parts, bytes) in &chunks { + size_bytes += bytes.len() as u64; + if let Err(e) = doc.load_incremental(bytes) { + bad_chunks += 1; + tracing::warn!( + doc_id = %doc_id, + key = %parts.join("/"), + error = %e, + "scan: bad storage chunk (doc may classify as unknown)" + ); + } + } + let kind = classify(&doc); + out.push(LoadedDoc { + doc_id: doc_id.clone(), + kind, + doc, + size_bytes, + bad_chunks, + }); + } + out +} + +/// Compute the live set: closure of the reference graph from the +/// roots (project indexes + project sets). Today only roots have +/// outgoing references, but the closure is a proper worklist so a +/// future referencing kind stays correct automatically. +pub fn live_doc_ids(docs: &[LoadedDoc]) -> HashSet { + let by_id: HashMap<&str, &LoadedDoc> = docs.iter().map(|d| (d.doc_id.as_str(), d)).collect(); + + let mut live: HashSet = HashSet::new(); + let mut worklist: Vec<&LoadedDoc> = docs.iter().filter(|d| d.kind.is_root()).collect(); + for d in &worklist { + live.insert(d.doc_id.clone()); + } + while let Some(d) = worklist.pop() { + for referenced in referenced_doc_ids(&d.doc) { + if live.insert(referenced.clone()) + && let Some(next) = by_id.get(referenced.as_str()) + { + worklist.push(next); + } + } + } + live +} + +/// Whether a capture doc passes the age gate. Returns +/// `(passes, evidence-string)`. +fn age_gate( + meta_created_at: Option<&str>, + now: chrono::DateTime, + opts: &ScanOptions, +) -> (bool, String) { + match meta_created_at { + Some(created_at) => match chrono::DateTime::parse_from_rfc3339(created_at) { + Ok(t) => { + let age_days = (now - t.with_timezone(&chrono::Utc)).num_days(); + ( + age_days > opts.older_than_days, + format!("age {age_days}d vs gate {}d", opts.older_than_days), + ) + } + Err(_) => (false, format!("unparseable createdAt {created_at:?}")), + }, + None => ( + opts.include_unstamped, + if opts.include_unstamped { + "unstamped (pre-envelope), included by --include-unstamped".to_string() + } else { + "unstamped (pre-envelope), excluded by default".to_string() + }, + ), + } +} + +/// Run a scan and produce the manifest. +/// +/// `data_dir` is recorded (canonicalized by the caller) so `collect` +/// can refuse a manifest pointed at the wrong server. +pub async fn scan( + storage: &S, + doc_ids: &[String], + data_dir: &str, + opts: &ScanOptions, +) -> ScanManifest { + let docs = load_all_docs(storage, doc_ids).await; + let live = live_doc_ids(&docs); + let now = chrono::Utc::now(); + + let mut inventory: BTreeMap = BTreeMap::new(); + let mut candidates = Vec::new(); + let mut not_collectible = Vec::new(); + + for d in &docs { + let stats = inventory.entry(d.kind.as_str().to_string()).or_default(); + stats.count += 1; + stats.bytes += d.size_bytes; + + if live.contains(&d.doc_id) { + continue; + } + // Unreferenced. Only engine captures are ever collectible + // (allowlist); everything else is reported. + if d.kind != DocKind::EngineCapture { + not_collectible.push(ReportedDoc { + doc_id: d.doc_id.clone(), + kind: d.kind.as_str().to_string(), + size_bytes: d.size_bytes, + reason: format!( + "unreferenced, but kind {:?} is not collectible in v1", + d.kind.as_str() + ), + }); + continue; + } + let meta = read_capture_meta(&d.doc); + let created_at = meta.as_ref().and_then(|m| m.created_at.clone()); + let (passes, age_evidence) = age_gate(created_at.as_deref(), now, opts); + let manifest_meta = meta.map(|m| CandidateMeta { + created_at: m.created_at, + source_path: m.source_path, + engines: m.engines, + }); + if passes { + candidates.push(CandidateDoc { + doc_id: d.doc_id.clone(), + kind: d.kind.as_str().to_string(), + size_bytes: d.size_bytes, + meta: manifest_meta, + reason: format!( + "capture MIME; not referenced by any project index, captures \ + sidecar, or project set; {age_evidence}" + ), + }); + } else { + not_collectible.push(ReportedDoc { + doc_id: d.doc_id.clone(), + kind: d.kind.as_str().to_string(), + size_bytes: d.size_bytes, + reason: format!("unreferenced capture, but {age_evidence}"), + }); + } + } + + ScanManifest { + manifest_version: MANIFEST_VERSION, + tool: format!("hub admin scan {}", env!("CARGO_PKG_VERSION")), + scanned_at: now.to_rfc3339(), + data_dir: data_dir.to_string(), + older_than_days: opts.older_than_days, + include_unstamped: opts.include_unstamped, + inventory, + candidates, + not_collectible, + } +} + +#[cfg(test)] +mod tests { + use super::*; + use automerge::transaction::Transactable; + use automerge::{ObjType, ROOT}; + use samod::storage::InMemoryStorage; + + use crate::resource::{ + CAPTURE_MIME_TYPE, CaptureDocMeta, create_binary_document, create_capture_document, + create_capture_document_at, + }; + + /// Store `doc` as a single snapshot chunk under samod's key + /// layout. (Scan treats doc ids as opaque strings, so tests can + /// use readable ids.) + async fn put_doc(storage: &InMemoryStorage, doc_id: &str, doc: &mut Automerge) { + let key = StorageKey::from_parts([doc_id, "snapshot", "h1"]).unwrap(); + storage.put(key, doc.save()).await; + } + + fn index_doc(files: &[(&str, &str)], captures: &[(&str, &str)]) -> Automerge { + let mut doc = Automerge::new(); + doc.transact::<_, _, automerge::AutomergeError>(|tx| { + let files_obj = tx.put_object(ROOT, "files", ObjType::Map)?; + for (path, id) in files { + tx.put(&files_obj, *path, *id)?; + } + if !captures.is_empty() { + let caps_obj = tx.put_object(ROOT, "captures", ObjType::Map)?; + for (path, id) in captures { + let entry = tx.put_object(&caps_obj, *path, ObjType::Map)?; + tx.put(&entry, "captureDocId", *id)?; + } + } + Ok(()) + }) + .unwrap(); + doc + } + + fn old_capture() -> Automerge { + create_capture_document_at( + b"gz", + &CaptureDocMeta { + source_path: "a.qmd".into(), + engines: vec!["knitr".into()], + }, + "2020-01-01T00:00:00+00:00", + ) + .unwrap() + } + + /// A storage with one project: index → {text file, live capture}, + /// plus one ORPHANED old capture and one legacy (unstamped) + /// orphaned capture. + async fn fixture_storage() -> (InMemoryStorage, Vec) { + let storage = InMemoryStorage::new(); + put_doc( + &storage, + "idx1", + &mut index_doc(&[("a.qmd", "fileA")], &[("a.qmd", "capLive")]), + ) + .await; + let mut text = Automerge::new(); + text.transact::<_, _, automerge::AutomergeError>(|tx| { + let t = tx.put_object(ROOT, "text", ObjType::Text)?; + tx.update_text(&t, "hello")?; + Ok(()) + }) + .unwrap(); + put_doc(&storage, "fileA", &mut text).await; + put_doc(&storage, "capLive", &mut old_capture()).await; + put_doc(&storage, "capOrphan", &mut old_capture()).await; + put_doc( + &storage, + "capLegacy", + &mut create_binary_document(b"gz", CAPTURE_MIME_TYPE).unwrap(), + ) + .await; + let ids = ["idx1", "fileA", "capLive", "capOrphan", "capLegacy"] + .map(str::to_string) + .to_vec(); + (storage, ids) + } + + #[tokio::test] + async fn scan_finds_only_the_orphaned_stamped_capture() { + let (storage, ids) = fixture_storage().await; + let m = scan(&storage, &ids, "/tmp/x", &ScanOptions::default()).await; + + let ids: Vec<&str> = m.candidates.iter().map(|c| c.doc_id.as_str()).collect(); + assert_eq!(ids, vec!["capOrphan"], "manifest: {m:#?}"); + // The live capture and the file doc are inventoried, not candidates. + assert_eq!(m.inventory["engine-capture"].count, 3); + assert_eq!(m.inventory["project-index"].count, 1); + assert_eq!(m.inventory["text-file"].count, 1); + // The legacy capture is reported as protected. + assert!( + m.not_collectible + .iter() + .any(|r| r.doc_id == "capLegacy" && r.reason.contains("unstamped")), + "manifest: {m:#?}" + ); + // Evidence strings are present. + assert!(m.candidates[0].reason.contains("not referenced")); + assert_eq!( + m.candidates[0] + .meta + .as_ref() + .unwrap() + .source_path + .as_deref(), + Some("a.qmd") + ); + } + + #[tokio::test] + async fn include_unstamped_makes_legacy_capture_a_candidate() { + let (storage, ids) = fixture_storage().await; + let m = scan( + &storage, + &ids, + "/tmp/x", + &ScanOptions { + include_unstamped: true, + ..Default::default() + }, + ) + .await; + let mut ids: Vec<&str> = m.candidates.iter().map(|c| c.doc_id.as_str()).collect(); + ids.sort_unstable(); + assert_eq!(ids, vec!["capLegacy", "capOrphan"]); + } + + #[tokio::test] + async fn young_capture_is_protected_by_age_gate() { + let storage = InMemoryStorage::new(); + // Freshly-stamped orphan (createdAt = now). + put_doc( + &storage, + "capYoung", + &mut create_capture_document( + b"gz", + &CaptureDocMeta { + source_path: "y.qmd".into(), + engines: vec![], + }, + ) + .unwrap(), + ) + .await; + let m = scan( + &storage, + &["capYoung".to_string()], + "/tmp/x", + &ScanOptions::default(), + ) + .await; + assert!(m.candidates.is_empty()); + assert!( + m.not_collectible + .iter() + .any(|r| r.doc_id == "capYoung" && r.reason.contains("age")), + "manifest: {m:#?}" + ); + } + + #[tokio::test] + async fn unreferenced_unknown_and_binary_docs_are_protected() { + let storage = InMemoryStorage::new(); + let mut unknown = Automerge::new(); + unknown + .transact::<_, _, automerge::AutomergeError>(|tx| { + tx.put_object(ROOT, "widgets", ObjType::Map)?; + Ok(()) + }) + .unwrap(); + put_doc(&storage, "mystery", &mut unknown).await; + put_doc( + &storage, + "img", + &mut create_binary_document(b"png", "image/png").unwrap(), + ) + .await; + + let m = scan( + &storage, + &["mystery".to_string(), "img".to_string()], + "/tmp/x", + &ScanOptions::default(), + ) + .await; + assert!(m.candidates.is_empty()); + assert_eq!(m.not_collectible.len(), 2); + } + + #[tokio::test] + async fn project_set_keeps_its_indexes_and_their_captures_live() { + let storage = InMemoryStorage::new(); + let mut set = Automerge::new(); + set.transact::<_, _, automerge::AutomergeError>(|tx| { + let projects = tx.put_object(ROOT, "projects", ObjType::Map)?; + let entry = tx.put_object(&projects, "idx1", ObjType::Map)?; + tx.put(&entry, "indexDocId", "idx1")?; + Ok(()) + }) + .unwrap(); + put_doc(&storage, "set1", &mut set).await; + put_doc(&storage, "idx1", &mut index_doc(&[], &[("a.qmd", "capA")])).await; + put_doc(&storage, "capA", &mut old_capture()).await; + + let m = scan( + &storage, + &["set1".to_string(), "idx1".to_string(), "capA".to_string()], + "/tmp/x", + &ScanOptions::default(), + ) + .await; + assert!( + m.candidates.is_empty(), + "capA is live via set1 → idx1 → captures; manifest: {m:#?}" + ); + } + + #[tokio::test] + async fn manifest_roundtrips_through_json() { + let (storage, ids) = fixture_storage().await; + let m = scan(&storage, &ids, "/tmp/x", &ScanOptions::default()).await; + let json = serde_json::to_string_pretty(&m).unwrap(); + let back: ScanManifest = serde_json::from_str(&json).unwrap(); + assert_eq!(back.manifest_version, MANIFEST_VERSION); + assert_eq!(back.candidates.len(), m.candidates.len()); + // Wire format is camelCase (hub-client admin page contract). + assert!(json.contains("\"manifestVersion\"")); + assert!(json.contains("\"sizeBytes\"")); + } +} + +/// Human summary of a manifest for terminal output. +pub fn human_summary(m: &ScanManifest) -> String { + let mut out = String::new(); + out.push_str(&format!( + "Scanned {} at {}\n\nInventory:\n", + m.data_dir, m.scanned_at + )); + for (kind, stats) in &m.inventory { + out.push_str(&format!( + " {kind:<16} {:>6} docs {:>12} bytes\n", + stats.count, stats.bytes + )); + } + let reclaimable: u64 = m.candidates.iter().map(|c| c.size_bytes).sum(); + out.push_str(&format!( + "\nRemovable candidates: {} ({} bytes reclaimable)\n", + m.candidates.len(), + reclaimable + )); + for c in &m.candidates { + let path = c + .meta + .as_ref() + .and_then(|meta| meta.source_path.as_deref()) + .unwrap_or(""); + out.push_str(&format!( + " {} {:>10} bytes {}\n", + c.doc_id, c.size_bytes, path + )); + } + if !m.not_collectible.is_empty() { + out.push_str(&format!( + "\nUnreferenced but protected (not collectible): {}\n", + m.not_collectible.len() + )); + for r in &m.not_collectible { + out.push_str(&format!(" {} {} {}\n", r.doc_id, r.kind, r.reason)); + } + } + out +} diff --git a/crates/quarto-hub/src/lib.rs b/crates/quarto-hub/src/lib.rs index 3a44cd588..e227ac277 100644 --- a/crates/quarto-hub/src/lib.rs +++ b/crates/quarto-hub/src/lib.rs @@ -7,6 +7,7 @@ //! - REST API for document operations pub mod access_policy; +pub mod admin; pub mod auth; pub mod context; pub mod discovery; diff --git a/crates/quarto-hub/src/main.rs b/crates/quarto-hub/src/main.rs index 46cdb371a..86ba51288 100644 --- a/crates/quarto-hub/src/main.rs +++ b/crates/quarto-hub/src/main.rs @@ -5,7 +5,7 @@ use std::path::PathBuf; -use clap::Parser; +use clap::{Parser, Subcommand}; use tracing::info; use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt}; @@ -15,6 +15,12 @@ use quarto_hub::{StorageManager, auth, context::HubConfig, default_standalone_da #[command(name = "hub")] #[command(about = "Collaborative sync server for Quarto projects")] struct Args { + /// Maintainer subcommands (`hub admin …`). When omitted, the hub + /// runs as a server with the flags below — the pre-subcommand CLI + /// is unchanged. + #[command(subcommand)] + command: Option, + /// Increase log verbosity. Repeat for more detail: /// `-v` adds info, `-vv` adds debug + samod=info, /// `-vvv` adds trace + samod=debug + tower_http=debug. @@ -119,10 +125,231 @@ struct Args { additional_audiences: Vec, } +#[derive(Subcommand, Debug)] +enum Command { + /// Sync-server maintainer tools (storage hygiene, bd-eiku4ymo). + /// See claude-notes/instructions/hub-storage-hygiene.md. + Admin { + #[command(subcommand)] + cmd: AdminCommand, + }, +} + +#[derive(Subcommand, Debug)] +enum AdminCommand { + /// Read-only orphan analysis: inventory every stored doc and emit + /// a manifest of safely-removable engine-capture docs. Safe to run + /// against a live server. + Scan { + /// Hub data directory (contains `automerge/` and `hub.lock`). + #[arg(long)] + data_dir: PathBuf, + /// Only captures older than this many days are candidates. + /// Negative values (e.g. `--older-than-days=-1`) disable the + /// age gate — every orphaned stamped capture qualifies. + #[arg(long, default_value_t = 30, allow_hyphen_values = true)] + older_than_days: i64, + /// Also consider captures without a `meta.createdAt` stamp + /// (recorded before the audit envelope existed). + #[arg(long)] + include_unstamped: bool, + /// Write the manifest JSON here (the input `collect` takes). + #[arg(long)] + output: Option, + /// Print the manifest JSON to stdout instead of the summary. + #[arg(long)] + json: bool, + }, + /// Quarantine the docs a scan manifest deems removable, after + /// re-verifying each against current storage. Moves doc chunks to + /// `/trash//`; never deletes. Dry-run unless + /// --execute. Refuses while a server holds the data dir. + Collect { + #[arg(long)] + data_dir: PathBuf, + /// Manifest produced by `hub admin scan --output`. + #[arg(long)] + manifest: PathBuf, + /// Actually quarantine (default is a dry-run report). + #[arg(long)] + execute: bool, + }, + /// Move a quarantined batch (or named docs from it) back into the + /// store, verifying chunk hashes recorded at collection time. + Restore { + #[arg(long)] + data_dir: PathBuf, + /// Batch directory under `/trash/`. + #[arg(long)] + batch: PathBuf, + /// Restore only these doc ids (default: the whole batch). + doc_ids: Vec, + }, + /// Delete trash batches older than the retention window. The only + /// operation that permanently removes bytes. Dry-run unless + /// --execute. + Purge { + #[arg(long)] + data_dir: PathBuf, + /// Negative values disable the retention gate. + #[arg(long, default_value_t = 30, allow_hyphen_values = true)] + retention_days: i64, + /// Actually delete eligible batches (default: list them). + #[arg(long)] + execute: bool, + }, +} + +/// Dispatch `hub admin …`. Exits the process with a non-zero status +/// on failure so scripts can gate on it. +async fn run_admin(cmd: AdminCommand) -> anyhow::Result<()> { + use quarto_hub::admin::{collect as collect_mod, scan as scan_mod}; + match cmd { + AdminCommand::Scan { + data_dir, + older_than_days, + include_unstamped, + output, + json, + } => { + let canonical = data_dir + .canonicalize() + .map_err(|e| anyhow::anyhow!("cannot canonicalize {}: {e}", data_dir.display()))?; + let automerge_dir = canonical.join("automerge"); + if !automerge_dir.is_dir() { + anyhow::bail!( + "{} has no automerge/ directory — is this a hub data dir?", + canonical.display() + ); + } + let storage = samod::storage::TokioFilesystemStorage::new(&automerge_dir); + let doc_ids = scan_mod::list_doc_ids_filesystem(&automerge_dir); + let manifest = scan_mod::scan( + &storage, + &doc_ids, + &canonical.to_string_lossy(), + &scan_mod::ScanOptions { + older_than_days, + include_unstamped, + }, + ) + .await; + if let Some(path) = &output { + std::fs::write(path, serde_json::to_vec_pretty(&manifest)?)?; + eprintln!("Manifest written to {}", path.display()); + } + if json { + println!("{}", serde_json::to_string_pretty(&manifest)?); + } else { + println!("{}", scan_mod::human_summary(&manifest)); + } + Ok(()) + } + AdminCommand::Collect { + data_dir, + manifest, + execute, + } => { + let manifest: quarto_hub::admin::manifest::ScanManifest = + serde_json::from_slice(&std::fs::read(&manifest)?)?; + let outcome = collect_mod::collect(&data_dir, &manifest, execute) + .await + .map_err(|e| anyhow::anyhow!(e))?; + for (doc_id, reason) in &outcome.skipped { + println!("SKIP {doc_id}: {reason}"); + } + for c in &outcome.verified { + println!( + "{} {} ({} bytes)", + if execute { + "QUARANTINED" + } else { + "WOULD COLLECT" + }, + c.doc_id, + c.size_bytes + ); + } + match &outcome.batch_dir { + Some(dir) => println!( + "Batch: {} (restore with `hub admin restore --data-dir {} --batch {}`)", + dir.display(), + data_dir.display(), + dir.display() + ), + None if !execute && !outcome.verified.is_empty() => { + println!("Dry-run only. Re-run with --execute to quarantine."); + } + None => {} + } + Ok(()) + } + AdminCommand::Restore { + data_dir, + batch, + doc_ids, + } => { + let results = collect_mod::restore(&data_dir, &batch, &doc_ids) + .map_err(|e| anyhow::anyhow!(e))?; + let mut failed = false; + for (doc_id, result) in &results { + match result { + Ok(()) => println!("RESTORED {doc_id}"), + Err(reason) => { + failed = true; + println!("FAILED {doc_id}: {reason}"); + } + } + } + if failed { + anyhow::bail!("some docs were not restored"); + } + Ok(()) + } + AdminCommand::Purge { + data_dir, + retention_days, + execute, + } => { + let candidates = collect_mod::purge(&data_dir, retention_days, execute) + .map_err(|e| anyhow::anyhow!(e))?; + if candidates.is_empty() { + println!("No trash batches."); + } + for c in &candidates { + println!( + "{} {} (created {:?}, age {:?} days)", + match (c.eligible, execute) { + (true, true) => "PURGED", + (true, false) => "WOULD PURGE", + (false, _) => "KEPT", + }, + c.batch_dir.display(), + c.created_at.as_deref().unwrap_or(""), + c.age_days, + ); + } + Ok(()) + } + } +} + #[tokio::main] async fn main() -> anyhow::Result<()> { let args = Args::parse(); + // `hub admin …`: maintainer tools, no server startup. + if let Some(Command::Admin { cmd }) = args.command { + tracing_subscriber::registry() + .with( + tracing_subscriber::EnvFilter::try_from_default_env() + .unwrap_or_else(|_| quarto_util::verbose_to_filter(args.verbose).into()), + ) + .with(tracing_subscriber::fmt::layer()) + .init(); + return run_admin(cmd).await; + } + // Initialize tracing. `-v` chooses a default filter directive (see // `quarto_util::verbose_to_filter`); `RUST_LOG`, when set, takes // precedence and is parsed directly by `try_from_default_env`. diff --git a/crates/quarto-hub/src/resource.rs b/crates/quarto-hub/src/resource.rs index e14a6961b..32f152a9e 100644 --- a/crates/quarto-hub/src/resource.rs +++ b/crates/quarto-hub/src/resource.rs @@ -136,6 +136,121 @@ pub fn create_binary_document(content: &[u8], mime_type: &str) -> Result, +} + +/// Schema version of the capture-doc `meta` map. +const CAPTURE_META_SCHEMA_VERSION: i64 = 1; + +/// Create an engine-capture binary document: the standard binary-doc +/// schema (`content`/`mimeType`/`hash`, MIME = [`CAPTURE_MIME_TYPE`]) +/// plus the uncompressed `meta` audit map: +/// +/// ```text +/// ROOT +/// ├── content: Bytes +/// ├── mimeType: String +/// ├── hash: String +/// └── meta: Map +/// ├── kind: "engine-capture" +/// ├── schemaVersion: 1 +/// ├── createdAt: String // RFC 3339 UTC +/// ├── sourcePath: String +/// └── engines: List +/// ``` +/// +/// Docs created before this envelope (no `meta`) remain valid; readers +/// of `content` are unaffected either way. +pub fn create_capture_document(gzipped: &[u8], meta: &CaptureDocMeta) -> Result { + create_capture_document_at(gzipped, meta, &chrono::Utc::now().to_rfc3339()) +} + +/// Test seam: like [`create_capture_document`] with an explicit +/// `createdAt` so age-gate tests can fabricate old captures. +pub fn create_capture_document_at( + gzipped: &[u8], + meta: &CaptureDocMeta, + created_at: &str, +) -> Result { + use automerge::ObjType; + let mut doc = create_binary_document(gzipped, CAPTURE_MIME_TYPE)?; + doc.transact::<_, _, automerge::AutomergeError>(|tx| { + let meta_obj = tx.put_object(ROOT, "meta", ObjType::Map)?; + tx.put(&meta_obj, "kind", "engine-capture")?; + tx.put(&meta_obj, "schemaVersion", CAPTURE_META_SCHEMA_VERSION)?; + tx.put(&meta_obj, "createdAt", created_at)?; + tx.put(&meta_obj, "sourcePath", meta.source_path.as_str())?; + let engines_obj = tx.put_object(&meta_obj, "engines", ObjType::List)?; + for (i, engine) in meta.engines.iter().enumerate() { + tx.insert(&engines_obj, i, engine.as_str())?; + } + Ok(()) + }) + .map_err(|e| { + crate::error::Error::IndexDocument(format!("failed to stamp capture meta: {:?}", e)) + })?; + Ok(doc) +} + +/// Read the `meta` audit map back from a capture doc. `None` when the +/// doc predates the envelope (or isn't a capture doc). Used by +/// `hub admin scan`'s age gate and inventory. +pub fn read_capture_meta(doc: &Automerge) -> Option { + use automerge::ReadDoc; + let (_, meta_obj) = doc.get(ROOT, "meta").ok().flatten()?; + let get_str = |key: &str| -> Option { + doc.get(&meta_obj, key) + .ok() + .flatten() + .and_then(|(value, _)| value.to_str().map(str::to_string)) + }; + let engines = match doc.get(&meta_obj, "engines").ok().flatten() { + Some((_, engines_obj)) => (0..doc.length(&engines_obj)) + .filter_map(|i| { + doc.get(&engines_obj, i) + .ok() + .flatten() + .and_then(|(value, _)| value.to_str().map(str::to_string)) + }) + .collect(), + None => Vec::new(), + }; + Some(CaptureDocMetaRead { + kind: get_str("kind"), + created_at: get_str("createdAt"), + source_path: get_str("sourcePath"), + engines, + }) +} + +/// The `meta` map as read back from a doc — every field optional +/// because a capture written by a future (or buggy) writer must +/// still be inspectable. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct CaptureDocMetaRead { + pub kind: Option, + pub created_at: Option, + pub source_path: Option, + pub engines: Vec, +} + /// Document type enumeration for distinguishing text and binary documents. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum DocumentType { @@ -338,4 +453,57 @@ mod tests { let empty_doc = Automerge::new(); assert_eq!(detect_document_type(&empty_doc), DocumentType::Invalid); } + + // ── bd-eiku4ymo: capture-doc meta envelope ───────────────────── + + fn sample_meta() -> CaptureDocMeta { + CaptureDocMeta { + source_path: "posts/analysis.qmd".to_string(), + engines: vec!["knitr".to_string(), "jupyter".to_string()], + } + } + + #[test] + fn capture_document_has_binary_schema_and_capture_mime() { + use automerge::ReadDoc; + let doc = create_capture_document(b"gzipped-bytes", &sample_meta()).unwrap(); + // Standard binary-doc fields intact. + assert_eq!(detect_document_type(&doc), DocumentType::Binary); + let (mime, _) = doc.get(ROOT, "mimeType").unwrap().unwrap(); + assert_eq!(mime.to_str(), Some(CAPTURE_MIME_TYPE)); + let (hash, _) = doc.get(ROOT, "hash").unwrap().unwrap(); + assert_eq!(hash.to_str(), Some(compute_hash(b"gzipped-bytes").as_str())); + } + + #[test] + fn capture_document_meta_roundtrips() { + let doc = create_capture_document(b"gz", &sample_meta()).unwrap(); + let meta = read_capture_meta(&doc).expect("meta map present"); + assert_eq!(meta.kind.as_deref(), Some("engine-capture")); + assert_eq!(meta.source_path.as_deref(), Some("posts/analysis.qmd")); + assert_eq!(meta.engines, vec!["knitr", "jupyter"]); + // createdAt parses as RFC 3339 — the scan age gate depends on it. + let created_at = meta.created_at.expect("createdAt present"); + chrono::DateTime::parse_from_rfc3339(&created_at) + .unwrap_or_else(|e| panic!("createdAt must be RFC 3339, got {created_at}: {e}")); + } + + #[test] + fn capture_document_at_uses_explicit_timestamp() { + let doc = + create_capture_document_at(b"gz", &sample_meta(), "2020-01-02T03:04:05+00:00").unwrap(); + let meta = read_capture_meta(&doc).unwrap(); + assert_eq!( + meta.created_at.as_deref(), + Some("2020-01-02T03:04:05+00:00") + ); + } + + #[test] + fn legacy_capture_doc_without_meta_reads_none() { + // Pre-envelope captures (plain binary docs with the capture + // MIME) must read back as "no meta", not error. + let doc = create_binary_document(b"gz", CAPTURE_MIME_TYPE).unwrap(); + assert!(read_capture_meta(&doc).is_none()); + } } diff --git a/crates/quarto-hub/tests/integration/admin_collect_lifecycle.rs b/crates/quarto-hub/tests/integration/admin_collect_lifecycle.rs new file mode 100644 index 000000000..aa5707cf6 --- /dev/null +++ b/crates/quarto-hub/tests/integration/admin_collect_lifecycle.rs @@ -0,0 +1,259 @@ +//! Full collect → restore → purge lifecycle against a REAL samod +//! filesystem store (bd-eiku4ymo), exercising every damage- +//! minimization gate: dry-run inertness, re-verification skips, +//! quarantine (no unlink), hash-verified restore, retention-gated +//! purge, and the hub.lock guard. + +use std::path::{Path, PathBuf}; + +use quarto_hub::admin::collect::{collect, purge, restore}; +use quarto_hub::admin::scan::{ScanOptions, list_doc_ids_filesystem, scan}; +use quarto_hub::context::{HubConfig, HubContext}; +use quarto_hub::index::CaptureRef; +use quarto_hub::resource::{CaptureDocMeta, create_capture_document_at}; +use quarto_hub::storage::StorageManager; +use samod::storage::TokioFilesystemStorage; +use tempfile::TempDir; + +fn old_capture_doc() -> automerge::Automerge { + create_capture_document_at( + b"gzipped-capture-bytes", + &CaptureDocMeta { + source_path: "a.qmd".into(), + engines: vec!["knitr".into()], + }, + "2020-01-01T00:00:00+00:00", + ) + .unwrap() +} + +/// Build a real store with one live and one orphaned capture; return +/// (project tempdir, hub data dir, orphan doc id, live doc id). +async fn build_store() -> (TempDir, PathBuf, String, String) { + let temp = TempDir::new().unwrap(); + let project_root = temp.path().join("project"); + std::fs::create_dir(&project_root).unwrap(); + std::fs::write(project_root.join("a.qmd"), "# Hello\n").unwrap(); + + let storage = StorageManager::new(&project_root).unwrap(); + let hub_dir = storage.hub_dir().to_path_buf(); + let ctx = HubContext::new(storage, HubConfig::default()) + .await + .unwrap(); + + let live = ctx.repo().create(old_capture_doc()).await.unwrap(); + let live_id = live.document_id().to_string(); + ctx.index() + .set_capture( + "a.qmd", + &CaptureRef { + capture_doc_id: live_id.clone(), + staleness: Some(false), + state: None, + last_error: None, + }, + ) + .unwrap(); + let orphan = ctx.repo().create(old_capture_doc()).await.unwrap(); + let orphan_id = orphan.document_id().to_string(); + drop(orphan); + drop(live); + ctx.repo().stop().await; + (temp, hub_dir, orphan_id, live_id) +} + +/// Recursive sorted listing of (relative path, size) — a cheap tree +/// fingerprint for dry-run inertness checks. +fn tree_fingerprint(dir: &Path) -> Vec<(String, u64)> { + fn walk(root: &Path, dir: &Path, out: &mut Vec<(String, u64)>) { + let Ok(entries) = std::fs::read_dir(dir) else { + return; + }; + for entry in entries.flatten() { + let path = entry.path(); + if path.is_dir() { + walk(root, &path, out); + } else if let Ok(meta) = path.metadata() { + let rel = path + .strip_prefix(root) + .unwrap() + .to_string_lossy() + .into_owned(); + // The (empty) lockfile is admin infrastructure: + // AdminLock creates it if absent, exactly as a server + // start would. It is not data. + if rel == "hub.lock" { + continue; + } + out.push((rel, meta.len())); + } + } + } + let mut out = Vec::new(); + walk(dir, dir, &mut out); + out.sort(); + out +} + +async fn scan_store(hub_dir: &Path) -> quarto_hub::admin::manifest::ScanManifest { + let automerge_dir = hub_dir.join("automerge"); + let storage = TokioFilesystemStorage::new(&automerge_dir); + let ids = list_doc_ids_filesystem(&automerge_dir); + // The manifest's dataDir must be the HUB dir (collect's contract). + scan( + &storage, + &ids, + &hub_dir.canonicalize().unwrap().to_string_lossy(), + &ScanOptions::default(), + ) + .await +} + +#[tokio::test(flavor = "multi_thread")] +async fn collect_lifecycle_quarantine_restore_purge() { + let (_temp, hub_dir, orphan_id, live_id) = build_store().await; + let manifest = scan_store(&hub_dir).await; + assert_eq!(manifest.candidates.len(), 1); + assert_eq!(manifest.candidates[0].doc_id, orphan_id); + + // ── Dry-run changes NOTHING on disk ───────────────────────────── + let before = tree_fingerprint(&hub_dir); + let dry = collect(&hub_dir, &manifest, false).await.unwrap(); + assert_eq!(dry.verified.len(), 1); + assert!(dry.batch_dir.is_none()); + assert_eq!( + tree_fingerprint(&hub_dir), + before, + "dry-run must not touch the data dir" + ); + + // ── Execute: quarantine, never unlink ─────────────────────────── + let outcome = collect(&hub_dir, &manifest, true).await.unwrap(); + let batch_dir = outcome.batch_dir.expect("batch created"); + // collect canonicalizes the data dir (macOS: /var → /private/var), + // so compare against the canonical form. + assert!(batch_dir.starts_with(hub_dir.canonicalize().unwrap().join("trash"))); + // Orphan's chunks moved (not copied, not deleted). + let orphan_quarantined = batch_dir.join("docs").join(&orphan_id); + assert!(orphan_quarantined.is_dir()); + assert!(!list_doc_ids_filesystem(&hub_dir.join("automerge")).contains(&orphan_id)); + // batch.json embeds the manifest + chunk hashes. + let record: quarto_hub::admin::collect::BatchRecord = + serde_json::from_slice(&std::fs::read(batch_dir.join("batch.json")).unwrap()).unwrap(); + assert_eq!(record.manifest.candidates[0].doc_id, orphan_id); + assert!(!record.docs[0].chunks.is_empty()); + + // The project still opens and serves everything live. + { + let storage = StorageManager::new(_temp.path().join("project")).unwrap(); + let ctx = HubContext::new(storage, HubConfig::default()) + .await + .unwrap(); + let cap = ctx.index().get_capture("a.qmd").unwrap(); + assert_eq!(cap.capture_doc_id, live_id); + let handle = ctx + .repo() + .find(samod::DocumentId::from_str(&live_id).unwrap()) + .await + .unwrap(); + assert!(handle.is_some(), "live capture must still load"); + ctx.repo().stop().await; + } + + // ── Restore: hash-verified, byte-identical ────────────────────── + let results = restore(&hub_dir, &batch_dir, &[]).unwrap(); + assert_eq!(results.len(), 1); + assert!(results[0].1.is_ok(), "restore failed: {:?}", results[0].1); + assert!( + list_doc_ids_filesystem(&hub_dir.join("automerge")).contains(&orphan_id), + "orphan chunks back in place" + ); + // Restored doc loads as a valid capture again. + let m2 = scan_store(&hub_dir).await; + assert_eq!( + m2.candidates.len(), + 1, + "restored orphan is a candidate again" + ); + + // ── Purge: retention-gated, batch-level ───────────────────────── + // Re-collect so there is a batch to purge. + let outcome2 = collect(&hub_dir, &m2, true).await.unwrap(); + let batch2 = outcome2.batch_dir.unwrap(); + // Young batch: listed, not eligible, survives --execute. + let purged = purge(&hub_dir, 30, true).unwrap(); + assert!(purged.iter().all(|p| !p.eligible)); + assert!(batch2.exists()); + // Age the batch by rewriting its createdAt, then purge for real. + let mut record2: quarto_hub::admin::collect::BatchRecord = + serde_json::from_slice(&std::fs::read(batch2.join("batch.json")).unwrap()).unwrap(); + record2.created_at = "2020-01-01T00:00:00+00:00".to_string(); + std::fs::write( + batch2.join("batch.json"), + serde_json::to_vec(&record2).unwrap(), + ) + .unwrap(); + let purged2 = purge(&hub_dir, 30, true).unwrap(); + assert!(purged2.iter().any(|p| p.eligible)); + assert!(!batch2.exists(), "aged batch purged"); +} + +#[tokio::test(flavor = "multi_thread")] +async fn collect_reverification_skips_rereferenced_candidate() { + let (_temp, hub_dir, orphan_id, _live_id) = build_store().await; + let manifest = scan_store(&hub_dir).await; + assert_eq!(manifest.candidates[0].doc_id, orphan_id); + + // Between scan and collect, the orphan becomes referenced again + // (e.g. a client synced back an older index state). + { + let storage = StorageManager::new(_temp.path().join("project")).unwrap(); + let ctx = HubContext::new(storage, HubConfig::default()) + .await + .unwrap(); + ctx.index() + .set_capture( + "a.qmd", + &CaptureRef { + capture_doc_id: orphan_id.clone(), + staleness: Some(false), + state: None, + last_error: None, + }, + ) + .unwrap(); + ctx.repo().stop().await; + } + + let outcome = collect(&hub_dir, &manifest, true).await.unwrap(); + assert!(outcome.verified.is_empty()); + assert!(outcome.batch_dir.is_none()); + assert_eq!(outcome.skipped.len(), 1); + assert!( + outcome.skipped[0].1.contains("referenced"), + "skip reason: {}", + outcome.skipped[0].1 + ); + // Nothing moved. + assert!(list_doc_ids_filesystem(&hub_dir.join("automerge")).contains(&orphan_id)); +} + +#[tokio::test(flavor = "multi_thread")] +async fn collect_refuses_wrong_data_dir_and_held_lock() { + let (_temp, hub_dir, _orphan_id, _live_id) = build_store().await; + let manifest = scan_store(&hub_dir).await; + + // Wrong data dir (a manifest from server A pointed at server B). + let other = TempDir::new().unwrap(); + std::fs::create_dir_all(other.path().join("automerge")).unwrap(); + let err = collect(other.path(), &manifest, true).await.unwrap_err(); + assert!(err.contains("refusing"), "got: {err}"); + + // Held lock (a live server): StorageManager holds the exclusive + // flock for its lifetime — collect must refuse. + let _held = StorageManager::new(_temp.path().join("project")).unwrap(); + let err = collect(&hub_dir, &manifest, true).await.unwrap_err(); + assert!(err.contains("locked"), "got: {err}"); +} + +use std::str::FromStr as _; diff --git a/crates/quarto-hub/tests/integration/admin_scan_real_store.rs b/crates/quarto-hub/tests/integration/admin_scan_real_store.rs new file mode 100644 index 000000000..0b1c9d892 --- /dev/null +++ b/crates/quarto-hub/tests/integration/admin_scan_real_store.rs @@ -0,0 +1,105 @@ +//! `hub admin scan` against a REAL samod filesystem store +//! (bd-eiku4ymo) — not fabricated chunks: docs are created through a +//! live `HubContext` repo (the same path the server uses), flushed to +//! disk via `Repo::stop()`, and then scanned offline through +//! `TokioFilesystemStorage`, exactly as the CLI will. + +use quarto_hub::admin::scan::{ScanOptions, list_doc_ids_filesystem, scan}; +use quarto_hub::context::{HubConfig, HubContext}; +use quarto_hub::index::CaptureRef; +use quarto_hub::resource::{CaptureDocMeta, create_capture_document_at}; +use quarto_hub::storage::StorageManager; +use samod::storage::TokioFilesystemStorage; +use tempfile::TempDir; + +/// An old-stamped capture doc (safely past any age gate). +fn old_capture_doc() -> automerge::Automerge { + create_capture_document_at( + b"gzipped-capture-bytes", + &CaptureDocMeta { + source_path: "a.qmd".into(), + engines: vec!["knitr".into()], + }, + "2020-01-01T00:00:00+00:00", + ) + .unwrap() +} + +#[tokio::test(flavor = "multi_thread")] +async fn scan_real_store_finds_orphaned_capture_only() { + let temp = TempDir::new().unwrap(); + let project_root = temp.path().join("project"); + std::fs::create_dir(&project_root).unwrap(); + std::fs::write(project_root.join("a.qmd"), "# Hello\n").unwrap(); + + let storage = StorageManager::new(&project_root).unwrap(); + let automerge_dir = storage.automerge_dir(); + let ctx = HubContext::new(storage, HubConfig::default()) + .await + .unwrap(); + + // A "current" capture, referenced from the index sidecar → live. + let live = ctx.repo().create(old_capture_doc()).await.unwrap(); + ctx.index() + .set_capture( + "a.qmd", + &CaptureRef { + capture_doc_id: live.document_id().to_string(), + staleness: Some(false), + state: None, + last_error: None, + }, + ) + .unwrap(); + + // A superseded capture nobody references → the orphan. This is + // exactly what perform_re_execute leaves behind when it repoints + // the sidecar. + let orphan = ctx.repo().create(old_capture_doc()).await.unwrap(); + let orphan_id = orphan.document_id().to_string(); + drop(orphan); + drop(live); + + // Flush the repo to disk and let go of the store. + ctx.repo().stop().await; + + // Scan offline through the same adapter the server uses. + let fs_storage = TokioFilesystemStorage::new(&automerge_dir); + let doc_ids = list_doc_ids_filesystem(&automerge_dir); + assert!( + doc_ids.len() >= 4, + "expected at least index + file + 2 captures; got {doc_ids:?}" + ); + let manifest = scan( + &fs_storage, + &doc_ids, + &automerge_dir.to_string_lossy(), + &ScanOptions::default(), + ) + .await; + + let ids: Vec<&str> = manifest + .candidates + .iter() + .map(|c| c.doc_id.as_str()) + .collect(); + assert_eq!( + ids, + vec![orphan_id.as_str()], + "exactly the superseded capture is removable; manifest: {manifest:#?}" + ); + // Evidence carried from the Phase A meta envelope. + let meta = manifest.candidates[0].meta.as_ref().unwrap(); + assert_eq!(meta.source_path.as_deref(), Some("a.qmd")); + assert_eq!(meta.engines, vec!["knitr"]); + + // The index, the file doc, and the live capture are all + // inventoried and protected. + assert_eq!(manifest.inventory["project-index"].count, 1); + assert_eq!(manifest.inventory["engine-capture"].count, 2); + assert!( + manifest.inventory.contains_key("text-file"), + "the synced a.qmd file doc should be inventoried; got {:?}", + manifest.inventory + ); +} diff --git a/crates/quarto-hub/tests/integration/main.rs b/crates/quarto-hub/tests/integration/main.rs index 6b8b1f488..be38ecf13 100644 --- a/crates/quarto-hub/tests/integration/main.rs +++ b/crates/quarto-hub/tests/integration/main.rs @@ -3,9 +3,11 @@ //! One `integration` binary per crate (see //! `.claude/rules/integration-tests.md`): add new integration tests as //! `pub mod ;` here, never as top-level `tests/.rs` files. -//! Shared fixtures (mock OIDC provider, test hub, tracing capture) -//! live in [`support`]. +//! Keep the module list alphabetized. Shared fixtures (mock OIDC +//! provider, test hub, tracing capture) live in [`support`]. +pub mod admin_collect_lifecycle; +pub mod admin_scan_real_store; pub mod auth_bearer; pub mod session_auth; pub mod support; diff --git a/crates/quarto-preview/src/capture_driver.rs b/crates/quarto-preview/src/capture_driver.rs index d4fcbf23c..e215b1f66 100644 --- a/crates/quarto-preview/src/capture_driver.rs +++ b/crates/quarto-preview/src/capture_driver.rs @@ -29,7 +29,6 @@ use quarto_core::project::ProjectContext; use quarto_error_reporting::DiagnosticMessageBuilder; use quarto_hub::HubContext; use quarto_hub::index::CaptureRef; -use quarto_hub::resource::create_binary_document; use quarto_system_runtime::SystemRuntime; use quarto_trace::EngineCapture; @@ -40,7 +39,9 @@ use crate::diagnostics; /// MIME type marker written into the capture binary doc so the /// browser-side reader can verify it's looking at the right payload /// shape (gzipped EngineCapture JSON) instead of an unrelated resource. -pub const CAPTURE_MIME_TYPE: &str = "application/x-engine-capture+gzip"; +/// Re-exported from quarto-hub, the single source of truth +/// (bd-eiku4ymo). +pub use quarto_hub::resource::CAPTURE_MIME_TYPE; /// Walk the project's `.qmd` files and record engine captures for any /// file that has code cells but no existing sidecar entry yet. @@ -180,7 +181,7 @@ async fn record_one( return Ok(false); } - let capture_doc_id = write_capture_doc(ctx, &captures).await?; + let capture_doc_id = write_capture_doc(ctx, rel_path, &captures).await?; let capture_ref = CaptureRef { capture_doc_id, @@ -322,8 +323,14 @@ pub async fn recompute_staleness( /// Serialize + gzip + store the EngineCapture sequence as a samod binary /// doc (bd-5yff4: a JSON array, one per engine). Returns the new doc's /// stringified DocumentId for use in the sidecar `captureDocId` field. +/// +/// `rel_path` is the project-relative source path, stamped (with the +/// engine names) into the doc's uncompressed `meta` audit map +/// (bd-eiku4ymo) so `hub admin scan` can read provenance without +/// gunzipping. async fn write_capture_doc( ctx: &Arc, + rel_path: &str, captures: &[EngineCapture], ) -> Result { // Shared wire format (serialize + gzip + 10MB size warning, @@ -332,7 +339,11 @@ async fn write_capture_doc( let gzipped = quarto_core::engine::capture_files::gzip_captures(captures) .map_err(|e| RecordError::Gzip(format!("{}", e)))?; - let automerge_doc = create_binary_document(&gzipped, CAPTURE_MIME_TYPE) + let meta = quarto_hub::resource::CaptureDocMeta { + source_path: rel_path.to_string(), + engines: captures.iter().map(|c| c.engine_name.clone()).collect(), + }; + let automerge_doc = quarto_hub::resource::create_capture_document(&gzipped, &meta) .map_err(|e| RecordError::CreateBinaryDoc(format!("{}", e)))?; let handle = ctx diff --git a/crates/quarto-preview/src/re_execute.rs b/crates/quarto-preview/src/re_execute.rs index cd21e024c..9a3c45736 100644 --- a/crates/quarto-preview/src/re_execute.rs +++ b/crates/quarto-preview/src/re_execute.rs @@ -33,13 +33,11 @@ use quarto_error_reporting::DiagnosticMessageBuilder; use quarto_hub::HubContext; use quarto_hub::context::SharedContext; use quarto_hub::index::{CaptureRef, CaptureState}; -use quarto_hub::resource::create_binary_document; use quarto_system_runtime::{NativeRuntime, SystemRuntime}; use quarto_trace::EngineCapture; use serde::{Deserialize, Serialize}; use crate::cache::record_capture_cached; -use crate::capture_driver::CAPTURE_MIME_TYPE; use crate::diagnostics; /// Process-wide set of paths currently being re-executed. Used to @@ -312,7 +310,7 @@ async fn perform_re_execute( return Err("engine produced no capture (no code cells?)".to_string()); } - let new_doc_id = write_capture_doc(&ctx, &captures) + let new_doc_id = write_capture_doc(&ctx, rel_path, &captures) .await .map_err(|e| format!("failed to store capture binary doc: {e}"))?; @@ -337,11 +335,16 @@ async fn perform_re_execute( /// bd-qbhp2cvv). async fn write_capture_doc( ctx: &Arc, + rel_path: &str, captures: &[EngineCapture], ) -> Result { let gzipped = quarto_core::engine::capture_files::gzip_captures(captures) .map_err(|e| format!("serialize/gzip: {e}"))?; - let doc = create_binary_document(&gzipped, CAPTURE_MIME_TYPE) + let meta = quarto_hub::resource::CaptureDocMeta { + source_path: rel_path.to_string(), + engines: captures.iter().map(|c| c.engine_name.clone()).collect(), + }; + let doc = quarto_hub::resource::create_capture_document(&gzipped, &meta) .map_err(|e| format!("binary doc: {e}"))?; let handle = ctx .repo() diff --git a/crates/quarto-preview/tests/integration/eager_capture.rs b/crates/quarto-preview/tests/integration/eager_capture.rs index ba8910f8e..4da760b82 100644 --- a/crates/quarto-preview/tests/integration/eager_capture.rs +++ b/crates/quarto-preview/tests/integration/eager_capture.rs @@ -20,6 +20,7 @@ //! short-circuit and trigger capture emission. use std::net::TcpListener as StdTcpListener; +use std::str::FromStr as _; use std::sync::Arc; use std::time::{Duration, Instant}; @@ -148,6 +149,26 @@ async fn eager_capture_populates_index_sidecar() { markdown ); + // bd-eiku4ymo: the capture doc carries the uncompressed `meta` + // audit map — readable without gunzipping the payload. + let doc_id = samod::DocumentId::from_str(&entry.capture_doc_id).unwrap(); + let handle = ctx + .repo() + .find(doc_id) + .await + .expect("repo running") + .expect("capture doc present"); + let meta = handle + .with_document(|doc| quarto_hub::resource::read_capture_meta(doc)) + .expect("capture doc must carry the meta envelope"); + assert_eq!(meta.kind.as_deref(), Some("engine-capture")); + assert_eq!(meta.source_path.as_deref(), Some("doc.qmd")); + assert_eq!(meta.engines, vec!["test-passthrough"]); + assert!( + meta.created_at.is_some(), + "createdAt must be stamped for the scan age gate" + ); + server_handle.abort(); let _ = server_handle.await; }