From 5e8d894a0c9d32096f72018ab0e149862c56619f Mon Sep 17 00:00:00 2001 From: Edouard Vanbelle Date: Sun, 12 Jul 2026 23:44:35 +0200 Subject: [PATCH] chore(plan): add job-registry + consistency check purpose is to design a job registry with a scheduler thi aim to drive in the same way any services requiring execution of periodic background tasks plugins could benefeciate it purpose is mostly to add a normative way to implement consistency check per services this is to edutcate implementors adding any new services goal is to ensure data quality with oxicloud and resumable jobs by default --- docs/plan/consistency-check.md | 810 +++++++++++++++++++++++++++++++++ docs/plan/job-registry.md | 798 ++++++++++++++++++++++++++++++++ 2 files changed, 1608 insertions(+) create mode 100644 docs/plan/consistency-check.md create mode 100644 docs/plan/job-registry.md diff --git a/docs/plan/consistency-check.md b/docs/plan/consistency-check.md new file mode 100644 index 00000000..c625b332 --- /dev/null +++ b/docs/plan/consistency-check.md @@ -0,0 +1,810 @@ +# Plan — Resumable consistency checks + `StatefulAdapter` contract + +## Context + +OxiCloud persists state in several independent subsystems: content-addressable +blobs on disk / S3, thumbnails (server-generated per blob + user-uploaded per +file), text-extraction cache, audio metadata cache, `storage.folders` + `file_metadata`, +`storage.trash`, `storage.drives.used_bytes`, WebDAV dead properties, and more. +Each has invariants that can silently drift: + +- A blob on disk with no `file_blobs` reference (leak — wasted disk). +- A `file_blobs` row whose bytes are gone from storage (**data loss** — GET + returns 500). +- A `drives.used_bytes` counter that no longer matches `SUM(size)`. +- A `folders.parent_id` pointing at a deleted row (historical raw-SQL fix). +- A thumbnail cache entry with no live file id (leak) or a missing entry + the user actually uploaded (data loss). + +Today the only "consistency" primitives are targeted point solutions — +`dedup_service` GC (orphan blob reap with 1 h grace), `storage_usage_service` +reconciliation (rebuild `used_bytes` from `SUM(size)`), and the trash cleaner. +None of them SURFACE inconsistencies for operators; they act blindly and +best-effort. `tests/api/storage_cleanup_check.sh` polls with a 5 s window +and races the 1 h GC grace (memory note `project_dedup_gc_test_trigger`). + +At scale — Ed's example: 1000 users × ~1000 files each = 1M files — an ad-hoc +"is my disk usage accurate?" check must be **resumable** across restarts, +cooperative on cancellation, and non-blocking to live traffic. A batch job +that starts over from scratch after a container restart or SIGTERM never +completes. + +This plan lands: + +1. Two **traits** — `ConsistencyCheck` (one check) and `StatefulAdapter` + (marker + registration on every state-owning port). +2. A **contract** every new adapter must satisfy at compile time — via a + supertrait bound on existing state-owning ports, no new adapter + compiles without declaring its consistency contract. +3. An **educational surface** in trait doc-comments — decision axes + (severity, direction, grace, cursor) and canonical-example pointers. +4. **Consistency-specific persistence** — `admin.consistency_findings`, + idempotent-on-`(run_id, kind, resource_id)`. +5. A **first check** — `BlobConsistencyCheck` (both directions, + blob-keyed cursor, severity split). + +**Layer boundary — the runtime is not in this plan.** The resumable +execution engine (cursor persistence, exclusivity, cancel protocol, +crash recovery, `admin.background_runs` schema, `JobStore`, +`RunOutcome`, `run_or_resume`) lives in `docs/plan/job-registry.md` +Part 2. This plan describes what `ConsistencyCheck` implementors +write and how the check-specific bits (findings, severity, +`StatefulAdapter` supertrait wiring) compose on top of that engine. + +**Order:** ships **after** the job-registry Part 2 engine lands. +Consistency closes an operator-visibility gap today, but it depends +on Part 2's `RecoverableJob` + `JobStore` + `admin.background_runs` +primitives — those come first. Once both are in, consistency runs +are admin-triggered v1, becoming periodic-triggered when a +`JobRegistry` (Part 1) tenant wraps `run_or_resume` for each +`ConsistencyCheck`. + +## Design decisions + +### Two-trait split + +`ConsistencyCheck` = one check (implements `run_resumable`). +`StatefulAdapter` = a subsystem that CONTRIBUTES checks (one or more). + +This split is load-bearing: +- Some subsystems emit **multiple** checks (`ThumbnailStore` emits four — + server-generated × 2 directions, user-uploaded × 2 directions). +- Some checks are **composed across** adapters (`UsedBytesConsistencyCheck` + reads from both `FileMetadataRepository` and `DriveRepository`). + +Bundling them into one trait would over-constrain the shape. + +### Compile-time enforcement via supertrait bound + +`StatefulAdapter` is added as a **supertrait** on every port that persists +state: + +```rust +pub trait BlobStorage: StatefulAdapter { … } +pub trait ThumbnailStore: StatefulAdapter { … } +pub trait FileBlobReadRepository: StatefulAdapter { … } +pub trait FolderRepository: StatefulAdapter { … } +``` + +Any new impl of these ports — a new S3-alike backend, a new mock in tests, +a plugin-provided storage backend — will not compile without providing +`subsystem()` and `consistency_checks()`. The compiler is the enforcement; +reviewers cannot merge a stateful adapter without an answer to "what can +go wrong with this state, and how do you check it?" + +### The severity axis + +Every finding carries a `Severity` so the admin UI can order results and +operators can dismiss the low-impact ones without hiding real risk. + +| Severity | Meaning | Examples | +|---|---|---| +| `DataLoss` | User-visible impact (500 on GET, missing user bytes) | Missing blob for a live `file_blobs` row; missing user-uploaded thumbnail | +| `Reclaimable` | Disk waste, no user impact | Orphan blob on storage, orphan thumbnail file | +| `Regenerable` | Auto-heals on next request | Missing server-generated thumbnail (server rebuilds), missing text-index row | +| `Drift` | Accounting mismatch, no user impact | `drives.used_bytes` vs `SUM(size)` | + +Rule of thumb: if a human user notices, it's `DataLoss`. If only the disk +accountant notices, it's `Reclaimable` or `Drift`. If the next automatic +regeneration will fix it, it's `Regenerable`. + +### Bidirectional in every check + +Every check emits BOTH directions where they exist: + +- **Backward (storage → DB) — orphan detection.** Wasted disk. `Reclaimable`. +- **Forward (DB → storage) — missing detection.** User-visible data loss. + `DataLoss`. Higher severity — a single missing content-addressable blob + silently breaks every file that referenced it. + +Skipping the forward direction is the single most common consistency-check +mistake. It's easy because Pass 1 (list storage, cross-check DB) LOOKS +complete. Pass 2 (list DB, cross-check storage) is where data-loss surfaces. + +### Report shape for missing findings — blob-level, not file-level + +`MissingInStorage { blob_hash, ref_count, affected_file_ids: Vec }`. +One row per missing blob with the fan-out of broken files. Operator gets +a triage-ordered "biggest impact first" list. File-per-line reports lose +that ordering. + +### Two-pass discipline eliminates the need for maintenance mode + +Pass 1 — build candidate list from a snapshot read (storage listing for +orphan direction, DB SELECT for missing direction). Exclude anything younger +than `grace_window`. + +Pass 2 — per candidate, re-read the OTHER side's state right before +flagging. If it transitioned (ref went up, blob just landed, row was +deleted, etc.), silently drop. + +Race matrix — orphan direction: +- **Upload lands mid-scan** (dedup hit → ref_count ↑ after we sampled) — + grace window skips young objects. +- **Last ref deleted mid-scan** (ref_count → 0, GC not yet) — cross-reference + `blobs.orphaned_at`; expected transient state, not flagged. +- **Deep hash on partial upload** — deep mode runs only on rows older + than a LONGER grace (24 h). + +Race matrix — missing direction: +- **Blob written but DB row not yet inserted** (young file looks missing + at flag time) — grace window skips DB rows younger than 1 h. +- **File deleted mid-scan** — Pass 2 re-reads `file_metadata` by id; + if gone, drop the finding. +- **Blob just now landed** — Pass 2 re-verifies storage `HEAD`; if now + present, drop. + +Nothing before "byte-exact whole-table snapshot verification" needs a +quiescent server. Reserve `concurrent_safe() = false` for that one. + +### Resumability — runs live in Part 2's `background_runs` + +Consistency runs are ordinary `RecoverableJob`s. The runtime plumbing +— cursor persistence, exclusivity, cancel protocol, crash recovery, +`admin.background_runs` schema, `JobStore` trait, `RunOutcome`, +`run_or_resume` helper — lives in `docs/plan/job-registry.md` Part 2. +This plan does not redefine any of it. + +**How consistency slots into the shared table:** +- Each consistency check registers under `job_name = + 'consistency_'` (e.g. `'consistency_blobs'`, + `'consistency_thumbnails'`). Naming convention lets a single + `SELECT DISTINCT ON (job_name)` return the last run of every check + alongside every other background job. +- Consistency's per-check knobs — `grace_window_secs`, `batch_size`, + `concurrent_safe` — live inside `background_runs.params` JSONB at + run-start time. The check reads them back via + `serde_json::from_value(store.params()?)`. +- `background_runs.stats` accumulates `{"scanned_count": …, + "findings_this_run": …}`; readers call + `(stats->>'scanned_count')::bigint`. + +The findings themselves are Layer C (this plan) — they don't +generalise to storage-migration or reextract: + +```sql +CREATE TABLE admin.consistency_findings ( + id UUID PRIMARY KEY, + run_id UUID NOT NULL REFERENCES admin.background_runs(id) ON DELETE CASCADE, + kind TEXT NOT NULL, -- OrphanBlob / MissingBlob / ... + severity TEXT NOT NULL, -- DataLoss / Reclaimable / ... + resource_id TEXT NOT NULL, + detail JSONB, + found_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + UNIQUE (run_id, kind, resource_id) -- idempotent re-scan on resume +); +CREATE INDEX ON admin.consistency_findings (run_id, severity); +``` + +FK on `background_runs.id` links a finding back to the run that +produced it; `ON DELETE CASCADE` clears findings when their run row +is pruned by a future retention job. + +`admin.*` is a NEW schema, created by Part 2's migration — keep it +distinct from `auth.*` / `storage.*` so operational tables don't +pollute domain schemas. + +### Non-obvious traps + +Recorded here (and in the trait doc-comments) because every one has been +learned the hard way in similar systems: + +1. **Grace window uses `scan_started_at`, NOT `NOW()`.** A resumable scan + spanning 6 h must snapshot its grace boundary at start. Otherwise items + uploaded 30 min in flip from "young, skip" (Pass 1's view) to "old, flag" + (Pass 2's view) mid-flight — the scan produces false positives against + itself. +2. **Cursor is per-check, opaque bytes.** Blob check cursors on BLAKE3 + hash (fixed 64 hex chars — natural lex order). Thumbnail cursors on + file_id UUID. Folder-tree cursor on ltree path. The trait treats it + as `Vec`; each impl serializes what it needs. +3. **Findings are idempotent on `(run_id, kind, resource_id)`.** Resume + revisit must not double-count. Pass 2 can also DELETE findings that + transitioned (was `MissingBlob`, blob has since landed → drop the + finding, not the whole run). +4. **Cooperative cancellation ONLY.** Between batches, poll + `background_runs.status`. A `tokio::spawn` abort mid-batch leaks — + cursor unpersisted, findings half-written. Cancel path writes + `status='Paused'` + current cursor before returning. +5. **Crash recovery on boot.** Any `status='Running'` at server start = + server died mid-scan. Auto-transition to `Paused`; DON'T auto-resume + (the bug that killed the last run may still be present). Admin decides. +6. **Batch size 1000 items or 30 s, whichever comes first.** Cursor commit + per-row makes DB write cost dominate at 1M items; longer batches leak + more progress on crash. +7. **Two directions don't share a cursor.** `BlobConsistencyCheck` orphan + side walks storage listing (S3 continuation token / readdir); missing + side walks `file_blobs` by hash. Sequence them (orphan phase → missing + phase); cursor encodes current phase. + `ThumbnailConsistencyCheck` is worse — four phases (2 subspaces × 2 + directions), each with its own natural cursor. Cursor encodes + `(subspace, direction, key)`. + +## Trait shapes + +### `ConsistencyCheck` + +```rust +/// A single consistency check with a resumable, cursor-based scan. +/// +/// # For implementors +/// +/// Every implementation must decide five things before the first line of +/// code. Answer them in comments at the top of the impl: +/// +/// 1. **Direction.** Backward (storage → DB) surfaces orphans; forward +/// (DB → storage) surfaces missing. Most checks do BOTH — sequence +/// them and encode the current phase in the cursor. +/// +/// 2. **Severity per finding kind.** `DataLoss` (user impact) / +/// `Reclaimable` (disk waste) / `Regenerable` (auto-heals) / +/// `Drift` (accounting). The single most common mistake is treating +/// a missing user-uploaded thumbnail as `Regenerable` — it's not, +/// the server can't recreate what the user provided. It's `DataLoss`. +/// +/// 3. **Cursor format.** Opaque `Vec` to the framework. Yours to +/// serialize. Content-addressable blobs → 32-byte BLAKE3. UUID rows → +/// 16-byte UUID. Path rows → the path bytes. Multi-phase check → +/// prepend a phase byte. +/// +/// 4. **Grace window.** Default 1 h (matches dedup GC). Deep checks +/// (hash verification) use 24 h. Grace ALWAYS refers to +/// `scan_started_at`, never `NOW()` — see trap #1 below. +/// +/// 5. **Batch boundary.** 1000 items or 30 s. Call `store.checkpoint` +/// and `store.should_cancel` between batches — cancellation is +/// cooperative, never task-abort. +/// +/// # Two-pass discipline +/// +/// Pass 1 — build candidate list from a snapshot read, excluding items +/// younger than `grace_window`. +/// +/// Pass 2 — per candidate, re-read the OTHER side's state right before +/// flagging. If it transitioned (ref went up, blob just landed, row was +/// deleted), silently drop. +/// +/// Pass 1 alone LOOKS complete but produces false positives on every +/// race. Never skip Pass 2. +/// +/// # Canonical example +/// +/// See `BlobConsistencyCheck` in +/// `src/infrastructure/services/consistency/blob_check.rs` — it exercises +/// every axis (both directions, both severities, grace window, cursor, +/// cooperative cancel, blob-level report shape for missing findings). +#[async_trait] +pub trait ConsistencyCheck: Send + Sync { + /// Machine-readable name — appears in the admin endpoint slug and in + /// audit `event` values. Lowercase snake_case, one per check. + fn name(&self) -> &'static str; + + fn grace_window(&self) -> Duration { Duration::from_secs(3600) } + + /// `true` (default) → safe to run against live traffic; the check + /// respects grace window + two-pass re-verify. Only false for a + /// check that genuinely needs a quiescent DB (whole-table snapshot + /// verification of hashes) — not required for anything in v1-v5. + fn concurrent_safe(&self) -> bool { true } + + /// `cursor: None` → fresh run. `Some(bytes)` → resume from last + /// persisted checkpoint. Impls MUST: + /// - call `store.checkpoint(cursor).await` between batches + /// (~1000 items or 30 s, whichever comes first); + /// - call `store.should_cancel().await` between batches — return + /// `RunOutcome::Paused { cursor }` when it returns `true`; + /// - use `store.scan_started_at()` (not `now()`) as the grace + /// window reference. + async fn run_resumable( + &self, + opts: &CheckOptions, + cursor: Option>, + store: &dyn CheckStore, + ) -> Result; +} + +#[derive(Debug)] +pub enum RunOutcome { + Completed, + Paused { cursor: Vec }, + Failed(DomainError), +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Severity { + DataLoss, // user impact — top of triage + Reclaimable, // disk waste, no user impact + Regenerable, // auto-heals on next request + Drift, // accounting mismatch, no user impact +} + +pub struct Inconsistency { + pub kind: &'static str, // "OrphanBlob", "MissingBlob", ... + pub severity: Severity, + pub resource_id: String, // opaque + pub detail: serde_json::Value, +} +``` + +### `CheckStore` + +The framework hands each check a `CheckStore` — the only side effect a +check performs on shared state. + +```rust +#[async_trait] +pub trait CheckStore: Send + Sync { + fn run_id(&self) -> Uuid; + fn scan_started_at(&self) -> chrono::DateTime; + + /// Persist the cursor + last-progress timestamp. Called between + /// batches. If a crash happens after this returns, the next resume + /// starts from `cursor`. + async fn checkpoint(&self, cursor: Vec, scanned_count: u64) + -> Result<(), DomainError>; + + /// Poll the run's `status` column. Returns `true` when an admin + /// requested cancellation. The check MUST return `Paused` with the + /// current cursor. + async fn should_cancel(&self) -> Result; + + /// Upsert a finding. `UNIQUE (run_id, kind, resource_id)` in the + /// schema means re-scanning the same resource on resume is safe. + async fn record_finding(&self, finding: Inconsistency) + -> Result<(), DomainError>; + + /// Delete a previously-recorded finding — used when Pass 2 sees + /// the resource transitioned out of the inconsistent state. + async fn drop_finding(&self, kind: &str, resource_id: &str) + -> Result<(), DomainError>; +} +``` + +### `StatefulAdapter` + +```rust +/// Marker + registration trait for any adapter that persists state OUTSIDE +/// process memory: blobs on disk / S3, DB tables, on-disk caches, message +/// queues you own. +/// +/// Added as a SUPERTRAIT on every state-owning port +/// (`trait BlobStorage: StatefulAdapter`, `trait ThumbnailStore: +/// StatefulAdapter`, `trait FolderRepository: StatefulAdapter`, …), which +/// means: NO NEW ADAPTER CAN COMPILE without declaring its consistency +/// contract. The compiler is the enforcement; these doc-comments are the +/// education. +/// +/// # For implementors adding a new stateful adapter +/// +/// You cannot skip this trait. If you're reading this because your PR +/// won't compile, work through: +/// +/// 1. **Am I actually stateful?** State means "bytes or rows outside +/// process memory that can desync from other subsystems". Config, +/// caches keyed by session, and derived indexes are NOT stateful +/// for this purpose (they can be dropped and rebuilt). If you're +/// not stateful, drop the `StatefulAdapter` impl entirely — but +/// then your port shouldn't have `StatefulAdapter` as a supertrait +/// either, so this compile error means the port author already +/// decided you were. +/// +/// 2. **What are the DIRECTIONS of drift I can detect?** Almost every +/// stateful adapter has both: +/// - Backward (my storage → the DB that references it): orphans. +/// - Forward (the DB → my storage): missing. +/// Return one check that covers both by sequencing phases, OR two +/// checks (one per direction). The former is easier to operate. +/// +/// 3. **What's the SEVERITY of each finding?** See `Severity` in +/// `consistency_check.rs`. Missing user-uploaded data is `DataLoss`; +/// missing server-derived data is `Regenerable`; orphan bytes are +/// `Reclaimable`; accounting drift is `Drift`. +/// +/// 4. **What CURSOR fits my walk?** Content-addressable → hash prefix. +/// UUID-keyed → UUID lex. Path-keyed → path bytes. Whatever you pick, +/// it's opaque `Vec` to the framework — decode inside your check. +/// +/// See `BlobConsistencyCheck` for the canonical impl to copy-adapt. +pub trait StatefulAdapter: Send + Sync { + /// Subsystem slug — appears in `POST /api/admin/internal/consistency/{name}` + /// and in audit log `event` values. Lowercase snake_case, unique + /// per adapter. Convention: `"blobs"`, `"thumbnails"`, `"trash"`, + /// `"folder_tree"`, `"used_bytes"`. + fn subsystem(&self) -> &'static str; + + /// REQUIRED (no default impl). Return every consistency check + /// this adapter contributes. Most adapters return exactly one. + /// Multi-keying subsystems return more — `ThumbnailStore` returns + /// FOUR checks (server-generated + user-uploaded, each in both + /// directions). + /// + /// Returning `vec![]` is a red flag. If your adapter has state but + /// no check, either: + /// - Your state is fully covered by another adapter's check + /// (rare — document exactly WHERE in a comment on this method). + /// - You haven't written the check yet — return + /// `vec![]` with a `TODO(consistency): add ConsistencyCheck` + /// comment, ship the trait wiring, add the check in a follow-up PR. + /// + /// Reviewers will grep `TODO(consistency)` and ask when it lands. + fn consistency_checks(&self) -> Vec>; +} +``` + +### `ConsistencyRegistry` + +```rust +/// Collects `StatefulAdapter`s at wire-up time. Instantiated once in +/// `AppServiceFactory`, exposed on `AppState`, consumed by the admin +/// handler + (when JobRegistry lands) the scheduler. +pub struct ConsistencyRegistry { + adapters: Vec>, +} + +impl ConsistencyRegistry { + pub fn register(&mut self, adapter: Arc) { + // Trait bound forces `subsystem()` + `consistency_checks()` to exist. + self.adapters.push(adapter); + } + + /// Every check contributed by every registered adapter, flat. + pub fn all_checks(&self) -> Vec> { + self.adapters + .iter() + .flat_map(|a| a.consistency_checks()) + .collect() + } + + pub fn get(&self, name: &str) -> Option> { + self.all_checks().into_iter().find(|c| c.name() == name) + } +} +``` + +## Admin surface + +``` +POST /api/admin/internal/consistency/{name} + → 202 { run_id } (starts a new run) + +POST /api/admin/internal/consistency/runs/{id}/cancel + → 200 { status: "CancelRequested" } + (cooperative — check finishes its current batch and returns Paused) + +POST /api/admin/internal/consistency/runs/{id}/resume + → 202 { run_id } (picks up cursor) + +GET /api/admin/internal/consistency/runs?check=&status= + → 200 [{ id, check_name, status, scanned_count, last_progress_at, … }] + +GET /api/admin/internal/consistency/runs/{id} + → 200 { run: {...}, findings: [...paginated] } +``` + +Gated by `OXICLOUD_ENABLE_ADMIN_INTERNAL_ENDPOINTS` — same admin-guard +middleware as `trigger-sweep`, `trigger-gc`, `trigger-grant-cleanup`. + +## Approach + +### 1. Traits + framework in isolation + +`src/application/ports/consistency.rs` +- Define `ConsistencyCheck`, `RunOutcome`, `Severity`, `Inconsistency`, + `CheckStore`, `StatefulAdapter`, `CheckOptions`. +- Full doc-comments as sketched above — these are the educational + surface, don't cut them. + +`src/infrastructure/services/consistency/mod.rs` +- `ConsistencyRegistry` (data structure only). +- `PgCheckStore` — impl of `CheckStore` reading/writing + `admin.background_runs` (filtered to `job_name LIKE 'consistency_%'`) + + `admin.consistency_findings`. +- `run_check(check, cursor, store)` — the runner that calls + `run_resumable`, applies timeout, records outcome. + +### 2. Schema migration + +`migrations/YYYYMMDDHHMMSS_background_runs_admin_schema.sql` — creates +the merged `admin.background_runs` table shared with the JobRegistry +plan. Consistency checks own the `admin.consistency_findings` table +alone and reference `background_runs.id` via FK. + +```sql +CREATE SCHEMA IF NOT EXISTS admin; + +CREATE TABLE admin.background_runs ( + id UUID PRIMARY KEY, + job_name TEXT NOT NULL, -- 'consistency_blobs', 'storage_migration', 'reextract_audio', ... + status TEXT NOT NULL, -- Running / Paused / Completed / Failed / CancelRequested + started_at TIMESTAMPTZ NOT NULL, + last_progress_at TIMESTAMPTZ NOT NULL, + completed_at TIMESTAMPTZ, + cursor BYTEA, + stats JSONB NOT NULL DEFAULT '{}'::jsonb, -- e.g. {"scanned_count": 12345} + params JSONB NOT NULL DEFAULT '{}'::jsonb, -- e.g. {"grace_window_secs": 3600} + error_message TEXT +); +CREATE UNIQUE INDEX one_active_run_per_job + ON admin.background_runs (job_name) + WHERE status IN ('Running', 'Paused'); +CREATE INDEX ON admin.background_runs (last_progress_at) WHERE status = 'Running'; + +CREATE TABLE admin.consistency_findings ( + id UUID PRIMARY KEY, + run_id UUID NOT NULL REFERENCES admin.background_runs(id) ON DELETE CASCADE, + kind TEXT NOT NULL, + severity TEXT NOT NULL, + resource_id TEXT NOT NULL, + detail JSONB NOT NULL DEFAULT '{}'::jsonb, + found_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + UNIQUE (run_id, kind, resource_id) +); +CREATE INDEX ON admin.consistency_findings (run_id, severity); +``` + +### 3. Supertrait bounds on existing state-owning ports + +Add `StatefulAdapter` as a supertrait on: + +- `src/application/ports/storage_ports.rs::BlobStorage` + (or wherever the blob-storage port lives). +- `src/application/ports/thumbnails.rs::ThumbnailStore` (both server-generated + and user-uploaded paths). +- `src/application/ports/text_extraction.rs::TextExtractionCache`. +- `src/application/ports/audio_metadata.rs::AudioMetadataCache` (if a + distinct port exists). +- `src/domain/repositories/file_blob_read_repository.rs::FileBlobReadRepository` + (via the port trait it exposes to application services). +- `src/domain/repositories/folder_repository.rs::FolderRepository`. +- `src/domain/repositories/trash_repository.rs::TrashRepository`. +- `src/infrastructure/services/webdav_dead_property_store.rs` + (`DeadPropertyStore` — has its own leak class per the deferred-rekey + memory note). + +Each of these will trigger compile errors in its impls. Each impl gets +a two-line stub: + +```rust +impl StatefulAdapter for LocalFsBlobStorage { + fn subsystem(&self) -> &'static str { "blobs" } + fn consistency_checks(&self) -> Vec> { + // TODO(consistency): add BlobConsistencyCheck once impl lands. + vec![] + } +} +``` + +Ship this PR without the actual checks. Grep `TODO(consistency)` = punch list. + +### 4. First real check — `BlobConsistencyCheck` + +`src/infrastructure/services/consistency/blob_check.rs` + +- Depends on `BlobStorage` (storage listing) + `FileBlobReadRepository` + (DB SELECT). +- Phase 1 (orphan direction): walk storage listing, cursor on hash prefix. + Batch of 1000, checkpoint, cancel-poll. For each batch: SELECT ref_count + FROM `storage.file_blobs` WHERE hash IN (…). Pass 2 re-verifies at flag + time. Severity: `Reclaimable`. +- Phase 2 (missing direction): walk `file_blobs` ordered by hash, cursor + on hash. Batch of 1000. For each row: `HEAD` on storage backend. If + missing AND row hasn't disappeared AND row is older than + `grace_window`, flag `MissingInStorage` with `affected_file_ids` from + a JOIN to `file_metadata`. Severity: `DataLoss`. +- Cursor format: `[phase: u8, hash_key: 32 bytes]`. +- `LocalFsBlobStorage::consistency_checks()` returns + `vec![Arc::new(BlobConsistencyCheck::new(self.clone(), ...))]`. + +### 5. Admin handlers + +`src/interfaces/api/handlers/admin_handler.rs` + +- `start_consistency_check(name, force)` — insert an + `admin.background_runs` row with `job_name = 'consistency_'` + and `status = 'Running'`, spawn a tokio task calling `run_check`, + return `run_id`. Concurrent triggers hit the partial unique index + and short-circuit to returning the surviving row. +- `cancel_run(id)` — UPDATE status = 'CancelRequested'. +- `resume_run(id)` — verify status == 'Paused', spawn task with the + persisted cursor. +- `list_runs(filter)` — SELECT with filters + paginate. +- `get_run(id)` — SELECT run + paginated findings. + +Same admin-guard + `OXICLOUD_ENABLE_ADMIN_INTERNAL_ENDPOINTS` gate as +existing internal endpoints. + +### 6. Boot-time crashed-run recovery + +In `AppServiceFactory` init, after DB pool is up: + +```rust +sqlx::query!( + "UPDATE admin.background_runs + SET status = 'Paused', + error_message = COALESCE(error_message, 'server restart mid-run') + WHERE job_name LIKE 'consistency_%' + AND (status = 'Running' OR status = 'CancelRequested')" +).execute(&pool).await?; +``` + +Filtering on `job_name LIKE 'consistency_%'` scopes the sweep to +consistency runs; other tenants of `background_runs` (storage +migration, reextract-*) run the same auto-Pause sweep from their own +boot-time hook. The JobRegistry supervisor's boot check may +generalise this into a single scheduler-wide sweep — until then, one +per subsystem is fine. + +Do NOT auto-resume — the bug that killed the last run may still be there. +Log a warning if any rows were flipped so operators notice. + +### 7. Hurl regression — `tests/api/consistency_check.hurl` + +- Setup: login admin, seed one file (which creates one blob). +- Trigger `blobs` check with `force=true` (grace_days=0). Poll runs + list until `status='Completed'`. Assert 0 findings. +- Manually orphan a blob (SQL: `DELETE FROM file_metadata WHERE …`, + leave `file_blobs` + storage in place). Trigger check again. + Assert 1 finding with `kind='OrphanInStorage'`, `severity='Reclaimable'`. +- Manually break a blob (SQL: leave `file_blobs` alone, wipe the + storage backend for that hash — actually, use the storage service's + test hook if one exists; otherwise skip this in Hurl and cover in + integration tests). +- Cancel a run mid-scan (large seed, poll for scanned_count > 0, POST + cancel, poll until status='Paused'). Resume. Assert scanned_count + after resume > checkpoint. + +### 8. Follow-up PRs (remaining checks) + +Priority order: + +| # | Check | Direction | Complexity | +|---|---|---|---| +| 1 | `BlobConsistencyCheck` | both | high (canonical) | +| 2 | `ThumbnailConsistencyCheck` | both × 2 subspaces = 4 sub-scans | high | +| 3 | `UsedBytesConsistencyCheck` | pure SQL | low — wrap existing reconciliation diff | +| 4 | `FolderTreeConsistencyCheck` | pure SQL | low — closure over `folders.parent_id` | +| 5 | Deep-hash sub-mode on `BlobConsistencyCheck` | forward | medium — 24 h grace, opt-in | +| 6 | `DeadPropertyConsistencyCheck` | forward | low, blocked on rekey (see `project_webdav_dead_properties_drive_rekey`) | +| 7 | `TrashConsistencyCheck` | both | medium — trash rows vs `file_metadata` soft-delete flags | + +Each is a separate PR against the stable trait. `TODO(consistency)` +count decreases by one per PR. + +## Critical files + +**Create:** +- `src/application/ports/consistency.rs` (~250 lines — traits + doc-comments) +- `src/infrastructure/services/consistency/mod.rs` (~40 lines — pub types) +- `src/infrastructure/services/consistency/registry.rs` (~80 lines) +- `src/infrastructure/services/consistency/pg_check_store.rs` (~150 lines) +- `src/infrastructure/services/consistency/runner.rs` (~100 lines) +- `src/infrastructure/services/consistency/blob_check.rs` (~300 lines — canonical impl) +- `migrations/YYYYMMDDHHMMSS_consistency_check_admin_schema.sql` (~30 lines) +- `tests/api/consistency_check.hurl` (~150 lines) + +**Modify (add supertrait bound):** +- `src/application/ports/storage_ports.rs` — `BlobStorage: StatefulAdapter`. +- `src/application/ports/thumbnails.rs` — `ThumbnailStore: StatefulAdapter`. +- `src/application/ports/text_extraction.rs`. +- `src/application/ports/audio_metadata.rs` (if applicable). +- `src/domain/repositories/file_blob_read_repository.rs`. +- `src/domain/repositories/folder_repository.rs`. +- `src/domain/repositories/trash_repository.rs`. +- `src/infrastructure/services/webdav_dead_property_store.rs`. + +**Modify (add `StatefulAdapter` stubs):** +- Every impl of the above ports. Each gets `subsystem()` + a `vec![]` stub + with `TODO(consistency)`. + +**Modify (wire up admin surface):** +- `src/common/di.rs` — build `Arc`, expose on + `AppState`, register every stateful adapter. +- `src/interfaces/api/handlers/admin_handler.rs` — five handlers. +- `src/interfaces/api/routes.rs` — five routes. +- `src/interfaces/api/mod.rs` — utoipa paths. +- `tests/api/run.sh` — register `consistency_check.hurl`. + +## Reused existing utilities + +- **Admin-guard + gate pattern** at + `src/interfaces/api/handlers/admin_handler.rs::internal_trigger_gc` — + same shape for the new endpoints. +- **`OXICLOUD_ENABLE_ADMIN_INTERNAL_ENDPOINTS` gate** — same env var. +- **Dedup GC's orphan-detection logic** (`dedup_service.rs`) — the + algorithmic template for `BlobConsistencyCheck`'s orphan phase. + Reference impl, not a callsite — the check needs its own two-pass + discipline; GC currently reap-and-forgets. +- **Reconciliation SQL diff** in `storage_usage_service.rs` — becomes + `UsedBytesConsistencyCheck` almost verbatim, wrapped in report-only mode. +- **`AGENTS.md` audit convention** — every finding double-logs to + `target: "audit"`, `event: "consistency.{check}.finding"`, plus + operational log to `target: "oxicloud::consistency"`. + +## Verification + +1. **Compile**: `cargo check --all-features --all-targets` + + `cargo clippy -- -D warnings` clean. +2. **Schema**: `just fe-nothing … cargo run` starts; migration lands + the `admin` schema; `psql -c "\dt admin.*"` shows the two tables. +3. **Boot line**: `consistency: N adapter(s) registered, M check(s) + available`. Grep `TODO(consistency)` in the source; count should + equal M in v1 minus the shipped `BlobConsistencyCheck`. +4. **Hurl** (`tests/api/consistency_check.hurl`): + - clean state → 0 findings + - forced orphan → 1 `OrphanInStorage` finding, severity `Reclaimable` + - cancel + resume round-trip preserves `scanned_count` +5. **Crash recovery**: kill server mid-scan (`kill -9`); restart; + confirm the row is `Paused` with `error_message='server restart + mid-run'`; POST resume; check completes. +6. **Trait enforcement**: add a new dummy adapter impl of `BlobStorage` + without `StatefulAdapter` — compile MUST fail. Add the stub; compile + succeeds. This is the load-bearing property of the design. +7. **Grace-window sanity**: run against a fresh 10 s window; upload a + file mid-scan; confirm the young blob does NOT surface as + `MissingInStorage` (grace window covers it). +8. **Env-flag off**: `OXICLOUD_ENABLE_ADMIN_INTERNAL_ENDPOINTS=false` + → endpoints return 404, no leakage in the audit channel. + +## Out of scope + +- **JobRegistry integration**. Consistency checks are admin-triggered + in v1. When `docs/plan/job-registry.md` lands, JobRegistry will + consume `ConsistencyRegistry::all_checks()` for scheduled execution + — no code change needed here. +- **Auto-repair**. Findings are reported, not fixed. Repair primitives + live in the existing services (dedup GC's reaper, storage_usage + reconciler); a future admin surface could trigger targeted repair + after human review. +- **Distributed scheduling**. Single-process. If OxiCloud ever runs + multi-node, add `SELECT … FOR UPDATE SKIP LOCKED` on the run rows. +- **Byte-exact whole-table snapshot verification**. The `concurrent_safe + = false` case — reserved for a future `DeepBlobConsistencyCheck` that + requires either `pg_export_snapshot` + S3-consistent list OR a + read-only mode. Not needed for v1-v5. +- **Cursor pagination on the `GET /runs/{id}` findings list**. Simple + offset/limit for v1. Add cursor only if operators actually hit a + 10k-findings run. +- **Findings retention**. Runs + findings accumulate forever until an + operator manually deletes. Add a background cleaner once volume + actually matters — most likely alongside JobRegistry. +- **Auto-scheduling in v1**. No `tokio::spawn` interval loop. Admin + triggers only. Every scheduled invocation goes through JobRegistry + when it lands. + +## Related memory notes + +- `feedback_no_abbreviated_env_vars` — full-word env var names if any + land (e.g. `OXICLOUD_CONSISTENCY_BATCH_SIZE`, not + `OXICLOUD_CC_BATCH`). +- `project_consistency_check_trait` — the memory that captures this + design's decisions and the traps that shape the trait. +- `project_dedup_gc_test_trigger` — motivates the check's grace-window + discipline; also the source of the algorithmic template for the + orphan-blob direction. +- `project_webdav_dead_properties_drive_rekey` — `DeadPropertyStore` + will get a check, but only after the rekey lands. +- `bug_thumbnail_dedup`, `bug_folder_cascade_hooks_missing` — surface + the four-sub-scan complexity of `ThumbnailConsistencyCheck`. +- `bug_orphan_seed_null_orphaned_at_flaky` — reminds implementors that + the orphan-blob direction MUST check `orphaned_at`, not just + `ref_count = 0`. diff --git a/docs/plan/job-registry.md b/docs/plan/job-registry.md new file mode 100644 index 00000000..c6cb1b83 --- /dev/null +++ b/docs/plan/job-registry.md @@ -0,0 +1,798 @@ +# Plan — Job engines (periodic + recoverable) + admin surface + +## Context + +OxiCloud runs several fire-and-forget background daemons today, each +spawned by a service factory in `src/common/di.rs` at startup: + +| Service | Cadence | Shape | +|---|---|---| +| `TrashCleanupService` | every 24 h | Fixed interval, no per-run state | +| `StorageUsageService::start_reconciliation_job` | every 600 s | Fixed interval, no per-run state | +| `db_pool_monitor` | every N s | Fixed interval, no per-run state | +| `dedup_service` GC | on demand + inline | Fixed interval, no per-run state | +| `GrantCleanupService` | every 24 h | Fixed interval, no per-run state | +| `tree_etag_flush_job` | every ~500 ms | Fixed interval, no per-run state | +| `content_index` worker | continuous | Fixed interval, no per-run state | +| Blob storage backend migration | admin-triggered | Long-running, cursor, resumable, in-memory state today | +| `admin/audio/metadata/reextract` | admin-triggered | Long-running, blocks HTTP request today | +| `admin/photos/metadata/reextract` | admin-triggered | Long-running, blocks HTTP request today | +| `ConsistencyCheck` runs (see `docs/plan/consistency-check.md`) | admin-triggered v1 | Long-running, cursor, resumable, needs DB state | + +Two shapes bleed together in the current codebase but shouldn't. Each +daemon reinvents its own env var pattern, admin trigger endpoint, +logging schema, and (for the long-running ones) its own in-memory +progress state that vanishes on restart. + +## Two engines, one file + +This plan is intentionally two plans in one file (Ed 2026-07-27), +because the two engines share an admin URL prefix, a config-var +convention, and a logging target — but nothing else: + +- **Part 1 — Periodic Scheduler.** In-memory registration + tokio + interval loop. Serves fixed-interval jobs an operator might trigger + manually. No DB tables, no cursor, no per-run persistence. +- **Part 2 — Recoverable-Run Engine.** DB-backed cursor persistence + + exclusivity + crash recovery. Serves the four long-running tenants + (storage-migration, reextract-audio, reextract-image, consistency + check runs) and any future work that iterates over a large space + with restart tolerance. + +A recoverable job CAN optionally be periodically-triggered (register +once in each engine; Part 1's tick calls Part 2's `run_or_resume` +instead of a bare handler). Most Layer B tenants are admin-triggered +only. + +Cross-cutting concerns (admin URL taxonomy, env vars, logging target, +plugin future) live in a shared section at the bottom so we're not +duplicating them between parts. + +## Migration criterion — the trigger question + +Not every background loop belongs in JobRegistry. The single question +that decides: + +> **"Would an operator plausibly `POST /trigger-job/{name}` to make +> it run right now?"** + +**Yes → migrate.** The whole payoff of JobRegistry is a uniform +*operator surface* — list, trigger, last-outcome, log line, config +knobs. If nobody would ever manually trigger the job, the surface +delivers no value; you're paying framework overhead for nothing. +Anything an operator would manually trigger is by definition +periodic + discrete + meaningful. + +**No → leave it as its own loop.** Continuous drains and +event-reactive workers ("core workers") fail this test — "trigger +the content-index worker" makes no sense; it's already running. +Standardise their env var naming and log target as a light +convention (see [Cross-cutting](#cross-cutting) below) but do NOT +wedge them into the scheduler. + +Secondary confirmation questions — if the primary is yes and any of +these is no, migrate anyway but flag the mismatch: + +1. Does each invocation report a meaningful `count` (rows swept, + blobs GC'd, bytes reclaimed)? Continuous workers don't have + discrete invocations to count. +2. Does the operator tune it via env vars beyond enable/disable? +3. Would an operator want a "did this run within the last N?" health + signal? Periodic jobs benefit from `last_outcome`; always-on + workers need liveness signals of a different shape. + +**Cadence is NOT the trigger** — it's a symptom. Sub-second jobs +almost always fail the primary question (nobody manually triggers +something that fires 2× per second), but a hypothetical 1 s periodic +job that operators do want to kick still belongs in JobRegistry. +Cadence tells you "probably no"; the operator-trigger question is +what decides. + +### Applied to the current daemons + +| Service | Operator-trigger? | Destination | +|---|---|---| +| `TrashCleanupService` | Yes — "purge expired trash now" | Part 1 | +| `StorageUsageService::start_reconciliation_job` | Yes — "recompute quotas now" | Part 1 | +| `dedup_service` GC | Yes — already has `trigger-gc` | Part 1 | +| `GrantCleanupService` | Yes — already has `trigger-grant-cleanup` | Part 1 | +| `tree_etag_flush_job` | No — a "flush now" is meaningless (queue drains itself) | Core worker, unchanged | +| `content_index` worker | No — continuous drain, no discrete invocation | Core worker, unchanged | +| `db_pool_monitor` | No — "log stats now" is either grep-existing-logs or attach-a-debugger, not a scheduled job trigger | Core worker, unchanged | +| Blob storage backend migration | Yes — already admin-triggered | Part 2 | +| `admin/audio/metadata/reextract` | Yes — currently admin-triggered (synchronously) | Part 2 | +| `admin/photos/metadata/reextract` | Yes — currently admin-triggered (synchronously) | Part 2 | +| `ConsistencyCheck` runs | Yes — needs a trigger endpoint | Part 2 | + +The `db_pool_monitor` case is illustrative: cadence-wise it *could* +fit Part 1 (10-30 s periodic, bounded work), but the operator-trigger +question kills it. Nobody manually triggers a stats-log because logs +are already there. Keeping it as its own loop is right. + +## Implementation order + +1. **Part 1 lands first** — small, self-contained, unblocks migration + of trash-cleanup + storage-usage + db_pool_monitor + dedup GC + + grant-cleanup + tree-etag flush + content-index. High mechanical + payoff, zero new schema, minimal review surface. +2. **Part 2 lands next** — introduces `admin.background_runs` schema + + `RecoverableJob` trait + `JobStore` port + `run_or_resume`. On + its own PR (schema change deserves independent review). +3. **Consistency-check framework (`docs/plan/consistency-check.md`)** + lands third, consuming Part 2 as its runtime. +4. **Storage-migration and reextract-* migrated to Part 2** as + follow-ups. + +--- + +## Part 1 — Periodic Scheduler + +### Contract — `JobHandler` trait + +The implementor-facing surface for a fixed-interval job: + +```rust +#[async_trait] +pub trait JobHandler: Send + Sync { + /// Stable snake_case identifier. Must be unique across the process. + /// Log lines, admin listing, env vars, and trigger URLs all key on + /// this name. + fn name(&self) -> &str; + + /// One execution. Called by the supervisor at the registered + /// interval and (optionally) on admin trigger. Return `Ok { count, + /// extra }` on success — the count is the primary scalar the job + /// reports (rows swept, ETags flushed, blobs GC'd). Return + /// `Err(msg)` on failure; the supervisor logs it and continues. + async fn run(&self) -> JobOutcome; +} +``` + +Native services implement this trait on an existing service type (no +new wrapper) and register a single `Arc` with the +scheduler. + +### `JobOutcome` + +```rust +pub enum JobOutcome { + Ok { count: u64, extra: serde_json::Value }, + Err(String), +} +``` + +Two variants only. Every reason a run can fail (handler returned an +error, wall-clock timeout, panic caught by the supervisor) collapses +to `Err(String)`, with the *cause* encoded in the message AND in a +`cause` tracing field the supervisor sets when it emits the log line: + +- Handler returned `Err(msg)` → `cause = "handler"`, message = `msg`. +- `tokio::time::timeout` tripped → `cause = "timeout"`. +- `catch_unwind` caught a panic → `cause = "panicked"`, message = the + payload as a string. + +Handlers never construct the cause themselves; they either return +`Ok { count, extra }` or `Err(String)`. Keeping the enum to two +variants prevents every consumer of `match outcome` from having to +distinguish diagnostic sub-cases that behave identically for logging, +persistence, retry, and admin display. + +### Runtime model + +- **One `tokio::spawn`** at startup runs the scheduler main loop. + Sleeps until the earliest due job, dispatches, sleeps again. +- Per-run **panic catching** via `tokio::spawn` inside the dispatch + (or `AssertUnwindSafe` + `catch_unwind`). A bad handler crashes + its own run, not the scheduler. +- **Sequential dispatch within a tick** by default. Two jobs due at + the same instant run one after the other. Parallel dispatch can + layer on later as a per-job toggle if a real need appears — most + handlers touch the DB and don't benefit from concurrency. +- **`ScheduledJob.timeout: Option`** is applied by the + supervisor via `tokio::time::timeout` when set. Optional; use it + when the handler has a real wall-clock bound. None means "let it + run to completion." + +Single supervisor is chosen for **operational** clarity, not runtime +cost: one place to observe, one panic-containment boundary, one +config surface, one plugin-registration hook when plugins land. + +### Exclusivity — one in-flight run per `job_name` + +Mirrors Part 2's exclusivity invariant, enforced in-memory since +Part 1 has no DB row: + +- Each `RegisteredJob` carries an `is_running` flag (an + `AtomicBool` or single-permit `Semaphore`). +- Before dispatching a tick, the supervisor tries to acquire the + flag. If it's already held (the previous run is still executing), + the tick is **skipped, not queued**: + + ```rust + tracing::warn!( + target: "oxicloud::scheduler", + event = "job.tick_skipped", + job = %name, + interval_ms = interval.as_millis(), + running_for_ms = current_run_start.elapsed().as_millis(), + "{name} still running past its interval — tick skipped" + ); + ``` + + `next_run_at` advances by one interval so the schedule stays on + its cadence rather than queueing backlog. +- On completion (or panic caught by the supervisor), the flag is + released. The next tick is free to fire. +- **Diagnostic value.** A `job.tick_skipped` line on every interval + is the operator signal that either the job is chronically slower + than its cadence (retune the interval) or hung (attach a debugger + / set a timeout / kill the process). Without this warning a slow + or hung handler would silently starve. +- **Interaction with timeout.** If a job has a `timeout` configured + and it trips, the supervisor kills the run and releases the flag. + Timeouts prevent hangs from permanently silencing a job. + Handlers without a timeout can, in principle, hang forever — the + repeated `tick_skipped` warning is the only signal. + +Cross-job concurrency is unchanged — different `job_name`s can run +sequentially per tick as described above. Exclusivity is per +job_name, not global. + +### `JobRegistry` + +```rust +pub struct JobRegistry { + jobs: RwLock>, +} + +struct RegisteredJob { + handler: Arc, + interval: Duration, + timeout: Option, + /// Single-permit gate that enforces the "one in-flight run per + /// `job_name`" invariant (see Exclusivity above). A tick that + /// finds the permit taken emits `job.tick_skipped` and does not + /// spawn. + in_flight: Arc, // capacity = 1 + /// Set when a run starts, cleared when it ends. Used to include + /// `running_for_ms` in the skip warning. + current_run_start: Arc>>, + last_outcome: Option<(chrono::DateTime, JobOutcome)>, + next_run_at: chrono::DateTime, +} +``` + +`Arc` lives on `AppState`. Native services register +themselves during DI: + +```rust +registry.register( + Arc::clone(&trash_cleanup) as Arc, + Duration::from_secs(interval_hours * 3600), + None, // no timeout +); +``` + +### Engine loop + +```rust +async fn run(registry: Arc) { + loop { + let next = registry.pick_next().await; // earliest next_run_at + let sleep = next.deadline().saturating_duration_since(Instant::now()); + tokio::time::sleep(sleep).await; + + let outcome = registry.dispatch(&next.name).await; + registry.record_outcome(&next.name, outcome).await; + } +} +``` + +`dispatch` grabs the handler under a read lock, spawns a task, applies +the timeout, catches panics, and returns the `JobOutcome`. Sequential +dispatch is intentional; two jobs due at the same instant run +one-after-the-other. + +### Native tenants and migration order + +Four services satisfy the operator-trigger criterion above and migrate: + +1. **trash-cleanup** — simplest self-contained loop; reference for the + migration shape. Ships with Part 1's landing PR. +2. **storage-usage reconciliation** — same shape, different service. +3. **dedup GC** — already has `trigger-gc`; the shim forwards to the + new registry-backed trigger. +4. **grant-cleanup** — already has `trigger-grant-cleanup`; same shim + pattern. + +Three services are **core workers** and STAY on their own loops +(fail the operator-trigger question — see the criterion table above): + +- `tree_etag_flush_job` — 500 ms queue-drain, coalescing semantics. +- `content_index` worker — continuous channel drain, event-reactive. +- `db_pool_monitor` — periodic stats-log with no discrete-invocation + count and no operator use for manual trigger. + +Standardise their env var naming (`OXICLOUD_JOB__*`) and +tracing target for uniform operator ergonomics, but do NOT wedge them +into the scheduler. + +### Verification (Part 1) + +1. **Compile**: `cargo check --all-features --all-targets` + + `cargo clippy -- -D warnings` clean. +2. **Boot**: start server; expect `scheduler started, N job(s) registered`. +3. **Admin listing**: + ``` + curl -s http://localhost:8086/api/admin/internal/jobs -H "Authorization: Bearer $TOKEN" + ``` + returns a JSON array with each registered job, its `interval_ms`, + `next_run_at`, and `last_outcome` (null until first tick). +4. **Trigger**: `POST /api/admin/internal/trigger-job/trash_cleanup` + invokes the handler immediately, records the outcome. +5. **Panic containment**: unit test a handler that panics; `last_outcome` + records `Err(...)` with `cause = "panicked"` in the log; the scheduler + is still alive (verified by triggering another job); the in-flight + permit is released so the next tick can fire. +6. **Timeout enforcement**: unit test a handler that blocks longer than + its declared timeout; `last_outcome` records `Err(...)` with + `cause = "timeout"`; the in-flight permit is released. +7. **Overrun exclusivity**: unit test a handler with a 100 ms interval + that sleeps 300 ms. Assert exactly ONE run is in flight at any moment + (no parallel dispatch), and that two `job.tick_skipped` log events + fire (one at each missed tick) with `running_for_ms` monotonically + increasing. +8. **Shim compatibility**: existing per-service trigger endpoints + (`trigger-sweep`, `trigger-gc`, `trigger-grant-cleanup`) keep working + as thin forwards. Existing api-test Hurl suites pass unchanged. + +--- + +## Part 2 — Recoverable-Run Engine + +### Contract — `RecoverableJob` trait + +Sibling to `JobHandler`, NOT a subtrait. A stateless job that only +implements `JobHandler` never needs to know Part 2 exists. + +```rust +#[async_trait] +pub trait RecoverableJob: Send + Sync { + /// Stable snake_case identifier — matches the `job_name` column + /// in `admin.background_runs`. + fn name(&self) -> &str; + + /// Long-running, cooperative scan. The store is the job's ONLY + /// side effect: cursor checkpointing, cancel polling, run-state + /// updates all go through it. + /// + /// Between batches the handler MUST poll `store.status()` — a + /// `CancelRequested` return means the operator asked for a pause + /// and the handler should return `Paused { cursor }` at the next + /// safe boundary. A mid-batch `tokio::spawn` abort corrupts the + /// cursor and MUST NEVER happen — that's why the supervisor does + /// not apply `tokio::time::timeout` to recoverable jobs (Part 1's + /// timeout policy does not apply here). + async fn run_resumable(&self, store: &dyn JobStore) -> RunOutcome; +} +``` + +### `RunOutcome` + +```rust +pub enum RunOutcome { + Completed, + Paused { cursor: Vec }, + Failed { message: String }, +} +``` + +- `Completed` — walked the whole space. Engine writes `status = Completed`. +- `Paused { cursor }` — cooperative pause (cancel poll or graceful + shutdown). Engine persists cursor + writes `status = Paused` so a + future resume picks up here. +- `Failed { message }` — irrecoverable error. Cursor NOT advanced; + engine writes `status = Failed` and captures the message. + +### `JobStore` trait + +The port the engine passes to a recoverable job. Backed by +`admin.background_runs` in production; can be mocked for unit tests. + +```rust +#[async_trait] +pub trait JobStore: Send + Sync { + /// The `run_id` this handler was invoked with. Uniquely identifies + /// the row in `admin.background_runs`. + fn run_id(&self) -> Uuid; + + /// Fixed at run start; used by consistency checks (and any other + /// job with a grace boundary) as the reference `NOW()` — NOT + /// `chrono::Utc::now()`, which would drift across a multi-hour + /// scan. See `docs/plan/consistency-check.md` trap #1. + fn started_at(&self) -> chrono::DateTime; + + /// Read the current `status` from the row. Between batches the + /// handler polls this; a return of `CancelRequested` means the + /// operator asked for a pause. + async fn status(&self) -> Result; + + /// The last-persisted cursor (raw bytes, per-job schema), or + /// `None` on a fresh run. The handler decodes into its own key + /// type (blob hash, file_id UUID, ltree path, …). + async fn load_cursor(&self) -> Result>, DomainError>; + + /// Advance cursor + stats, bump `last_progress_at`. Called between + /// batches, typically every ~30 s OR every ~1 000 rows, whichever + /// comes first. See `docs/plan/consistency-check.md` trap #6. + async fn checkpoint(&self, cursor: Vec, delta_count: u64) + -> Result<(), DomainError>; +} +``` + +Domain-specific extensions (consistency-check's finding sink, for +instance) are separate traits the impl composes on top of `JobStore`. +`JobStore` itself carries no findings/severity concept — those are +Layer C in the consistency-check plan, not the engine's concern. + +### Schema — `admin.background_runs` + +```sql +CREATE SCHEMA IF NOT EXISTS admin; + +CREATE TABLE admin.background_runs ( + id UUID PRIMARY KEY, + job_name TEXT NOT NULL, + status TEXT NOT NULL, -- Running / Paused / CancelRequested / Completed / Failed + started_at TIMESTAMPTZ NOT NULL, -- fixed at run start + last_progress_at TIMESTAMPTZ NOT NULL, -- heartbeat + last-checkpoint marker + completed_at TIMESTAMPTZ, + cursor BYTEA, -- opaque, per-job resume key (NULL = fresh) + stats JSONB NOT NULL DEFAULT '{}'::jsonb, -- job-specific counters + params JSONB NOT NULL DEFAULT '{}'::jsonb, -- job-specific params + error_message TEXT +); + +CREATE UNIQUE INDEX one_active_run_per_job + ON admin.background_runs (job_name) + WHERE status IN ('Running', 'Paused', 'CancelRequested'); + +CREATE INDEX ON admin.background_runs (last_progress_at) + WHERE status = 'Running'; +``` + +**The partial unique index is load-bearing.** It enforces the "at +most one non-terminal run per `job_name`" invariant at the DB layer +so it survives concurrent triggers, admin-vs-scheduler races, and +transaction interleavings. The `CancelRequested` inclusion prevents +a second trigger during cancel from spawning a parallel run. + +`admin.*` is a NEW schema — kept distinct from `auth.*` / `storage.*` +so operational tables don't pollute domain schemas. Consistency +checks own their own `admin.consistency_findings` in the same +schema. + +Cursor is `BYTEA`, not JSONB, because per-job cursors are fixed-shape +opaque keys (32-byte BLAKE3, 16-byte UUID, ltree bytes) — JSONB adds +encoding overhead and a keying convention every impl has to agree on. +`stats` and `params` ARE JSONB because they carry human-readable +key/value pairs read by observability code, not compared inside SQL. + +### Cursor semantics + +- **`NULL` cursor** = fresh run, no rows processed yet. Handler + interprets as "start from the beginning." Every keyset-pagination + helper handles this as `WHERE ($1::bytea IS NULL OR key > $1)`. +- **Non-NULL cursor** = last-processed key. On resume, `key > cursor` + in the ORDER BY key ASC iteration. +- **Advance rule** = handler updates its in-memory cursor to the LAST + row it successfully processed at the end of each batch, checkpoints + periodically. On crash: at most one batch of work replays. Idempotent + processing (e.g. `UNIQUE (run_id, kind, resource_id)` on findings) + makes replay a no-op for anything already recorded. + +### Checkpoint mechanics + +One `UPDATE` per checkpoint. Cheap, no row-lock contention (this +process owns the row): + +```sql +UPDATE admin.background_runs + SET cursor = $2, + stats = jsonb_set( + stats, + '{scanned_count}', + ((COALESCE(stats->>'scanned_count','0')::bigint + $3)::text)::jsonb + ), + last_progress_at = NOW() + WHERE id = $1; +``` + +- `cursor` advances to the last row we processed. +- `stats.scanned_count` accumulates the delta — not overwritten. Each + job's handler picks its own key names inside `stats`. There's ONE + convention: a top-level `count` field mirroring the value carried + in `JobOutcome::Ok.count` (see next section) — everything else is + free-form. +- `last_progress_at` doubles as heartbeat. Boot recovery uses it to + spot stale-Running rows. + +### `RunOutcome` → `JobOutcome` bridge + +The supervisor translates so a periodic-triggered recoverable job +records the same `JobOutcome` shape as any other tick: + +- `Completed` → `Ok { count, extra: json!({"completed": true}) }` +- `Paused { cursor }` → `Ok { count, extra: json!({"paused": true, "cursor_hex": …}) }` +- `Failed { message }` → `Err(message)` + +Paused is deliberately NOT an error — the run cooperatively yielded, +that's a success. Log lines stay meaningful (`outcome=ok`, +`extra.paused=true` distinguishes from full completion). Only +`Failed` alerts an operator. + +### `run_or_resume` helper + +The engine module exposes: + +```rust +pub async fn run_or_resume( + job: Arc, + store_factory: &dyn JobStoreFactory, +) -> JobOutcome +``` + +Body: + +1. Look up the latest row for `job.name()`. +2. If `Completed`/`Failed` or nothing → `INSERT` a new `Running` row + with `started_at = NOW()`, cursor NULL. On unique-index conflict + (rare race), read the winning row and continue from step 3. +3. If `Paused` → `UPDATE ... SET status='Running'` on that row. +4. If `Running`/`CancelRequested` → short-circuit + `Ok { count: 0, extra: {"skipped": "already_running"} }`. +5. Build a `JobStore` bound to the row's `run_id` and pass it to + `job.run_resumable(store).await`. +6. Translate the returned `RunOutcome`, write the terminal status + (`Completed` / `Paused` / `Failed`) with the final cursor/stats + snapshot, return the `JobOutcome`. + +### Concurrency policy — exclusive-by-default + +**At most one non-terminal run per `job_name` may exist at any time.** +Non-terminal = `status IN ('Running', 'Paused', 'CancelRequested')`. +This is the default, not opt-in — a job runs to completion, gets +manually paused, or fails; a second trigger while one is active +never spawns a parallel run. + +- A storage-migration cannot run twice at once. Neither can a + reextract-audio, a reextract-image, or a consistency-check. +- The registry's trigger endpoint is idempotent: called while a run + is active it returns the existing `run_id` + status; called while + the latest run is `Paused` it resumes it (same cursor, same stats + accumulator); called when no non-terminal run exists it starts + fresh. +- The DB-level partial unique index makes the invariant impossible + to violate even under concurrent triggers or scheduler-vs-operator + races. +- The scheduler's periodic tick honours the same rule — if the + latest row for a job is non-terminal, the tick does not spawn + another. For long-running jobs "interval" effectively means "check + every N whether a run needs starting", not "start every N." +- Cross-job concurrency is unchanged — different `job_name`s can + run in parallel subject to Part 1's sequential-dispatch default. + Exclusivity is per job_name, not global. + +### Boot-time crashed-run recovery + +At `AppServiceFactory` init, after DB pool is up: + +```rust +sqlx::query!( + "UPDATE admin.background_runs + SET status = 'Paused', + error_message = COALESCE(error_message, 'server restart mid-run') + WHERE status IN ('Running', 'CancelRequested')" +).execute(&pool).await?; +``` + +Do NOT auto-resume — the bug that killed the last run may still be +present. Operators decide. The next scheduler tick (or an explicit +trigger) resumes any `Paused` row per the normal flow. + +Consistency-check.md's existing consistency-scoped sweep collapses +into this general one. + +### Admin surface (recoverable runs) + +Same URL taxonomy as Part 1, extended for run identity: + +``` +POST /api/admin/internal/trigger-job/{name} + → { run_id, status } # starts or resumes; idempotent +POST /api/admin/internal/trigger-job/{name}/cancel + → { run_id, status: "CancelRequested" } +GET /api/admin/internal/jobs/{name}/runs + → [{ run_id, status, started_at, last_progress_at, stats, ... }] +GET /api/admin/internal/jobs/{name}/runs/{id} + → { run_id, status, cursor_hex, stats, params, error_message, ... } +``` + +### Native tenants (Part 2) + +- **Blob storage backend migration.** `migration_job.rs` becomes a + `RecoverableJob` impl. Cursor = last processed blob hash. Retires + the `Arc>` in-memory struct. +- **Reextract audio metadata.** Currently synchronous inside the + admin HTTP request. Becomes a `RecoverableJob` iterating audio + files by `file_id`. +- **Reextract image/video capture dates.** Same as above. +- **Consistency-check runs.** Every `ConsistencyCheck` impl gets + wrapped by a `RecoverableJob` adapter; the wrapper writes to + `admin.background_runs` via `JobStore`, and separately writes + findings to `admin.consistency_findings` via a check-specific + extension trait. See `docs/plan/consistency-check.md`. + +### Verification (Part 2) + +1. **Compile + schema-migration idempotence.** +2. **Fresh run:** `POST /trigger-job/storage_migration` → new row with + `status='Running'`, `cursor=NULL`. +3. **Concurrent trigger:** second `POST` while the first is running + returns the SAME `run_id` (idempotent, DB unique index enforces). +4. **Cancel + resume round-trip:** `trigger-job/…/cancel` flips to + `CancelRequested`; handler polls, returns `Paused { cursor }`; + engine writes `Paused`. `POST /trigger-job/…` again resumes; cursor + picks up where left off; `stats.count` continues accumulating. +5. **Crash recovery:** stop the server mid-run; restart; boot sweep + flips the row to `Paused` with `error_message = 'server restart mid-run'`; + admin triggers again and it resumes. +6. **Idempotent replay:** for consistency-check specifically, verify + that re-processing the last unpersisted batch does NOT double-record + findings (`UNIQUE (run_id, kind, resource_id)` on + `admin.consistency_findings`). +7. **`RunOutcome` bridge log lines:** completed run logs + `outcome=ok, extra.completed=true`; paused logs + `outcome=ok, extra.paused=true`; failed logs `outcome=err, cause=handler`. + +--- + +## Cross-cutting + +### Admin URL taxonomy + +All under `/api/admin/internal/*`, gated by the existing +`OXICLOUD_ENABLE_ADMIN_INTERNAL_ENDPOINTS` env var — reuses the same +admin-guard middleware and the same "disabled → 404" contract as +today's per-service triggers. + +**Existing per-service shims** (`trigger-sweep`, `trigger-gc`, +`trigger-grant-cleanup`) stay as thin forwards to `trigger-job/{name}` +during migration so the existing Hurl suites keep working. +Deprecation surfaces via a `Deprecation: true` response header +operators can grep for. + +### Config surface — env vars + +Canonical form for every job (Part 1 or Part 2 alike, AND for core +workers even though they don't register with the scheduler): + +``` +OXICLOUD_JOB__ENABLED +OXICLOUD_JOB__INTERVAL_HOURS # or _INTERVAL_SECS for sub-hour cadences +OXICLOUD_JOB__... # e.g. _GRACE_HOURS, _BATCH_SIZE +``` + +Core workers reuse this naming purely for uniform operator ergonomics +(e.g. `OXICLOUD_JOB_TREE_ETAG_FLUSH_INTERVAL_MS`) — the convention is +what operators grep for; whether the loop is scheduler-driven or a +dedicated `tokio::spawn` is an implementation detail they don't see. + +Existing per-service env vars keep working as **aliases** during +migration — `OXICLOUD_GRANT_CLEANUP_INTERVAL_HOURS` reads first, falls +back to `OXICLOUD_JOB_GRANT_CLEANUP_INTERVAL_HOURS`. Deprecated aliases +warn once on startup and stay recognised through one minor version. + +### Logging schema + +Uniform structured target across both engines: + +```rust +tracing::info!( + target: "oxicloud::scheduler", + event = "job.run", + job = %name, + outcome = %outcome_kind, // "ok" | "err" + cause = %cause, // omitted on ok; "handler" | "timeout" | "panicked" + count = ..., + elapsed_ms = ..., + // extras from the JobOutcome::Ok.extra map, flattened + ..., + "job {name} ran" +); +``` + +Security-relevant jobs (grant cleanup, authz cache invalidation) still +double-log to `target: "audit"` — the scheduler channel is for +observability; the audit channel is for compliance. + +For Part 2 handlers, the same log line fires at run completion. The +`extra` map surfaces `completed`/`paused`/`cursor_hex` per the +`RunOutcome` bridge above. + +### Composability + +A recoverable job CAN also be periodically-triggered — register with +both engines. Part 1's tick calls Part 2's `run_or_resume(job, store_factory).await` +as its handler. The exclusivity index in Part 2 makes this safe even +if the interval is short enough that a tick fires while a previous +run is still going: the second tick's `run_or_resume` short-circuits +to "already running." + +### Ordering and dependencies (deferred) + +Cross-job dependencies (e.g. "trash cleanup runs before dedup GC") +are not modelled. Every job runs independently. If a real ordering +constraint appears, we add a `depends_on: Vec` field and +topological scheduling then. + +### Shutdown coordination (deferred) + +Matches the existing daemons: no cancellation channel. The scheduler +task dies with the runtime. Recoverable jobs surviving a hard shutdown +land as `Paused` on the next boot via the sweep. If graceful shutdown +lands elsewhere in the codebase, the scheduler and all jobs migrate +together. + +### Future extension — plugins + +Once these engines exist they become the natural place for Extism +plugins to declare scheduled work — manifest `[[jobs]]` entries, +registered on `on_plugin_loaded`, unregistered on unload. Deliberately +deferred: no plugin needs it today, and adding +`JobOwner { Native | Plugin { id } }` + `unregister_by_owner` is a +small type extension the day one does. Nothing in the v1 design +precludes it. + +### Job-history observability + +`admin.background_runs` already carries the latest run per Part 2 job +— "last run time + status" is a `SELECT DISTINCT ON (job_name) …` +query. Deeper history (retention window, per-run drill-down UI) is +deferred; the log stream is the source of truth for older runs. + +Part 1's periodic jobs only carry the last outcome IN MEMORY — no +DB row. If a periodic-only job needs persisted last-run visibility, +either promote it to a "trivial" recoverable job (immediate +`Completed`) or add a small `admin.periodic_runs_last` table later. +No such need today. + +## Out of scope + +- **Cross-job dependencies.** Register-time ordering only, not runtime + graph. +- **Retention pruning of terminal `background_runs` rows.** Deferred + until the volume warrants a policy. +- **Prometheus / OpenMetrics export.** Log-only for now. +- **Distributed scheduling.** Single-process. If OxiCloud ever runs + multi-node, `SELECT … FOR UPDATE SKIP LOCKED` on the runs table is + the pattern; not now. +- **Backfill on startup.** If the process is down when a Part 1 job's + tick was due, we do NOT catch up — the job runs at its next + interval. Matches every existing daemon's behaviour today. +- **Cron expressions.** Fixed intervals only. +- **Rate limiting the admin trigger endpoint.** It's already + admin-gated. + +## Related memory notes + +- `feedback_no_abbreviated_env_vars` — full-word env var names + (`OXICLOUD_JOB_TRASH_CLEANUP_INTERVAL_HOURS`, not + `OXICLOUD_JOB_TC_INTERVAL_H`). +- The grant-cleanup implementation is the closest reference for the + Part 1 daemon → tenant migration shape: three env vars, one impl of + an authz trait method, one daemon service, one admin trigger. +- `project_consistency_check_trait` — the consistency framework + described in `docs/plan/consistency-check.md` is a *consumer* of + Part 2 (the recoverable-run engine), not a peer. It ships after + Part 2 lands.