diff --git a/AGENTS.md b/AGENTS.md
index 73f177b3..3796c3a1 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -32,7 +32,7 @@ OmniGraph is a typed property-graph engine built as a coordination layer over ma
- **Languages**: a `.pg` schema language and a `.gq` query language, both Pest-based, with a typed IR.
- **Multi-modal querying**: vector ANN (`nearest`), full-text (`search`/`fuzzy`/`match_text`/`bm25`), Reciprocal Rank Fusion (`rrf`), and graph traversal (`Expand`, anti-join `not { … }`) in one runtime.
- **Branches and commits across the whole graph**: Git-style — every successful publish appends to a commit DAG; merges are three-way at the row level.
-- **Atomic per-query writes**: `mutate_as` and `load` accumulate insert/update batches into an in-memory `MutationStaging.pending` per touched table; one `stage_*` + `commit_staged` per table runs at end-of-query, then `ManifestBatchPublisher::publish` commits the manifest atomically with per-table `expected_table_versions` CAS. A mid-query failure leaves Lance HEAD untouched on staged tables — no drift, no run state machine, no staging branches. Deletes stage through the same path (MR-A: `stage_delete` via Lance 7.0 `DeleteBuilder::execute_uncommitted`), so they no longer advance Lance HEAD inline. D₂ at parse time is a deliberate boundary — one mutation query is constructive (insert/update) XOR destructive (delete) — so read-your-writes within a query stays unambiguous and each table commits at most one version; compose mixed operations via separate mutations, or a branch for single-commit atomicity.
+- **Atomic per-query writes**: `mutate_as` and `load` accumulate insert/update batches into an in-memory `MutationStaging.pending` per touched table. 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. It arms a schema-v3 sidecar containing exact Lance transaction identities + pre-minted lineage, commits each table with zero transparent conflict retries, confirms the achieved effects, then publishes under the same token. Retryable pre-effect authority conflicts fully reprepare; strict conflicts return `ReadSetChanged`; an unresolved intent or any post-effect publish error returns `RecoveryRequired`. Deletes stage through the same path. D₂ at parse time remains the constructive (insert/update) XOR destructive (delete) boundary.
- **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.
- **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).
@@ -53,7 +53,7 @@ CLI (omnigraph) HTTP Server (omnigraph-server, Axum)
omnigraph (engine) ── ManifestCoordinator, CommitGraph, RunRegistry, GraphIndex (CSR/CSC), exec
│
▼
- Lance 7.x ── columnar Arrow, fragments, per-dataset versions/branches, indexes
+ Lance 9.x ── columnar Arrow, fragments, per-dataset versions/branches, indexes
│
▼
Object store (file / s3 / RustFS / MinIO / S3-compat)
@@ -253,11 +253,11 @@ omnigraph policy explain --cluster ./company-brain --graph knowledge --actor act
| Columnar storage on object store | ✅ Arrow/Lance | URI normalization, S3 env-var plumbing |
| Per-dataset versioning + time travel | ✅ | `snapshot_at_version`, `entity_at`, snapshot-pinned reads across many tables |
| Per-dataset branches | ✅ | **Graph-level** branches (atomic across all sub-tables), lazy fork, system branch filtering |
-| Atomic single-dataset commits | ✅ | **Multi-table publish via three layers**, NOT a single Lance primitive: (1) per-table Lance `commit_staged` for the data write, (2) `__manifest` row-level CAS via `ManifestBatchPublisher` for cross-table ordering, (3) the open-time recovery sweep for the residual gap between (1) and (2). All three layers ship; the five migrated writers (`MutationStaging::finalize`, `schema_apply`, `branch_merge`, `ensure_indices`, `optimize_all_tables`) write a `__recovery/{ulid}.json` sidecar before Phase B and delete it after Phase C. The next `Omnigraph::open` (gated on `OpenMode::ReadWrite`) runs the sweep in `db/manifest/recovery.rs`: classify, decide all-or-nothing per sidecar, roll forward via single `ManifestBatchPublisher::publish` or roll back via `Dataset::restore` followed by a manifest publish of the restored version (so both directions converge to `manifest == HEAD` — no residual drift), and record an audit row in `_graph_commit_recoveries.lance` (queryable via `omnigraph commit list --filter actor=omnigraph:recovery`). The write entry points (`load_as`, `mutate_as`, `apply_schema_as`, `branch_merge_as`) and `refresh` additionally run an in-process roll-forward-only heal (serialized against live writers via the per-table write queues), so a long-lived server converges on its next write without restart; only rollback-eligible sidecars still defer to the next read-write open (a future background reconciler's goal). Engine writes route through a sealed `TableStorage` trait (`db.storage()`) exposing only `stage_*` + `commit_staged` + reads; the sole inline-commit residual (`create_vector_index`) is split onto a separate sealed `InlineCommitResidual` trait reached via `db.storage_inline_residual()` (MR-854), so the default surface cannot couple a write with a HEAD advance — §1 holds by construction. `delete` migrated to the staged path in MR-A (`stage_delete` via Lance 7.0 `DeleteBuilder::execute_uncommitted`, [#6658](https://github.com/lance-format/lance/issues/6658)); `create_vector_index` stays inline until upstream Lance ships a public two-phase API ([#6666](https://github.com/lance-format/lance/issues/6666)); `LoadMode::Overwrite` uses Lance `Overwrite` staged transactions. |
+| Atomic single-dataset commits | ✅ | **Multi-table publish via three layers**, NOT a single Lance primitive: (1) per-table Lance `commit_staged` for the data write, (2) `__manifest` row-level CAS via `ManifestBatchPublisher` for cross-table ordering, (3) the open-time recovery sweep for the residual gap between (1) and (2). All three layers ship; the five migrated writers (`MutationStaging::commit_all`, `schema_apply`, `branch_merge`, `ensure_indices`, `optimize_all_tables`) write a `__recovery/{ulid}.json` sidecar before Phase B and delete it after Phase C. The next `Omnigraph::open` (gated on `OpenMode::ReadWrite`) runs the sweep in `db/manifest/recovery.rs`: classify, decide all-or-nothing per sidecar, roll forward via single `ManifestBatchPublisher::publish` or roll back via `Dataset::restore` followed by a manifest publish of the restored version (so both directions converge to `manifest == HEAD` — no residual drift), and record an internal audit row in `_graph_commit_recoveries.lance`. A v3 roll-forward preserves the interrupted writer's fixed commit lineage and actor; rollback and legacy recovery commits use `omnigraph:recovery`. There is currently no public CLI query for the recovery-audit table, and ordinary commit history is not a complete recovery enumeration. The write entry points (`load_as`, `mutate_as`, `apply_schema_as`, `branch_merge_as`) and `refresh` additionally run an in-process roll-forward-only heal, serialized against same-process live writers through the shared root-scoped gate manager, so a long-lived server converges on its next write without restart; only rollback-eligible sidecars still defer to the next read-write open (a future background reconciler's goal). Engine writes route through a sealed `TableStorage` trait (`db.storage()`) exposing only `stage_*` + `commit_staged` + reads; the sole inline-commit residual (`create_vector_index`) is split onto a separate sealed `InlineCommitResidual` trait reached via `db.storage_inline_residual()` (MR-854), so the default surface cannot couple a write with a HEAD advance — §1 holds by construction. `delete` migrated to the staged path in MR-A (`stage_delete` via Lance 7.0 `DeleteBuilder::execute_uncommitted`, [#6658](https://github.com/lance-format/lance/issues/6658)); `create_vector_index` stays inline until upstream Lance ships a public two-phase API ([#6666](https://github.com/lance-format/lance/issues/6666)); `LoadMode::Overwrite` uses Lance `Overwrite` staged transactions. |
| Compaction (`compact_files`) + reindex (`optimize_indices`) | ✅ | `omnigraph optimize` orchestrates over all node/edge tables, bounded concurrency; per table runs `compact_files` **then Lance `optimize_indices`** (folds appended/rewritten fragments back into existing indexes — incremental merge, not retrain) and **publishes the resulting version to `__manifest`** (so the manifest tracks the Lance HEAD — required for reads to observe the work and for schema apply / strict writes to pass their HEAD-vs-manifest precondition), under the per-`(table, main)` write queue with `SidecarKind::Optimize` recovery coverage spanning both ops; **commits even with no compaction work if index coverage is stale**; **refuses on an unrecovered graph**; **skips uncovered HEAD > manifest drift** with `DriftNeedsRepair`; **compacts blob-bearing tables** (the pre-9 `LANCE_SUPPORTS_BLOB_COMPACTION` skip was removed once Lance 8.0.0+ shipped blob-v2 compaction — see [docs/dev/invariants.md](docs/dev/invariants.md) Known Gaps) |
| Repair uncovered drift | — | `omnigraph repair` explicitly classifies uncovered table `HEAD > manifest` drift: verified maintenance drift (`ReserveFragments`/`Rewrite`) can be published with `--confirm`; suspicious or unverifiable drift requires `--force --confirm`. Sidecar-covered crash residuals still recover automatically on open. |
| Cleanup (`cleanup_old_versions`) | ✅ | `omnigraph cleanup` with `--keep` / `--older-than` policy |
-| BTREE / inverted (FTS) / vector indexes | ✅ | `@index`/`@key` declares intent; the physical index is derived state that never fails a logical op. Built per column through one chokepoint (`build_indices_on_dataset_for_catalog`, type-dispatched by `node_prop_index_kind`: enum + orderable scalar → BTREE, free-text String → FTS, Vector → vector); idempotent; lazy across branches. **Schema apply builds nothing** (records intent only); `load`/`mutate` build inline but **defer an untrainable Vector column** (no trainable vectors yet) as *pending* rather than aborting. `ensure_indices`/`optimize` is the reconciler that materializes declared-but-missing indexes and restores coverage of appended/rewritten fragments (`optimize_indices`), reporting still-pending columns (see Compaction row). |
+| BTREE / inverted (FTS) / vector indexes | ✅ | `@index`/`@key` declares intent; the physical index is derived state that never fails a logical op. Built per column through one chokepoint (`build_indices_on_dataset_for_catalog`, type-dispatched by `node_prop_index_kind`: enum + orderable scalar → BTREE, free-text String → FTS, Vector → vector); idempotent; lazy across branches. **Schema apply and mutation/load build no indexes inline**: the latter publish only their exact data effects, leaving physical intent pending. `ensure_indices`/`optimize` materializes declared-but-missing indexes, restores fragment coverage, and continues to report untrainable Vector columns as pending. |
| `merge_insert` upsert | ✅ | `LoadMode::Merge`, mutation `update`/`insert`/`delete` lowering |
| Vector search | ✅ | `nearest()` query op; embedding pipeline (Gemini / OpenAI clients); `@embed` in schema |
| Full-text search | ✅ | `search/fuzzy/match_text/bm25` query ops |
@@ -267,7 +267,7 @@ omnigraph policy explain --cluster ./company-brain --graph knowledge --actor act
| Query language | — | `.gq` + Pest grammar + IR + lowering + linter |
| Schema migration planning | — | `plan_schema_migration` + `apply_schema` step types + `__schema_apply_lock__` |
| Commit graph (DAG) across whole graph | — | Lineage (linear + merge parents, ULID ids, actor) stored as `graph_commit`/`graph_head` rows in `__manifest`, written in the same publish CAS as the table-version rows (RFC-013 Phase 7 — atomic with the graph commit). The in-memory commit graph is a pure projection of those rows; the legacy `_graph_commits.lance` / `_graph_commit_actors.lance` tables are **retired** (a fresh graph creates neither) |
-| Per-query atomic writes | — | In-memory `MutationStaging.pending` accumulator + `stage_*` / `commit_staged` per touched table at end-of-query + publisher CAS via `commit_with_expected` (single manifest commit per `mutate_as` / `load`); D₂ parse-time rule keeps inserts/updates and deletes from mixing |
+| Per-query atomic writes | — | In-memory `MutationStaging.pending` accumulator + one exact staged Lance transaction per touched table + schema-v3 recovery confirmation + exact branch-head publisher CAS (single manifest commit per `mutate_as` / `load`); D₂ parse-time rule keeps inserts/updates and deletes from mixing |
| 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. |
diff --git a/crates/omnigraph-api-types/src/lib.rs b/crates/omnigraph-api-types/src/lib.rs
index cff10f2a..feb4a777 100644
--- a/crates/omnigraph-api-types/src/lib.rs
+++ b/crates/omnigraph-api-types/src/lib.rs
@@ -577,6 +577,21 @@ pub struct ManifestConflictOutput {
pub actual: u64,
}
+/// Structured authority mismatch for an RFC-022 prepared write. Values are
+/// strings because members include optional graph commit ids and future
+/// authority tokens, not only numeric table versions.
+#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
+pub struct ReadSetConflictOutput {
+ pub member: String,
+ pub expected: Option,
+ pub actual: Option,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
+pub struct RecoveryRequiredOutput {
+ pub operation_id: String,
+}
+
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct ErrorOutput {
pub error: String,
@@ -590,6 +605,13 @@ pub struct ErrorOutput {
/// manifest is now at `actual`. Refresh and retry.
#[serde(skip_serializing_if = "Option::is_none")]
pub manifest_conflict: Option,
+ /// Set when a prepared write's logical authority changed before effects.
+ #[serde(skip_serializing_if = "Option::is_none")]
+ pub read_set_conflict: Option,
+ /// Set when an overlapping durable recovery intent must be resolved before
+ /// retry. Its table effects may or may not have started.
+ #[serde(skip_serializing_if = "Option::is_none")]
+ pub recovery_required: Option,
}
pub fn snapshot_payload(
diff --git a/crates/omnigraph-server/src/handlers.rs b/crates/omnigraph-server/src/handlers.rs
index 1d0a5360..c226a029 100644
--- a/crates/omnigraph-server/src/handlers.rs
+++ b/crates/omnigraph-server/src/handlers.rs
@@ -788,8 +788,9 @@ pub(crate) async fn run_query(
(status = 400, description = "Bad request", body = ErrorOutput),
(status = 401, description = "Unauthorized", body = ErrorOutput),
(status = 403, description = "Forbidden", body = ErrorOutput),
- (status = 409, description = "Merge conflict", body = ErrorOutput),
+ (status = 409, description = "Write-authority conflict", body = ErrorOutput),
(status = 429, description = "Per-actor admission cap exceeded; honor `Retry-After` header", body = ErrorOutput),
+ (status = 503, description = "An overlapping durable recovery intent must be resolved before retry", body = ErrorOutput),
),
security(("bearer_token" = [])),
)]
@@ -837,8 +838,9 @@ pub(crate) async fn server_change(
(status = 400, description = "Bad request", body = ErrorOutput),
(status = 401, description = "Unauthorized", body = ErrorOutput),
(status = 403, description = "Forbidden", body = ErrorOutput),
- (status = 409, description = "Merge conflict", body = ErrorOutput),
+ (status = 409, description = "Write-authority conflict", body = ErrorOutput),
(status = 429, description = "Per-actor admission cap exceeded; honor `Retry-After` header", body = ErrorOutput),
+ (status = 503, description = "An overlapping durable recovery intent must be resolved before retry", body = ErrorOutput),
),
security(("bearer_token" = [])),
)]
@@ -847,7 +849,8 @@ pub(crate) async fn server_change(
/// Writes to the named `branch` (defaults to `main`). Mutations are atomic
/// per call and produce a new commit. Returns counts of nodes and edges
/// affected. **Destructive**: on success the branch is updated; rejected
-/// mutations may still acquire locks briefly. Returns 409 on merge conflict.
+/// mutations may still acquire locks briefly. Returns 409 when the prepared
+/// write authority changes before effects.
///
/// Pairs with `POST /query` (read-only). The legacy `POST /change` route
/// has identical semantics and is kept as a deprecated alias.
@@ -904,9 +907,10 @@ pub(crate) fn parse_optional_invoke_body(
(status = 401, description = "Unauthorized", body = ErrorOutput),
(status = 403, description = "Forbidden (the inner `change` gate for a stored mutation)", body = ErrorOutput),
(status = 404, description = "Unknown stored query, or `invoke_query` denied — indistinguishable to a caller without the grant", body = ErrorOutput),
- (status = 409, description = "Merge conflict", body = ErrorOutput),
+ (status = 409, description = "Stored mutation write-authority conflict", body = ErrorOutput),
(status = 429, description = "Per-actor admission cap exceeded; honor `Retry-After` header", body = ErrorOutput),
(status = 500, description = "Policy evaluation error (a denial is reported as 404, not 500)", body = ErrorOutput),
+ (status = 503, description = "A stored mutation is blocked by a durable recovery intent", body = ErrorOutput),
),
security(("bearer_token" = [])),
)]
@@ -1203,25 +1207,11 @@ pub(crate) async fn server_schema_apply(
.await
.map_err(ApiError::from_omni)?
};
- // Prompt index convergence (iss-848): schema apply records `@index` intent
- // but defers the physical build. On a long-lived server, materialize it
- // promptly rather than waiting for the next `optimize` cron — spawned
- // detached so it never blocks or fails the apply response. Best-effort: a
- // failure is logged and the index still converges on the next optimize.
- // The CLI is one-shot, so it has no equivalent; its convergence path is the
- // operator's optimize cadence.
- if result.applied {
- let engine = Arc::clone(&handle.engine);
- tokio::spawn(async move {
- if let Err(err) = engine.ensure_indices().await {
- tracing::warn!(
- target: "omnigraph::server",
- error = %err,
- "post-apply ensure_indices failed; indexes will converge on the next optimize",
- );
- }
- });
- }
+ // Physical indexes are derived state. Schema apply records intent only;
+ // explicit `ensure_indices` / `optimize` maintenance owns convergence on
+ // every surface, including a long-lived server. Keeping the handler free
+ // of detached physical writes also makes a successful response describe
+ // the complete effect envelope of this request.
Ok(Json(schema_apply_output(handle.uri.as_str(), result)))
}
@@ -1313,7 +1303,9 @@ async fn run_ingest(
(status = 400, description = "Bad request", body = ErrorOutput),
(status = 401, description = "Unauthorized", body = ErrorOutput),
(status = 403, description = "Forbidden", body = ErrorOutput),
+ (status = 409, description = "Prepared load authority changed before effects", body = ErrorOutput),
(status = 429, description = "Per-actor admission cap exceeded; honor `Retry-After` header", body = ErrorOutput),
+ (status = 503, description = "An overlapping durable recovery intent must be resolved before retry", body = ErrorOutput),
),
security(("bearer_token" = [])),
)]
@@ -1357,7 +1349,9 @@ pub(crate) async fn server_load(
(status = 400, description = "Bad request", body = ErrorOutput),
(status = 401, description = "Unauthorized", body = ErrorOutput),
(status = 403, description = "Forbidden", body = ErrorOutput),
+ (status = 409, description = "Prepared load authority changed before effects", body = ErrorOutput),
(status = 429, description = "Per-actor admission cap exceeded; honor `Retry-After` header", body = ErrorOutput),
+ (status = 503, description = "An overlapping durable recovery intent must be resolved before retry", body = ErrorOutput),
),
security(("bearer_token" = [])),
)]
@@ -1437,6 +1431,7 @@ pub(crate) async fn server_branch_list(
(status = 403, description = "Forbidden", body = ErrorOutput),
(status = 409, description = "Branch already exists", body = ErrorOutput),
(status = 429, description = "Per-actor admission cap exceeded; honor `Retry-After` header", body = ErrorOutput),
+ (status = 503, description = "An overlapping durable recovery intent must be resolved before retry", body = ErrorOutput),
),
security(("bearer_token" = [])),
)]
@@ -1518,6 +1513,7 @@ pub(crate) struct BranchPath {
(status = 403, description = "Forbidden", body = ErrorOutput),
(status = 404, description = "Branch not found", body = ErrorOutput),
(status = 429, description = "Per-actor admission cap exceeded; honor `Retry-After` header", body = ErrorOutput),
+ (status = 503, description = "An overlapping durable recovery intent must be resolved before retry", body = ErrorOutput),
),
security(("bearer_token" = [])),
)]
@@ -1579,6 +1575,7 @@ pub(crate) async fn server_branch_delete(
(status = 403, description = "Forbidden", body = ErrorOutput),
(status = 409, description = "Merge conflict", body = ErrorOutput),
(status = 429, description = "Per-actor admission cap exceeded; honor `Retry-After` header", body = ErrorOutput),
+ (status = 503, description = "An overlapping durable recovery intent must be resolved before retry", body = ErrorOutput),
),
security(("bearer_token" = [])),
)]
diff --git a/crates/omnigraph-server/src/lib.rs b/crates/omnigraph-server/src/lib.rs
index d60a0dc4..7a91abcd 100644
--- a/crates/omnigraph-server/src/lib.rs
+++ b/crates/omnigraph-server/src/lib.rs
@@ -286,10 +286,12 @@ impl Write for ExportStreamWriter {
#[derive(Debug)]
pub struct ApiError {
status: StatusCode,
- code: ErrorCode,
+ code: Option,
message: String,
merge_conflicts: Vec,
manifest_conflict: Option,
+ read_set_conflict: Option,
+ recovery_required: Option,
}
impl AppState {
@@ -616,40 +618,48 @@ impl ApiError {
pub fn unauthorized(message: impl Into) -> Self {
Self {
status: StatusCode::UNAUTHORIZED,
- code: ErrorCode::Unauthorized,
+ code: Some(ErrorCode::Unauthorized),
message: message.into(),
merge_conflicts: Vec::new(),
manifest_conflict: None,
+ read_set_conflict: None,
+ recovery_required: None,
}
}
pub fn forbidden(message: impl Into) -> Self {
Self {
status: StatusCode::FORBIDDEN,
- code: ErrorCode::Forbidden,
+ code: Some(ErrorCode::Forbidden),
message: message.into(),
merge_conflicts: Vec::new(),
manifest_conflict: None,
+ read_set_conflict: None,
+ recovery_required: None,
}
}
pub fn bad_request(message: impl Into) -> Self {
Self {
status: StatusCode::BAD_REQUEST,
- code: ErrorCode::BadRequest,
+ code: Some(ErrorCode::BadRequest),
message: message.into(),
merge_conflicts: Vec::new(),
manifest_conflict: None,
+ read_set_conflict: None,
+ recovery_required: None,
}
}
pub fn not_found(message: impl Into) -> Self {
Self {
status: StatusCode::NOT_FOUND,
- code: ErrorCode::NotFound,
+ code: Some(ErrorCode::NotFound),
message: message.into(),
merge_conflicts: Vec::new(),
manifest_conflict: None,
+ read_set_conflict: None,
+ recovery_required: None,
}
}
@@ -660,30 +670,36 @@ impl ApiError {
pub fn method_not_allowed(message: impl Into) -> Self {
Self {
status: StatusCode::METHOD_NOT_ALLOWED,
- code: ErrorCode::MethodNotAllowed,
+ code: Some(ErrorCode::MethodNotAllowed),
message: message.into(),
merge_conflicts: Vec::new(),
manifest_conflict: None,
+ read_set_conflict: None,
+ recovery_required: None,
}
}
pub fn conflict(message: impl Into) -> Self {
Self {
status: StatusCode::CONFLICT,
- code: ErrorCode::Conflict,
+ code: Some(ErrorCode::Conflict),
message: message.into(),
merge_conflicts: Vec::new(),
manifest_conflict: None,
+ read_set_conflict: None,
+ recovery_required: None,
}
}
pub fn internal(message: impl Into) -> Self {
Self {
status: StatusCode::INTERNAL_SERVER_ERROR,
- code: ErrorCode::Internal,
+ code: Some(ErrorCode::Internal),
message: message.into(),
merge_conflicts: Vec::new(),
manifest_conflict: None,
+ read_set_conflict: None,
+ recovery_required: None,
}
}
@@ -694,10 +710,12 @@ impl ApiError {
pub fn too_many_requests(message: impl Into) -> Self {
Self {
status: StatusCode::TOO_MANY_REQUESTS,
- code: ErrorCode::TooManyRequests,
+ code: Some(ErrorCode::TooManyRequests),
message: message.into(),
merge_conflicts: Vec::new(),
manifest_conflict: None,
+ read_set_conflict: None,
+ recovery_required: None,
}
}
@@ -715,20 +733,51 @@ impl ApiError {
fn merge_conflict(conflicts: Vec) -> Self {
Self {
status: StatusCode::CONFLICT,
- code: ErrorCode::Conflict,
+ code: Some(ErrorCode::Conflict),
message: summarize_merge_conflicts(&conflicts),
merge_conflicts: conflicts,
manifest_conflict: None,
+ read_set_conflict: None,
+ recovery_required: None,
}
}
fn manifest_version_conflict(message: String, details: api::ManifestConflictOutput) -> Self {
Self {
status: StatusCode::CONFLICT,
- code: ErrorCode::Conflict,
+ code: Some(ErrorCode::Conflict),
message,
merge_conflicts: Vec::new(),
manifest_conflict: Some(details),
+ read_set_conflict: None,
+ recovery_required: None,
+ }
+ }
+
+ fn read_set_conflict(message: String, details: api::ReadSetConflictOutput) -> Self {
+ Self {
+ status: StatusCode::CONFLICT,
+ code: Some(ErrorCode::Conflict),
+ message,
+ merge_conflicts: Vec::new(),
+ manifest_conflict: None,
+ read_set_conflict: Some(details),
+ recovery_required: None,
+ }
+ }
+
+ fn recovery_required(message: String, operation_id: String) -> Self {
+ Self {
+ status: StatusCode::SERVICE_UNAVAILABLE,
+ // `ErrorCode` is a closed rolling wire contract. The additive
+ // `recovery_required` field carries the new meaning while older
+ // clients continue to deserialize the otherwise familiar body.
+ code: None,
+ message,
+ merge_conflicts: Vec::new(),
+ manifest_conflict: None,
+ read_set_conflict: None,
+ recovery_required: Some(api::RecoveryRequiredOutput { operation_id }),
}
}
@@ -752,6 +801,18 @@ impl ApiError {
actual,
},
),
+ Some(ManifestConflictDetails::ReadSetChanged {
+ member,
+ expected,
+ actual,
+ }) => Self::read_set_conflict(
+ err.message,
+ api::ReadSetConflictOutput {
+ member,
+ expected,
+ actual,
+ },
+ ),
_ => Self::conflict(err.message),
},
ManifestErrorKind::Internal => Self::internal(err.message),
@@ -762,6 +823,13 @@ impl ApiError {
.map(api::MergeConflictOutput::from)
.collect(),
),
+ OmniError::RecoveryRequired {
+ operation_id,
+ reason,
+ } => Self::recovery_required(
+ format!("recovery required for operation {operation_id}: {reason}"),
+ operation_id,
+ ),
OmniError::Lance(message) => Self::internal(format!("storage: {message}")),
OmniError::Io(err) => Self::internal(format!("io: {err}")),
// Engine-layer policy enforcement (MR-722). All denials and
@@ -815,7 +883,7 @@ const RETRY_AFTER_SECONDS: &str = "60";
impl IntoResponse for ApiError {
fn into_response(self) -> Response {
let mut headers = axum::http::HeaderMap::new();
- if matches!(self.code, ErrorCode::TooManyRequests) {
+ if matches!(self.code, Some(ErrorCode::TooManyRequests)) {
headers.insert(
axum::http::header::RETRY_AFTER,
axum::http::HeaderValue::from_static(RETRY_AFTER_SECONDS),
@@ -826,15 +894,42 @@ impl IntoResponse for ApiError {
headers,
Json(ErrorOutput {
error: self.message,
- code: Some(self.code),
+ code: self.code,
merge_conflicts: self.merge_conflicts,
manifest_conflict: self.manifest_conflict,
+ read_set_conflict: self.read_set_conflict,
+ recovery_required: self.recovery_required,
}),
)
.into_response()
}
}
+#[cfg(test)]
+mod api_error_tests {
+ use super::*;
+
+ #[tokio::test]
+ async fn recovery_required_503_omits_closed_error_code() {
+ let response = ApiError::from_omni(OmniError::RecoveryRequired {
+ operation_id: "01JTESTRECOVERY".to_string(),
+ reason: "pending recovery intent".to_string(),
+ })
+ .into_response();
+
+ assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE);
+ let body = axum::body::to_bytes(response.into_body(), usize::MAX)
+ .await
+ .unwrap();
+ let error: ErrorOutput = serde_json::from_slice(&body).unwrap();
+ assert_eq!(error.code, None);
+ assert_eq!(
+ error.recovery_required.unwrap().operation_id,
+ "01JTESTRECOVERY"
+ );
+ }
+}
+
pub fn init_tracing() {
let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));
let _ = tracing_subscriber::fmt().with_env_filter(filter).try_init();
diff --git a/crates/omnigraph-server/tests/data_routes.rs b/crates/omnigraph-server/tests/data_routes.rs
index 3e533288..3db0ed78 100644
--- a/crates/omnigraph-server/tests/data_routes.rs
+++ b/crates/omnigraph-server/tests/data_routes.rs
@@ -1099,18 +1099,15 @@ query vector_search_string($q: String) {
}
#[tokio::test(flavor = "multi_thread")]
-async fn change_conflict_returns_manifest_conflict_409() {
- // A write that races with another writer surfaces as HTTP 409 with
- // a structured `manifest_conflict` body — `table_key`, `expected`,
- // and `actual` — so clients can detect-and-retry without parsing
- // the message.
+async fn change_long_lived_handle_refreshes_before_preparing_write() {
+ // A handle that merely predates another committed write is not stale
+ // authority: open_write_txn probes the manifest incarnation and prepares
+ // from the fresh head. ReadSetChanged is reserved for movement *during* an
+ // already-prepared attempt (covered by the concurrent test below).
let temp = init_loaded_graph().await;
let graph = graph_path(temp.path());
- // Build the server first so its handle pins the pre-mutation manifest
- // version. Then advance the manifest from outside the server. The
- // server's next /change call will capture stale `expected_versions`
- // (from its still-pinned snapshot) and the publisher's CAS rejects.
+ // Build the server first, then advance the graph through another handle.
let state = AppState::open(graph.to_string_lossy().to_string())
.await
.unwrap();
@@ -1154,34 +1151,17 @@ async fn change_conflict_returns_manifest_conflict_409() {
)
.await;
- assert_eq!(status, StatusCode::CONFLICT);
- let error: ErrorOutput = serde_json::from_value(body).unwrap();
- assert_eq!(error.code, Some(omnigraph_server::api::ErrorCode::Conflict));
- let conflict = error
- .manifest_conflict
- .expect("publisher CAS rejection must populate manifest_conflict body");
- assert_eq!(conflict.table_key, "node:Person");
- assert!(
- conflict.actual > conflict.expected,
- "actual ({}) should be ahead of expected ({})",
- conflict.actual,
- conflict.expected,
- );
+ assert_eq!(status, StatusCode::OK);
+ assert_eq!(body["affected_nodes"], 1);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn change_concurrent_inserts_same_key_serialize_without_409() {
- // PR 2 Phase 2 (MR-686): pin the design fix for the same-key
- // concurrency hazard. Pre-fix, in-process concurrent inserts on
- // the same `(table, branch)` rejected with 409 manifest_conflict
- // because `ensure_expected_version` fired before the per-table
- // queue was acquired and saw Lance HEAD already advanced by a
- // peer writer. Post-fix, Insert/Merge skip the strict pre-stage
- // check (see `MutationOpKind::strict_pre_stage_version_check`);
- // the queue serializes commit_staged; Lance's natural rebase
- // handles the in-flight stage; the publisher's CAS on a fresh
- // per-branch snapshot under the queue catches genuine cross-
- // process drift.
+ // RFC-022 preservation guard: concurrent retryable inserts still all
+ // succeed, but not by rebasing an already-validated Lance transaction.
+ // The coarse branch gate serializes effects; a waiter whose authority
+ // token changed discards its complete attempt and reprepares from the
+ // winner's committed branch state.
//
// This test spawns N concurrent /change inserts on a single
// node type and asserts: every request returns 200 (no 409),
@@ -1270,36 +1250,10 @@ async fn change_concurrent_inserts_same_key_serialize_without_409() {
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
-async fn change_concurrent_updates_same_key_serialize_via_publisher_cas() {
- // Pin Update RYW semantics under in-process concurrency on the same
- // `(table, branch)`. With per-table queue serialization and op-kind-aware
- // drift detection at commit time, exactly one of N concurrent UPDATEs
- // on the same row commits; the rest are rejected as 409 manifest_conflict.
- //
- // Pre-fix bug class: in `MutationStaging::commit_all`, after queue
- // acquisition, the staged Lance transaction is handed straight to
- // `commit_staged`. For a writer whose staged dataset is at V0 but
- // Lance HEAD has advanced to V1 (because the queue's prior winner
- // already published), Lance's transaction conflict resolver fires
- // `RetryableCommitConflict` on Update vs Update on the same row.
- // That error gets wrapped as `OmniError::Lance()` and the
- // API surfaces it as **500 internal**, not 409. Users see "internal
- // server error" instead of a retryable conflict, breaking the
- // documented 409 contract for in-process drift.
- //
- // Post-fix invariant: `commit_all` does an op-kind-aware drift check
- // before each `commit_staged`. For tables whose tracked op_kind has
- // `strict_pre_stage_version_check() == true` (Update / Delete /
- // SchemaRewrite), if the staged dataset's version doesn't match the
- // fresh manifest pin, return `OmniError::manifest_expected_version_mismatch`
- // → 409 ExpectedVersionMismatch. The N-1 losers see a clean 409
- // before Lance's commit_staged ever runs.
- //
- // Why correct-by-design: closing the class "Lance internal conflict
- // surfaces as 500 instead of 409" rather than mapping the specific
- // Lance error variant. The drift check fires at the right architectural
- // layer (engine boundary, under the queue) and respects the existing
- // `MutationOpKind` policy.
+async fn change_concurrent_updates_same_key_return_typed_pre_effect_conflicts() {
+ // Strict read-modify-write attempts are never automatically reprepared.
+ // Exactly one concurrent UPDATE commits; once it changes branch authority,
+ // every waiter reports a typed 409 before any of its Lance effects begin.
let temp = init_loaded_graph().await;
let graph = graph_path(temp.path());
let state = AppState::open(graph.to_string_lossy().to_string())
@@ -1373,27 +1327,37 @@ async fn change_concurrent_updates_same_key_serialize_via_publisher_cas() {
assert_eq!(
ok_count,
1,
- "expected exactly one update to commit and N-1 to receive 409 manifest_conflict \
- (op-kind-aware drift check rejects stale-V0 staged datasets at commit_all entry). \
- Got {} OK + {} 409 + {} other. \
- Pre-fix symptom: 1 OK + (N-1) x 500 because Lance's RetryableCommitConflict for \
- Update vs Update on the same row bubbles up as `OmniError::Lance()` and \
- the API maps it to 500 internal, not 409. Statuses: {:?}",
+ "expected exactly one update to commit and N-1 to receive typed 409 conflicts \
+ before effects. Got {} OK + {} 409 + {} other. Statuses: {:?}",
ok_count,
conflict_count,
statuses.len() - ok_count - conflict_count,
statuses,
);
+
+ for (status, bytes) in &results {
+ if *status != StatusCode::CONFLICT {
+ continue;
+ }
+ let error: ErrorOutput = serde_json::from_slice(bytes).unwrap();
+ assert_eq!(error.code, Some(omnigraph_server::api::ErrorCode::Conflict));
+ let conflict = error
+ .read_set_conflict
+ .expect("strict OCC loser must include structured read-set authority");
+ assert_eq!(conflict.member, "graph_head:main");
+ assert_ne!(conflict.actual, conflict.expected);
+ assert!(error.manifest_conflict.is_none());
+ assert!(error.recovery_required.is_none());
+ }
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
-async fn change_disjoint_table_concurrency_succeeds_at_http_level() {
- // HTTP-level pin for MR-686's disjoint-table promise: concurrent /change
- // requests touching different node types must coexist without admission
- // rejection or publisher-CAS conflict. The bench harness measures
- // throughput; this test is the regression sentinel that catches a
- // future change which accidentally re-introduces graph-wide
- // serialization on the disjoint path.
+async fn change_disjoint_table_concurrency_succeeds_under_branch_occ_gate() {
+ // RFC-022 intentionally serializes effect publication per branch because
+ // graph-head authority protects validation dependencies across tables.
+ // Disjoint retryable inserts must nevertheless all succeed through bounded
+ // full-attempt repreparation, without admission rejection or a user-visible
+ // publisher conflict.
//
// Setup: test.jsonl seeds 4 Persons + 2 Companies. Spawn N=4 concurrent
// /change inserts on `node:Person` and N=4 concurrent inserts on
diff --git a/crates/omnigraph-server/tests/openapi.rs b/crates/omnigraph-server/tests/openapi.rs
index 2341bbc2..060869b3 100644
--- a/crates/omnigraph-server/tests/openapi.rs
+++ b/crates/omnigraph-server/tests/openapi.rs
@@ -391,7 +391,9 @@ const EXPECTED_SCHEMAS: &[&str] = &[
"MergeConflictOutput",
"ReadOutput",
"ReadRequest",
+ "ReadSetConflictOutput",
"ReadTargetOutput",
+ "RecoveryRequiredOutput",
"ManifestConflictOutput",
"SchemaApplyOutput",
"SchemaApplyRequest",
@@ -614,6 +616,8 @@ fn error_output_schema_has_expected_fields() {
assert!(props.contains_key("code"));
assert!(props.contains_key("merge_conflicts"));
assert!(props.contains_key("manifest_conflict"));
+ assert!(props.contains_key("read_set_conflict"));
+ assert!(props.contains_key("recovery_required"));
}
#[test]
@@ -626,6 +630,24 @@ fn manifest_conflict_output_schema_has_expected_fields() {
assert!(props.contains_key("actual"));
}
+#[test]
+fn read_set_conflict_output_schema_has_expected_fields() {
+ let doc = openapi_json();
+ let schema = &doc["components"]["schemas"]["ReadSetConflictOutput"];
+ let props = schema["properties"].as_object().unwrap();
+ assert!(props.contains_key("member"));
+ assert!(props.contains_key("expected"));
+ assert!(props.contains_key("actual"));
+}
+
+#[test]
+fn recovery_required_output_schema_has_expected_fields() {
+ let doc = openapi_json();
+ let schema = &doc["components"]["schemas"]["RecoveryRequiredOutput"];
+ let props = schema["properties"].as_object().unwrap();
+ assert!(props.contains_key("operation_id"));
+}
+
#[test]
fn commit_output_schema_has_expected_fields() {
let doc = openapi_json();
@@ -694,12 +716,21 @@ fn error_code_schema_has_expected_variants() {
let schema = &doc["components"]["schemas"]["ErrorCode"];
let variants = schema["enum"].as_array().unwrap();
let values: HashSet<&str> = variants.iter().map(|v| v.as_str().unwrap()).collect();
- assert!(values.contains("unauthorized"));
- assert!(values.contains("forbidden"));
- assert!(values.contains("bad_request"));
- assert!(values.contains("not_found"));
- assert!(values.contains("conflict"));
- assert!(values.contains("internal"));
+ assert_eq!(
+ values,
+ HashSet::from([
+ "unauthorized",
+ "forbidden",
+ "bad_request",
+ "not_found",
+ "method_not_allowed",
+ "conflict",
+ "too_many_requests",
+ "internal",
+ ]),
+ "ErrorCode is a rolling wire contract: new meanings belong in optional \
+ structured fields, not new closed-enum values",
+ );
}
#[test]
@@ -919,6 +950,32 @@ fn error_responses_reference_error_output_schema() {
}
}
+#[test]
+fn recovery_barrier_write_endpoints_document_recovery_required() {
+ let doc = openapi_json();
+ for (path, method) in [
+ ("/graphs/{graph_id}/change", "post"),
+ ("/graphs/{graph_id}/mutate", "post"),
+ ("/graphs/{graph_id}/queries/{name}", "post"),
+ ("/graphs/{graph_id}/load", "post"),
+ ("/graphs/{graph_id}/ingest", "post"),
+ ("/graphs/{graph_id}/branches", "post"),
+ ("/graphs/{graph_id}/branches/{branch}", "delete"),
+ ("/graphs/{graph_id}/branches/merge", "post"),
+ ] {
+ let response = &doc["paths"][path][method]["responses"]["503"];
+ assert!(
+ response.is_object(),
+ "{method} {path} must document the recovery-required 503 outcome"
+ );
+ assert_eq!(
+ response["content"]["application/json"]["schema"]["$ref"],
+ "#/components/schemas/ErrorOutput",
+ "{method} {path} 503 must use ErrorOutput"
+ );
+ }
+}
+
// ---------------------------------------------------------------------------
// Request body reference tests
// ---------------------------------------------------------------------------
diff --git a/crates/omnigraph-server/tests/schema_routes.rs b/crates/omnigraph-server/tests/schema_routes.rs
index c73591ca..9acc9018 100644
--- a/crates/omnigraph-server/tests/schema_routes.rs
+++ b/crates/omnigraph-server/tests/schema_routes.rs
@@ -371,8 +371,7 @@ async fn schema_apply_route_can_rename_property() {
#[tokio::test]
async fn schema_apply_route_can_add_index() {
- let (temp, app) = app_for_graph_with_auth_tokens_and_policy(
- &fs::read_to_string(fixture("test.pg")).unwrap(),
+ let (temp, app) = app_for_loaded_graph_with_auth_tokens_and_policy(
&[("act-ragnor", "admin-token")],
SCHEMA_APPLY_POLICY_YAML,
)
@@ -392,7 +391,9 @@ async fn schema_apply_route_can_add_index() {
.header("authorization", "Bearer admin-token")
.body(Body::from(
serde_json::to_vec(&SchemaApplyRequest {
- schema_source: indexed_name_schema(),
+ schema_source: fs::read_to_string(fixture("test.pg"))
+ .unwrap()
+ .replace("age: I32?", "age: I32? @index"),
..Default::default()
})
.unwrap(),
@@ -404,9 +405,24 @@ async fn schema_apply_route_can_add_index() {
assert_eq!(payload["applied"], true);
// iss-848: the /schema/apply route accepts the index-add and applies it as a
// metadata change — it records the `@index` intent in the catalog/IR but does
- // NOT build the physical index inline (the build is deferred to
- // ensure_indices/optimize; on this empty table nothing would build anyway).
- // So the physical index count is unchanged by the apply.
+ // NOT build the physical index inline or spawn a server-only build (the
+ // build is deferred to explicit ensure_indices/optimize maintenance).
+ // Pin the detached-work half structurally instead of waiting an arbitrary
+ // amount of wall-clock time for a task that must not exist. The physical
+ // assertion below independently covers synchronous/awaited index builds.
+ let handlers = include_str!("../src/handlers.rs");
+ let (_, after_apply_signature) = handlers
+ .split_once("pub(crate) async fn server_schema_apply")
+ .expect("server_schema_apply handler must remain present");
+ let (apply_handler, _) = after_apply_signature
+ .split_once("/// Shared body for `POST /load`")
+ .expect("schema apply handler must precede the load handler");
+ assert!(
+ !apply_handler.contains("tokio::spawn") && !apply_handler.contains(".ensure_indices("),
+ "schema apply must not schedule implicit physical index convergence"
+ );
+
+ // The physical index count is unchanged by the apply.
let reopened = Omnigraph::open(graph.to_str().unwrap()).await.unwrap();
let snapshot = reopened
.snapshot_of(ReadTarget::branch("main"))
diff --git a/crates/omnigraph-server/tests/support/mod.rs b/crates/omnigraph-server/tests/support/mod.rs
index 694db467..6a7c0f82 100644
--- a/crates/omnigraph-server/tests/support/mod.rs
+++ b/crates/omnigraph-server/tests/support/mod.rs
@@ -502,12 +502,6 @@ pub fn renamed_age_schema() -> String {
.replace("age: I32?", "years: I32? @rename_from(\"age\")")
}
-pub fn indexed_name_schema() -> String {
- fs::read_to_string(fixture("test.pg"))
- .unwrap()
- .replace("name: String @key", "name: String @key @index")
-}
-
pub fn unsupported_schema_change() -> String {
fs::read_to_string(fixture("test.pg"))
.unwrap()
diff --git a/crates/omnigraph/src/db/commit_graph.rs b/crates/omnigraph/src/db/commit_graph.rs
index ad6831ad..8a6814f1 100644
--- a/crates/omnigraph/src/db/commit_graph.rs
+++ b/crates/omnigraph/src/db/commit_graph.rs
@@ -49,6 +49,11 @@ impl CommitGraph {
/// keeps the cache consistent for same-handle reads, with no storage I/O.
/// Head selection matches the manifest-sourced load (`should_replace_head`).
pub fn insert_committed(&mut self, commit: GraphCommit) {
+ debug_assert_eq!(
+ commit.manifest_branch.as_deref(),
+ self.active_branch.as_deref(),
+ "published lineage must target the commit graph's active branch"
+ );
if should_replace_head(self.head_commit.as_ref(), &commit) {
self.head_commit = Some(commit.clone());
}
@@ -71,7 +76,8 @@ impl CommitGraph {
let root = root_uri.trim_end_matches('/');
// `load_commit_cache_for_branch` opens the branch's `__manifest` (the
// authoritative table), so a truly absent branch fails loudly here.
- let (commit_by_id, head_commit) = load_commit_cache_for_branch(root, Some(branch)).await?;
+ let (commit_by_id, head_commit) =
+ load_commit_cache_for_branch(root, Some(branch)).await?;
Ok(Self {
root_uri: root.to_string(),
active_branch: Some(branch.to_string()),
@@ -180,7 +186,7 @@ async fn load_commit_cache_from_manifest(
root_uri: &str,
branch: Option<&str>,
) -> Result<(HashMap, Option)> {
- let (rows, _heads) =
+ let (rows, _) =
crate::db::manifest::ManifestCoordinator::read_graph_lineage_at(root_uri, branch).await?;
let mut commit_by_id = HashMap::with_capacity(rows.len());
let mut head_commit = None;
diff --git a/crates/omnigraph/src/db/graph_coordinator.rs b/crates/omnigraph/src/db/graph_coordinator.rs
index 0b016f4d..ca9b9d3f 100644
--- a/crates/omnigraph/src/db/graph_coordinator.rs
+++ b/crates/omnigraph/src/db/graph_coordinator.rs
@@ -11,7 +11,8 @@ use crate::storage::{StorageAdapter, normalize_root_uri};
use super::commit_graph::{CommitGraph, GraphCommit};
use super::is_internal_system_branch;
use super::manifest::{
- ManifestChange, ManifestCoordinator, ManifestIncarnation, Snapshot, SubTableUpdate,
+ LineageIntent, ManifestChange, ManifestCoordinator, ManifestIncarnation, PublishPrecondition,
+ Snapshot, SubTableUpdate,
};
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
@@ -167,6 +168,20 @@ impl GraphCoordinator {
self.manifest.incarnation()
}
+ /// Lance-native identity of the active `__manifest` branch. Stable across
+ /// commits; changes when a named branch is deleted and recreated.
+ pub(crate) async fn branch_identifier(&self) -> Result {
+ self.manifest.branch_identifier().await
+ }
+
+ /// Exact `graph_head:` pointer, preserving `None` for a
+ /// freshly-created named branch even though its inherited commit history has
+ /// an inferred head. Sourced from the manifest coordinator's SAME pinned
+ /// state as [`Self::snapshot`], not the separately refreshed lineage cache.
+ pub(crate) fn exact_graph_head(&self) -> Option {
+ self.manifest.exact_graph_head()
+ }
+
pub fn snapshot(&self) -> Snapshot {
self.manifest.snapshot()
}
@@ -393,10 +408,35 @@ impl GraphCoordinator {
actor_id: Option<&str>,
) -> Result {
let intent = self.new_lineage_intent(actor_id, None)?;
+ self.commit_changes_with_intent_and_expected(
+ changes,
+ expected_table_versions,
+ intent,
+ &PublishPrecondition::Any,
+ )
+ .await
+ }
+
+ /// Publish a pre-minted lineage intent under an explicit authority
+ /// precondition. The intent's identity and timestamp remain stable across
+ /// publisher retries and can also be persisted by the caller's recovery
+ /// protocol before this method is invoked.
+ pub(crate) async fn commit_changes_with_intent_and_expected(
+ &mut self,
+ changes: &[ManifestChange],
+ expected_table_versions: &HashMap,
+ intent: LineageIntent,
+ precondition: &PublishPrecondition,
+ ) -> Result {
failpoints::maybe_fail(crate::failpoints::names::GRAPH_PUBLISH_BEFORE_COMMIT_APPEND)?;
let outcome = self
.manifest
- .commit_changes_with_lineage(changes, expected_table_versions, Some(&intent))
+ .commit_changes_with_lineage_and_precondition(
+ changes,
+ expected_table_versions,
+ Some(&intent),
+ precondition,
+ )
.await?;
failpoints::maybe_fail(crate::failpoints::names::GRAPH_PUBLISH_AFTER_MANIFEST_COMMIT)?;
let snapshot_id = self.apply_lineage_to_cache(intent, &outcome);
@@ -434,12 +474,12 @@ impl GraphCoordinator {
/// fresh ULID (stable across the publisher's CAS retries) and a timestamp.
/// The parent is NOT chosen here — the publisher resolves it per attempt
/// against the manifest it commits against.
- fn new_lineage_intent(
+ pub(crate) fn new_lineage_intent(
&self,
actor_id: Option<&str>,
merged_parent_commit_id: Option,
- ) -> Result {
- Ok(crate::db::manifest::LineageIntent {
+ ) -> Result {
+ Ok(LineageIntent {
graph_commit_id: ulid::Ulid::new().to_string(),
branch: self.current_branch().map(str::to_string),
actor_id: actor_id.map(str::to_string),
diff --git a/crates/omnigraph/src/db/manifest.rs b/crates/omnigraph/src/db/manifest.rs
index 9b9853d2..2458cb25 100644
--- a/crates/omnigraph/src/db/manifest.rs
+++ b/crates/omnigraph/src/db/manifest.rs
@@ -28,25 +28,26 @@ mod recovery;
mod state;
use graph::{init_manifest_graph, open_manifest_graph, snapshot_state_at};
-use layout::{open_manifest_dataset, table_uri_for_path, type_name_hash};
pub(crate) use layout::manifest_uri;
+use layout::{open_manifest_dataset, table_uri_for_path, type_name_hash};
pub(crate) use metadata::TableVersionMetadata;
#[cfg(test)]
use metadata::{OMNIGRAPH_ROW_COUNT_KEY, table_version_metadata_for_state};
#[cfg(test)]
use namespace::{branch_manifest_namespace, staged_table_namespace};
-pub(crate) use publisher::LineageIntent;
+pub(crate) use publisher::{GraphHeadExpectation, LineageIntent, PublishPrecondition};
use publisher::{GraphNamespacePublisher, ManifestBatchPublisher, PublishOutcome};
pub(crate) use recovery::{
- RecoveryMode, RecoverySidecar, RecoverySidecarHandle, SidecarKind, SidecarTablePin,
- SidecarTableRegistration, SidecarTombstone, confirm_sidecar_phase_b, delete_sidecar,
- has_schema_apply_sidecar, heal_pending_sidecars_roll_forward, list_sidecars, new_sidecar,
- recover_manifest_drift, schema_apply_serial_queue_key, write_sidecar,
+ HealPendingOutcome, RecoveryAuthorityToken, RecoveryLineageIntent, RecoveryMode,
+ RecoverySidecar, RecoverySidecarHandle, SidecarKind, SidecarTablePin, SidecarTableRegistration,
+ SidecarTombstone, confirm_occ_sidecar_phase_b, confirm_sidecar_phase_b, delete_sidecar,
+ has_schema_apply_sidecar, heal_pending_sidecars_roll_forward, list_sidecars, new_occ_sidecar,
+ new_sidecar, recover_manifest_drift, schema_apply_serial_queue_key, write_sidecar,
};
pub use state::SubTableEntry;
-pub(crate) use state::{GraphLineageRow, read_graph_lineage};
#[cfg(test)]
use state::string_column;
+pub(crate) use state::{GraphLineageRow, read_graph_lineage};
use state::{ManifestState, read_manifest_state};
/// The internal-schema (storage-format) version this binary writes and reads.
@@ -498,7 +499,30 @@ impl ManifestCoordinator {
expected_table_versions: &HashMap,
lineage: Option<&LineageIntent>,
) -> Result {
- if changes.is_empty() && expected_table_versions.is_empty() && lineage.is_none() {
+ self.commit_changes_with_lineage_and_precondition(
+ changes,
+ expected_table_versions,
+ lineage,
+ &PublishPrecondition::Any,
+ )
+ .await
+ }
+
+ /// The token-aware form of [`commit_changes_with_lineage`]. Exact authority
+ /// is checked by the publisher from every CAS attempt's existing one-scan
+ /// state; legacy callers continue through `Any` above.
+ pub(crate) async fn commit_changes_with_lineage_and_precondition(
+ &mut self,
+ changes: &[ManifestChange],
+ expected_table_versions: &HashMap,
+ lineage: Option<&LineageIntent>,
+ precondition: &PublishPrecondition,
+ ) -> Result {
+ if changes.is_empty()
+ && expected_table_versions.is_empty()
+ && lineage.is_none()
+ && matches!(precondition, PublishPrecondition::Any)
+ {
return Ok(CommitOutcome {
version: self.version(),
parent_commit_id: None,
@@ -511,7 +535,7 @@ impl ManifestCoordinator {
known_state,
} = self
.publisher
- .publish(changes, expected_table_versions, lineage)
+ .publish_with_precondition(changes, expected_table_versions, lineage, precondition)
.await?;
// RFC-013 PR2 #1b: the publisher folded the new visible state in-memory
// (byte-identical to a re-scan via the shared `assemble_manifest_state`),
@@ -550,6 +574,28 @@ impl ManifestCoordinator {
.map_err(|e| OmniError::Lance(e.to_string()))
}
+ /// Lance-native stable identity for the active manifest branch. Unlike a
+ /// manifest version/eTag, this remains stable across ordinary commits and
+ /// changes when a named branch is deleted and recreated (ABA protection).
+ pub(crate) async fn branch_identifier(&self) -> Result {
+ self.dataset
+ .branch_identifier()
+ .await
+ .map_err(|e| OmniError::Lance(e.to_string()))
+ }
+
+ /// Exact materialized `graph_head:` from the same pinned
+ /// manifest version as [`Self::snapshot`]. This is write authority, not a
+ /// lineage-cache query: a read may refresh only the manifest, so consulting
+ /// `CommitGraph` here would combine a fresh table snapshot with a stale head.
+ pub(crate) fn exact_graph_head(&self) -> Option {
+ let branch_key = self
+ .active_branch
+ .as_deref()
+ .unwrap_or(MAIN_BRANCH_HEAD_KEY);
+ self.known_state.graph_heads.get(branch_key).cloned()
+ }
+
pub(crate) fn incarnation(&self) -> ManifestIncarnation {
ManifestIncarnation {
version: self.version(),
diff --git a/crates/omnigraph/src/db/manifest/publisher.rs b/crates/omnigraph/src/db/manifest/publisher.rs
index 39d7932a..e448344b 100644
--- a/crates/omnigraph/src/db/manifest/publisher.rs
+++ b/crates/omnigraph/src/db/manifest/publisher.rs
@@ -36,12 +36,12 @@ use super::metadata::{TableVersionMetadata, parse_namespace_version_request};
use super::migrations::{read_stamp, refuse_if_stamp_unsupported};
use super::state::{
GraphLineageRow, GraphLineageRowPart, ManifestState, assemble_manifest_state,
- graph_lineage_row_parts, head_lineage_row, manifest_rows_batch, manifest_schema,
- read_manifest_state, read_publish_scan,
+ graph_head_object_id, graph_lineage_row_parts, head_lineage_row, manifest_rows_batch,
+ manifest_schema, read_manifest_state, read_publish_scan,
};
use super::{
- ManifestChange, OBJECT_TYPE_TABLE, OBJECT_TYPE_TABLE_TOMBSTONE, OBJECT_TYPE_TABLE_VERSION,
- SubTableEntry, TableRegistration, TableTombstone,
+ MAIN_BRANCH_HEAD_KEY, ManifestChange, OBJECT_TYPE_TABLE, OBJECT_TYPE_TABLE_TOMBSTONE,
+ OBJECT_TYPE_TABLE_VERSION, SubTableEntry, TableRegistration, TableTombstone,
};
/// Bound on the publisher-level retry loop that wraps Lance's row-level CAS
@@ -54,11 +54,10 @@ const PUBLISHER_RETRY_BUDGET: u32 = 5;
/// The graph-lineage commit to record atomically with a manifest publish
/// (RFC-013 Phase 7). One logical commit per publish: the `graph_commit_id` is
/// minted once by the caller and stays stable across the publisher's CAS
-/// retries; only the parent re-resolves per attempt (against the freshly loaded
-/// `__manifest`), so a retry after a concurrent commit parents off the new head
-/// — the TOCTOU the dual-write era's `commit_graph.refresh()` guarded is closed
-/// by construction.
-#[derive(Debug, Clone)]
+/// retries. Legacy [`PublishPrecondition::Any`] publishes re-resolve the parent
+/// per attempt. An exact-head publish instead rejects a retry once that authority
+/// changed, so a prepared write can never be silently re-parented.
+#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub(crate) struct LineageIntent {
/// ULID minted once before the publish loop; the graph commit's identity.
pub graph_commit_id: String,
@@ -73,6 +72,55 @@ pub(crate) struct LineageIntent {
pub created_at: i64,
}
+/// The exact mutable graph-head authority a prepared write observed. A missing
+/// row is first-class: a freshly-created named branch inherits lineage commits
+/// but has no `graph_head:` until its first graph commit.
+#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
+pub(crate) struct GraphHeadExpectation {
+ /// `None` means main; `Some("main")` is normalized to `None` by [`new`].
+ pub(crate) branch: Option,
+ /// Lance-native stable branch identity. This detects delete/recreate ABA;
+ /// manifest versions/eTags are deliberately not branch identity.
+ pub(crate) branch_identifier: lance::dataset::refs::BranchIdentifier,
+ /// Exact commit id stored in the branch's head row, or `None` when absent.
+ pub(crate) head_commit_id: Option,
+}
+
+impl GraphHeadExpectation {
+ pub(crate) fn new(
+ branch: Option<&str>,
+ branch_identifier: lance::dataset::refs::BranchIdentifier,
+ head_commit_id: Option,
+ ) -> Self {
+ Self {
+ branch: branch
+ .filter(|branch| *branch != "main")
+ .map(ToOwned::to_owned),
+ branch_identifier,
+ head_commit_id,
+ }
+ }
+
+ fn object_id(&self) -> String {
+ graph_head_object_id(self.branch.as_deref())
+ }
+}
+
+/// Authority checked by the manifest publisher on every CAS attempt.
+///
+/// `Any` preserves the legacy dispatcher semantics: row-level contention may
+/// retry and re-parent a lineage intent. `ExactGraphHead` is the RFC-022
+/// foundation for prepared writes: after contention, any head movement becomes
+/// `ReadSetChanged` rather than a transparent re-parent. Its native branch-id
+/// check detects delete/recreate ABA on every attempt; it is not a distributed
+/// ref-control fence (Lance branch create/delete still lacks conditional CAS),
+/// so branch control remains within the documented single-writer-process bound.
+#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
+pub(crate) enum PublishPrecondition {
+ Any,
+ ExactGraphHead(GraphHeadExpectation),
+}
+
/// The result of a manifest publish that may have folded in a graph commit.
#[derive(Debug)]
pub(super) struct PublishOutcome {
@@ -92,11 +140,29 @@ pub(super) struct PublishOutcome {
#[async_trait]
pub(super) trait ManifestBatchPublisher: Send + Sync {
+ /// Legacy publish behavior. All existing writers deliberately retain
+ /// `Any` until enrolled in their RFC-022 adapter.
async fn publish(
&self,
changes: &[ManifestChange],
expected_table_versions: &HashMap,
lineage: Option<&LineageIntent>,
+ ) -> Result {
+ self.publish_with_precondition(
+ changes,
+ expected_table_versions,
+ lineage,
+ &PublishPrecondition::Any,
+ )
+ .await
+ }
+
+ async fn publish_with_precondition(
+ &self,
+ changes: &[ManifestChange],
+ expected_table_versions: &HashMap,
+ lineage: Option<&LineageIntent>,
+ precondition: &PublishPrecondition,
) -> Result;
}
@@ -119,15 +185,16 @@ struct PendingVersionRow {
/// Everything one CAS attempt needs out of a single `__manifest` scan
/// (RFC-013 P2): the open dataset, table state for the pre-check + pending-row
-/// build, and the `graph_commit` lineage rows for parent resolution. Folding the
-/// lineage into this struct is what lets `resolve_lineage_rows` skip its own
-/// `read_graph_lineage` scan.
+/// build, `graph_commit` lineage rows for parent resolution, and exact
+/// `graph_head` rows for OCC. Folding lineage authority into this struct is what
+/// lets both checks skip their own `read_graph_lineage` scan.
struct LoadedPublishState {
dataset: Dataset,
registered_tables: HashMap,
existing_versions: HashMap<(String, u64), SubTableEntry>,
existing_tombstones: HashMap<(String, u64), ()>,
lineage_rows: Vec,
+ graph_heads: HashMap,
}
impl GraphNamespacePublisher {
@@ -159,12 +226,11 @@ impl GraphNamespacePublisher {
// `db/manifest/migrations.rs`.
refuse_if_stamp_unsupported(read_stamp(&dataset))?;
// ONE `__manifest` scan for everything the publish needs: table
- // locations, version entries, tombstones, AND the `graph_commit` lineage
- // rows for parent resolution (RFC-013 P2). The lineage extraction rides
- // this pass instead of a second `read_graph_lineage` scan in
- // `resolve_lineage_rows`; the per-attempt re-read is preserved because
- // `load_publish_state` runs once per CAS attempt, so a retry sees the
- // advanced head and re-parents correctly.
+ // locations, version entries, tombstones, `graph_commit` lineage rows
+ // for parent resolution, AND exact `graph_head` rows for OCC (RFC-013
+ // P2 / RFC-022). Extraction rides this pass instead of a second
+ // `read_graph_lineage` scan; the per-attempt re-read is preserved because
+ // `load_publish_state` runs once per CAS attempt.
let scan = read_publish_scan(&dataset).await?;
let existing_versions = scan
.version_entries
@@ -183,6 +249,7 @@ impl GraphNamespacePublisher {
existing_versions,
existing_tombstones,
lineage_rows: scan.lineage_rows,
+ graph_heads: scan.graph_heads,
})
}
@@ -506,12 +573,15 @@ impl GraphNamespacePublisher {
))
})?;
let table_path =
- table_locations.get(&row.table_key).cloned().ok_or_else(|| {
- OmniError::manifest_internal(format!(
- "post-publish fold: missing table row for {}",
- row.table_key
- ))
- })?;
+ table_locations
+ .get(&row.table_key)
+ .cloned()
+ .ok_or_else(|| {
+ OmniError::manifest_internal(format!(
+ "post-publish fold: missing table row for {}",
+ row.table_key
+ ))
+ })?;
let metadata_json = row.metadata.as_deref().ok_or_else(|| {
OmniError::manifest_internal(format!(
"post-publish fold: table_version row missing metadata for {}",
@@ -572,6 +642,80 @@ impl GraphNamespacePublisher {
Ok(())
}
+ /// Check authority inside the publisher retry loop before pending rows are
+ /// built. The graph head comes from the SAME scan used to build this CAS
+ /// attempt; Lance's native branch identifier is re-read from the ref. Thus a
+ /// row-level-CAS loser with an exact expectation cannot silently re-parent
+ /// on its next attempt.
+ async fn check_publish_precondition(
+ &self,
+ dataset: &Dataset,
+ graph_heads: &HashMap,
+ precondition: &PublishPrecondition,
+ ) -> Result<()> {
+ let PublishPrecondition::ExactGraphHead(expected) = precondition else {
+ return Ok(());
+ };
+
+ let expected_branch = expected
+ .branch
+ .as_deref()
+ .filter(|branch| *branch != "main");
+ if expected_branch != self.branch.as_deref() {
+ return Err(OmniError::manifest_internal(format!(
+ "publish graph-head precondition targets branch '{}' but publisher is bound to '{}'",
+ expected_branch.unwrap_or("main"),
+ self.branch.as_deref().unwrap_or("main"),
+ )));
+ }
+
+ let branch_identity_member =
+ format!("branch_identifier:{}", expected_branch.unwrap_or("main"));
+ let expected_branch_identifier = serde_json::to_string(&expected.branch_identifier)
+ .map_err(|e| {
+ OmniError::manifest_internal(format!(
+ "failed to encode expected Lance branch identifier: {e}"
+ ))
+ })?;
+ let actual_branch_identifier = match dataset.branch_identifier().await {
+ Ok(identifier) => identifier,
+ Err(LanceError::RefNotFound { .. }) => {
+ return Err(OmniError::manifest_read_set_changed(
+ branch_identity_member,
+ Some(expected_branch_identifier),
+ None,
+ ));
+ }
+ Err(err) => return Err(OmniError::Lance(err.to_string())),
+ };
+ if actual_branch_identifier != expected.branch_identifier {
+ let actual = serde_json::to_string(&actual_branch_identifier).map_err(|e| {
+ OmniError::manifest_internal(format!(
+ "failed to encode current Lance branch identifier: {e}"
+ ))
+ })?;
+ return Err(OmniError::manifest_read_set_changed(
+ branch_identity_member,
+ Some(expected_branch_identifier),
+ Some(actual),
+ ));
+ }
+
+ let object_id = expected.object_id();
+ let branch_key = object_id
+ .strip_prefix("graph_head:")
+ .expect("graph_head_object_id always supplies the prefix");
+ let actual = graph_heads.get(branch_key).cloned();
+ if actual != expected.head_commit_id {
+ return Err(OmniError::manifest_read_set_changed(
+ object_id,
+ expected.head_commit_id.clone(),
+ actual,
+ ));
+ }
+ Ok(())
+ }
+
async fn merge_rows(&self, dataset: Dataset, rows: Vec) -> Result {
let batch = Self::pending_rows_to_batch(rows)?;
let reader = RecordBatchIterator::new(vec![Ok(batch)], manifest_schema());
@@ -671,13 +815,18 @@ pub(crate) fn map_lance_publish_error(err: LanceError) -> OmniError {
#[async_trait]
impl ManifestBatchPublisher for GraphNamespacePublisher {
- async fn publish(
+ async fn publish_with_precondition(
&self,
changes: &[ManifestChange],
expected_table_versions: &HashMap,
lineage: Option<&LineageIntent>,
+ precondition: &PublishPrecondition,
) -> Result {
- if changes.is_empty() && expected_table_versions.is_empty() && lineage.is_none() {
+ if changes.is_empty()
+ && expected_table_versions.is_empty()
+ && lineage.is_none()
+ && matches!(precondition, PublishPrecondition::Any)
+ {
// Defensive no-op (never reached from `commit_changes_with_lineage`,
// which short-circuits the all-empty case): state is unchanged, so a
// re-scan here is acceptable.
@@ -709,8 +858,16 @@ impl ManifestBatchPublisher for GraphNamespacePublisher {
existing_versions,
existing_tombstones,
lineage_rows,
+ graph_heads,
} = loaded;
+ // Exact logical authority is checked on EVERY attempt from this
+ // attempt's single manifest scan. In particular, a CAS retry after
+ // another writer creates or advances `graph_head:` fails
+ // here instead of transparently re-parenting the prepared intent.
+ self.check_publish_precondition(&dataset, &graph_heads, precondition)
+ .await?;
+
let latest_per_table =
Self::latest_visible_per_table(&existing_versions, &existing_tombstones);
// Pre-check on every attempt against freshly loaded state so a
@@ -754,6 +911,7 @@ impl ManifestBatchPublisher for GraphNamespacePublisher {
existing_tombstones
.keys()
.map(|(key, version)| (key.clone(), *version)),
+ graph_heads,
);
return Ok(PublishOutcome {
dataset,
@@ -765,8 +923,23 @@ impl ManifestBatchPublisher for GraphNamespacePublisher {
// Build the post-publish fold inputs from the pre-publish state ∪ the
// rows we are about to commit, BEFORE `rows` is moved into merge_rows
// (RFC-013 PR2 #1b). Recomputed per attempt from freshly-loaded state.
- let (fold_entries, fold_tombstones) =
- Self::fold_inputs(&existing_versions, &existing_tombstones, &rows, &known_tables)?;
+ let (fold_entries, fold_tombstones) = Self::fold_inputs(
+ &existing_versions,
+ &existing_tombstones,
+ &rows,
+ &known_tables,
+ )?;
+ let mut fold_graph_heads = graph_heads;
+ if let Some(intent) = lineage {
+ fold_graph_heads.insert(
+ intent
+ .branch
+ .as_deref()
+ .unwrap_or(MAIN_BRANCH_HEAD_KEY)
+ .to_string(),
+ intent.graph_commit_id.clone(),
+ );
+ }
match self.merge_rows(dataset, rows).await {
Ok(new_dataset) => {
@@ -774,6 +947,7 @@ impl ManifestBatchPublisher for GraphNamespacePublisher {
new_dataset.version().version,
fold_entries,
fold_tombstones,
+ fold_graph_heads,
);
return Ok(PublishOutcome {
dataset: new_dataset,
diff --git a/crates/omnigraph/src/db/manifest/recovery.rs b/crates/omnigraph/src/db/manifest/recovery.rs
index fbc02be1..e272149e 100644
--- a/crates/omnigraph/src/db/manifest/recovery.rs
+++ b/crates/omnigraph/src/db/manifest/recovery.rs
@@ -28,13 +28,12 @@
//! CreateIndex/Merge — see `check_restore_txn` at lance-6.0.1
//! `src/io/commit/conflict_resolver.rs:986`. The hazard is documented
//! by `tests/staged_writes.rs::lance_restore_loses_to_concurrent_append_via_orphaning`.
-//! The open-time sweep sidesteps the hazard by running before any
-//! other writers can race; the in-process heal
-//! ([`heal_pending_sidecars_roll_forward`]) never restores (roll-
-//! forward only) and guards via per-(table_key, branch) queue
-//! acquisition.
+//! The open-time sweep and live healer both join the root-scoped ordered
+//! gates. The in-process healer ([`heal_pending_sidecars_roll_forward`])
+//! additionally never restores (roll-forward only); foreign processes remain
+//! outside this serialization boundary.
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
use serde::{Deserialize, Serialize};
use tracing::warn;
@@ -44,30 +43,38 @@ use crate::db::recovery_audit::{RecoveryAudit, RecoveryAuditRecord, RecoveryKind
use crate::db::schema_state::SchemaStateRecovery;
use crate::error::{OmniError, Result};
use crate::storage::StorageAdapter;
+use crate::table_store::StagedTransactionIdentity;
use super::Snapshot;
-use super::publisher::{GraphNamespacePublisher, LineageIntent, ManifestBatchPublisher};
-use super::{ManifestChange, SubTableUpdate, TableRegistration, TableTombstone};
+use super::publisher::{
+ GraphHeadExpectation, GraphNamespacePublisher, LineageIntent, ManifestBatchPublisher,
+ PublishPrecondition,
+};
+use super::{
+ ManifestChange, ManifestCoordinator, SubTableUpdate, TableRegistration, TableTombstone,
+};
-/// System actor identifier recorded on every recovery commit. Operators
-/// distinguish recovery commits from user commits in `omnigraph commit list`
-/// by filtering on this actor; the original sidecar's actor (if any) flows
-/// into the audit row's `recovery_for_actor` field.
+/// System actor identifier for recovery-owned lineage: legacy recovery,
+/// v3 rollback, and orphan discard. A v3 roll-forward deliberately publishes
+/// the original writer's fixed lineage and actor instead. The recovery audit
+/// row is the authoritative attribution surface in both cases; the original
+/// sidecar actor flows into its `recovery_for_actor` field.
pub(crate) const RECOVERY_ACTOR: &str = "omnigraph:recovery";
-/// Publish a recovery action's manifest `updates` AND its recovery commit in one
-/// CAS (RFC-013 Phase 7). The recovery commit's lineage (`graph_commit` +
-/// `graph_head`) rides the same merge-insert as the table-version re-pin — there
-/// is no separate `_graph_commits.lance` write and no manifest→commit-graph gap.
+/// Publish a recovery action's manifest `updates` AND its lineage in one CAS
+/// (RFC-013 Phase 7). Legacy recovery and v3 rollback publish a recovery commit;
+/// v3 roll-forward publishes the original writer's fixed lineage intent. The
+/// lineage (`graph_commit` + `graph_head`) rides the same merge-insert as the
+/// table-version re-pin — there is no separate `_graph_commits.lance` write and
+/// no manifest→commit-graph gap.
/// `updates` is empty for the no-table-change recovery paths (all-NoMovement
/// roll-back, stale-sidecar cleanup, orphaned-branch discard); the lineage rows
/// still publish, so the recovery commit is always durable.
///
-/// The commit's first parent is resolved by the publisher (the live head of the
-/// recovery's branch); its merged-in parent is the sidecar's recorded source
-/// head for a rolled-forward branch merge, matching the pre-Phase-7 merge-commit
-/// shape. Returns the new manifest version and the minted recovery commit id
-/// (which the audit row references).
+/// Legacy commits resolve their parent from the live branch head. A v3
+/// roll-forward instead carries an exact graph-head precondition, so it cannot
+/// be silently re-parented after authority changes. Returns the new manifest
+/// version and the stable commit id the audit row references.
async fn publish_recovery_commit(
root_uri: &str,
sidecar: &RecoverySidecar,
@@ -75,21 +82,53 @@ async fn publish_recovery_commit(
updates: &[ManifestChange],
expected: &HashMap,
) -> Result<(u64, String)> {
- let merged_parent_commit_id = match (sidecar.writer_kind, kind) {
- (SidecarKind::BranchMerge, RecoveryKind::RolledForward) => {
- sidecar.merge_source_commit_id.clone()
+ let intent = match (sidecar.protocol_v3.as_ref(), kind) {
+ (Some(protocol), RecoveryKind::RolledForward) => LineageIntent::from(&protocol.lineage),
+ (Some(protocol), RecoveryKind::RolledBack) => LineageIntent {
+ graph_commit_id: protocol.rollback_graph_commit_id.clone(),
+ branch: sidecar.branch.clone(),
+ actor_id: Some(RECOVERY_ACTOR.to_string()),
+ merged_parent_commit_id: None,
+ created_at: sidecar
+ .started_at
+ .parse::()
+ .unwrap_or(crate::db::now_micros()?),
+ },
+ (Some(_), RecoveryKind::OrphanedBranchDiscarded) => {
+ return Err(OmniError::manifest_internal(
+ "orphaned-branch discard cannot publish through the v3 OCC recovery path",
+ ));
+ }
+ (None, _) => {
+ let merged_parent_commit_id = match (sidecar.writer_kind, kind) {
+ (SidecarKind::BranchMerge, RecoveryKind::RolledForward) => {
+ sidecar.merge_source_commit_id.clone()
+ }
+ _ => None,
+ };
+ LineageIntent {
+ graph_commit_id: ulid::Ulid::new().to_string(),
+ branch: sidecar.branch.clone(),
+ actor_id: Some(RECOVERY_ACTOR.to_string()),
+ merged_parent_commit_id,
+ created_at: crate::db::now_micros()?,
+ }
}
- _ => None,
- };
- let intent = LineageIntent {
- graph_commit_id: ulid::Ulid::new().to_string(),
- branch: sidecar.branch.clone(),
- actor_id: Some(RECOVERY_ACTOR.to_string()),
- merged_parent_commit_id,
- created_at: crate::db::now_micros()?,
};
let publisher = GraphNamespacePublisher::new(root_uri, sidecar.branch.as_deref());
- let outcome = publisher.publish(updates, expected, Some(&intent)).await?;
+ let precondition = match (sidecar.protocol_v3.as_ref(), kind) {
+ (Some(protocol), RecoveryKind::RolledForward) => {
+ PublishPrecondition::ExactGraphHead(GraphHeadExpectation::new(
+ sidecar.branch.as_deref(),
+ protocol.authority.branch_identifier.clone(),
+ protocol.authority.graph_head.clone(),
+ ))
+ }
+ _ => PublishPrecondition::Any,
+ };
+ let outcome = publisher
+ .publish_with_precondition(updates, expected, Some(&intent), &precondition)
+ .await?;
Ok((outcome.dataset.version().version, intent.graph_commit_id))
}
@@ -109,7 +148,18 @@ pub(crate) const RECOVERY_DIR_NAME: &str = "__recovery";
/// back); at v1 it predates confirmation and keeps the loose roll-forward. The
/// reader must distinguish the two, so this is a real version bump, not an
/// additive field.
-pub(crate) const SIDECAR_SCHEMA_VERSION: u32 = 2;
+///
+/// v2 → v3: RFC-022 exact mutation/load effects. An enrolled sidecar carries
+/// the captured authority, stable lineage/rollback ids, planned Lance
+/// transaction identities, a canonical manifest delta, and an explicit
+/// Armed → EffectsConfirmed transition. Legacy writers deliberately continue
+/// producing v2 until their adapter opts into [`new_occ_sidecar`].
+pub(crate) const SIDECAR_SCHEMA_VERSION: u32 = 3;
+
+/// Schema version emitted by the legacy constructor. Fixed at v2 so merely
+/// teaching this binary to understand v3 does not silently enroll every writer
+/// in new recovery semantics.
+pub(crate) const LEGACY_SIDECAR_SCHEMA_VERSION: u32 = 2;
/// The version at which Phase-B confirmation shipped. A `BranchMerge` sidecar is
/// confirmation-aware (strict classification) iff `schema_version >=` this.
@@ -117,6 +167,16 @@ pub(crate) const SIDECAR_SCHEMA_VERSION: u32 = 2;
/// v3+ still treats v2 sidecars as confirmation-aware.
pub(crate) const CONFIRMATION_SCHEMA_VERSION: u32 = 2;
+/// Mutation/load exact-effect semantics shipped at v3. Fixed rather than tied
+/// to the current maximum so future sidecar versions retain this floor.
+pub(crate) const EXACT_EFFECT_IDENTITY_SCHEMA_VERSION: u32 = 3;
+
+/// Bound the cold-path transaction-history probe used to find a v3 sidecar's
+/// UUID after unexpected drift. Normal recovery reads one version; a larger
+/// gap indicates foreign activity and is surfaced as unverifiable rather than
+/// turning open into an unbounded history walk.
+const MAX_EFFECT_IDENTITY_SCAN_VERSIONS: u64 = 1024;
+
/// Selects which recovery actions are allowed in a sweep.
///
/// Open-time recovery (`Omnigraph::open` with `OpenMode::ReadWrite`)
@@ -184,6 +244,10 @@ pub(crate) enum ClassificationMode {
/// CONFIRMATION_SCHEMA_VERSION`): roll forward ONLY to the confirmed
/// version; an unconfirmed moved HEAD is a partial publish and rolls back.
Confirmed,
+ /// RFC-022 mutation/load effect: the sidecar must be durably confirmed and
+ /// the observed Lance HEAD transaction `(read_version, uuid)` must exactly
+ /// equal the planned and confirmed transaction identity.
+ ExactEffect,
}
impl SidecarKind {
@@ -192,7 +256,13 @@ impl SidecarKind {
/// compile error here until its recovery semantics are declared.
pub(crate) fn classification_mode(self, schema_version: u32) -> ClassificationMode {
match self {
- SidecarKind::Mutation | SidecarKind::Load => ClassificationMode::Strict,
+ SidecarKind::Mutation | SidecarKind::Load => {
+ if schema_version >= EXACT_EFFECT_IDENTITY_SCHEMA_VERSION {
+ ClassificationMode::ExactEffect
+ } else {
+ ClassificationMode::Strict
+ }
+ }
// BranchMerge gained two-phase confirmation at
// `CONFIRMATION_SCHEMA_VERSION`. A sidecar written before that
// carries no `confirmed_version` and must keep the prior loose
@@ -284,6 +354,114 @@ pub(crate) struct SidecarTombstone {
pub tombstone_version: u64,
}
+/// Authority captured before an RFC-022 mutation/load attempt was prepared.
+/// Recovery compares this complete token before it may roll confirmed effects
+/// forward. The native Lance branch identifier closes delete/recreate ABA;
+/// `graph_head: None` deliberately preserves absence on a fresh branch.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub(crate) struct RecoveryAuthorityToken {
+ pub branch_identifier: lance::dataset::refs::BranchIdentifier,
+ pub graph_head: Option,
+ pub schema_ir_hash: String,
+ pub schema_identity_version: u32,
+}
+
+/// Serializable copy of the original graph lineage intent. It is minted once
+/// before recovery is armed and never re-parented or replaced during recovery.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub(crate) struct RecoveryLineageIntent {
+ pub graph_commit_id: String,
+ pub branch: Option,
+ pub actor_id: Option,
+ pub merged_parent_commit_id: Option,
+ pub created_at: i64,
+}
+
+impl From<&RecoveryLineageIntent> for LineageIntent {
+ fn from(intent: &RecoveryLineageIntent) -> Self {
+ Self {
+ graph_commit_id: intent.graph_commit_id.clone(),
+ branch: intent.branch.clone(),
+ actor_id: intent.actor_id.clone(),
+ merged_parent_commit_id: intent.merged_parent_commit_id.clone(),
+ created_at: intent.created_at,
+ }
+ }
+}
+
+/// Durable phase of an RFC-022 effect set. `Armed` is written before the first
+/// HEAD advance. `EffectsConfirmed` is written only after every planned table
+/// effect landed with its exact transaction identity and full manifest update.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "PascalCase")]
+pub(crate) enum RecoveryEffectPhase {
+ Armed,
+ EffectsConfirmed,
+}
+
+/// One exact physical effect enrolled in an RFC-022 sidecar.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub(crate) struct RecoveryTableEffect {
+ pub table_key: String,
+ pub planned_transaction: StagedTransactionIdentity,
+ #[serde(default, skip_serializing_if = "Option::is_none")]
+ pub confirmed_transaction: Option,
+}
+
+/// The confirmed value for one table-version output slot in the intended
+/// manifest delta. Metadata is copied from `SubTableUpdate` so recovery can
+/// publish exactly what the original writer prepared rather than re-deriving a
+/// wider delta from an arbitrary live HEAD.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub(crate) struct RecoveryConfirmedTableUpdate {
+ pub table_version: u64,
+ pub table_branch: Option,
+ pub row_count: u64,
+ pub version_metadata: super::TableVersionMetadata,
+}
+
+/// Canonical manifest-delta output slot for one touched table.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub(crate) struct RecoveryTableUpdateSlot {
+ pub table_key: String,
+ pub expected_version: u64,
+ pub table_branch: Option,
+ #[serde(default, skip_serializing_if = "Option::is_none")]
+ pub confirmed: Option,
+}
+
+/// Complete manifest delta carried by an RFC-022 mutation/load sidecar. The
+/// table keys and expectations are immutable at arm time; physical output
+/// slots are bound exactly once by Phase-B confirmation.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub(crate) struct RecoveryManifestDelta {
+ pub table_updates: Vec,
+ #[serde(default, skip_serializing_if = "Vec::is_empty")]
+ pub registrations: Vec,
+ #[serde(default, skip_serializing_if = "Vec::is_empty")]
+ pub tombstones: Vec,
+}
+
+/// v3 payload. Kept optional on [`RecoverySidecar`] so v1/v2 continue to parse
+/// with their original semantics; [`parse_sidecar`] requires it for v3.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub(crate) struct RecoveryProtocolV3 {
+ pub authority: RecoveryAuthorityToken,
+ pub lineage: RecoveryLineageIntent,
+ /// Stable id for the rollback lineage outcome. A recovery retry reuses it
+ /// instead of manufacturing an unbounded series of indistinguishable ids.
+ pub rollback_graph_commit_id: String,
+ /// Exact operator-facing outcomes for the fixed rollback commit. Recovery
+ /// persists this plan before the first restore or rollback publish and
+ /// reuses it verbatim after a crash. `None` means no rollback has been
+ /// prepared; `Some([])` is a prepared lineage-only rollback.
+ #[serde(default, skip_serializing_if = "Option::is_none")]
+ pub rollback_audit_outcomes: Option>,
+ pub effect_phase: RecoveryEffectPhase,
+ pub effects: Vec,
+ pub intended_delta: RecoveryManifestDelta,
+}
+
/// In-memory representation of the on-disk JSON sidecar.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub(crate) struct RecoverySidecar {
@@ -318,6 +496,9 @@ pub(crate) struct RecoverySidecar {
/// non-SchemaApply writers.
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub tombstones: Vec,
+ /// RFC-022 exact-effect protocol. Absent for every v1/v2 sidecar.
+ #[serde(default, skip_serializing_if = "Option::is_none")]
+ pub protocol_v3: Option,
}
/// Opaque handle returned by [`write_sidecar`] so the caller can delete
@@ -384,6 +565,11 @@ pub(crate) enum TableClassification {
/// previous restore attempt or an external mutation. Roll back to
/// the manifest pin.
UnexpectedMultistep,
+ /// The observed HEAD version may have the expected number, but the Lance
+ /// transaction identity does not equal the exact planned+confirmed effect.
+ /// A foreign writer or transparent rebase produced it; never roll it
+ /// forward as this sidecar's result.
+ UnexpectedEffectIdentity,
/// A confirmation-using writer (`BranchMerge`) advanced this table's HEAD
/// (`lance_head > manifest_pinned`) but the sidecar carries no
/// `confirmed_version` — its multi-commit publish crashed mid-flight, so
@@ -407,7 +593,8 @@ pub(crate) enum TableClassification {
/// based on the worst classification:
///
/// - Any `InvariantViolation` → `Abort` (operator action required).
-/// - Any `UnexpectedAtP1` / `UnexpectedMultistep` / `NoMovement` →
+/// - Any `UnexpectedAtP1` / `UnexpectedMultistep` /
+/// `UnexpectedEffectIdentity` / `NoMovement` →
/// `RollBack` all drifted tables to the manifest pin.
/// - All `RolledPastExpected` → `RollForward` every table in one
/// manifest publish.
@@ -450,8 +637,9 @@ pub(crate) async fn write_sidecar(
// Failpoint: models a storage put failure (S3 PutObject / fs write)
// in Phase A — every writer must abort before any HEAD advance.
crate::failpoints::maybe_fail(crate::failpoints::names::RECOVERY_SIDECAR_WRITE)?;
- debug_assert_eq!(sidecar.schema_version, SIDECAR_SCHEMA_VERSION);
+ debug_assert!(sidecar.schema_version <= SIDECAR_SCHEMA_VERSION);
let uri = sidecar_uri(root_uri, &sidecar.operation_id);
+ validate_sidecar_shape(&uri, sidecar)?;
let json = serde_json::to_string_pretty(sidecar).map_err(|err| {
OmniError::manifest_internal(format!("failed to serialize recovery sidecar: {}", err))
})?;
@@ -564,6 +752,53 @@ pub(crate) async fn list_sidecars(
Ok(out)
}
+/// Re-read one listed sidecar after its process-local recovery gates are held.
+///
+/// `list_sidecars` necessarily runs before gate acquisition so the caller knows
+/// which branch/table keys to take. The sidecar body is mutable, however:
+/// BranchMerge stamps `confirmed_version`, and v3 writers move Armed ->
+/// EffectsConfirmed (or persist a rollback audit plan) while holding those same
+/// gates. Existence alone therefore is not a sufficient post-wait recheck. A
+/// recovery pass must classify the body that is durable *after* it wins the
+/// gates, not the stale body it used only for coordination discovery.
+async fn reread_sidecar_under_gates(
+ root_uri: &str,
+ storage: &dyn StorageAdapter,
+ listed: &RecoverySidecar,
+) -> Result