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
This commit is contained in:
Edouard Vanbelle
@@ -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<Uuid> }`.
|
||||
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_<check_name>'` (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<u8>`; 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<u8>` 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<Vec<u8>>,
|
||||
store: &dyn CheckStore,
|
||||
) -> Result<RunOutcome, DomainError>;
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
pub enum RunOutcome {
|
||||
Completed,
|
||||
Paused { cursor: Vec<u8> },
|
||||
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<chrono::Utc>;
|
||||
|
||||
/// 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<u8>, 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<bool, DomainError>;
|
||||
|
||||
/// 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<u8>` 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 <Name>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<Arc<dyn ConsistencyCheck>>;
|
||||
}
|
||||
```
|
||||
|
||||
### `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<Arc<dyn StatefulAdapter>>,
|
||||
}
|
||||
|
||||
impl ConsistencyRegistry {
|
||||
pub fn register(&mut self, adapter: Arc<dyn StatefulAdapter>) {
|
||||
// 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<Arc<dyn ConsistencyCheck>> {
|
||||
self.adapters
|
||||
.iter()
|
||||
.flat_map(|a| a.consistency_checks())
|
||||
.collect()
|
||||
}
|
||||
|
||||
pub fn get(&self, name: &str) -> Option<Arc<dyn ConsistencyCheck>> {
|
||||
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=<name>&status=<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<Arc<dyn ConsistencyCheck>> {
|
||||
// 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_<name>'`
|
||||
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<ConsistencyRegistry>`, 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`.
|
||||
@@ -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<dyn JobHandler>` 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<Duration>`** 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<HashMap<String, RegisteredJob>>,
|
||||
}
|
||||
|
||||
struct RegisteredJob {
|
||||
handler: Arc<dyn JobHandler>,
|
||||
interval: Duration,
|
||||
timeout: Option<Duration>,
|
||||
/// 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<tokio::sync::Semaphore>, // 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<parking_lot::Mutex<Option<Instant>>>,
|
||||
last_outcome: Option<(chrono::DateTime<Utc>, JobOutcome)>,
|
||||
next_run_at: chrono::DateTime<Utc>,
|
||||
}
|
||||
```
|
||||
|
||||
`Arc<JobRegistry>` lives on `AppState`. Native services register
|
||||
themselves during DI:
|
||||
|
||||
```rust
|
||||
registry.register(
|
||||
Arc::clone(&trash_cleanup) as Arc<dyn JobHandler>,
|
||||
Duration::from_secs(interval_hours * 3600),
|
||||
None, // no timeout
|
||||
);
|
||||
```
|
||||
|
||||
### Engine loop
|
||||
|
||||
```rust
|
||||
async fn run(registry: Arc<JobRegistry>) {
|
||||
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_<NAME>_*`) 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<u8> },
|
||||
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<chrono::Utc>;
|
||||
|
||||
/// 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<RunStatus, DomainError>;
|
||||
|
||||
/// 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<Option<Vec<u8>>, 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<u8>, 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<J: RecoverableJob + ?Sized>(
|
||||
job: Arc<J>,
|
||||
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<RwLock<MigrationState>>` 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_<NAME>_ENABLED
|
||||
OXICLOUD_JOB_<NAME>_INTERVAL_HOURS # or _INTERVAL_SECS for sub-hour cadences
|
||||
OXICLOUD_JOB_<NAME>_<CUSTOM>... # 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<String>` 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.
|
||||
Reference in New Issue
Block a user