diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 0c279808..0c555e17 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -306,6 +306,7 @@ jobs: echo "$output" | grep -Fq "fold_delta_remeasure_bytes=402653184" \ || { echo "::error::Gate R0 RSS evidence line or 384-MiB tripwire changed"; exit 1; } cargo test --locked -p omnigraph-cluster --features failpoints --test failpoints + cargo test --locked -p omnigraph-server --features failpoints --test failpoints - name: Commit regenerated openapi.json to PR branch if: | diff --git a/AGENTS.md b/AGENTS.md index d32a25da..eef59464 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -35,7 +35,7 @@ OmniGraph is a typed property-graph engine built as a coordination layer over ma - **Atomic per-query writes**: `mutate_as` and `load` accumulate insert/update batches into an in-memory `MutationStaging.pending` per touched table. Strict insert and upsert both route through the sealed exact-`id`, filter-bearing adapter; bare Lance Append is test-only. Their RFC-022 adapter resolves or rejects relevant recovery intents before base capture, captures `(native branch id, exact graph_head, schema identity)`, then rechecks recovery and authority under schema → branch → table gates. Mutation/Load keeps one keyed transaction per table and rejects more than 8,192 rows or 32 MiB before recovery arm; mutation update scans stream into the remaining table budget after pending-key shadowing, with blob sizes checked before payload reads. It then arms an identity-bearing recovery-v9 sidecar containing exact Lance transaction identities + pre-minted lineage, commits each table with zero transparent conflict retries, confirms the achieved effects, and publishes under the same token. Unrelated retryable pre-effect authority movement may fully reprepare without changing logical mode. A proven effect-free strict conflict returns `KeyConflict` only after a fresh exact-ID probe; no exact match triggers bounded full strict-mode reprepare, while an effect-free upsert conflict also fully reprepares. Any earlier effect or ambiguity returns `RecoveryRequired` with the sidecar retained. Strict read-set conflicts return `ReadSetChanged`. Deletes stage through the same path. D₂ at parse time remains the constructive (insert/update) XOR destructive (delete) boundary. - **RFC-026 private streaming core**: Phase A's main-only adapter enrolls one exact-`id` table into one empty unsharded Lance MemWAL under recovery-v10, then publishes its physical binding and `OPEN` lifecycle. Phase B1 adds a root-singleflight worker for one no-roll generation (8,192 rows / 32 MiB of logical dense-slice Arrow data), watcher success plus the same writer's post-durability `check_fenced()` before clean acknowledgement, conservative replay/fold-only recovery, and the bounded fold mechanics. Cheap raw bounds and exact post-tombstone validation run before recovery; put ownership follows charge → shared admission → same-key queue → worker mode; cold replay installs exact fold-only accounting; exclusive admission spans seal, drain proof, both table effects, and publication. Admission, replay, and fold all charge `ArrayData::get_slice_memory_size`; fold densifies retained arrays. Backing-buffer capacity and physical allocation are not the limit, while isolated fold RSS remains a 384-MiB remeasurement tripwire. The 8,192-row high-entropy near-cap closure cell remains green. Internal schema v9 adds the private B2 compare-and-chain core: canonical payload and token digests, grammar-impossible trusted hidden row metadata (`__omnigraph_stream_v1$`), same-generation token overlays, and a manifest-selected graph-global `_stream_tokens.lance` authority. Exact base/token lookups and recovery validation materialize at most the requested rows plus one and cap both Arrow batch bytes and cumulative retained bytes. Admission recaptures lifecycle/binding/HEAD authority only after acquiring shared admission and the same-key queue, so a stale provisional capture cannot authorize a WAL put. Recovery-v12 owns exact pre-minted base and token transactions; only their exact joint outcome may drive the sole `__manifest` CAS that advances both pointers, lifecycle state-v2, graph lineage, and a durable fold-attribution summary. Recovery-v11 is historical v8 state and is refused under v9 because it lacks the token participant and complete state-v2 authority. Post-invocation ambiguity is `AckUnknown`; unresolved recovery blocks progress. The topology remains main-only, unsharded, one resident writer and one live writer process, with no fresh reads or generation GC. The private **B2a unbounded retain-all profile is active**: OmniGraph imposes no retained-byte, object-count, file-count, or history quota and never deletes a canonical durable `_mem_wal` object. Lance may clean only its losing `.binpb.tmp.` atomic-CAS staging. Complete and partial unreferenced generation residue stays non-authoritative and untouched through recovery/reopen; provider exhaustion is loud. The genuine v8↔v9 refusal/rebuild gate preserves logical rows—including an ordinary v8 `__omnigraph_stream_v1` user property—vectors, and PK metadata while omitting hidden trusted stream metadata from export. The core remains crate-private behind one feature-gated, doc-hidden test seam. Explicit production enrollment, lifecycle management/correction/status, authorization, and SDK/CLI/HTTP/OpenAPI surfaces remain inactive. - **RFC-026 B2 preprocessing bound**: before blob materialization or canonical encoding, the private adapter reserves a 128-MiB worst-case envelope (original 32-MiB Arrow row + possible 32-MiB replacement + 64-MiB canonical payload) and an inflight slot. The private profile admits exactly two envelopes (256 MiB root-wide), preserving the two-caller stale-authority race without unbounded preprocessing. The slot transfers into queued/worker ownership; scratch releases after digest derivation. Pressure fails effect-free as typed `stream_b2_preprocessing_bytes`; this is process-memory admission, not retained-storage GC or quota. -- **HTTP server**: Axum + utoipa OpenAPI, bearer auth (SHA-256 hashed, optional AWS Secrets Manager). Cedar policy enforcement is engine-wide — every `_as` writer calls `Omnigraph::enforce(action, scope, actor)`, so HTTP, CLI, and embedded SDK consumers all hit the same gate. **Cluster-only boot** (RFC-011): the server always boots from a cluster directory (`--cluster `, RFC-005) and serves N graphs (N ≥ 1) under multi-graph routes (`/graphs/{graph_id}/...` + read-only `GET /graphs` enumeration); there are no single-graph flat routes and no positional-URI boot. Per-graph + server-level Cedar policies. Runtime add/remove (`POST /graphs`, `DELETE /graphs/{id}`) is not exposed — operators run `cluster apply` and restart. +- **HTTP server**: Axum + utoipa OpenAPI, bearer auth (SHA-256 hashed, optional AWS Secrets Manager). Cedar policy enforcement is engine-wide — every `_as` writer calls `Omnigraph::enforce(action, scope, actor)`, so HTTP, CLI, and embedded SDK consumers all hit the same gate. **Cluster-only boot** (RFC-011): the server always boots from a cluster directory (`--cluster `, RFC-005) and serves N configured graphs (N ≥ 1; the serving subset may transiently be empty in lenient mode while open-quarantined graphs converge under supervision) under multi-graph routes (`/graphs/{graph_id}/...` + read-only `GET /graphs` enumeration with per-graph serving/quarantined status); there are no single-graph flat routes and no positional-URI boot. Per-graph + server-level Cedar policies. Runtime add/remove (`POST /graphs`, `DELETE /graphs/{id}`) is not exposed — operators run `cluster apply` and restart. - **CLI** with two-surface config (RFC-007/008): the team-owned cluster directory (`cluster.yaml`) plus the per-operator `~/.omnigraph/config.yaml` (servers, clusters, credentials, actor, profiles, aliases, defaults). Graphs are addressed via `--store`/`--server`/`--cluster`/`--profile`/operator defaults (RFC-011). Multi-format output (json/jsonl/csv/kv/table). Throughout the docs, capabilities are split into **L1 — Inherited from Lance** vs **L2 — Added by OmniGraph**. @@ -294,7 +294,7 @@ omnigraph policy explain --cluster ./company-brain --graph knowledge --actor act | Three-way row-level merge | — | `OrderedTableCursor` + `StagedTableWriter`, structured `MergeConflictKind` | | Change feeds | — | `diff_between` / `diff_commits` with manifest fast path + ID streaming | | Cedar policy | — | Per-graph actions plus server-scoped actions (see [docs/user/operations/policy.md](docs/user/operations/policy.md) for the current list), branch / target_branch / protected scopes, validate/test/explain CLI. **Engine-wide enforcement** (MR-722): every `_as` writer (`apply_schema_as`, `mutate_as`, `load_as` — the deprecated `ingest_as` shims route through it — `branch_create_as` / `branch_create_from_as`, `branch_delete_as`, `branch_merge_as`) calls `Omnigraph::enforce(action, scope, actor)` — HTTP, CLI, embedded SDK all hit the same gate. | -| HTTP server | — | Axum, OpenAPI via utoipa, bearer auth (SHA-256, AWS Secrets Manager option), `authorize_request` at the HTTP boundary (resolves bearer→actor, applies admission control), NDJSON streaming export, **cluster-only boot (RFC-011): always `--cluster `, serving N graphs (N ≥ 1) under multi-graph routes + read-only `GET /graphs` enumeration + per-graph + server-level Cedar policies. Add/remove graphs via `cluster apply` and restart.** | +| HTTP server | — | Axum, OpenAPI via utoipa, bearer auth (SHA-256, AWS Secrets Manager option), `authorize_request` at the HTTP boundary (resolves bearer→actor, applies admission control), NDJSON streaming export, **cluster-only boot (RFC-011): always `--cluster `, serving N configured graphs (N ≥ 1; lenient mode may transiently serve none while quarantined graphs converge) under multi-graph routes + read-only `GET /graphs` enumeration with per-graph serving/quarantined status + per-graph + server-level Cedar policies. Add/remove graphs via `cluster apply` and restart.** | | CLI with config | — | two-surface config (team `cluster.yaml` dir + per-operator `~/.omnigraph/config.yaml`), scope addressing (`--store`/`--server`/`--cluster`/`--profile`/defaults, RFC-011), aliases, multi-format output (json/jsonl/csv/kv/table) | | Audit / actor tracking | — | `_as` write APIs + actor map in commit graph | | Local S3 testing | — | run RustFS/MinIO + the `AWS_*` env; see [docs/user/deployment.md](docs/user/deployment.md) → *Testing against S3 locally* | diff --git a/crates/omnigraph-api-types/src/lib.rs b/crates/omnigraph-api-types/src/lib.rs index bba7ec8c..56670014 100644 --- a/crates/omnigraph-api-types/src/lib.rs +++ b/crates/omnigraph-api-types/src/lib.rs @@ -761,14 +761,49 @@ pub fn read_target_output(target: &ReadTarget) -> ReadTargetOutput { // ─── MR-668 — management endpoint shapes ────────────────────────────────── +/// Serving state of a configured graph (RFC-030 W3). +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize, ToSchema)] +#[serde(rename_all = "snake_case")] +pub enum GraphStatus { + /// Open and answering requests. + #[default] + Serving, + /// Configured but not serving: the open failed and the server's + /// supervision loop is retrying with capped backoff. Routes under the + /// graph answer 503 meanwhile. + Quarantined, +} + +/// Quarantine detail on a [`GraphInfo`] whose `status` is `quarantined` +/// (RFC-030 W3). Timestamps are seconds since the Unix epoch. +#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] +pub struct GraphQuarantineInfo { + /// When the graph entered quarantine (this boot), Unix seconds. + pub since_unix_secs: u64, + /// Open attempts so far (the failed boot open counts as the first). + pub attempts: u32, + /// The most recent open error, verbatim. + pub last_error: String, + /// When the supervision loop will retry, Unix seconds. + pub retry_at_unix_secs: u64, +} + /// One entry in the response from `GET /graphs`. Cluster operators /// consume this list to discover which graphs the server is currently -/// serving. The shape is intentionally minimal — `graph_id` and `uri` -/// are the only fields a routing client needs. +/// serving — including graphs that are configured but quarantined +/// (RFC-030 W3), so an unhealthy graph is visible instead of silently +/// absent. #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] pub struct GraphInfo { pub graph_id: String, pub uri: String, + /// Additive (RFC-030): absent in older responses, so deserialization + /// defaults to `serving` for wire compatibility. + #[serde(default)] + pub status: GraphStatus, + /// Present only when `status` is `quarantined`. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub quarantine: Option, } /// Response from `GET /graphs`. Lists every graph registered with the diff --git a/crates/omnigraph-server/Cargo.toml b/crates/omnigraph-server/Cargo.toml index 9ee201fd..34149885 100644 --- a/crates/omnigraph-server/Cargo.toml +++ b/crates/omnigraph-server/Cargo.toml @@ -17,6 +17,12 @@ default = [] # Enables the AWS Secrets Manager bearer-token source. Off by default — on-prem # and local-dev builds don't pay the AWS SDK compile cost. aws = ["dep:aws-config", "dep:aws-sdk-secretsmanager"] +# Test-only fault injection: exposes the engine's failpoint registry +# (`omnigraph::failpoints::{names, ScopedFailPoint}`) to this crate's +# feature-gated integration tests (`tests/failpoints.rs`). The server defines +# no failpoint hooks of its own, so unlike omnigraph-cluster there is no +# `dep:fail` here. Never enabled in production builds. +failpoints = ["omnigraph/failpoints"] [dependencies] omnigraph = { package = "omnigraph-engine", path = "../omnigraph", version = "0.9.0" } diff --git a/crates/omnigraph-server/src/handlers.rs b/crates/omnigraph-server/src/handlers.rs index 042ce541..310f2ffc 100644 --- a/crates/omnigraph-server/src/handlers.rs +++ b/crates/omnigraph-server/src/handlers.rs @@ -77,8 +77,29 @@ pub(crate) async fn server_graphs_list( .map(|handle| GraphInfo { graph_id: handle.key.graph_id.as_str().to_string(), uri: handle.uri.clone(), + status: GraphStatus::Serving, + quarantine: None, }) .collect(); + // RFC-030 W3: quarantined graphs are configured state and must be + // visible, not silently absent. Timestamps flatten to Unix seconds. + let unix_secs = |t: std::time::SystemTime| { + t.duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or(0) + }; + let snapshot = registry.snapshot_ref(); + graphs.extend(snapshot.quarantined.iter().map(|(key, info)| GraphInfo { + graph_id: key.graph_id.as_str().to_string(), + uri: info.uri.clone(), + status: GraphStatus::Quarantined, + quarantine: Some(GraphQuarantineInfo { + since_unix_secs: unix_secs(info.since), + attempts: info.attempts, + last_error: info.last_error.clone(), + retry_at_unix_secs: unix_secs(info.retry_at), + }), + })); graphs.sort_by(|a, b| a.graph_id.cmp(&b.graph_id)); Ok(Json(GraphListResponse { graphs })) } @@ -277,6 +298,15 @@ pub(crate) async fn resolve_graph_handle( let key = GraphKey::cluster(graph_id.clone()); let handle = match registry.get(&key) { RegistryLookup::Ready(handle) => handle, + // RFC-030 W3: configured-but-healing is "retry later", not "no such + // resource" — the supervision loop is re-driving the open. + RegistryLookup::Quarantined(info) => { + return Err(ApiError::service_unavailable(format!( + "graph '{graph_id}' is quarantined (last open error: {}); the server is \ + retrying — retry later", + info.last_error + ))); + } RegistryLookup::Gone => { return Err(ApiError::not_found(format!("graph '{graph_id}' not found"))); } @@ -643,6 +673,47 @@ pub(crate) async fn server_export( .into_response()) } +/// RFC-030 W1 Stage 1: run an armed write protocol on a detached task so +/// dropping the request future (client disconnect) cannot abandon the +/// protocol mid-flight. The caller still awaits the result — the HTTP +/// contract is unchanged; only abandonment semantics change. The admission +/// guard is acquired by the caller and moved into `work`, so the slot +/// releases at protocol completion, not at disconnect. A panic inside the +/// protocol surfaces as `JoinError` → 500, the same observability as an +/// unwind through the handler. Cancellation of the join cannot occur: +/// nothing aborts the spawned handle, and the requester dropping the join +/// future leaves the task running — that is the point. +pub(crate) async fn shielded_write( + state: &AppState, + key: &GraphKey, + work: F, +) -> std::result::Result +where + T: Send + 'static, + F: std::future::Future> + Send + 'static, +{ + // RFC-030 effect-envelope shielding: the trigger-B notification lives + // INSIDE the spawned task, so it is owned by the protocol and survives + // caller cancellation — a disconnected client's write that surfaces + // RecoveryRequired still arms the supervised reopen. (An unresolved + // rollback-class residual wedges this graph's writes until a read-write + // open runs the Full sweep; every shielded writer funnels through this + // one chokepoint.) The handler-side await below is pure observation. + let state = state.clone(); + let key = key.clone(); + tokio::spawn(async move { + let result = work.await; + if let Err(err) = &result + && err.recovery_required.is_some() + { + state.request_reopen(&key); + } + result + }) + .await + .map_err(|join_err| ApiError::internal(format!("write protocol task failed: {join_err}")))? +} + /// Shared implementation behind `POST /mutate` (canonical) and /// `POST /change` (deprecated alias). Returns the bare `ChangeOutput`; /// each route handler wraps it (the alias also attaches Deprecation @@ -683,7 +754,7 @@ pub(crate) async fn run_mutate( + params_json .map(|p| p.to_string().len() as u64) .unwrap_or(0); - let _admission = state + let admission = state .workload .try_admit(&actor_arc, est_bytes) .map_err(ApiError::from_workload_reject)?; @@ -693,10 +764,20 @@ pub(crate) async fn run_mutate( .map_err(|err| ApiError::bad_request(err.to_string()))?; let result = { - let db = &handle.engine; - db.mutate_as(&branch, query, &selected_name, ¶ms, actor_id) - .await - .map_err(ApiError::from_omni)? + let handle = Arc::clone(&handle); + let branch = branch.clone(); + let query = query.to_string(); + let selected_name = selected_name.clone(); + let actor_id: Option = actor_id.map(str::to_string); + shielded_write(&state, &handle.key.clone(), async move { + let _admission = admission; + handle + .engine + .mutate_as(&branch, &query, &selected_name, ¶ms, actor_id.as_deref()) + .await + .map_err(ApiError::from_omni) + }) + .await? }; Ok(ChangeOutput { branch, @@ -1180,35 +1261,42 @@ pub(crate) async fn server_schema_apply( )); } let est_bytes = request.schema_source.len() as u64; - let _admission = state + let admission = state .workload .try_admit(&actor_arc, est_bytes) .map_err(ApiError::from_workload_reject)?; let result = { - let db = &handle.engine; - let registry = handle.queries.as_deref(); - let label = handle.key.graph_id.as_str().to_string(); - // Engine-layer policy enforcement (MR-722): pass the resolved - // actor through so apply_schema_as can call enforce() with the - // authoritative identity. With a policy installed in AppState, - // engine-side enforcement re-checks the same decision the - // HTTP-layer authorize_request just made above. PR #3 collapses - // the redundancy. - db.apply_schema_as_with_catalog_check( - &request.schema_source, - omnigraph::db::SchemaApplyOptions { - allow_data_loss: request.allow_data_loss, - }, - actor_id, - |catalog| { - if let Some(registry) = registry { - validate_registry_against_catalog(registry, catalog, &label)?; - } - Ok(()) - }, - ) - .await - .map_err(ApiError::from_omni)? + let handle = Arc::clone(&handle); + let schema_source = request.schema_source; + let allow_data_loss = request.allow_data_loss; + let actor_id: Option = actor_id.map(str::to_string); + shielded_write(&state, &handle.key.clone(), async move { + let _admission = admission; + let registry = handle.queries.as_deref(); + let label = handle.key.graph_id.as_str().to_string(); + // Engine-layer policy enforcement (MR-722): pass the resolved + // actor through so apply_schema_as can call enforce() with the + // authoritative identity. With a policy installed in AppState, + // engine-side enforcement re-checks the same decision the + // HTTP-layer authorize_request just made above. PR #3 collapses + // the redundancy. + handle + .engine + .apply_schema_as_with_catalog_check( + &schema_source, + omnigraph::db::SchemaApplyOptions { allow_data_loss }, + actor_id.as_deref(), + |catalog| { + if let Some(registry) = registry { + validate_registry_against_catalog(registry, catalog, &label)?; + } + Ok(()) + }, + ) + .await + .map_err(ApiError::from_omni) + }) + .await? }; // Physical indexes are derived state. Schema apply records intent only; // explicit `ensure_indices` / `optimize` maintenance owns convergence on @@ -1275,16 +1363,26 @@ async fn run_ingest( }, )?; let est_bytes = request.data.len() as u64; - let _admission = state + let admission = state .workload .try_admit(&actor_arc, est_bytes) .map_err(ApiError::from_workload_reject)?; let result = { - let db = &handle.engine; - db.load_as(&branch, from.as_deref(), &request.data, mode, actor_id) - .await - .map_err(ApiError::from_omni)? + let handle = Arc::clone(&handle); + let branch = branch.clone(); + let from = from.clone(); + let data = request.data; + let actor_id: Option = actor_id.map(str::to_string); + shielded_write(&state, &handle.key.clone(), async move { + let _admission = admission; + handle + .engine + .load_as(&branch, from.as_deref(), &data, mode, actor_id.as_deref()) + .await + .map_err(ApiError::from_omni) + }) + .await? }; Ok(ingest_output( @@ -1468,19 +1566,26 @@ pub(crate) async fn server_branch_create( // Branch metadata only — small constant bytes estimate. The Lance // shallow-clone work is bounded by the parent's manifest size, not // the request body. - let _admission = state + let admission = state .workload .try_admit(&actor_arc, 256) .map_err(ApiError::from_workload_reject)?; { - let db = &handle.engine; - db.branch_create_from_as( - ReadTarget::branch(&from), - &request.name, - actor.as_ref().map(|Extension(a)| a.actor_id.as_ref()), - ) - .await - .map_err(ApiError::from_omni)?; + let handle = Arc::clone(&handle); + let from = from.clone(); + let name = request.name.clone(); + let actor_id: Option = actor + .as_ref() + .map(|Extension(a)| a.actor_id.as_ref().to_string()); + shielded_write(&state, &handle.key.clone(), async move { + let _admission = admission; + handle + .engine + .branch_create_from_as(ReadTarget::branch(&from), &name, actor_id.as_deref()) + .await + .map_err(ApiError::from_omni) + }) + .await?; } Ok(Json(BranchCreateOutput { uri: handle.uri.clone(), @@ -1550,15 +1655,23 @@ pub(crate) async fn server_branch_delete( }, )?; // Metadata-only manifest tombstone — small constant estimate. - let _admission = state + let admission = state .workload .try_admit(&actor_arc, 256) .map_err(ApiError::from_workload_reject)?; { - let db = &handle.engine; - db.branch_delete_as(&branch, actor_id) - .await - .map_err(ApiError::from_omni)?; + let handle = Arc::clone(&handle); + let branch = branch.clone(); + let actor_id: Option = actor_id.map(str::to_string); + shielded_write(&state, &handle.key.clone(), async move { + let _admission = admission; + handle + .engine + .branch_delete_as(&branch, actor_id.as_deref()) + .await + .map_err(ApiError::from_omni) + }) + .await?; } Ok(Json(BranchDeleteOutput { uri: handle.uri.clone(), @@ -1622,25 +1735,48 @@ pub(crate) async fn server_branch_merge( // Merge body is small JSON; the heavy work is in the engine but is // bounded per-(table, branch) by the writer queue. Small constant // estimate suffices for the actor in-flight count. - let _admission = state + let admission = state .workload .try_admit(&actor_arc, 256) .map_err(ApiError::from_workload_reject)?; - let outcome = { - let db = &handle.engine; - db.branch_merge_as(&request.source, &target, actor_id) - .await - .map_err(ApiError::from_omni)? - }; - let (branch_deleted, branch_delete_error) = if request.delete_branch { - match delete_merged_source_branch(&handle, actor.as_ref().map(|Extension(a)| a), &request.source) - .await - { - Ok(()) => (Some(true), None), - Err(message) => (Some(false), Some(message)), - } - } else { - (None, None) + // RFC-030 effect-envelope shielding: `merge` and the optional + // `delete_branch` follow-up are ONE composite operation from the + // client's perspective, so ONE shielded task owns the admission guard + // and both protocols. A disconnect can no longer split the composite + // (merge landing, deletion silently skipped), and the admission slot + // spans both protocols as it did before the shield. + let (outcome, branch_deleted, branch_delete_error) = { + let handle = Arc::clone(&handle); + let source = request.source.clone(); + let target = target.clone(); + let actor_owned: Option = + actor.as_ref().map(|Extension(actor)| actor.clone()); + let delete_branch = request.delete_branch; + let state_task = state.clone(); + shielded_write(&state, &handle.key.clone(), async move { + let _admission = admission; + let outcome = handle + .engine + .branch_merge_as( + &source, + &target, + actor_owned.as_ref().map(|actor| actor.actor_id.as_ref()), + ) + .await + .map_err(ApiError::from_omni)?; + let (branch_deleted, branch_delete_error) = if delete_branch { + match delete_merged_source_branch(&state_task, &handle, actor_owned.as_ref(), &source) + .await + { + Ok(()) => (Some(true), None), + Err(message) => (Some(false), Some(message)), + } + } else { + (None, None) + }; + Ok((outcome, branch_deleted, branch_delete_error)) + }) + .await? }; Ok(Json(BranchMergeOutput { source: request.source, @@ -1658,7 +1794,8 @@ pub(crate) async fn server_branch_merge( /// operational error — into a message instead of an error status: the merge is /// already durable, so the request must not report failure for it. async fn delete_merged_source_branch( - handle: &GraphHandle, + state: &AppState, + handle: &Arc, actor: Option<&ResolvedActor>, source: &str, ) -> std::result::Result<(), String> { @@ -1675,12 +1812,29 @@ async fn delete_merged_source_branch( Ok(Authz::Denied(message)) => return Err(message), Err(err) => return Err(err.message), } - let actor_id = actor.map(|actor| actor.actor_id.as_ref()); - handle - .engine - .branch_delete_as(source, actor_id) - .await - .map_err(|err| err.to_string()) + // Runs inside the merge's shielded composite task (RFC-030), so it is + // already detached from the request future and covered by the merge's + // admission guard. The engine call still routes through its own + // `shielded_write` for one reason: this helper's contract stringifies + // failures into `branch_delete_error` on a 200 response, so the + // composite completes `Ok` and the OUTER chokepoint can never see a + // typed delete error — a delete-phase `RecoveryRequired` must pass the + // trigger-B chokepoint BEFORE stringification or the supervised reopen + // is never armed for it. + let key = handle.key.clone(); + let handle = Arc::clone(handle); + let source_task = source.to_string(); + let actor_id: Option = actor.map(|actor| actor.actor_id.as_ref().to_string()); + shielded_write(state, &key, async move { + handle + .engine + .branch_delete_as(&source_task, actor_id.as_deref()) + .await + .map_err(ApiError::from_omni) + }) + .await + .map(|_| ()) + .map_err(|err| err.message) } #[utoipa::path( diff --git a/crates/omnigraph-server/src/lib.rs b/crates/omnigraph-server/src/lib.rs index e533182d..220bc3fd 100644 --- a/crates/omnigraph-server/src/lib.rs +++ b/crates/omnigraph-server/src/lib.rs @@ -10,11 +10,15 @@ pub mod identity; pub mod policy; pub mod queries; pub mod registry; +pub mod supervisor; pub mod workload; pub use graph_id::GraphId; pub use identity::{AuthSource, GraphKey, ResolvedActor, Scope, TenantId}; -pub use registry::{GraphHandle, GraphRegistry, InsertError, RegistryLookup, RegistrySnapshot}; +pub use registry::{ + GraphHandle, GraphRegistry, InsertError, QuarantineInfo, RegistryLookup, RegistrySnapshot, +}; +pub use supervisor::SupervisorConfig; use crate::queries::{QueryRegistry, check, format_check_breakages}; @@ -29,6 +33,7 @@ use api::{ BranchCreateOutput, BranchCreateRequest, BranchDeleteOutput, BranchListOutput, BranchMergeOutput, BranchMergeRequest, ChangeOutput, ChangeRequest, CommitListOutput, CommitListQuery, ErrorCode, ErrorOutput, ExportRequest, GraphInfo, GraphListResponse, + GraphQuarantineInfo, GraphStatus, HealthOutput, IngestOutput, IngestRequest, InvokeStoredQueryRequest, InvokeStoredQueryResponse, QueriesCatalogOutput, QueryRequest, ReadOutput, ReadRequest, SchemaApplyOutput, SchemaApplyRequest, SchemaOutput, SnapshotQuery, ingest_output, schema_apply_output, @@ -244,6 +249,11 @@ pub struct GraphStartupConfig { pub struct GraphRouting { pub registry: Arc, pub config_path: Option, + /// Clones of every configured graph's startup config, keyed for the + /// RFC-030 supervision loop (W3 boot retry + W2(b) reopen) — healthy + /// graphs included, since a supervised reopen needs the config too. + /// Empty in single mode (no supervision surface there). + pub startup_configs: Arc>, } #[derive(Clone)] @@ -258,6 +268,10 @@ pub struct AppState { /// Per-actor admission control. Process-wide (not per-graph) — /// see MR-668 decision Q6. workload: Arc, + /// RFC-030 supervision notify channel. Set once by `spawn_supervision`; + /// `request_reopen` sends on it (and silently drops when supervision is + /// not running — single mode, or tests that don't spawn it). + supervisor_tx: Arc>>, bearer_tokens: Arc<[(BearerTokenHash, Arc)]>, /// Server-level Cedar policy. Used by management endpoints (`GET /// /graphs`) which act on the registry resource, not on a per-graph @@ -539,8 +553,10 @@ impl AppState { routing: GraphRouting { registry, config_path: None, + startup_configs: Arc::new(HashMap::new()), }, workload, + supervisor_tx: Arc::new(std::sync::OnceLock::new()), bearer_tokens, server_policy: None, } @@ -558,15 +574,42 @@ impl AppState { server_policy: Option, workload: workload::WorkloadController, config_path: Option, + ) -> std::result::Result { + Self::new_multi_with_boot( + handles, + Vec::new(), + Arc::new(HashMap::new()), + bearer_tokens, + server_policy, + workload, + config_path, + ) + } + + /// Multi-mode constructor carrying RFC-030 boot supervision state: + /// quarantine entries for graphs whose open failed, and the retained + /// startup configs the supervision loop reopens from. `new_multi` + /// delegates here with empty supervision state. + #[allow(clippy::too_many_arguments)] + pub fn new_multi_with_boot( + handles: Vec>, + quarantined: Vec<(GraphKey, QuarantineInfo)>, + startup_configs: Arc>, + bearer_tokens: Vec<(String, String)>, + server_policy: Option, + workload: workload::WorkloadController, + config_path: Option, ) -> std::result::Result { let bearer_tokens = hash_bearer_tokens(bearer_tokens); - let registry = Arc::new(GraphRegistry::from_handles(handles)?); + let registry = Arc::new(GraphRegistry::from_boot(handles, quarantined)?); Ok(Self { routing: GraphRouting { registry, config_path, + startup_configs, }, workload: Arc::new(workload), + supervisor_tx: Arc::new(std::sync::OnceLock::new()), bearer_tokens, server_policy: server_policy.map(Arc::new), }) @@ -579,6 +622,39 @@ impl AppState { &self.routing } + /// Start the RFC-030 supervision loop (W3 boot retry + W2(b) supervised + /// reopen) for this state's registry. Idempotent: a second call warns + /// and returns a completed no-op handle. `serve()` calls this with + /// production pacing; in-process tests call it directly with + /// `SupervisorConfig::fast_for_tests()` since they never run `serve()`. + pub fn spawn_supervision( + &self, + config: supervisor::SupervisorConfig, + ) -> tokio::task::JoinHandle<()> { + let (tx, rx) = tokio::sync::mpsc::unbounded_channel(); + if self.supervisor_tx.set(tx).is_err() { + warn!("graph supervision already running; ignoring second spawn"); + return tokio::spawn(async {}); + } + let supervisor = supervisor::GraphSupervisor { + registry: Arc::clone(&self.routing.registry), + configs: Arc::clone(&self.routing.startup_configs), + config, + rx, + }; + tokio::spawn(supervisor.run()) + } + + /// RFC-030 W2(b) trigger B: a shielded write surfaced `RecoveryRequired` + /// for this graph — ask the supervision loop to re-drive the open (whose + /// Full sweep resolves the residual). Silently drops when supervision is + /// not running; the documented reopen-or-restart remedy still applies. + pub(crate) fn request_reopen(&self, key: &GraphKey) { + if let Some(tx) = self.supervisor_tx.get() { + let _ = tx.send(key.clone()); + } + } + fn requires_bearer_auth(&self) -> bool { if !self.bearer_tokens.is_empty() { return true; @@ -691,6 +767,23 @@ impl ApiError { } } + /// 503 without a closed `ErrorCode` — mirrors the recovery-required + /// 503 shape (RFC-030 W3: quarantined graphs are configured-but-healing, + /// "retry later", not "no such resource"). + pub fn service_unavailable(message: impl Into) -> Self { + Self { + status: StatusCode::SERVICE_UNAVAILABLE, + code: None, + message: message.into(), + merge_conflicts: Vec::new(), + manifest_conflict: None, + read_set_conflict: None, + key_conflict: None, + resource_limit: None, + recovery_required: None, + } + } + pub fn conflict(message: impl Into) -> Self { Self { status: StatusCode::CONFLICT, @@ -1267,6 +1360,10 @@ pub async fn serve(config: ServerConfig) -> Result<()> { } }; + // RFC-030 W3/W2(b): start the supervision loop before serving. Its task + // exits on its own when the state (and with it the notify sender) drops + // at shutdown; graphs quarantined at boot begin converging immediately. + let _supervision = state.spawn_supervision(supervisor::SupervisorConfig::production()); let listener = TcpListener::bind(&bind).await?; axum::serve(listener, build_app(state)) .with_graceful_shutdown(shutdown_signal()) @@ -1314,26 +1411,53 @@ pub async fn open_multi_graph_state( }; let configured_graphs = graphs.len(); + // Retain every config for the RFC-030 supervision loop (healthy graphs + // included: a supervised reopen after RecoveryRequired needs them too). + // A config whose graph id cannot form a key is unquarantinable; its open + // fails below and it keeps the historical warn-and-drop behavior. + let mut startup_configs: HashMap = HashMap::new(); + for cfg in &graphs { + if let Ok(graph_id) = GraphId::try_from(cfg.graph_id.clone()) { + startup_configs.insert(GraphKey::cluster(graph_id), cfg.clone()); + } + } let results = futures::stream::iter(graphs.into_iter()) .map(|cfg| async move { - let graph_id = cfg.graph_id.clone(); - open_single_graph(cfg).await.map_err(|err| (graph_id, err)) + match open_single_graph(cfg.clone()).await { + Ok(handle) => Ok(handle), + Err(err) => Err((cfg, err)), + } }) .buffer_unordered(4) .collect::>() .await; let mut handles = Vec::new(); + let mut quarantined: Vec<(GraphKey, QuarantineInfo)> = Vec::new(); let mut failed = 0usize; for result in results { match result { Ok(handle) => handles.push(handle), - Err((graph_id, err)) => { + Err((cfg, err)) => { failed += 1; warn!( - graph_id = %graph_id, + graph_id = %cfg.graph_id, error = %err, "graph quarantined during startup" ); + if let Ok(graph_id) = GraphId::try_from(cfg.graph_id.clone()) { + let now = std::time::SystemTime::now(); + quarantined.push(( + GraphKey::cluster(graph_id), + QuarantineInfo { + uri: cfg.uri.clone(), + since: now, + attempts: 1, + last_error: err.to_string(), + retry_at: now, + policy_configured: cfg.policy.is_some(), + }, + )); + } } } } @@ -1344,7 +1468,13 @@ pub async fn open_multi_graph_state( failed ); } - if handles.is_empty() { + // RFC-030 W3: zero SERVING graphs is a valid lenient-mode boot state as + // long as quarantine entries exist for supervision to converge — aborting + // here would re-create the incident's offline-until-redeploy failure for + // single-graph deployments with a transient open fault. Zero configured + // graphs (and settings-time config-class quarantine, which supervision + // cannot heal) remain fail-fast upstream. + if handles.is_empty() && quarantined.is_empty() { bail!( "no healthy graphs opened from multi-graph startup config ({} configured, {} failed)", configured_graphs, @@ -1353,8 +1483,16 @@ pub async fn open_multi_graph_state( } let workload = workload::WorkloadController::from_env(); - let state = AppState::new_multi(handles, tokens, server_policy, workload, Some(config_path)) - .map_err(|err| color_eyre::eyre::eyre!("multi-graph registry: {err}"))?; + let state = AppState::new_multi_with_boot( + handles, + quarantined, + Arc::new(startup_configs), + tokens, + server_policy, + workload, + Some(config_path), + ) + .map_err(|err| color_eyre::eyre::eyre!("multi-graph registry: {err}"))?; Ok(state) } diff --git a/crates/omnigraph-server/src/registry.rs b/crates/omnigraph-server/src/registry.rs index 54115e4d..c295e8a9 100644 --- a/crates/omnigraph-server/src/registry.rs +++ b/crates/omnigraph-server/src/registry.rs @@ -24,13 +24,39 @@ use std::sync::Arc; use arc_swap::ArcSwap; use omnigraph::db::Omnigraph; use omnigraph::storage::normalize_root_uri; -#[cfg(test)] use tokio::sync::Mutex; use crate::identity::GraphKey; use crate::policy::PolicyEngine; use crate::queries::QueryRegistry; +/// Supervision state for a configured graph whose open failed (RFC-030 W3). +/// +/// A quarantined graph is *configured but not serving*: its startup config is +/// retained, the supervision loop retries the full `open_single_graph` with +/// capped backoff, and this record is what `GET /graphs` and the routing +/// middleware surface meanwhile. +#[derive(Debug, Clone)] +pub struct QuarantineInfo { + /// The configured graph URI (the open never succeeded, so there is no + /// handle to carry it). + pub uri: String, + /// When the graph first entered quarantine (this boot). + pub since: std::time::SystemTime, + /// Open attempts so far (boot's failed open counts as the first). + pub attempts: u32, + /// The most recent open error, verbatim. + pub last_error: String, + /// When the supervision loop will try again. + pub retry_at: std::time::SystemTime, + /// Whether the graph's startup config declares a per-graph policy. + /// Folded into [`RegistrySnapshot::any_per_graph_policy`] so bearer auth + /// stays required while the policy-bearing graph is quarantined — today + /// such a graph silently vanishes from the registry, so this is strictly + /// safer than the pre-RFC-030 behavior. + pub policy_configured: bool, +} + /// Open handle for a single graph in the registry. Cheap to clone (`Arc`-wrapped /// engine + policy). Cluster-mode handlers extract this via /// `Extension>` injected by the routing middleware. @@ -64,22 +90,44 @@ pub struct GraphHandle { /// (or `Default`) so the field stays in sync with `graphs`. pub struct RegistrySnapshot { pub graphs: HashMap>, - /// `true` iff any registered graph has a per-graph policy installed. - /// Used by `AppState::requires_bearer_auth` to decide whether the - /// auth middleware should challenge a request — a per-graph policy - /// implies bearer auth is required even when no server-level tokens - /// or policy are configured. + /// Configured graphs whose open failed and are under supervision + /// (RFC-030 W3). Disjoint from `graphs` by construction: the writer + /// methods maintain "a key is never in both maps" under the `mutate` + /// mutex, and [`RegistrySnapshot::with_quarantined`] debug-asserts it. + pub quarantined: HashMap, + /// `true` iff any registered graph has a per-graph policy installed, + /// OR any quarantined graph's config declares one. Used by + /// `AppState::requires_bearer_auth` to decide whether the auth + /// middleware should challenge a request — a per-graph policy implies + /// bearer auth is required even when no server-level tokens or policy + /// are configured, and a quarantined policy-bearing graph must not + /// flap auth off while it heals. pub any_per_graph_policy: bool, } impl RegistrySnapshot { /// Build a snapshot from a graph map, deriving cached fields. - /// The only construction path — direct struct-literal use elsewhere - /// would let derived state drift from `graphs`. + /// The only construction paths are this and + /// [`RegistrySnapshot::with_quarantined`] — direct struct-literal use + /// elsewhere would let derived state drift from `graphs`. pub fn new(graphs: HashMap>) -> Self { - let any_per_graph_policy = graphs.values().any(|h| h.policy.is_some()); + Self::with_quarantined(graphs, HashMap::new()) + } + + /// Build a snapshot carrying quarantine entries (RFC-030 W3). + pub fn with_quarantined( + graphs: HashMap>, + quarantined: HashMap, + ) -> Self { + debug_assert!( + quarantined.keys().all(|key| !graphs.contains_key(key)), + "a graph key must never be both serving and quarantined", + ); + let any_per_graph_policy = graphs.values().any(|h| h.policy.is_some()) + || quarantined.values().any(|q| q.policy_configured); Self { graphs, + quarantined, any_per_graph_policy, } } @@ -91,10 +139,14 @@ impl Default for RegistrySnapshot { } } -/// Result of a registry lookup. Two-valued — `Tombstoned` deferred with DELETE. +/// Result of a registry lookup. `Tombstoned` remains deferred with DELETE. pub enum RegistryLookup { /// Graph is open and ready to serve. Ready(Arc), + /// Graph is configured but not serving: its open failed and the + /// supervision loop is retrying (RFC-030 W3). Handlers respond 503 — + /// "retry later", not "no such resource". + Quarantined(QuarantineInfo), /// Graph is not in the registry (never existed, or was unregistered in a /// future release). Handlers respond with 404. Gone, @@ -118,11 +170,11 @@ pub enum InsertError { pub struct GraphRegistry { snapshot: ArcSwap, - /// Serializes runtime mutations through [`GraphRegistry::insert`]. - /// Gated with `insert` because they share a single contract — if - /// the consumer goes away, so does the lock. Re-introducing one - /// requires re-introducing the other. - #[cfg(test)] + /// Serializes runtime mutations (`publish`, `set_quarantined`, and the + /// test-only `insert`) so read-modify-write cycles over the `ArcSwap` + /// snapshot are atomic. Ungated from `#[cfg(test)]` by RFC-030 W3/W2(b): + /// the supervision loop is the anticipated production consumer the + /// original gate's doc comment named. mutate: Mutex<()>, } @@ -131,7 +183,6 @@ impl GraphRegistry { pub fn new() -> Self { Self { snapshot: ArcSwap::from_pointee(RegistrySnapshot::default()), - #[cfg(test)] mutate: Mutex::new(()), } } @@ -139,6 +190,20 @@ impl GraphRegistry { /// Build a registry from a startup-time list of open handles. /// Rejects duplicate `GraphKey`s and duplicate URIs. pub fn from_handles(handles: Vec>) -> Result { + Self::from_boot(handles, Vec::new()) + } + + /// Boot-time constructor carrying both healthy handles and quarantined + /// entries (RFC-030 W3). Rejects duplicate `GraphKey`s among the + /// handles and duplicate canonical URIs across the whole configured set + /// — serving and quarantined alike, so an invalid config fails boot the + /// same way regardless of which opens succeeded. Quarantined keys must + /// be disjoint from the serving keys (the boot loop guarantees it — a + /// graph either opened or it didn't). + pub fn from_boot( + handles: Vec>, + quarantined: Vec<(GraphKey, QuarantineInfo)>, + ) -> Result { let mut graphs: HashMap> = HashMap::with_capacity(handles.len()); let mut seen_uris: HashMap = HashMap::with_capacity(handles.len()); for handle in handles { @@ -152,9 +217,34 @@ impl GraphRegistry { seen_uris.insert(canonical_uri, handle.key.clone()); graphs.insert(handle.key.clone(), handle); } + let mut quarantined_map: HashMap = + HashMap::with_capacity(quarantined.len()); + for (key, info) in quarantined { + if graphs.contains_key(&key) { + continue; + } + // The "distinct canonical URIs across configured graphs" + // invariant must not depend on which opens succeeded: when both + // colliding graphs open, the handle loop above already fails + // boot, and a half-open collision would otherwise boot into an + // endless supervised-retry loop (the reopen succeeds but + // `publish` forever returns `DuplicateUri`). A URI that cannot + // be normalized is skipped: its reopen fails at normalization + // too, so it can never reach `publish`. + if let Ok(canonical_uri) = normalize_root_uri(&info.uri) { + if seen_uris.contains_key(&canonical_uri) { + return Err(InsertError::DuplicateUri(info.uri.clone())); + } + seen_uris.insert(canonical_uri, key.clone()); + } + quarantined_map.insert(key, info); + } + let quarantined = quarantined_map; Ok(Self { - snapshot: ArcSwap::from_pointee(RegistrySnapshot::new(graphs)), - #[cfg(test)] + snapshot: ArcSwap::from_pointee(RegistrySnapshot::with_quarantined( + graphs, + quarantined, + )), mutate: Mutex::new(()), }) } @@ -167,14 +257,18 @@ impl GraphRegistry { self.snapshot.load() } - /// Lock-free read. Returns `Ready` if the graph is in the current snapshot, + /// Lock-free read. Returns `Ready` if the graph is serving, + /// `Quarantined` if it is configured-but-healing (RFC-030 W3), and /// `Gone` otherwise. pub fn get(&self, key: &GraphKey) -> RegistryLookup { let snapshot = self.snapshot.load(); - match snapshot.graphs.get(key) { - Some(handle) => RegistryLookup::Ready(Arc::clone(handle)), - None => RegistryLookup::Gone, + if let Some(handle) = snapshot.graphs.get(key) { + return RegistryLookup::Ready(Arc::clone(handle)); + } + if let Some(info) = snapshot.quarantined.get(key) { + return RegistryLookup::Quarantined(info.clone()); } + RegistryLookup::Gone } /// Snapshot the full set of currently-registered handles. Ordering @@ -201,11 +295,12 @@ impl GraphRegistry { /// and duplicate `uri`. /// /// **Test-only surface.** No production code reaches this — startup - /// uses `from_handles`, and runtime add/remove is deferred. The - /// race-contract tests below pin the mutex linearization point so - /// that when a real consumer ships (managed cluster catalog), the - /// concurrency contract is already proven. Ungate by removing - /// `#[cfg(test)]` once that consumer is in scope. + /// uses `from_boot`, and production runtime mutation goes through + /// [`GraphRegistry::publish`] (replace-allowed) / + /// [`GraphRegistry::set_quarantined`], the RFC-030 consumers that + /// ungated the `mutate` mutex. `insert` stays test-only because its + /// add-only duplicate-key semantics exist to pin the mutex + /// linearization contract, not to serve traffic. /// /// Race semantics (pinned by `concurrent_insert_same_key_exactly_one_succeeds`): /// under N concurrent calls with the same key, exactly one returns @@ -215,7 +310,12 @@ impl GraphRegistry { let _guard = self.mutate.lock().await; let current = self.snapshot.load(); let (canonical_uri, handle) = canonicalize_handle_uri(handle)?; - if current.graphs.contains_key(&handle.key) { + // A quarantined key is CONFIGURED — add-only insert rejects it like + // any registered key (publish is the heal path that may replace it). + // Without this guard an insert could put one key in both maps, + // tripping the snapshot's disjointness invariant. + if current.graphs.contains_key(&handle.key) || current.quarantined.contains_key(&handle.key) + { return Err(InsertError::DuplicateKey(handle.key.clone())); } for existing in current.graphs.values() { @@ -230,10 +330,64 @@ impl GraphRegistry { } let mut new_graphs = current.graphs.clone(); new_graphs.insert(handle.key.clone(), handle); - self.snapshot - .store(Arc::new(RegistrySnapshot::new(new_graphs))); + self.snapshot.store(Arc::new(RegistrySnapshot::with_quarantined( + new_graphs, + current.quarantined.clone(), + ))); + Ok(()) + } + + /// RCU publish (RFC-030 W3/W2(b)): install `handle` for its key, clearing + /// any quarantine entry and replacing any prior serving handle. Same-key + /// replacement is the supervised-reopen swap — in-flight requests on the + /// old handle finish on their own `Arc` clone (the engine-survival + /// contract in this module's docs); new requests resolve the new handle. + /// The duplicate-URI check skips the key being replaced. + pub async fn publish(&self, handle: Arc) -> Result<(), InsertError> { + let _guard = self.mutate.lock().await; + let current = self.snapshot.load(); + let (canonical_uri, handle) = canonicalize_handle_uri(handle)?; + for (key, existing) in ¤t.graphs { + if *key == handle.key { + continue; + } + let existing_uri = + normalize_root_uri(&existing.uri).map_err(|err| InsertError::InvalidUri { + uri: existing.uri.clone(), + message: err.to_string(), + })?; + if existing_uri == canonical_uri { + return Err(InsertError::DuplicateUri(handle.uri.clone())); + } + } + let mut new_graphs = current.graphs.clone(); + new_graphs.insert(handle.key.clone(), Arc::clone(&handle)); + let mut new_quarantined = current.quarantined.clone(); + new_quarantined.remove(&handle.key); + self.snapshot.store(Arc::new(RegistrySnapshot::with_quarantined( + new_graphs, + new_quarantined, + ))); Ok(()) } + + /// Record a new or updated quarantine entry for a key with no serving + /// handle (RFC-030 W3). **No-op if the key is currently serving**: a + /// failed supervised reopen of a still-healthy graph must not take it + /// down — the old handle keeps serving and the retry reschedules. + pub async fn set_quarantined(&self, key: GraphKey, info: QuarantineInfo) { + let _guard = self.mutate.lock().await; + let current = self.snapshot.load(); + if current.graphs.contains_key(&key) { + return; + } + let mut new_quarantined = current.quarantined.clone(); + new_quarantined.insert(key, info); + self.snapshot.store(Arc::new(RegistrySnapshot::with_quarantined( + current.graphs.clone(), + new_quarantined, + ))); + } } fn canonicalize_handle_uri( @@ -306,6 +460,7 @@ mod tests { RegistryLookup::Ready(found) => { assert!(Arc::ptr_eq(&found, &handle)); } + RegistryLookup::Quarantined(_) => panic!("expected Ready, got Quarantined"), RegistryLookup::Gone => panic!("expected Ready, got Gone"), } } @@ -316,6 +471,7 @@ mod tests { let key = GraphKey::cluster(GraphId::try_from("ghost").unwrap()); match registry.get(&key) { RegistryLookup::Gone => {} + RegistryLookup::Quarantined(_) => panic!("expected Gone, got Quarantined"), RegistryLookup::Ready(_) => panic!("expected Gone"), } } @@ -553,8 +709,8 @@ mod tests { RegistryLookup::Ready(found) => { assert!(Arc::ptr_eq(&found, handle)); } - RegistryLookup::Gone => panic!( - "snapshot listed key {} but get() returned Gone", + RegistryLookup::Quarantined(_) | RegistryLookup::Gone => panic!( + "snapshot listed key {} but get() did not return Ready", handle.key.graph_id ), } @@ -567,4 +723,216 @@ mod tests { reader.await.unwrap(); assert_eq!(registry.len(), N_WRITES); } + + fn quarantine_info(uri: &str, policy_configured: bool) -> QuarantineInfo { + let now = std::time::SystemTime::now(); + QuarantineInfo { + uri: uri.to_string(), + since: now, + attempts: 1, + last_error: "open failed (test)".to_string(), + retry_at: now, + policy_configured, + } + } + + /// RFC-030 W3/W2(b): `publish` installs a handle, clears its quarantine + /// entry, and replaces a prior serving handle for the same key (the + /// supervised-reopen swap `insert` deliberately refuses). + #[tokio::test] + async fn publish_replaces_serving_handle_and_clears_quarantine() { + let dir = TempDir::new().unwrap(); + let key = GraphKey::cluster(GraphId::try_from("alpha").unwrap()); + let registry = GraphRegistry::from_boot( + Vec::new(), + vec![(key.clone(), quarantine_info("file:///nowhere/alpha", false))], + ) + .unwrap(); + match registry.get(&key) { + RegistryLookup::Quarantined(info) => assert_eq!(info.attempts, 1), + _ => panic!("boot quarantine entry must surface as Quarantined"), + } + + // Heal: publish clears the quarantine entry and serves. + let healed = build_handle("alpha", dir.path()).await; + registry.publish(Arc::clone(&healed)).await.unwrap(); + assert!(registry.snapshot_ref().quarantined.is_empty()); + match registry.get(&key) { + RegistryLookup::Ready(found) => assert!(Arc::ptr_eq(&found, &healed)), + _ => panic!("published handle must be Ready"), + } + + // Swap: publishing again for the same key replaces the handle + // (same-key replace skips the duplicate-URI check). + let dir2 = TempDir::new().unwrap(); + let replacement = build_handle("alpha", dir2.path()).await; + registry.publish(Arc::clone(&replacement)).await.unwrap(); + match registry.get(&key) { + RegistryLookup::Ready(found) => assert!(Arc::ptr_eq(&found, &replacement)), + _ => panic!("replacement handle must be Ready"), + } + assert_eq!(registry.len(), 1); + + // A different key colliding on URI is still rejected. + let colliding = Arc::new(GraphHandle { + key: GraphKey::cluster(GraphId::try_from("beta").unwrap()), + uri: replacement.uri.clone(), + engine: Arc::clone(&replacement.engine), + policy: None, + queries: None, + }); + match registry.publish(colliding).await { + Err(InsertError::DuplicateUri(_)) => {} + other => panic!("expected DuplicateUri, got {other:?}"), + } + } + + /// RFC-030 W3: a failed supervised reopen of a still-serving graph must + /// not take it down — `set_quarantined` is a no-op while the key serves. + #[tokio::test] + async fn set_quarantined_is_noop_while_key_is_serving() { + let dir = TempDir::new().unwrap(); + let handle = build_handle("alpha", dir.path()).await; + let key = handle.key.clone(); + let registry = GraphRegistry::from_handles(vec![handle]).unwrap(); + + registry + .set_quarantined(key.clone(), quarantine_info("file:///nowhere/alpha", false)) + .await; + assert!( + registry.snapshot_ref().quarantined.is_empty(), + "serving graph must not gain a quarantine entry", + ); + assert!(matches!(registry.get(&key), RegistryLookup::Ready(_))); + + // And for a non-serving key it records/updates the entry. + let ghost = GraphKey::cluster(GraphId::try_from("ghost").unwrap()); + registry + .set_quarantined(ghost.clone(), quarantine_info("file:///nowhere/ghost", false)) + .await; + let mut updated = quarantine_info("file:///nowhere/ghost", false); + updated.attempts = 3; + registry.set_quarantined(ghost.clone(), updated).await; + match registry.get(&ghost) { + RegistryLookup::Quarantined(info) => assert_eq!(info.attempts, 3), + _ => panic!("non-serving key must surface Quarantined"), + } + } + + /// Test-only `insert` is add-only over CONFIGURED keys: a quarantined + /// key is configured, so insert rejects it (publish is the heal path) — + /// one key can never sit in both snapshot maps. + #[tokio::test] + async fn insert_rejects_quarantined_key() { + let dir = TempDir::new().unwrap(); + let handle = build_handle("alpha", dir.path()).await; + let key = handle.key.clone(); + let registry = GraphRegistry::from_boot( + Vec::new(), + vec![(key, quarantine_info("file:///nowhere/alpha", false))], + ) + .unwrap(); + match registry.insert(handle).await { + Err(InsertError::DuplicateKey(_)) => {} + other => panic!("expected DuplicateKey for a quarantined key, got {other:?}"), + } + } + + /// RFC-030 review: the "distinct canonical URIs across configured + /// graphs" invariant must not depend on which opens succeed at boot. + /// When both colliding graphs open, the handle check already fails + /// boot; a half-open collision must fail the same way — otherwise the + /// quarantined side boots into an endless supervised-retry loop (its + /// reopen succeeds but `publish` forever returns `DuplicateUri`). + #[tokio::test] + async fn from_boot_rejects_uri_collisions_involving_quarantined_entries() { + // Serving handle and quarantined entry on the same canonical URI. + let dir = TempDir::new().unwrap(); + let handle = build_handle("alpha", dir.path()).await; + let uri = handle.uri.clone(); + let beta = GraphKey::cluster(GraphId::try_from("beta").unwrap()); + match GraphRegistry::from_boot( + vec![handle], + vec![(beta.clone(), quarantine_info(&uri, false))], + ) { + Err(InsertError::DuplicateUri(_)) => {} + Err(other) => panic!("expected DuplicateUri, got {other:?}"), + Ok(_) => panic!("expected DuplicateUri for serving/quarantined collision, got Ok"), + } + + // Two quarantined entries on the same canonical URI. + let gamma = GraphKey::cluster(GraphId::try_from("gamma").unwrap()); + match GraphRegistry::from_boot( + Vec::new(), + vec![ + (beta, quarantine_info("file:///nowhere/shared", false)), + (gamma, quarantine_info("file:///nowhere/shared", false)), + ], + ) { + Err(InsertError::DuplicateUri(_)) => {} + Err(other) => panic!("expected DuplicateUri, got {other:?}"), + Ok(_) => panic!("expected DuplicateUri for quarantined/quarantined collision, got Ok"), + } + } + + /// RFC-030 W3 auth-flap closure: a quarantined graph whose config + /// declares a per-graph policy keeps `any_per_graph_policy` true, so + /// bearer auth stays required while it heals. + #[tokio::test] + async fn quarantined_policy_bearing_graph_keeps_auth_required() { + let key = GraphKey::cluster(GraphId::try_from("secured").unwrap()); + let registry = GraphRegistry::from_boot( + Vec::new(), + vec![(key, quarantine_info("file:///nowhere/secured", true))], + ) + .unwrap(); + assert!( + registry.snapshot_ref().any_per_graph_policy, + "quarantined policy-bearing graph must keep bearer auth required", + ); + } + + /// Concurrent `publish` calls for the same key serialize on the mutate + /// mutex: all succeed (replace-allowed), the registry ends with exactly + /// one handle, and it is one of the published ones. + #[tokio::test(flavor = "multi_thread")] + async fn concurrent_publish_same_key_all_succeed_last_wins() { + const N: usize = 8; + let registry = Arc::new(GraphRegistry::new()); + let mut handles = Vec::with_capacity(N); + let mut dirs = Vec::with_capacity(N); + for _ in 0..N { + let d = TempDir::new().unwrap(); + handles.push(build_handle("contested", d.path()).await); + dirs.push(d); + } + let published: Vec<_> = handles.clone(); + + let barrier = Arc::new(tokio::sync::Barrier::new(N)); + let mut tasks = Vec::with_capacity(N); + for handle in handles { + let registry = Arc::clone(®istry); + let barrier = Arc::clone(&barrier); + tasks.push(tokio::spawn(async move { + barrier.wait().await; + registry.publish(handle).await + })); + } + for t in tasks { + t.await.unwrap().unwrap(); + } + assert_eq!(registry.len(), 1); + let winner = match registry.get(&published[0].key) { + RegistryLookup::Ready(handle) => handle, + _ => panic!("contested key must be Ready"), + }; + assert!( + published.iter().any(|h| { + // publish may canonicalize the handle; compare engines. + Arc::ptr_eq(&h.engine, &winner.engine) + }), + "winner must be one of the published handles", + ); + drop(dirs); + } } diff --git a/crates/omnigraph-server/src/supervisor.rs b/crates/omnigraph-server/src/supervisor.rs new file mode 100644 index 00000000..749006d4 --- /dev/null +++ b/crates/omnigraph-server/src/supervisor.rs @@ -0,0 +1,229 @@ +//! RFC-030 unified graph supervision (W3 boot retry + W2(b) supervised +//! reopen). +//! +//! One server-owned task with per-graph retry state, fed by two triggers: +//! quarantine entries seeded at boot (a graph whose open failed), and +//! `RecoveryRequired` notifications from the shielded write path (a served +//! graph carrying an unresolved rollback-class recovery residual). In both +//! cases the action is identical: re-drive the full `open_single_graph` — +//! whose read-write open runs the engine's Full recovery sweep under the +//! shared root-scoped write queue — and RCU-publish the healed handle into +//! the registry. In-flight requests on a replaced handle finish on their own +//! `Arc` clone (the registry's engine-survival contract). +//! +//! Deliberately ONE task, not one per graph: a single loop dedups +//! notification storms for the same graph and guarantees at most one +//! in-flight open per root (two concurrent read-write opens of one root +//! would run redundant Full sweeps). Shutdown is free: the only sender +//! lives in `AppState`; when the last state clone drops, `recv()` yields +//! `None` and the loop exits. Because each in-flight shielded write holds +//! an `AppState` clone (for the in-task reopen notification), the loop may +//! outlive `serve()` by up to one write's duration — consistent with those +//! writes themselves being detached, and bounded by the engine's commit +//! timeout. + +use std::collections::HashMap; +use std::sync::Arc; +use std::time::Duration; + +use tokio::sync::mpsc; + +use crate::identity::GraphKey; +use crate::registry::GraphRegistry; +use crate::GraphStartupConfig; + +/// Retry pacing for the supervision loop. Production uses capped +/// exponential backoff with jitter; tests inject millisecond-scale +/// timings so convergence tests complete in bounded wall-clock. +#[derive(Debug, Clone)] +pub struct SupervisorConfig { + pub initial_backoff: Duration, + pub multiplier: u32, + pub max_backoff: Duration, + /// Jitter as a fraction of the computed delay (0.0 disables). + pub jitter: f64, +} + +impl SupervisorConfig { + pub fn production() -> Self { + Self { + initial_backoff: Duration::from_secs(5), + multiplier: 2, + max_backoff: Duration::from_secs(600), + jitter: 0.1, + } + } + + /// Millisecond-scale pacing for in-process tests. + pub fn fast_for_tests() -> Self { + Self { + initial_backoff: Duration::from_millis(10), + multiplier: 2, + max_backoff: Duration::from_millis(200), + jitter: 0.0, + } + } +} + +/// Per-graph retry bookkeeping. +struct RetryState { + attempts: u32, + next_at: tokio::time::Instant, +} + +pub(crate) struct GraphSupervisor { + pub(crate) registry: Arc, + pub(crate) configs: Arc>, + pub(crate) config: SupervisorConfig, + pub(crate) rx: mpsc::UnboundedReceiver, +} + +impl GraphSupervisor { + /// The RFC-030 supervision loop. Seeds retry state from the registry's + /// boot-time quarantine entries (trigger A), then selects over reopen + /// notifications (trigger B — retried immediately, deduped while + /// pending) and the earliest scheduled retry. A due graph gets one full + /// `open_single_graph`; success RCU-publishes the healed handle + /// (clearing quarantine or swapping the serving handle), failure + /// reschedules with capped exponential backoff and refreshes the + /// quarantine record (a no-op while the graph still serves, so a failed + /// W2(b) reopen never takes a healthy graph down). + pub(crate) async fn run(mut self) { + use tokio::time::{Instant, sleep_until}; + use tracing::{info, warn}; + + let mut pending: HashMap = HashMap::new(); + for key in self.registry.snapshot_ref().quarantined.keys() { + pending.insert( + key.clone(), + RetryState { + attempts: 1, + next_at: Instant::now(), + }, + ); + } + loop { + // With nothing pending, park solely on the channel; the + // far-future fallback only exists to give select! a future. + let next_due = pending.values().map(|r| r.next_at).min(); + let deadline = + next_due.unwrap_or_else(|| Instant::now() + Duration::from_secs(24 * 3600)); + tokio::select! { + msg = self.rx.recv() => { + match msg { + // The only sender lives in AppState — its drop is + // server shutdown. + None => return, + Some(key) => { + // Dedup: a fresh trigger-B request retries + // immediately; an already-pending graph keeps + // its backoff schedule. Deliberate, not an + // oversight: the schedule is attempt-driven + // evidence about the OPEN path, and a new + // `RecoveryRequired` notification carries no + // evidence that path recovered — it only proves + // demand. Resetting on demand would let + // sustained 503 traffic collapse exponential + // backoff into a tight open loop exactly when + // the backend is struggling. Pending state only + // exists while an episode is unresolved + // (success removes it), so the worst-case added + // wait is one max_backoff interval. + pending.entry(key).or_insert(RetryState { + attempts: 0, + next_at: Instant::now(), + }); + } + } + } + _ = sleep_until(deadline), if next_due.is_some() => { + let now = Instant::now(); + let due: Vec = pending + .iter() + .filter(|(_, r)| r.next_at <= now) + .map(|(k, _)| k.clone()) + .collect(); + for key in due { + let Some(cfg) = self.configs.get(&key) else { + // No retained config (single mode, or a key from + // a foreign source) — nothing to reopen from. + warn!(graph_id = %key.graph_id, "no startup config retained; dropping supervision"); + pending.remove(&key); + continue; + }; + match crate::open_single_graph(cfg.clone()).await { + Ok(handle) => match self.registry.publish(handle).await { + Ok(()) => { + info!(graph_id = %key.graph_id, "supervised reopen healed graph"); + pending.remove(&key); + } + Err(err) => { + warn!(graph_id = %key.graph_id, error = %err, "supervised reopen could not publish"); + self.reschedule(&key, cfg, err.to_string(), &mut pending).await; + } + }, + Err(err) => { + warn!(graph_id = %key.graph_id, error = %err, "supervised reopen failed"); + self.reschedule(&key, cfg, err.to_string(), &mut pending).await; + } + } + } + } + } + } + } + + /// Backoff-reschedule `key` and refresh its quarantine record. + async fn reschedule( + &self, + key: &GraphKey, + cfg: &GraphStartupConfig, + last_error: String, + pending: &mut HashMap, + ) { + let Some(state) = pending.get_mut(key) else { + return; + }; + state.attempts = state.attempts.saturating_add(1); + let exp = state.attempts.saturating_sub(1).min(20); + let base = self + .config + .initial_backoff + .saturating_mul(self.config.multiplier.saturating_pow(exp)) + .min(self.config.max_backoff); + // Cheap deterministic-enough jitter without a rand dependency: + // subsecond nanos spread retries within ±jitter of the base delay. + let frac = f64::from( + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.subsec_nanos()) + .unwrap_or(0) + % 1000, + ) / 500.0 + - 1.0; + let delay = base.mul_f64((1.0 + self.config.jitter * frac).max(0.0)); + state.next_at = tokio::time::Instant::now() + delay; + + let now_sys = std::time::SystemTime::now(); + let since = self + .registry + .snapshot_ref() + .quarantined + .get(key) + .map(|q| q.since) + .unwrap_or(now_sys); + self.registry + .set_quarantined( + key.clone(), + crate::QuarantineInfo { + uri: cfg.uri.clone(), + since, + attempts: state.attempts, + last_error, + retry_at: now_sys + delay, + policy_configured: cfg.policy.is_some(), + }, + ) + .await; + } +} diff --git a/crates/omnigraph-server/tests/auth_policy.rs b/crates/omnigraph-server/tests/auth_policy.rs index 68e89b8d..60bcf496 100644 --- a/crates/omnigraph-server/tests/auth_policy.rs +++ b/crates/omnigraph-server/tests/auth_policy.rs @@ -827,7 +827,9 @@ async fn engine_layer_policy_fires_via_direct_arc_omnigraph_from_new_single() { ); let handle = match state.routing().registry.get(&key) { omnigraph_server::RegistryLookup::Ready(handle) => handle, - omnigraph_server::RegistryLookup::Gone => panic!("default graph must be registered"), + omnigraph_server::RegistryLookup::Quarantined(_) | omnigraph_server::RegistryLookup::Gone => { + panic!("default graph must be registered and serving") + } }; let engine = Arc::clone(&handle.engine); diff --git a/crates/omnigraph-server/tests/failpoints.rs b/crates/omnigraph-server/tests/failpoints.rs new file mode 100644 index 00000000..97c72de2 --- /dev/null +++ b/crates/omnigraph-server/tests/failpoints.rs @@ -0,0 +1,448 @@ +//! Feature-gated fault-injection tests at the HTTP boundary (RFC-030). +//! +//! Enabled with `--features failpoints` (a passthrough to the engine's +//! failpoint registry; the server defines no hooks of its own). Every test is +//! `#[serial]` because the `fail` crate registry is process-global, and uses +//! the multi-thread runtime because the rendezvous callback blocks a worker +//! thread until released. + +#![cfg(feature = "failpoints")] + +mod support; + +#[path = "../../omnigraph/tests/helpers/failpoint.rs"] +mod failpoint; + +use std::time::{Duration, Instant}; + +use axum::body::Body; +use axum::http::{Method, Request, StatusCode}; +use failpoint::Rendezvous; +use omnigraph_server::build_app; +use serde_json::json; +use serial_test::serial; +use support::*; +use tower::ServiceExt; + +fn mutate_request_on(branch: &str, name: &str, params: serde_json::Value) -> Request { + Request::builder() + .uri(g("/mutate")) + .method(Method::POST) + .header("content-type", "application/json") + .body( + Body::from( + serde_json::to_vec(&json!({ + "query": MUTATION_QUERIES, + "name": name, + "params": params, + "branch": branch, + })) + .unwrap(), + ), + ) + .unwrap() +} + +fn mutate_request(name: &str, params: serde_json::Value) -> Request { + mutate_request_on("main", name, params) +} + +fn branch_create_request(name: &str) -> Request { + Request::builder() + .uri(g("/branches")) + .method(Method::POST) + .header("content-type", "application/json") + .body(Body::from( + serde_json::to_vec(&json!({"name": name, "from": "main"})).unwrap(), + )) + .unwrap() +} + +fn merge_with_delete_request(source: &str) -> Request { + Request::builder() + .uri(g("/branches/merge")) + .method(Method::POST) + .header("content-type", "application/json") + .body(Body::from( + serde_json::to_vec(&json!({ + "source": source, + "target": "main", + "delete_branch": true, + })) + .unwrap(), + )) + .unwrap() +} + +fn branch_list_request() -> Request { + Request::builder() + .uri(g("/branches")) + .method(Method::GET) + .body(Body::empty()) + .unwrap() +} + +async fn branch_names(app: &axum::Router) -> Vec { + let response = app.clone().oneshot(branch_list_request()).await.unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let body = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .unwrap(); + let body: serde_json::Value = serde_json::from_slice(&body).unwrap(); + body["branches"] + .as_array() + .unwrap() + .iter() + .map(|b| b.as_str().unwrap().to_string()) + .collect() +} + +fn recovery_dir_entries(graph: &std::path::Path) -> Vec { + match std::fs::read_dir(graph.join("__recovery")) { + Ok(entries) => entries + .filter_map(|e| e.ok()) + .map(|e| e.file_name().to_string_lossy().into_owned()) + .collect(), + Err(_) => Vec::new(), + } +} + +/// RFC-030 W1 Stage 1: a mutation whose request future is dropped mid-protocol +/// (the client disconnected) must still run to its own terminal state — the +/// commit protocol publishes and deletes its recovery sidecar with NO further +/// request and NO graph reopen. +/// +/// The rendezvous parks the engine at the sidecar-confirmation failpoint — +/// after the recovery intent is armed and the table effect committed, before +/// publication — and the test aborts the task driving the request there, +/// which drops the axum handler future exactly as hyper does on client +/// disconnect (`tower::ServiceExt::oneshot` drives the handler inline, with +/// no connection task in between). +/// +/// The core assertion polls ONLY the filesystem: `__recovery/` must become +/// empty within the deadline. Sidecar deletion happens strictly after a +/// successful manifest publish, so an empty `__recovery/` proves the whole +/// protocol completed. The poll deliberately sends no requests and opens no +/// handles before the assertion: a follow-up write would run the write-entry +/// heal and a reopen would run the Full sweep, either of which resolves the +/// residual and would mask an unshielded server (the exact blind spot of the +/// engine's older cancellation test — see +/// `crates/omnigraph/tests/rfc030_probe.rs`). +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +#[serial] +async fn mutation_dropped_mid_protocol_completes_and_leaves_no_residual() { + let (temp, app) = app_for_loaded_graph().await; + let graph = graph_path(temp.path()); + + let rv = Rendezvous::park_first(omnigraph::failpoints::names::RECOVERY_SIDECAR_CONFIRM); + + // The doomed request: driven by its own task so aborting the task drops + // the in-flight handler (and engine) future, as a disconnect does. + let doomed_app = app.clone(); + let doomed = tokio::spawn(async move { + doomed_app + .oneshot(mutate_request("insert_person", json!({"name": "Eve", "age": 22}))) + .await + }); + + rv.wait_until_reached().await; + doomed.abort(); + rv.release(); + let join = doomed.await; + assert!( + join.is_err() && join.unwrap_err().is_cancelled(), + "the doomed request must have been cancelled mid-protocol, not completed", + ); + drop(rv); + + // Core shield property: the protocol finishes on its own. Poll the + // filesystem only — no requests, no reopen — until `__recovery/` is + // empty (sidecar deleted ⇒ manifest published) or the deadline expires. + let deadline = Instant::now() + Duration::from_secs(10); + loop { + let residue = recovery_dir_entries(&graph); + if residue.is_empty() { + break; + } + assert!( + Instant::now() < deadline, + "dropped mutation abandoned its armed protocol: __recovery/ still \ + holds {residue:?} after the deadline with no further requests — \ + the write was not shielded from caller cancellation (RFC-030 W1)", + ); + tokio::time::sleep(Duration::from_millis(25)).await; + } + + // The protocol completed, so Eve must be graph-visible: an update keyed + // on her row reports exactly one affected node. + let response = app + .clone() + .oneshot(mutate_request("set_age", json!({"name": "Eve", "age": 23}))) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK, "post-disconnect graph must be writable"); + let body = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .unwrap(); + let body: serde_json::Value = serde_json::from_slice(&body).unwrap(); + let affected = body + .get("affectedNodes") + .or_else(|| body.get("affected_nodes")) + .and_then(|v| v.as_u64()); + assert_eq!( + affected, + Some(1), + "the disconnected mutation's row must have been published (body: {body})", + ); + + // And the admission slot released at protocol completion: a fresh + // mutation admits and succeeds. + let response = app + .oneshot(mutate_request("insert_person", json!({"name": "Frank", "age": 33}))) + .await + .unwrap(); + assert_eq!( + response.status(), + StatusCode::OK, + "follow-up mutation must admit and succeed after the disconnected write completed", + ); +} + +/// RFC-030 review fix (effect-envelope shielding): a merge request with +/// `delete_branch: true` is ONE composite operation. A client disconnect +/// mid-merge must not split it — the already-durable merge AND the requested +/// source-branch deletion must both complete, and the admission guard must +/// span the composite. +/// +/// The rendezvous parks the engine inside the merge (post-authority-capture, +/// before any route work); aborting the request task there guarantees the +/// handler future is gone before the merge returns, so under the pre-fix +/// per-call shield the follow-up delete could never run. +/// +/// Polling uses only the read-only branch listing (reads run no write-entry +/// heal), so nothing can mask a skipped deletion. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +#[serial] +async fn merge_with_delete_branch_dropped_mid_merge_still_deletes_source() { + let (_temp, app) = app_for_loaded_graph().await; + + // Fixture: a source branch carrying one real change. + let response = app + .clone() + .oneshot(branch_create_request("feature")) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK, "branch create must succeed"); + let response = app + .clone() + .oneshot(mutate_request_on( + "feature", + "insert_person", + json!({"name": "Eve", "age": 22}), + )) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK, "branch write must succeed"); + + let rv = Rendezvous::park_first( + omnigraph::failpoints::names::BRANCH_MERGE_POST_AUTHORITY_CAPTURE, + ); + let doomed_app = app.clone(); + let doomed = tokio::spawn(async move { + doomed_app + .oneshot(merge_with_delete_request("feature")) + .await + }); + rv.wait_until_reached().await; + // Order matters for determinism: await the cancelled join BEFORE + // releasing the park. The engine-side task is still parked, so the + // request task's join is pending and the abort is processed cleanly — + // the handler future is provably gone before the merge resumes. (The + // reverse order races: a join woken by the released task can let the + // aborted handler run one final poll to completion.) + doomed.abort(); + let join = doomed.await; + assert!( + join.is_err() && join.unwrap_err().is_cancelled(), + "the doomed merge request must have been cancelled mid-merge", + ); + rv.release(); + drop(rv); + + // The composite must complete on its own: merge durable AND source + // branch deleted, with no further write request. + let deadline = Instant::now() + Duration::from_secs(10); + loop { + let names = branch_names(&app).await; + if !names.iter().any(|name| name == "feature") { + break; + } + assert!( + Instant::now() < deadline, + "disconnect split the merge composite: source branch 'feature' \ + still exists after the deadline — delete_branch was abandoned \ + with the dropped handler (RFC-030 effect-envelope shielding)", + ); + tokio::time::sleep(Duration::from_millis(25)).await; + } + + // And the merge itself landed on main: Eve is visible there. + let response = app + .clone() + .oneshot(mutate_request("set_age", json!({"name": "Eve", "age": 23}))) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let body = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .unwrap(); + let body: serde_json::Value = serde_json::from_slice(&body).unwrap(); + let affected = body + .get("affectedNodes") + .or_else(|| body.get("affected_nodes")) + .and_then(|v| v.as_u64()); + assert_eq!(affected, Some(1), "the merged row must be visible on main: {body}"); +} + +/// RFC-030 W2(b): a write surfacing `RecoveryRequired` (an unresolved +/// rollback-class recovery residual) triggers a supervised in-process reopen +/// whose Full sweep resolves the residual — the graph heals on the next +/// writes WITHOUT a process restart. This is the HTTP twin of +/// `rfc030_probe.rs` phases 2–3. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +#[serial] +async fn recovery_required_write_triggers_supervised_reopen_and_heals() { + use omnigraph::failpoints::ScopedFailPoint; + + // Multi-mode boot through open_multi_graph_state so the retained + // startup configs (the supervision loop's reopen source) are populated. + let temp = init_loaded_graph().await; + let graph = graph_path(temp.path()); + let cfg = omnigraph_server::GraphStartupConfig { + graph_id: "default".to_string(), + uri: graph.to_string_lossy().to_string(), + policy: None, + embedding: None, + queries: stored_query_registry(&[]), + }; + let state = omnigraph_server::open_multi_graph_state( + vec![cfg], + Vec::new(), + None, + temp.path().join("cluster.yaml"), + false, + ) + .await + .unwrap(); + // Supervision is deliberately NOT spawned yet: until `spawn_supervision` + // sets the notify sender, every `request_reopen` drops silently. That + // lets the manufacture step and the connected-wedge assertion run + // race-free (their own RecoveryRequired responses cannot arm a heal), + // isolating the doomed request as the ONLY possible notifier. + let app = build_app(state.clone()); + + // Manufacture an Armed residual through HTTP: the confirmation write + // fails (the failpoint's documented storage-crash model), leaving the + // sidecar armed with a committed table effect. The connected response IS + // the 503 recovery_required wedge — asserted race-free because no + // supervisor is listening yet. + { + let _fp = ScopedFailPoint::new( + omnigraph::failpoints::names::RECOVERY_SIDECAR_CONFIRM, + "return", + ); + let response = app + .clone() + .oneshot(mutate_request("insert_person", json!({"name": "Mallory", "age": 41}))) + .await + .unwrap(); + assert_eq!( + response.status(), + StatusCode::SERVICE_UNAVAILABLE, + "a failed confirmation write must surface RecoveryRequired", + ); + } + assert!( + !recovery_dir_entries(&graph).is_empty(), + "the Armed sidecar must remain on disk after the confirm failure", + ); + // The wedge, still race-free: a follow-up connected write is 503 too. + let response = app + .clone() + .oneshot(mutate_request("insert_person", json!({"name": "Frank", "age": 33}))) + .await + .unwrap(); + assert_eq!( + response.status(), + StatusCode::SERVICE_UNAVAILABLE, + "a write must surface RecoveryRequired while the residual is unresolved", + ); + assert!( + !recovery_dir_entries(&graph).is_empty(), + "with no supervisor spawned, nothing may heal the residual", + ); + + // Now start supervision. The ONLY notifier from here on is the doomed + // request below — the earlier connected 503s fired into the unset + // OnceLock and were dropped. + let _supervision = + state.spawn_supervision(omnigraph_server::SupervisorConfig::fast_for_tests()); + + // The doomed trigger-B request: park its write-entry heal (the residual + // is listed, gates not yet taken), abort the request task while parked + // (handler provably dropped), release. The shielded task then completes + // with RecoveryRequired with no waiter — the reopen signal must fire + // from INSIDE the shielded task for the graph to heal with NO further + // request. Poll the filesystem only. + { + let rv = Rendezvous::park_first( + omnigraph::failpoints::names::RECOVERY_POST_LIST_PRE_GATES, + ); + let doomed_app = app.clone(); + let doomed = tokio::spawn(async move { + doomed_app + .oneshot(mutate_request("insert_person", json!({"name": "Trent", "age": 50}))) + .await + }); + rv.wait_until_reached().await; + // abort -> await-cancelled -> release: the heal is still parked, so + // the handler future is provably dropped before the shielded task + // resumes and surfaces RecoveryRequired (see the merge test's + // ordering note). + doomed.abort(); + let join = doomed.await; + assert!( + join.is_err() && join.unwrap_err().is_cancelled(), + "the doomed trigger-B request must have been cancelled", + ); + rv.release(); + drop(rv); + + let deadline = Instant::now() + Duration::from_secs(15); + loop { + if recovery_dir_entries(&graph).is_empty() { + break; + } + assert!( + Instant::now() < deadline, + "disconnect dropped the trigger-B reopen signal: the Armed \ + residual survived the deadline with no further requests — \ + the notification must live inside the shielded task \ + (RFC-030 effect-envelope shielding)", + ); + tokio::time::sleep(Duration::from_millis(50)).await; + } + } + + // Healed without restart: a connected write succeeds. + let response = app + .clone() + .oneshot(mutate_request("insert_person", json!({"name": "Grace", "age": 44}))) + .await + .unwrap(); + assert_eq!( + response.status(), + StatusCode::OK, + "the graph must be writable after the supervised reopen healed the residual", + ); +} diff --git a/crates/omnigraph-server/tests/multi_graph.rs b/crates/omnigraph-server/tests/multi_graph.rs index 5679aef2..bfc75073 100644 --- a/crates/omnigraph-server/tests/multi_graph.rs +++ b/crates/omnigraph-server/tests/multi_graph.rs @@ -539,6 +539,11 @@ rules: assert_eq!(ready, vec!["good"]); let app = build_app(state); + // RFC-030 W3: a failed boot open quarantines the graph as OBSERVABLE, + // CONFIGURED state — not silent absence. `GET /graphs` lists both graphs + // with a status discriminator and quarantine detail; routes under the + // quarantined graph answer 503 (retry later), while a never-configured id + // stays 404 (no such resource). let (status, body) = json_response( &app, Request::builder() @@ -549,14 +554,35 @@ rules: ) .await; assert_eq!(status, StatusCode::OK, "{body}"); + let graphs_json = body["graphs"].as_array().unwrap(); assert_eq!( - body["graphs"] - .as_array() - .unwrap() + graphs_json .iter() .map(|graph| graph["graph_id"].as_str().unwrap()) .collect::>(), - vec!["good"] + vec!["broken", "good"], + "GET /graphs must list quarantined graphs alongside serving ones: {body}" + ); + assert_eq!( + graphs_json[1]["status"].as_str(), + Some("serving"), + "healthy graph must report status=serving: {body}" + ); + assert_eq!( + graphs_json[0]["status"].as_str(), + Some("quarantined"), + "failed-open graph must report status=quarantined: {body}" + ); + let quarantine = &graphs_json[0]["quarantine"]; + assert!( + quarantine["last_error"] + .as_str() + .is_some_and(|err| !err.is_empty()), + "quarantine detail must carry the open error: {body}" + ); + assert!( + quarantine["attempts"].as_u64().is_some_and(|n| n >= 1), + "quarantine detail must count the failed boot open: {body}" ); let (status, body) = json_response( @@ -568,9 +594,252 @@ rules: .unwrap(), ) .await; + assert_eq!( + status, + StatusCode::SERVICE_UNAVAILABLE, + "a quarantined graph is configured-but-healing: 503, not 404: {body}" + ); + + // A graph id that was never configured remains a plain 404 — pins the + // Quarantined-vs-Gone distinction. + let (status, body) = json_response( + &app, + Request::builder() + .uri("/graphs/never-configured/queries") + .header("authorization", "Bearer admin-token") + .body(Body::empty()) + .unwrap(), + ) + .await; assert_eq!(status, StatusCode::NOT_FOUND, "{body}"); } +/// RFC-030 W3: quarantine is a CONVERGING state, not a terminal one. A graph +/// whose boot open failed must reach `serving` without a process restart +/// once the fault clears — the supervision loop re-drives the full +/// `open_single_graph` with capped backoff and RCU-publishes the healed +/// handle. The "transient fault clears" model is deterministic: the graph +/// RFC-030 review fix (all-quarantined boot): the incident's own topology — +/// a single-graph deployment whose only graph fails its boot open on a +/// transient fault — must still boot in lenient mode, serve the quarantine +/// state observably, and converge once the fault clears. Zero SERVING graphs +/// with at least one quarantined entry is a valid, converging boot state; +/// only zero-configured (and settings-time config-class quarantine, which +/// supervision cannot heal) remain fail-fast. +#[tokio::test(flavor = "multi_thread")] +async fn all_quarantined_lenient_boot_serves_and_converges() { + let temp = tempfile::tempdir().unwrap(); + let schema = "\nnode Person {\n name: String @key\n}\n"; + let bad_uri = temp.path().join("solo.omni"); + let server_policy = omnigraph_server::PolicySource::Inline( + r#" +version: 1 +kind: server +groups: + admins: [act-admin] +rules: + - id: admins-list-graphs + allow: + actors: { group: admins } + actions: [graph_list] +"# + .to_string(), + ); + let graphs = vec![omnigraph_server::GraphStartupConfig { + graph_id: "solo".to_string(), + uri: bad_uri.to_string_lossy().to_string(), + policy: None, + embedding: None, + queries: stored_query_registry(&[]), + }]; + let state = omnigraph_server::open_multi_graph_state( + graphs, + vec![("act-admin".to_string(), "admin-token".to_string())], + Some(&server_policy), + temp.path().join("cluster.yaml"), + false, + ) + .await + .expect( + "lenient boot with every graph quarantined must serve so supervision \ + can converge (RFC-030 W3) — aborting here re-creates the incident's \ + offline-until-redeploy failure for single-graph deployments", + ); + let _supervision = + state.spawn_supervision(omnigraph_server::SupervisorConfig::fast_for_tests()); + let app = build_app(state); + + // Observable while nothing serves: the one graph is quarantined and its + // routes answer 503. + let (status, body) = json_response( + &app, + Request::builder() + .uri("/graphs") + .header("authorization", "Bearer admin-token") + .body(Body::empty()) + .unwrap(), + ) + .await; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!( + body["graphs"][0]["status"].as_str(), + Some("quarantined"), + "the solo graph must boot quarantined: {body}" + ); + let (status, body) = json_response( + &app, + Request::builder() + .uri("/graphs/solo/queries") + .header("authorization", "Bearer admin-token") + .body(Body::empty()) + .unwrap(), + ) + .await; + assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE, "{body}"); + + // The fault clears; supervision converges without a restart. + Omnigraph::init(bad_uri.to_string_lossy().as_ref(), schema) + .await + .unwrap(); + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10); + loop { + let (status, body) = json_response( + &app, + Request::builder() + .uri("/graphs") + .header("authorization", "Bearer admin-token") + .body(Body::empty()) + .unwrap(), + ) + .await; + assert_eq!(status, StatusCode::OK, "{body}"); + if body["graphs"][0]["status"].as_str() == Some("serving") { + break; + } + assert!( + std::time::Instant::now() < deadline, + "all-quarantined boot never converged to serving (RFC-030 W3); \ + last listing: {body}" + ); + tokio::time::sleep(std::time::Duration::from_millis(25)).await; + } +} + +/// root simply does not exist at boot and is initialized while the server +/// runs. +#[tokio::test(flavor = "multi_thread")] +async fn boot_quarantined_graph_converges_to_serving_without_restart() { + let temp = tempfile::tempdir().unwrap(); + let schema = "\nnode Person {\n name: String @key\n}\n"; + let good_uri = temp.path().join("good.omni"); + Omnigraph::init(good_uri.to_string_lossy().as_ref(), schema) + .await + .unwrap(); + let bad_uri = temp.path().join("late.omni"); + let server_policy = omnigraph_server::PolicySource::Inline( + r#" +version: 1 +kind: server +groups: + admins: [act-admin] +rules: + - id: admins-list-graphs + allow: + actors: { group: admins } + actions: [graph_list] +"# + .to_string(), + ); + let graphs = vec![ + omnigraph_server::GraphStartupConfig { + graph_id: "late".to_string(), + uri: bad_uri.to_string_lossy().to_string(), + policy: None, + embedding: None, + queries: stored_query_registry(&[]), + }, + omnigraph_server::GraphStartupConfig { + graph_id: "good".to_string(), + uri: good_uri.to_string_lossy().to_string(), + policy: None, + embedding: None, + queries: stored_query_registry(&[]), + }, + ]; + let state = omnigraph_server::open_multi_graph_state( + graphs, + vec![("act-admin".to_string(), "admin-token".to_string())], + Some(&server_policy), + temp.path().join("cluster.yaml"), + false, + ) + .await + .unwrap(); + let _supervision = + state.spawn_supervision(omnigraph_server::SupervisorConfig::fast_for_tests()); + let app = build_app(state); + + // Quarantined and visible while the root is missing. + let (status, body) = json_response( + &app, + Request::builder() + .uri("/graphs") + .header("authorization", "Bearer admin-token") + .body(Body::empty()) + .unwrap(), + ) + .await; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!( + body["graphs"][1]["status"].as_str(), + Some("quarantined"), + "late graph must boot quarantined: {body}" + ); + + // The fault clears: the graph root appears while the server runs. + Omnigraph::init(bad_uri.to_string_lossy().as_ref(), schema) + .await + .unwrap(); + + // Convergence: bounded poll of GET /graphs until `late` serves. Red + // without the supervision loop (the drain-only skeleton never reopens), + // green once the loop re-drives the open and publishes the handle. + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10); + loop { + let (status, body) = json_response( + &app, + Request::builder() + .uri("/graphs") + .header("authorization", "Bearer admin-token") + .body(Body::empty()) + .unwrap(), + ) + .await; + assert_eq!(status, StatusCode::OK, "{body}"); + if body["graphs"][1]["status"].as_str() == Some("serving") { + break; + } + assert!( + std::time::Instant::now() < deadline, + "quarantined graph never converged to serving without a restart \ + (RFC-030 W3); last listing: {body}" + ); + tokio::time::sleep(std::time::Duration::from_millis(25)).await; + } + + // And it actually serves. + let (status, body) = json_response( + &app, + Request::builder() + .uri("/graphs/late/queries") + .header("authorization", "Bearer admin-token") + .body(Body::empty()) + .unwrap(), + ) + .await; + assert_eq!(status, StatusCode::OK, "healed graph must serve: {body}"); +} + #[tokio::test(flavor = "multi_thread")] #[serial] async fn cluster_boot_injects_embedding_provider_config() { diff --git a/crates/omnigraph/tests/failpoint_names_guard.rs b/crates/omnigraph/tests/failpoint_names_guard.rs index df8fc1c8..d55c8982 100644 --- a/crates/omnigraph/tests/failpoint_names_guard.rs +++ b/crates/omnigraph/tests/failpoint_names_guard.rs @@ -33,16 +33,20 @@ fn manifest_dir() -> PathBuf { } /// Production call sites live under each crate's `src`; test call sites live -/// in the two failpoint integration binaries. This guard file is deliberately -/// not in the set (it names the patterns as literals itself). +/// in the failpoint integration binaries (engine, cluster, and — since +/// RFC-030 — the server's HTTP-boundary suite). This guard file is +/// deliberately not in the set (it names the patterns as literals itself). fn files_to_scan() -> Vec { let engine = manifest_dir(); let cluster = engine.join("../omnigraph-cluster"); + let server = engine.join("../omnigraph-server"); let mut out = Vec::new(); collect_rs(&engine.join("src"), &mut out); collect_rs(&cluster.join("src"), &mut out); + collect_rs(&server.join("src"), &mut out); out.push(engine.join("tests/failpoints.rs")); out.push(cluster.join("tests/failpoints.rs")); + out.push(server.join("tests/failpoints.rs")); out } diff --git a/crates/omnigraph/tests/rfc030_probe.rs b/crates/omnigraph/tests/rfc030_probe.rs new file mode 100644 index 00000000..1c3823f2 --- /dev/null +++ b/crates/omnigraph/tests/rfc030_probe.rs @@ -0,0 +1,185 @@ +//! RFC-030 behavior probe: pins the current cancellation-residual lifecycle +//! at the engine boundary. +//! +//! Three deterministic phases, each pinning one claim the RFC's design rests +//! on: +//! +//! 1. **Bug 1 (torn-by-cancellation):** dropping a mutation future parked at +//! the sidecar-confirmation failpoint leaves its recovery sidecar on disk — +//! no drop guard compensates. (Empirical nuance, recorded in RFC-030: the +//! residual's *class* is backend-dependent. On local FS the in-flight +//! confirmation write runs on `spawn_blocking` and completes even after the +//! future is dropped, so the leaked sidecar usually lands +//! `EffectsConfirmed` — self-healing on the next write's roll-forward-only +//! heal. On a network object store the in-flight PUT dies with the future, +//! leaving `Armed`. This phase therefore asserts only the leak, which is +//! deterministic; phase 2 manufactures the `Armed` class explicitly.) +//! 2. **Bug 2 (wedged-until-barrier):** a confirmation-write failure (the +//! documented crash model for `RECOVERY_SIDECAR_CONFIRM`) leaves an `Armed` +//! sidecar with a committed table effect. A subsequent write on the +//! still-live handle hits the roll-forward-only heal, cannot resolve the +//! rollback-class residual, and returns `RecoveryRequired`. +//! 3. **W2(b) (in-process reopen heals):** a fresh read-write +//! `Omnigraph::open` in the SAME process runs the Full sweep under the +//! shared root-scoped write queue, resolves the residual, and the +//! previously wedged handle becomes writable again — no process restart. + +#![cfg(feature = "failpoints")] + +mod helpers; + +use helpers::failpoint::Rendezvous; +use helpers::*; +use omnigraph::db::Omnigraph; +use omnigraph::error::OmniError; +use omnigraph::failpoints::ScopedFailPoint; +use omnigraph::loader::{LoadMode, load_jsonl}; +use serial_test::serial; + +fn recovery_dir_entries(uri: &str) -> Vec { + match std::fs::read_dir(format!("{uri}/__recovery")) { + Ok(entries) => entries + .filter_map(|e| e.ok()) + .map(|e| e.file_name().to_string_lossy().into_owned()) + .collect(), + Err(_) => Vec::new(), + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +#[serial] +async fn rfc030_cancellation_leaks_sidecar_armed_residual_wedges_and_reopen_heals() { + let dir = tempfile::tempdir().unwrap(); + let uri = dir.path().to_string_lossy().into_owned(); + + { + let mut db = Omnigraph::init(&uri, TEST_SCHEMA).await.unwrap(); + load_jsonl(&mut db, TEST_DATA, LoadMode::Overwrite) + .await + .unwrap(); + } + + // The long-lived "server" handle, opened before either incident. + let server_handle = Omnigraph::open(&uri).await.unwrap(); + + // ── Phase 1 / Bug 1: cancellation leaks the sidecar. ──────────────────── + { + let rv = Rendezvous::park_first(omnigraph::failpoints::names::RECOVERY_SIDECAR_CONFIRM); + let uri_task = uri.clone(); + let doomed = tokio::spawn(async move { + let db = Omnigraph::open(&uri_task).await.unwrap(); + db.mutate( + "main", + MUTATION_QUERIES, + "insert_person", + &mixed_params(&[("$name", "Eve")], &[("$age", 22)]), + ) + .await + }); + rv.wait_until_reached().await; + // The client hangs up: the request future is dropped mid-protocol. + doomed.abort(); + rv.release(); + let join = doomed.await; + assert!( + join.is_err() && join.unwrap_err().is_cancelled(), + "the doomed mutation must have been cancelled, not completed", + ); + drop(rv); + + let leaked = recovery_dir_entries(&uri); + println!("PROBE bug1: __recovery after cancellation = {leaked:?}"); + assert!( + !leaked.is_empty(), + "cancelling the mutation future mid-protocol must leak its recovery \ + sidecar (bug 1); found none", + ); + + // Clear to a clean slate for phase 2 via the open-time Full sweep, + // whichever class the phase-1 residual landed in. + drop(Omnigraph::open(&uri).await.unwrap()); + assert!( + recovery_dir_entries(&uri).is_empty(), + "open-time Full sweep must clear the phase-1 residual", + ); + } + + // ── Phase 2 / Bug 2: an Armed residual wedges the live handle. ────────── + { + // Deterministic Armed manufacture: the confirmation write FAILS (the + // failpoint's documented storage-crash model), so the table effect is + // committed while the sidecar stays pre-confirm. + let fp = ScopedFailPoint::new( + omnigraph::failpoints::names::RECOVERY_SIDECAR_CONFIRM, + "return", + ); + let crashed = { + let db = Omnigraph::open(&uri).await.unwrap(); + db.mutate( + "main", + MUTATION_QUERIES, + "insert_person", + &mixed_params(&[("$name", "Mallory")], &[("$age", 41)]), + ) + .await + }; + drop(fp); + assert!( + crashed.is_err(), + "a failed confirmation write must fail the mutation; got {crashed:?}", + ); + let armed = recovery_dir_entries(&uri); + println!("PROBE bug2 setup: __recovery after confirm failure = {armed:?}"); + assert!(!armed.is_empty(), "the Armed sidecar must remain on disk"); + + let wedged = server_handle + .mutate( + "main", + MUTATION_QUERIES, + "insert_person", + &mixed_params(&[("$name", "Frank")], &[("$age", 33)]), + ) + .await; + match &wedged { + Err(OmniError::RecoveryRequired { + operation_id, + reason, + }) => { + println!("PROBE bug2: RecoveryRequired op={operation_id} reason={reason}"); + } + other => panic!( + "a write on the live handle must return RecoveryRequired while \ + the rollback-class residual is unresolved (bug 2); got {other:?}", + ), + } + assert!( + !recovery_dir_entries(&uri).is_empty(), + "the roll-forward-only heal must leave the rollback-class sidecar on disk", + ); + } + + // ── Phase 3 / W2(b): an in-process reopen runs the Full sweep. ────────── + drop(Omnigraph::open(&uri).await.unwrap()); + let after_reopen = recovery_dir_entries(&uri); + println!("PROBE reopen: __recovery after in-process open = {after_reopen:?}"); + assert!( + after_reopen.is_empty(), + "a read-write open in the SAME process must run the Full sweep and \ + resolve the residual; sidecars remain: {after_reopen:?}", + ); + + // The previously wedged handle recovers without restart. + server_handle + .mutate( + "main", + MUTATION_QUERIES, + "insert_person", + &mixed_params(&[("$name", "Grace")], &[("$age", 44)]), + ) + .await + .expect( + "the original handle must be writable again after the in-process \ + reopen healed the residual", + ); + println!("PROBE recovered: original handle writable again without process restart"); +} diff --git a/crates/omnigraph/tests/writes.rs b/crates/omnigraph/tests/writes.rs index 2113568c..d2c03a10 100644 --- a/crates/omnigraph/tests/writes.rs +++ b/crates/omnigraph/tests/writes.rs @@ -308,19 +308,29 @@ async fn stale_non_strict_insert_reprepares_from_live_branch_state() { } /// The cancellation hole that motivated removing the Run state machine: dropping a mutation future -/// mid-flight must not leave any graph-level state behind. With the run -/// state machine gone, only orphaned Lance fragments can remain — and those -/// are reclaimed by `omnigraph cleanup`. +/// mid-flight must not leave any graph-level *staging* state behind (no +/// `__run__*` branches, no `_graph_runs.lance`). +/// +/// Scope note (RFC-030): "no graph-level state" is narrower than it sounds. +/// Under the RFC-022 protocol a drop after the recovery sidecar is armed CAN +/// leave that sidecar plus a committed-but-unpublished table effect on disk — +/// recovery-covered residue, resolved by the write-entry heal or the next +/// read-write open. This test cannot observe that residue: its final +/// `Omnigraph::open` runs the Full recovery sweep before any assertion runs. +/// The residue lifecycle is pinned by `rfc030_probe.rs`, and the HTTP +/// boundary shields cancellation entirely (`omnigraph-server`'s +/// `tests/failpoints.rs`), so an SDK caller dropping an engine future +/// mid-protocol is the only remaining cancellation source (RFC-030 W1 +/// Stage 2 territory). /// /// The test deliberately does NOT assert that the manifest version is /// unchanged: `handle.abort()` is racing the spawned task, and on a fast /// machine the mutation may complete before cancellation. That is acceptable -/// — what matters for cancel safety is that no `__run__*` staging branches -/// are ever created, that `_graph_runs.lance` is never written, and that -/// any partial state on disk is reachable through the regular manifest / -/// commit graph pipes (so `omnigraph cleanup` can reclaim it). Asserting -/// version equality would just be a flake on hosts where the abort lands -/// late. +/// — what matters here is that no `__run__*` staging branches are ever +/// created, that `_graph_runs.lance` is never written, and that any partial +/// state on disk is reachable through the regular manifest / commit graph / +/// recovery pipes. Asserting version equality would just be a flake on hosts +/// where the abort lands late. #[tokio::test] async fn cancelled_mutation_future_leaves_no_state() { let dir = tempfile::tempdir().unwrap(); diff --git a/docs/dev/invariants.md b/docs/dev/invariants.md index 76aa5f9a..80044c0d 100644 --- a/docs/dev/invariants.md +++ b/docs/dev/invariants.md @@ -389,6 +389,12 @@ them explicit. needs a cross-process serialization primitive (for example, a lease-based use of the schema-apply lock branch). Design and test that fence before promoting multi-process write topologies. + Within one process, RFC-030 narrows the operational cost of this boundary: + the write-entry heal remains roll-forward-only, but the HTTP server now + shields armed writes from caller cancellation and, on `RecoveryRequired`, + performs a supervised in-process reopen whose Full sweep resolves + rollback-class residuals without a process restart (a reopen IS the + documented resolution barrier — no recovery rule is relaxed). - **Fork ownership is durable, but Lance ref deletion is not conditional:** a first-touch Mutation/Load or BranchMerge table never creates its target ref before recovery ownership is durable. Under schema → branch → table gates it diff --git a/docs/dev/testing.md b/docs/dev/testing.md index dd068cfb..a314c79b 100644 --- a/docs/dev/testing.md +++ b/docs/dev/testing.md @@ -9,7 +9,7 @@ This file is the always-on map of the test surface. **Consult it before every ta | `omnigraph` (engine) | `crates/omnigraph/tests/` | Integration tests (one file per behavior area — see the table below), fixture-driven, share `tests/helpers/mod.rs` | | `omnigraph-cli` | `crates/omnigraph-cli/tests/` | Per-area suites (post-modularization): `cli_cluster.rs` (cluster command surface + operator-actor cascade), `cli_cluster_e2e.rs` (spawned-binary lifecycle compositions — lost-state re-import recovery, out-of-band drift, graph-root destruction, multi-graph mixed-disposition convergence), `cli_data.rs` (load/read/change/branch/commit/export/snapshot/policy/embed/maintenance + operator format cascade), `cli_schema_config.rs` (init/config, schema plan/apply), `cli_queries.rs`, `parity_matrix.rs` (RFC-009 Phase 1: the embedded-vs-remote referee — every forked verb run against both arms with matched Cedar policy and the same actor, scrubbed-JSON + exit-code equality; divergences are pinned in its `KNOWN_DIVERGENCES` ledger, never silently repaired), `system_local.rs` (full-cycle cluster lifecycle with a spawned `--cluster` server, applied-policy enforcement over HTTP, keyed-credential auth, operator aliases), `system_remote.rs`, `crossversion_upgrade.rs` (genuine v3→v9 and v4/v5/v6/v7/v8↔v9 rebuild/refusal harness — see below); share `tests/support/mod.rs` (hermetic `OMNIGRAPH_HOME` by default) | | `omnigraph-cluster` | mostly in-source `#[cfg(test)] mod tests`; `tests/failpoints.rs` (feature-gated); `tests/s3_cluster.rs` (bucket-gated full lifecycle on object storage) | Cluster config parser, local JSON state diff, state CAS/lock handling/recovery, read-only validate/plan/status plus explicit refresh/import graph observations, config-only apply (content-addressed payload publish, disposition gating, composite-digest convergence, idempotent re-apply), catalog payload verification (status read-only, refresh drift + self-heal), failpoint crash-mid-apply / CAS-race coverage, Stage 4A graph creation (create executor, recovery sidecars + sweep rows, create crash windows), Stage 4B schema apply (migration previews in plan, schema executor, schema-apply sweep classification, schema crash windows), Stage 4C gated deletes (digest-bound approvals, delete executor + tombstones, delete sweep rows, delete crash windows), and 5A policy binding metadata (applies_to in the applied revision, binding-change diffing + convergence, pre-5A backfill), and the 5B serving-snapshot read API (converged read, refusal rows) | -| `omnigraph-server` | `crates/omnigraph-server/tests/` | Per-area suites (post-modularization): `auth_policy.rs`, `data_routes.rs`, `schema_routes.rs`, `stored_queries.rs`, `multi_graph.rs` (cluster-mode boot — converged serving, policy binding wiring, boot refusals — + the concurrent branch-ops matrix), `boot_settings.rs` (mode inference, PolicySource), `s3.rs` (bucket-gated: single-graph serving + config-free `--cluster s3://` boot), `openapi.rs` (OpenAPI drift / regeneration); share `tests/support/mod.rs` | +| `omnigraph-server` | `crates/omnigraph-server/tests/` | Per-area suites (post-modularization): `auth_policy.rs`, `data_routes.rs`, `schema_routes.rs`, `stored_queries.rs`, `multi_graph.rs` (cluster-mode boot — converged serving, policy binding wiring, boot refusals, RFC-030 quarantine visibility/503 semantics and the boot-quarantine-converges-without-restart supervision cell — + the concurrent branch-ops matrix), `boot_settings.rs` (mode inference, PolicySource), `s3.rs` (bucket-gated: single-graph serving + config-free `--cluster s3://` boot), `openapi.rs` (OpenAPI drift / regeneration), `failpoints.rs` (feature-gated via the `failpoints` passthrough — HTTP-boundary fault injection; owns the RFC-030 W1 cancellation-shield cell — a request dropped mid-protocol must complete its armed write with no residue, asserted by filesystem-only polling of `__recovery/` so no follow-up request or reopen can mask an unshielded server — and the W2(b) cell: a confirm-failure-injected Armed residual wedges writes with 503 `recovery_required`, and the supervised in-process reopen heals it without restart); share `tests/support/mod.rs` (hermetic; `failpoints.rs` additionally reuses the engine's `Rendezvous` helper by path) | | `omnigraph-compiler` | mostly in-source `#[cfg(test)] mod tests` | Parser, type-checker, IR lowering, lint. Schema parser and SchemaIR validation tests both reject the five exact Lance virtual system-column property names while preserving near-miss identifiers | The engine's `tests/` is the principal coverage surface; most graph-shaped behavior is exercised there. @@ -529,12 +529,16 @@ sidecar arm or graph movement. - **Serialize and rendezvous, never sleep.** The `fail` registry is process-global, so every failpoint test carries `#[serial]` (`serial_test`). For concurrent tests, use `helpers::failpoint::Rendezvous` (`tests/helpers/failpoint.rs`): `park_first(name)` parks the first thread to hit the point until `release()`, and `wait_until_reached().await` blocks on that condition (it doubles as a fired-assertion). Do not coordinate threads with fixed `sleep`s. - Activated tests: `crates/omnigraph/tests/failpoints.rs`, `crates/omnigraph/tests/memwal_stream.rs`, - `crates/omnigraph/tests/memwal_stream_cost.rs`, and - `crates/omnigraph-cluster/tests/failpoints.rs` (integration binaries, never + `crates/omnigraph/tests/memwal_stream_cost.rs`, + `crates/omnigraph-cluster/tests/failpoints.rs`, and + `crates/omnigraph-server/tests/failpoints.rs` (integration binaries, never in-source — the fail registry is process-global). Run the main suites with `cargo test -p omnigraph-engine --features failpoints --test failpoints` / - `cargo test -p omnigraph-cluster --features failpoints --test failpoints`; - Gate R0's exact command is documented above. + `cargo test -p omnigraph-cluster --features failpoints --test failpoints` / + `cargo test -p omnigraph-server --features failpoints --test failpoints`; + Gate R0's exact command is documented above. The server's `failpoints` + feature is a passthrough to `omnigraph/failpoints` (no hooks of its own), + and `failpoint_names_guard.rs` walks the server crate too. ## RustFS / S3 integration diff --git a/docs/dev/writes.md b/docs/dev/writes.md index 91baa24c..fb224e28 100644 --- a/docs/dev/writes.md +++ b/docs/dev/writes.md @@ -1326,8 +1326,21 @@ on the staged tables, so the next mutation proceeds normally with no drift to reconcile. The cancellation case (future drop mid-mutation) inherits the same -guarantee — the in-memory accumulator evaporates with the dropped task -and no Lance write was ever issued. +guarantee **during op execution**: the in-memory accumulator evaporates +with the dropped task and no Lance write was ever issued. A drop +*after* the commit phase begins is a different animal (RFC-030): the +recovery sidecar is armed before the first durable effect, so a future +dropped between arm and sidecar delete leaves recovery-covered residue — +a sidecar plus possibly a committed-but-unpublished table effect — +resolved by the write-entry heal (roll-forward-eligible residue) or the +next read-write open's Full sweep (rollback-class residue). The residue +lifecycle is pinned by `tests/rfc030_probe.rs`. The HTTP server shields +its write handlers from caller cancellation entirely (RFC-030 W1 +Stage 1: the protocol runs on a server-owned task and completes +regardless of client fate — `omnigraph-server/src/handlers.rs:: +shielded_write`), so post-arm cancellation is reachable only by an +embedded SDK caller dropping an engine future mid-protocol (W1 Stage 2 +closes that). Delete-touching mutations now inherit the same guarantee (MR-A). Deletes accumulate as predicates and stage via `stage_delete` at end-of-query, so a diff --git a/docs/rfcs/0030-write-lifetime-and-recovery-barriers.md b/docs/rfcs/0030-write-lifetime-and-recovery-barriers.md new file mode 100644 index 00000000..dddf0091 --- /dev/null +++ b/docs/rfcs/0030-write-lifetime-and-recovery-barriers.md @@ -0,0 +1,633 @@ +--- +type: spec +title: "RFC-030 — Write-protocol lifetime ownership and recovery resolution barriers" +description: Shields armed write protocols from caller cancellation, resolves rollback-class recovery residuals at the write-entry gate instead of process restart, and turns server-boot quarantine into an observable, converging state. +status: draft +tags: [eng, rfc, writes, recovery, cancellation, server, availability, omnigraph] +timestamp: 2026-07-20 +owner: OmniGraph maintainers +--- + +# RFC-030: Write-protocol lifetime ownership and recovery resolution barriers + +**Status:** Draft +**Date:** 2026-07-20 +**Author track:** Maintainer design series +**Depends on:** [RFC-022](0022-unified-write-path.md)'s recovery-v9 envelope, +gate ordering, and `Armed`/`EffectsConfirmed` classification; +[RFC-028](0028-stable-schema-identity.md)'s identity-keyed sidecars +**Surveyed:** OmniGraph 0.8.1 (`main`); Lance 9.0.0-rc.1 at git rev +`cec0b7dffe2d85c7e66dbe9d1f3891c297903a1d` (upstream +`rust/lance/src/io/commit.rs`, `rust/lance/src/dataset.rs`, +`rust/lance-table/src/io/commit.rs` read at that rev) +**Audience:** engine write-path, recovery, and server maintainers + +--- + +## 0. Decision summary + +The RFC-022 commit protocol is sound: every write arms an identity-bearing +recovery intent before its first durable effect, publishes graph visibility in +one `__manifest` CAS, and resolves interruptions all-or-nothing. This RFC does +not change that protocol. It changes **who owns the protocol's lifetime** and +**where its residuals may be resolved**, closing two structural defects that +turn a correct protocol into an availability incident: + +1. **A cancellable caller future owns a multi-step durable protocol.** The + server awaits `mutate_as` inside the request future + (`crates/omnigraph-server/src/handlers.rs:697`), and no `Drop` guard exists + anywhere in the staging/publisher/recovery path. A client disconnect + therefore aborts the engine mid-protocol and parks an `Armed` sidecar. +2. **The only full-resolution barrier is process restart.** The live-handle + heal is deliberately roll-forward-only + (`crates/omnigraph/src/db/manifest/recovery.rs:425-429`): sidecars that + need restore or abort are "LEFT ON DISK for the next ReadWrite open". An + `Armed` residual on a live server wedges writes to its tables for the rest + of the process lifetime, and the restart that resolves it performs a single + unsupervised open attempt per graph — a transient object-store error during + that open quarantines the graph until the next restart + (`crates/omnigraph-server/src/lib.rs:1284-1343`). + +Three coordinated, independently landable changes: + +- **W1 — Engine-owned write lifetime (cancellation shielding).** Once a write + operation begins its gated protocol, its execution is owned by a spawned + task, not the caller's future. Caller cancellation means "stopped waiting + for the result", never "abandoned the protocol". Staged: the server boundary + first (the established spawn-and-clone idiom), then engine-level `'static` + write execution as the durable close for embedded SDK callers. +- **W2 — Residual resolution without process restart.** Two implementations, + staged. **W2(b), first:** a supervised in-process reopen — on + `RecoveryRequired`, the server rebuilds the graph handle via + `open_single_graph` and swaps it into the registry; the read-write open's + Full sweep runs under the *same root-scoped write queue* live handles + share, so it serializes correctly against live traffic and resolves the + residual using only existing, failpoint-tested machinery. Zero engine + changes, no stated rule relaxed. **W2(a), end state:** the write-entry heal + escalates rollback-class residuals to `RecoveryMode::Full` under + *exclusive* admission plus the existing schema → branch → table gates, + avoiding W2(b)'s per-heal cache teardown and covering embedded SDK callers. + Either way, the property restore actually requires is exclusive authority + over the affected tables; process start was only ever a proxy for it. + `RecoveryRequired` becomes a transient condition, not an operator event. +- **W3 — Boot supervision.** A graph whose open fails enters an observable + `quarantined` registry state with a capped-backoff re-open loop, instead of + silently dropping out of the handle set until redeploy. The engine's + fail-closed recovery classification is untouched; the retry unit is the + whole (idempotent) open. W3 and W2(b) are one server mechanism — a + supervised reopen triggered by boot failure or by `RecoveryRequired` — not + two. + +Together they establish one new standing rule, proposed for +[invariants.md](../dev/invariants.md): + +> **Caller fate is not a protocol participant.** An armed graph-write protocol +> completes or compensates under engine ownership regardless of caller +> cancellation, and resolving any recovery residual requires only exclusive +> gate acquisition — never process restart. + +## 1. Motivation: a validated field incident + +A production report against 0.8.1 described a client-timeout-interrupted +multi-statement delete that left "torn state", wedged writes with +`stale view of 'edge:ProjectAsksQuestion': expected manifest table version 2 +but current is 3`, and — after the next boot's cleanup hit a transient S3 +error — a hard-quarantined graph that stayed offline until a second redeploy. + +Every link of that chain was validated against current code, including Lance +upstream source at the pinned rev: + +| Incident observation | Validated mechanism | +|---|---| +| Client 5-minute timeout aborted the delete mid-flight | `run_mutate` awaits `mutate_as` inline in the request future; hyper drops the handler future on disconnect; no `Drop` compensation exists in `exec/staging.rs`, `db/manifest/publisher.rs`, or `db/manifest/recovery.rs` | +| "stale view … expected 2 but current is 3" on later writes | `OmniError::manifest_expected_version_mismatch` (`src/error.rs:247`) — the typed OCC conflict raised while the orphaned table HEAD sits ahead of the manifest | +| Wedged until restart | The live-handle heal runs `RecoveryMode::RollForwardOnly`; the `Armed` residual is rollback-class and is parked for the next ReadWrite open (`recovery.rs:420-429`) | +| Boot "cleanup" wrote `_versions/18446744073709551611.manifest` | Ordinary recovery table commit: Lance V2 manifest naming writes `_versions/{u64::MAX − version, 020-padded}.manifest`; that filename decodes to **version 4** — exactly what a restore at drifted HEAD 3 commits (upstream `lance-table/src/io/commit.rs`, `ManifestNamingScheme::V2`; `Dataset::restore` commits `Operation::Restore` at latest + 1, `lance/src/dataset.rs:1272`) | +| "Failed to clean up orphaned transaction file…" | Verbatim Lance warning on the best-effort transaction-file delete after a failed commit (upstream `lance/src/io/commit.rs:114`); the orphaned `_transactions/` file is harmless residue reclaimed by cleanup | +| "graph quarantined during startup" until a "lucky" redeploy | One open attempt per graph, `warn!` and exclude, no retry loop, no runtime re-open, no health surface (`lib.rs:1284-1343`); the second redeploy's Full sweep was the designed recovery path finally running to completion | + +No reader ever observed torn state — snapshot isolation over +manifest-published versions held throughout — and the final outcome was the +all-or-nothing one. What failed was availability: an interruption that the +protocol is designed to absorb instead cost the graph its writability for a +process lifetime and its entire service for a deploy cycle. + +The long-run liability argument (the project's first principle): today, every +`await` point between sidecar arm and sidecar delete is a latent torn-state +site, and cancel-safety is maintained *by audit* — every future change to the +write path must re-reason about cancellation at every await. W1 replaces that +recurring obligation with a structural property. Similarly, W2 replaces +"remember that `Armed` residuals need a restart" — operational knowledge that +lives in runbooks and pages — with a barrier the system crosses on its own. + +## 2. Current architecture (what this RFC preserves) + +For reviewers: the boundaries this RFC deliberately does **not** move. + +- The RFC-022 protocol order — resolve/refuse relevant intents, capture the + base under admission → schema → branch → table gates, arm the recovery-v9 + sidecar before the first durable effect, commit exact per-table + transactions, confirm achieved effects, publish once through the manifest + CAS, delete the sidecar — is unchanged in every branch. +- `Armed` remains rollback-only for the established writers; + `EffectsConfirmed` remains roll-forward-only under captured authority. + Ambiguity still fails closed. No classification rule is weakened. +- The documented single-writer-process support boundary is unchanged. W2 does + not make destructive recovery safe against a live *foreign process*; that + still requires the distributed fence tracked in + [invariants.md](../dev/invariants.md) Known Gaps. +- The sidecar wire format, recovery schema versions, and on-disk layout are + untouched. Nothing in this RFC is a format change. +- Rollback already ends published: a successful roll-back "appends one restore + commit and then publishes" the restored versions (`recovery.rs:3509-3512`), + and the bound plan persists before the first restore so a retry replays it + (`recovery.rs:1027`). W2 and W3 lean on that existing idempotency; they do + not add a new compensation mechanism. + +## 3. W1 — Engine-owned write-protocol lifetime + +### 3.1 Problem shape + +`Omnigraph`'s write entry points are `&self` async methods; the server holds +`Arc` and awaits them inline in the request handler. When the +client connection closes, hyper drops the handler future and the engine +future with it, at whatever await point it has reached. Post-arm, that parks +an `Armed` sidecar; the write-queue guards release on drop, so a *subsequent* +writer immediately trips over the residual (and, per current W2-less behavior, +cannot resolve it). + +The existing cancellation test +(`writes.rs::cancelled_mutation_future_leaves_no_state`) cannot see this: its +final `Omnigraph::open` runs the Full sweep, which silently heals any leaked +sidecar before the assertions run, and its doc comment ("only orphaned Lance +fragments can remain") predates the RFC-022 sidecar protocol. That comment is +corrected as part of this work. + +**Empirical nuance (probe-observed): the leaked residual's class is a +storage-backend-dependent race.** Which state the sidecar lands in depends on +where the drop's first pending yield falls relative to the confirmation +write — and on whether that write survives the drop. On local FS, storage +puts run on `spawn_blocking`, and an already-spawned blocking write completes +even after the awaiting future is dropped; the probe observed exactly this: +a mutation cancelled *at* the confirmation failpoint still produced a +durably-confirmed sidecar, which the next write's roll-forward-only heal +resolved in-process (write succeeded, no wedge). On a network object store +the in-flight PUT dies with the future, so the same cancellation leaves the +`Armed` class that wedges — consistent with the field incident occurring on +S3. The design consequence: cancellation's outcome today is a +backend-dependent roulette between "self-heals on next write" and "wedged +until a reopen barrier". W1 does not tune the roulette; it removes it. + +### 3.2 Stage 1 — server-boundary shield + +The server already documents and uses the exact idiom needed: +`registry.rs:14-19` specifies that a request's own `Arc` clone +keeps the engine alive across registry swaps, and `server_export` +(`handlers.rs:620-637`) spawns engine work that survives the request future. +Stage 1 applies the same spawn-and-clone shape to every `_as` writer handler +(`run_mutate`, `run_ingest`, schema apply, branch create/delete/merge, +maintenance): + +```rust +let admission = state.workload.try_admit(&actor_arc, est_bytes)?; +let engine = Arc::clone(&handle.engine); +let task = tokio::spawn(async move { + let _admission = admission; // released at protocol completion + engine.mutate_as(&branch, &query, &name, ¶ms, actor_id.as_deref()).await + // ...and the RecoveryRequired → request_reopen notification fires + // HERE, inside the task, before the result is returned. +}); +let result = task.await /* JoinError → 500 */ ?; +``` + +**The shielded unit is the request's effect envelope, not the engine call.** +Everything that must survive client fate lives inside the spawned task: the +admission guard, every armed protocol the request implies (a merge with +`delete_branch: true` is ONE composite task covering both the merge and the +follow-up delete), and post-outcome obligations such as the supervised-reopen +notification. Anything left in the handler is, by construction, cancellable — +the review-fix tranche closed exactly the three gaps that violated this rule +(handler-side delete, handler-side notify, admission released at first-protocol +completion). + +Semantics: + +- The HTTP response contract is unchanged: the handler still awaits the + result and returns the same status/body mapping. No 202/async-job surface + is introduced. +- On client disconnect, the request future drops but the spawned task runs + the protocol to its own terminal state: success (sidecar deleted) or error + (compensated by the operation's existing error path). The result is + discarded. +- The per-actor admission permit moves into the spawned task and is released + at protocol completion, not at disconnect. This is intentional: admission + bounds concurrent *work*, and the work genuinely continues. A disconnecting + actor cannot free its own slot early by hanging up. +- Panic in the task surfaces as `JoinError` → 500, the same observability as + today's unwind through the handler. + +Stage 1 closes the production HTTP surface — the only caller class that +routinely drops futures mid-flight. The CLI is not shielded by Stage 1 and +does not need to be: a CLI future is only dropped by process death, which is +crash-class and already recovery-covered. + +### 3.3 Stage 2 — engine-level `'static` write execution + +Embedded SDK callers can still drop futures (timeout combinators, select +loops). The durable close moves the shield into the engine so the property +holds for every consumer. + +**This is not a new pattern — it is the engine's newest-writer standard.** +The RFC-026 B1 stream-fold writer already detaches its *entire* protocol: +`stream_fold_phase_b1(self: &Arc, …)` clones the Arc, owns its inputs, +and spawns, with the doc comment "The entire operation is detached, not +merely the Lance seal. This keeps the exclusive admission token alive from +the cut through recovery arm, the exact base-table effect, and the one +manifest visibility CAS **even if the requesting task is cancelled**" +(`db/omnigraph/stream_ingest.rs`). It spawns through +`instrumentation::spawn_with_query_io_probes`, an existing crate-private +helper written for exactly this ("a few correctness-sensitive paths detach +physical work so cancellation cannot abandon it") that also re-scopes the +observation task-locals `tokio::spawn` would otherwise drop — so the cost +harness keeps seeing datasets opened inside the detached task, and +production (which never installs probes) gets a plain spawn. Stage 2 is +therefore precisely scoped: bring the five established writers (Mutation, +Load, SchemaApply, BranchMerge, EnsureIndices — and Optimize) up to the +cancellation-ownership standard the B1 writer already set, using the same +`Arc`-receiver shape and the same spawn helper. + +Validated constraint: `Omnigraph` is not `Clone`, and `commit_all` takes +borrowed `&WriteTxn` / `&LineageIntent`, so this is a real restructure, not a +wrapper. The struct is already Arc-shaped where it matters — `coordinator: +Arc>`, `storage: Arc`, +`read_caches: Arc<…>`, `schema_view: Arc>`, and the root-scoped +write queues are process-shared — so the design question is packaging, not +architecture. Two candidate shapes, resolved during implementation review: + +- **(a) `Arc` receivers:** shielded variants + `pub async fn mutate_as(self: &Arc, …)` clone the Arc into the + spawned protocol task; existing `&self` methods delegate where possible. +- **(b) Inner-handle split:** move the shared fields into one + `Arc`; `Omnigraph` becomes a cheap-clone façade and write + entry points spawn with a clone of the inner Arc. + +Either way, the write path's inputs become owned at the entry point (query +text, params, branch — all cheap clones), and the spawned future is +`Send + 'static` (engine futures already run inside hyper's spawned +connection tasks today, so `Send` holds). + +Contract note: `crates/omnigraph/tests/forbidden_apis.rs` classifies every +public async inherent `Omnigraph` method. Any new or re-signed entry point +updates that registry in the same commit, per its protocol. + +### 3.4 Non-goals + +No detached "fire-and-forget" write API, no job queue, no server-side +completion watchdog beyond the existing Lance 30-minute commit timeout. The +caller still waits; only abandonment semantics change. + +## 4. W2 — Full recovery at the write-entry barrier + +### 4.1 Design + +W1 prevents cancellation-created residuals, but crash-class residuals +(kill -9, OOM, power) remain, and "wedged until restart" must die +independently. The change: when the write-entry heal +(`heal_pending_sidecars_roll_forward`, `recovery.rs:3601`) encounters a +sidecar whose classification is rollback-class (today's parked case), it +escalates that sidecar to `RecoveryMode::Full` under strengthened gates +instead of parking it. + +Mechanics, building on what the heal already does: + +1. **Gates.** The heal already acquires admission → schema → branch → table + guards in the one total order shared by writers, Full recovery, and live + healing, and re-reads the sidecar under those gates so a completed writer's + deletion or durable confirmation is respected (`recovery.rs:3617-3666`). + One validated delta is required: ordinary sidecars currently take **shared** + stream admission (`recovery.rs:3644-3648`); a Full-mode heal performing + restore escalates to **exclusive** admission for the affected tables, the + same class the open-time sweep effectively enjoys and the same shape + RFC-026 enrollment/fold sidecars already use on this path. +2. **Dispatch.** `process_sidecar` already takes `RecoveryMode` as a + parameter; the heal passes `Full` for the escalated sidecar. Roll-forward + cases are unchanged (they already resolve in-process). +3. **Completion.** Rollback uses the existing machinery unchanged: persisted + bound plan, restore commit at HEAD + 1, publication of the restored + versions, recovery audit row, sidecar delete. No residual + `HEAD > manifest` drift survives the heal, because publication of restored + versions is already part of the rollback contract. +4. **Then the triggering write proceeds** from a fresh capture, exactly as if + it had opened a clean graph. + +### 4.2 Safety argument + +The rule being relaxed is "restore only at the open-time barrier". The +property restore actually needs is **exclusive authority over the affected +tables while the restore-and-publish completes**: + +- **In-process writers** are excluded by the exclusive admission + table + gates — stronger than the open-time situation, where the barrier is merely + "no traffic has started yet". +- **In-process readers** are safe by construction, not by exclusion: reads + pin manifest-published versions, restore *appends* a new version (upstream + `Dataset::restore`, validated at the pinned rev), and read caches are keyed + by version/e_tag. A reader mid-query during a heal observes exactly what it + observed before the residual existed. +- **Foreign processes** are exactly as (un)protected as at open time. The + documented single-writer-process boundary owns that risk today for the + restart path; W2 neither widens nor narrows it. The distributed fence + remains future work with its own gate. + +Process start was a proxy for exclusivity. The root-scoped +`WriteQueueManager` provides the real thing, so the barrier moves from +"restart" to "next write's gate acquisition". + +### 4.3 W2(b) — supervised in-process reopen (lands first) + +Investigation of the open path showed the restart barrier is already softer +than "process restart": a read-write `Omnigraph::open` in the *same process* +runs the Full sweep (`db/omnigraph.rs:628-651`), under exclusive recovery +admission (`:602-604`) and the same root-scoped `WriteQueueManager` that +every live handle for that root shares — `recover_manifest_drift` takes the +queue as a parameter and acquires the shared total-order gates per sidecar. +The open path's own comment states the design intent this RFC operationalizes: +"only rollback-eligible sidecars wait for this open-time sweep." + +W2(b) therefore needs no engine change at all: when a request surfaces +`RecoveryRequired`, the server's supervision loop (the same one W3 adds for +boot) rebuilds the graph handle via `open_single_graph` and swaps it into +the registry. Engine-instance survival across registry swaps is already +documented and relied upon (`registry.rs:14-19`): in-flight requests on the +old handle finish on their own Arc clone; new requests get the healed +handle. The behavior probe (§7) pins this end to end: a leaked `Armed` +sidecar wedges the live handle with `RecoveryRequired`, an in-process open +clears `__recovery/`, and the previously wedged handle writes again — no +restart. + +Costs, honestly stated: the reopened handle starts with cold read caches and +re-validates stored queries/policy for that graph (bounded, one-time, +per-residual — and residuals are crash-class-rare once W1 lands); and W2(b) +covers server deployments only. Embedded SDK callers keep the documented +"reopen read-write" remedy until W2(a). Those costs are why W2(a) remains +the end state rather than being dropped; W2(b) is why W2(a) is no longer +urgent. + +### 4.4 Contract changes + +`RecoveryRequired` (HTTP 503) becomes transient: the *next* write to the +affected tables resolves the residual instead of every write failing until +restart. The [errors.md](../user/operations/errors.md) guidance — "resolve the +sidecar through a read-write reopen/server restart" — is rewritten to +"resolves automatically on the next write or reopen; back off and retry". +Per Hyrum's law this narrowing is compatible (clients coded to the current +contract simply see their backoff succeed sooner), but it is an observable +behavior change and is documented as such in the release notes. + +## 5. W3 — Boot supervision: quarantine as a converging state + +### 5.1 Design + +The engine keeps deciding once and failing closed; the server gains the +supervision loop it currently lacks. Division of labor: + +- **One mechanism with W2(b).** The supervision loop is shared: it is armed + by a failed boot open (W3) or by a served graph surfacing + `RecoveryRequired` (W2(b)); in both cases the action is the same + supervised `open_single_graph` + registry swap. `open_single_graph` is a + pure function of its (clonable) `GraphStartupConfig`, so re-driving it is + mechanically safe. +- **Registry state.** A graph whose open fails enters + `Quarantined { since, attempts, last_error, retry_at }` in the registry + instead of vanishing from the handle set. (The registry's documented `Gone` + direction already anticipates non-serving states.) +- **Re-open loop.** One supervision task per quarantined graph retries the + full `open_single_graph` with capped exponential backoff plus jitter + (proposed: 5 s initial, ×2, 10 min cap; final values are an unresolved + question below). The retry unit is deliberately the *whole open*: recovery + is idempotent by its replay-bounded plans and pinned by the + convergence-idempotent roll-forward regression, so re-driving it is always + safe, and no per-I/O retry policy is smeared through the recovery module. +- **Observability.** `GET /graphs` gains a per-graph `status` field + (`serving` | `quarantined`) with the quarantine detail object. This is an + additive OpenAPI change (`openapi.json` regenerated in the same PR) and + answers the incident report's "quarantine is invisible except in logs". +- **Strict mode preserved.** `--require-all-graphs` still aborts startup on + any failed graph; supervision applies only to the default + serve-the-healthy-subset mode. + +In the field incident, W3 alone turns "offline until a lucky redeploy" into +"offline for one backoff interval". + +### 5.2 What W3 does not do + +No engine change. No weakening of fail-closed classification. No retry of +*ambiguity* — a sidecar that classifies as indeterminate keeps failing the +open on every attempt (loudly, now visibly in `GET /graphs`) until an +operator intervenes; what the loop absorbs is transient substrate I/O. + +## 6. Invariants & deny-list check + +Walked against [invariants.md](../dev/invariants.md): + +- **Inv 2 (one publication door):** untouched. W1/W2/W3 add no visibility + path; W2 uses the existing restore-then-publish rollback. +- **Inv 3 (one coherent accepted view):** untouched; the escalated heal + re-reads the sidecar and captures fresh authority under gates, as today. +- **Inv 5 (recovery is part of the commit protocol):** strengthened. Writers + still resolve or refuse every relevant intent; W2 widens *where* resolution + may complete, not *what* may be resolved. Ambiguity still fails closed. +- **Inv 6 (durability before acknowledgement):** untouched. W1 changes who + waits, not what is durable before a success response. +- **Inv 13 (failures bounded, typed, observable):** improved — quarantine + becomes typed observable state; `RecoveryRequired` keeps its type with a + narrower lifetime; the boot loop is bounded by backoff caps. +- **Inv 15 (one source of truth):** untouched; no new derived copies. The + quarantine registry entry is process-local supervision state, not graph + authority. + +Deny-list brushes, and why each is not a violation: + +- *"Job queue for manifest-derivable state"* — the W3 loop supervises + process-local open failures, not manifest-derivable graph state; the + resolution barriers (write entry, open) remain event-driven. No timer-driven + sidecar janitor is added (explicitly rejected, §9). +- *"Acknowledging before durable persistence"* — unchanged; W1's spawned task + completes the same durable protocol before the same response. +- *"Shipping observable behavior as if it weren't contract"* — the 503 + lifetime change and the disconnect-surviving mutation are called out as + contract changes with docs and release notes (§4.3, §8). + +One stated rule is relaxed under this RFC's review — by W2(a) only: the +recovery-mode doc comment's "restore … for the next ReadWrite open" becomes +"restore under exclusive gate acquisition (write-entry heal or open)". W2(b) +relaxes nothing — it *is* a ReadWrite open, exercised in-process, which the +recovery design already supports (the sweep takes the shared root-scoped +queue precisely so it can serialize against live handles). This RFC is the +invariant-review artifact for the W2(a) relaxation when it lands. + +## 7. Evidence plan + +Per the repo's test-first rule, each change lands as a red test commit +followed by the fix commit. + +**Baseline pin (checked in, green):** `tests/rfc030_probe.rs` — a +feature-gated failpoint test pinning the *current* lifecycle this RFC +changes, at the engine boundary, in three deterministic phases: (1) +cancelling a mutation future parked at the sidecar-confirmation failpoint +leaks the recovery sidecar — the residual's *class* being the +backend-dependent race documented in §3.1, phase 1 asserts only the leak; +(2) a failed confirmation write (the failpoint's documented storage-crash +model) deterministically manufactures the `Armed` class, and a subsequent +write on the still-live handle returns `RecoveryRequired` with the +documented "reopen the graph read-write" guidance; (3) an in-process +read-write open clears `__recovery/` and the previously wedged handle +writes again without a process restart — the W2(b) mechanism, proven +end to end. After W1 lands, phase 1 inverts by design — cancellation can no +longer create the residual — and is superseded by the W1 red test below; +phases 2–3 remain valid permanently (crash-class residuals survive W1). + +**W1 (red first):** +- New failpoint cell: rendezvous-park the mutation protocol *post-arm, + pre-confirmation*; drop the caller future; release; assert the operation + reaches a terminal state and `__recovery/` is empty **without any reopen** + (listing sidecars directly — the existing cancellation test is blind here + because its `Omnigraph::open` sweeps first). **Stage-2 scope (PR-3):** this + engine-boundary cell can only turn green under the engine-level shield; + PR-1 lands the server-boundary cell below, and `rfc030_probe.rs` phase 1 + remains the engine baseline pin until then. +- Server-level test: spawned-server mutation with the client connection + dropped mid-flight (failpoint-parked); assert the mutation publishes and a + follow-up read observes it. +- Same-PR docs: correct the stale + `cancelled_mutation_future_leaves_no_state` doc comment. + +**W2 (red first):** +- Failpoint cell: manufacture an `Armed` residual (crash-injection, not + cancellation, so it stays valid after W1); a *same-handle* write must fully + heal it and then succeed. Today this asserts `RecoveryRequired`; the + expectation flip is the deliberate, reviewed contract change. +- Concurrent-reader cell: a long-lived snapshot read spanning the heal + observes identical results before and after the restore-and-publish. +- Admission cell: the escalated heal takes exclusive admission (extend the + existing write-queue unit tests). +- Existing open-time cells (`ensure_indices_complete_armed_effects_roll_back`, + the SchemaApply/Optimize/BranchMerge rollback matrix) remain untouched and + green — open-time behavior is unchanged. + +**W3 (red first):** +- Server test: inject one transient storage fault into boot-time recovery; + assert the graph is `quarantined` in `GET /graphs`, then converges to + `serving` without a restart. +- Strict-mode test: `--require-all-graphs` still aborts. +- OpenAPI regeneration committed (`OMNIGRAPH_UPDATE_OPENAPI=1`). + +**Docs in the same PRs:** [writes.md](../dev/writes.md) (protocol lifetime +ownership), [invariants.md](../dev/invariants.md) (the new standing rule + +recovery-mode relaxation), [errors.md](../user/operations/errors.md) (503 +guidance), [server.md](../user/operations/server.md) (quarantine status, +disconnect semantics), release notes. + +## 8. Behavior-change ledger (Hyrum's law) + +Observable changes shipped by this RFC, stated as contract: + +| Change | Before | After | +|---|---|---| +| Mutation vs client disconnect | Work cancelled mid-protocol; outcome ambiguous *and* residual likely | Work completes server-side; outcome ambiguous to the disconnected client only (verify by read) | +| `RecoveryRequired` (503) lifetime | Until process restart | Until the next write to the affected tables (or reopen) | +| Admission slot on disconnect | Released at cancellation | Released at protocol completion | +| `GET /graphs` | Serving graphs only | All configured graphs with `status`; additive schema | +| Quarantine | Terminal until restart, log-only | Converging with backoff, visible in `GET /graphs` | + +## 9. Drawbacks & alternatives + +- **Do nothing.** The protocol is correct and self-heals at restart; clients + can be taught generous timeouts. Rejected: cancel-safety-by-audit is a + recurring cost on every write-path change, and the availability failure + mode is disproportionate to its trigger (one impatient client). +- **Per-await cancel-guards / `select!` hardening.** Symptomatic: every new + await point reopens the bug class. Rejected. +- **Drop-guard compensation (`impl Drop` spawning cleanup).** Compensation + from a `Drop` impl cannot await and would race the very gates that make + recovery safe. Rejected in favor of not being cancelled at all. +- **202-Accepted async job API.** A larger product-surface change that + weakens the synchronous contract every existing client relies on; not + needed to fix the defect. Rejected (revisitable independently). +- **Per-I/O retry inside recovery.** Blurs the fail-closed line between + transient error and ambiguity, and smears retry policy through the module. + Rejected: the idempotent retry unit is the whole open (W3). +- **Timer-driven background sidecar janitor.** Adds a third resolution path + to keep consistent with two event-driven ones, for no coverage the + write-entry barrier doesn't already give once W2 lands. Rejected. +- **Weakening `Armed` classification (roll forward unconfirmed effects).** + Violates the partial-publish rule outright. Rejected without ceremony. + +Costs accepted: one task spawn per write (microseconds, off the hot path's +semantics); a modest engine restructure in W1 Stage 2 with a +`forbidden_apis.rs` registry update; slightly longer admission occupancy for +disconnected writes (bounded by the write caps and the Lance commit timeout); +one supervision task per quarantined graph (zero when healthy). + +## 10. Reversibility + +Fully reversible. No on-disk, sidecar-schema, or wire-format change; the one +additive API field is `GET /graphs.status`. W2 can be reverted by passing +`RollForwardOnly` again at the heal site; W3 by removing the supervision +loop; W1 Stage 1 by unspawning the handlers. Per the project's reversibility +rule this demands test evidence rather than irreversibility-grade proof — the +RFC exists because W2 relaxes a stated recovery rule (invariant-review +process) and because the three changes form one contract users will observe +together. + +## 11. Rollout + +Independent PR series, in leverage order, each red-test-first with docs: + +1. **PR-1 (W1 Stage 1):** server-boundary shield + the two W1 tests + stale + comment fix. Smallest change, kills the incident's trigger. +2. **PR-2 (W3 + W2(b)):** the unified supervision mechanism — registry + quarantine state, the supervised `open_single_graph` + registry-swap loop, + its two triggers (boot failure, `RecoveryRequired`), `GET /graphs` status + + OpenAPI regen + server.md/errors.md updates. Zero engine changes. +3. **PR-3 (W1 Stage 2):** engine `'static` write execution via the + established `Arc`-receiver + `spawn_with_query_io_probes` pattern + (§3.3), `forbidden_apis.rs` registry update, SDK-level cancellation test. +4. **PR-4 (W2(a), optional end state):** exclusive-admission escalation + + `Full` at the write-entry heal + expectation flips + the invariants.md + relaxation. Motivated by W2(b)'s cache-teardown cost and by embedded + callers; scheduled on evidence that either matters in practice. + +## 12. Unresolved questions + +- **W1 Stage 2 shape — narrowed by precedent:** the B1 stream-fold writer + already uses `self: &Arc` receivers with + `spawn_with_query_io_probes` (§3.3), so the remaining question is only + whether the *public* SDK entry points adopt the same receiver shape or an + `Arc` split keeps them `&self`. +- **Whether W2(a) ships at all,** given W2(b) + W1 close the incident class: + the remaining motivations are per-heal cache teardown (rare once W1 lands) + and embedded SDK callers (who hold the reopen remedy themselves). +- **Backoff policy for W3** (initial/cap/jitter) and whether attempts should + be capped at all, or retry indefinitely at the cap interval. +- **Whether the shielded task needs its own deadline** beyond Lance's + 30-minute commit timeout, and what a deadline expiry should do (it must not + cancel the protocol — at most it can alert). +- **Whether quarantine should also surface via a health/readiness endpoint** + for orchestrators that never call `GET /graphs`. +- **Generation fencing for supervised publishes:** today `startup_configs` + are boot-frozen and runtime add/remove does not exist (RFC-011: `cluster + apply` + restart), so a supervised reopen can never publish from stale + configuration. When the registry's deferred runtime DELETE/reconfigure + lands, the supervisor's opens must carry a config-generation token and + `publish` must CAS against it, so an in-flight reopen from a removed or + reconfigured graph cannot resurrect an obsolete handle or URI. +- **Settings-time quarantine visibility:** graphs quarantined at + settings-build time (stored-query parse or embedding-provider config + failures — config-class, unfixable by retry, correctly fail-fast) never + reach the registry and are invisible to `GET /graphs`. Surfacing them with + a distinct status is natural follow-up work now the status field exists. +- **Interaction with the future distributed fence:** when cross-process + recovery fencing lands, the W2 exclusivity argument should be restated in + terms of the fence rather than the process-local queue; flagged so the + fence RFC picks it up. diff --git a/docs/user/operations/errors.md b/docs/user/operations/errors.md index 06c3b795..37648ba8 100644 --- a/docs/user/operations/errors.md +++ b/docs/user/operations/errors.md @@ -36,7 +36,7 @@ recovery arm and has no durable effect. HTTP returns **413** with `resource_limit.{resource,limit,actual}`. Reshape the input; it is not partial success. -- `RecoveryRequired { operation_id, reason }` — an overlapping durable recovery intent remains unresolved. Its physical effects may already have landed, or it may still be armed before the first effect. HTTP returns **503** with `recovery_required.operation_id`. Resolve the sidecar through a read-write reopen/server restart before retrying; this is intentionally not an ordinary OCC retry. +- `RecoveryRequired { operation_id, reason }` — an overlapping durable recovery intent remains unresolved. Its physical effects may already have landed, or it may still be armed before the first effect. HTTP returns **503** with `recovery_required.operation_id`. This is intentionally not an ordinary OCC retry: the residual is resolved by a read-write reopen's recovery sweep. Against the HTTP server this now happens **automatically** (RFC-030: the write that surfaced the 503 arms a supervised in-process reopen) — back off on a seconds scale and retry, and escalate only if the 503 persists. Embedded SDK callers resolve it themselves with a read-write `Omnigraph::open` (or process restart). For RFC-023 Mutation/Load keyed writes, `KeyConflict` is returned only after the writer proves that none of its planned table effects landed, finalizes the diff --git a/docs/user/operations/server.md b/docs/user/operations/server.md index 4d4749ce..b9b96882 100644 --- a/docs/user/operations/server.md +++ b/docs/user/operations/server.md @@ -1,6 +1,9 @@ # HTTP Server (`omnigraph-server`) -Axum 0.8 + tokio + utoipa-generated OpenAPI. **Cluster-only boot**: the server always boots from a cluster (`--cluster `) and serves N graphs (N ≥ 1) under cluster routes. There is no longer a single-graph flat-route mode, no positional `` boot, no `--target`, and no `omnigraph.yaml`-`graphs:`-map boot. All HTTP is nested under `/graphs/{graph_id}/...`; `/healthz` and the management `/graphs` enumeration stay flat. +Axum 0.8 + tokio + utoipa-generated OpenAPI. **Cluster-only boot**: the server always boots from a cluster (`--cluster `) and serves N configured graphs (N ≥ 1) under cluster routes. In default +(lenient) mode the *serving* subset may transiently be smaller — even empty — +while open-quarantined graphs converge under supervision; strict +`--require-all-graphs` keeps serving = configured. There is no longer a single-graph flat-route mode, no positional `` boot, no `--target`, and no `omnigraph.yaml`-`graphs:`-map boot. All HTTP is nested under `/graphs/{graph_id}/...`; `/healthz` and the management `/graphs` enumeration stay flat. ## Boot @@ -26,8 +29,21 @@ state, invalid/unattributable recovery sidecars, unreadable shared catalog payloads, cluster policy errors, or zero healthy graphs. Graph-attributed pending recovery sidecars and graph-specific startup failures quarantine that graph instead; the server logs startup diagnostics and serves the -remaining healthy graphs. `GET /graphs` enumerates ready/served graphs only, -so quarantined graphs are absent and their routes return 404. +remaining healthy graphs. + +Quarantine is a **converging, observable state** (RFC-030): the server's +supervision loop retries the full graph open with capped exponential backoff +(5 s initial, ×2, 10 min cap, ±10 % jitter), so a graph that failed on a +transient fault returns to service without a restart. Meanwhile `GET /graphs` +lists the quarantined graph with `status: "quarantined"` and a `quarantine` +detail object (`since_unix_secs`, `attempts`, `last_error`, +`retry_at_unix_secs`); healthy graphs report `status: "serving"`. Routes +under a quarantined graph answer **503** ("retrying — retry later"), while a +graph id that was never configured stays 404. The same supervision loop also +heals a served graph that surfaces `recovery_required` on a write: the +server reopens the graph in-process (running the engine's full recovery +sweep) and swaps the healed handle in — in-flight requests on the old handle +finish unaffected. Operators who want the original all-or-nothing boot contract can pass `--require-all-graphs` or set `OMNIGRAPH_REQUIRE_ALL_GRAPHS=1`. In that mode, @@ -228,6 +244,37 @@ and `/schema/apply`. Read-only endpoints (`/snapshot`, `/query`, `/read`, `/export`, `/branches` GET, `/commits`, `/schema` GET) are not admission-gated. +## Client disconnects during writes (RFC-030) + +Once a write handler passes authorization and admission, the engine's commit +protocol runs on a server-owned task, detached from the request connection. +A client that times out or disconnects mid-request does **not** abort the +write: the server runs the protocol to its own terminal state — publish, or a +compensated failure — exactly as if the client had stayed connected. Two +consequences for clients: + +- **A disconnect makes the outcome unknown to you, not failed.** Verify by + reading the affected rows before retrying. A retried strict insert is safe + by construction: it succeeds or returns `key_conflict` — and a + `key_conflict` right after an ambiguous disconnect usually means the first + attempt landed. +- **The admission slot is held until the protocol completes**, not until the + connection closes. A disconnecting actor cannot free its own in-flight + budget early by hanging up; slots release when the abandoned write + finishes. +- **Composite requests complete as a unit.** A merge with + `delete_branch: true` runs the merge and the follow-up source-branch + deletion in one server-owned task under one admission slot — a disconnect + cannot land the merge and silently skip the requested deletion. + +Two quarantine classes exist and behave differently: **open-class** +quarantine (a graph whose open failed, possibly transiently) is supervised — +visible in `GET /graphs`, routes answer 503, converges with capped backoff. +**Config-class** quarantine (stored-query parse or embedding-provider config +failures at settings time) is deterministic and unfixable by retry, so it +remains fail-fast: those graphs are skipped (or abort the boot when every +graph is config-broken) and need `cluster apply` + restart. + ## Body limits - Default: 1 MB diff --git a/openapi.json b/openapi.json index 543a09b3..480e7dda 100644 --- a/openapi.json +++ b/openapi.json @@ -2153,7 +2153,7 @@ }, "GraphInfo": { "type": "object", - "description": "One entry in the response from `GET /graphs`. Cluster operators\nconsume this list to discover which graphs the server is currently\nserving. The shape is intentionally minimal — `graph_id` and `uri`\nare the only fields a routing client needs.", + "description": "One entry in the response from `GET /graphs`. Cluster operators\nconsume this list to discover which graphs the server is currently\nserving — including graphs that are configured but quarantined\n(RFC-030 W3), so an unhealthy graph is visible instead of silently\nabsent.", "required": [ "graph_id", "uri" @@ -2162,6 +2162,21 @@ "graph_id": { "type": "string" }, + "quarantine": { + "oneOf": [ + { + "type": "null" + }, + { + "$ref": "#/components/schemas/GraphQuarantineInfo", + "description": "Present only when `status` is `quarantined`." + } + ] + }, + "status": { + "$ref": "#/components/schemas/GraphStatus", + "description": "Additive (RFC-030): absent in older responses, so deserialization\ndefaults to `serving` for wire compatibility." + }, "uri": { "type": "string" } @@ -2182,6 +2197,48 @@ } } }, + "GraphQuarantineInfo": { + "type": "object", + "description": "Quarantine detail on a [`GraphInfo`] whose `status` is `quarantined`\n(RFC-030 W3). Timestamps are seconds since the Unix epoch.", + "required": [ + "since_unix_secs", + "attempts", + "last_error", + "retry_at_unix_secs" + ], + "properties": { + "attempts": { + "type": "integer", + "format": "int32", + "description": "Open attempts so far (the failed boot open counts as the first).", + "minimum": 0 + }, + "last_error": { + "type": "string", + "description": "The most recent open error, verbatim." + }, + "retry_at_unix_secs": { + "type": "integer", + "format": "int64", + "description": "When the supervision loop will retry, Unix seconds.", + "minimum": 0 + }, + "since_unix_secs": { + "type": "integer", + "format": "int64", + "description": "When the graph entered quarantine (this boot), Unix seconds.", + "minimum": 0 + } + } + }, + "GraphStatus": { + "type": "string", + "description": "Serving state of a configured graph (RFC-030 W3).", + "enum": [ + "serving", + "quarantined" + ] + }, "HealthOutput": { "type": "object", "required": [