From 905a27c4bd38a01aa497f71ca226628f7ac66bd2 Mon Sep 17 00:00:00 2001 From: Andrew Altshuler Date: Sat, 11 Jul 2026 23:52:53 +0300 Subject: [PATCH] Harden RFC-022 pre-arm recovery ownership (#346) --- AGENTS.md | 2 +- crates/omnigraph/src/db/manifest.rs | 10 +- crates/omnigraph/src/db/manifest/recovery.rs | 670 ++++++++++++++++-- crates/omnigraph/src/db/omnigraph.rs | 158 +++-- .../src/db/omnigraph/schema_apply.rs | 200 ++++-- .../omnigraph/src/db/omnigraph/table_ops.rs | 382 ++++++---- crates/omnigraph/src/exec/merge.rs | 134 +++- crates/omnigraph/src/exec/staging.rs | 183 ++--- crates/omnigraph/src/failpoints.rs | 20 +- crates/omnigraph/tests/branching.rs | 5 +- crates/omnigraph/tests/failpoints.rs | 654 +++++++++++++++-- crates/omnigraph/tests/maintenance.rs | 146 +++- crates/omnigraph/tests/schema_apply.rs | 47 +- docs/dev/invariants.md | 5 + docs/dev/writes.md | 60 +- docs/rfcs/rfc-022-unified-write-path.md | 26 + docs/user/search/indexes.md | 2 +- 17 files changed, 2166 insertions(+), 538 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 69978ed7..59971b5f 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -253,7 +253,7 @@ 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** refs are logically atomic through authoritative `__manifest` `BranchContents`; native create/delete crash gaps are classified and reclaimed under a single-writer-process boundary; live names are path-prefix-disjoint; data-table forks are lazy; system branches are filtered | -| 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`. Schema-v3 Mutation/Load and schema-v4 BranchMerge roll-forward preserve the interrupted writer's fixed commit lineage and actor; rollback and legacy recovery commits use `omnigraph:recovery`. BranchMerge v4 also owns first-touch refs, pre-mints each table's exact ordered data-transaction chain with zero transparent conflict retries, carries pointer-only updates in its complete confirmed manifest delta, and recognizes an interrupted compensation restore on restart. 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. | +| 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. Under their final schema → branch → table gates, mutation/load, SchemaApply, BranchMerge, and EnsureIndices prove every existing effect target still equals its manifest pin before arming; first-touch refs are created only after the sidecar is durable. 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`. Schema-v3 Mutation/Load and schema-v4 BranchMerge roll-forward preserve the interrupted writer's fixed commit lineage and actor. SchemaApply v5 records the target schema identity plus durable Phase-C confirmation so metadata-only recovery is unambiguous. EnsureIndices v6 retains loose effect classification but pre-mints rollback lineage and persists the rollback audit plan before restore, making compensation retry-exact. Other rollback and legacy recovery commits use `omnigraph:recovery`. BranchMerge v4 also owns first-touch refs, pre-mints each table's exact ordered data-transaction chain with zero transparent conflict retries, carries pointer-only updates in its complete confirmed manifest delta, and recognizes an interrupted compensation restore on restart. 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` derives requested `--keep` / `--older-than` cutoffs from each table's available versions; Lance refs plus OmniGraph's live-lazy-branch and recovery floors may retain additional versions. It fails closed on unopenable pins, recovery intent, or uncovered main-table HEAD drift | diff --git a/crates/omnigraph/src/db/manifest.rs b/crates/omnigraph/src/db/manifest.rs index d6edf9bc..746702a2 100644 --- a/crates/omnigraph/src/db/manifest.rs +++ b/crates/omnigraph/src/db/manifest.rs @@ -40,11 +40,13 @@ use publisher::{GraphNamespacePublisher, ManifestBatchPublisher, PublishOutcome} pub(crate) use recovery::{ HealPendingOutcome, RecoveryAuthorityToken, RecoveryBranchMergeEffect, RecoveryBranchMergeEffectKind, RecoveryLineageIntent, RecoveryManifestDelta, RecoveryMode, - RecoverySidecar, RecoverySidecarHandle, RecoveryTableUpdateSlot, SidecarKind, SidecarTablePin, + RecoverySidecar, RecoverySidecarHandle, RecoveryTableUpdateSlot, + SCHEMA_APPLY_CONFIRMATION_SCHEMA_VERSION, SidecarKind, SidecarTablePin, SidecarTableRegistration, SidecarTombstone, confirm_branch_merge_sidecar_phase_b, - confirm_occ_sidecar_phase_b, delete_sidecar, has_schema_apply_sidecar, - heal_pending_sidecars_roll_forward, list_sidecars, new_branch_merge_sidecar, new_occ_sidecar, - new_sidecar, recover_manifest_drift, schema_apply_serial_queue_key, write_sidecar, + confirm_occ_sidecar_phase_b, confirm_schema_apply_manifest_published, delete_sidecar, + has_schema_apply_sidecar, heal_pending_sidecars_roll_forward, list_sidecars, + new_branch_merge_sidecar, new_ensure_indices_sidecar, new_occ_sidecar, new_sidecar, + recover_manifest_drift, schema_apply_serial_queue_key, write_sidecar, }; pub use state::SubTableEntry; #[cfg(test)] diff --git a/crates/omnigraph/src/db/manifest/recovery.rs b/crates/omnigraph/src/db/manifest/recovery.rs index 9090babb..1ad2dd14 100644 --- a/crates/omnigraph/src/db/manifest/recovery.rs +++ b/crates/omnigraph/src/db/manifest/recovery.rs @@ -40,7 +40,7 @@ use tracing::warn; use crate::db::graph_coordinator::GraphCoordinator; use crate::db::recovery_audit::{RecoveryAudit, RecoveryAuditRecord, RecoveryKind, TableOutcome}; -use crate::db::schema_state::SchemaStateRecovery; +use crate::db::schema_state::{SchemaStateRecovery, read_schema_state_identity}; use crate::error::{OmniError, Result}; use crate::storage::StorageAdapter; use crate::table_store::StagedTransactionIdentity; @@ -55,7 +55,8 @@ use super::{ }; /// System actor identifier for recovery-owned lineage: legacy recovery, -/// exact-protocol rollback, and orphan discard. A v3/v4 roll-forward +/// exact-protocol rollback, schema-v6 EnsureIndices rollback, and orphan +/// discard. A v3/v4 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` @@ -63,9 +64,10 @@ use super::{ pub(crate) const RECOVERY_ACTOR: &str = "omnigraph:recovery"; /// Publish a recovery action's manifest `updates` AND its lineage in one CAS -/// (RFC-013 Phase 7). Legacy recovery and v3/v4 rollback publish a recovery -/// commit; v3/v4 roll-forward publishes the original writer's fixed lineage -/// intent. The lineage (`graph_commit` + `graph_head`) rides the same +/// (RFC-013 Phase 7). Legacy recovery and rollback publish a recovery commit; +/// v3/v4 and v6 rollback reuse a pre-minted id, while v3/v4 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 @@ -102,6 +104,16 @@ async fn publish_recovery_commit( .protocol_v4 .as_ref() .map(|protocol| protocol.rollback_graph_commit_id.as_str()) + }) + .or_else(|| { + matches!(kind, RecoveryKind::RolledBack) + .then(|| { + sidecar + .ensure_indices_rollback_v6 + .as_ref() + .map(|protocol| protocol.rollback_graph_commit_id.as_str()) + }) + .flatten() }); let intent = match (exact_lineage, exact_rollback_id, kind) { (Some(lineage), _, RecoveryKind::RolledForward) => LineageIntent::from(lineage), @@ -199,7 +211,16 @@ pub(crate) const RECOVERY_DIR_NAME: &str = "__recovery"; /// effect or an explicit ref-only first-touch fork. Mutation/Load remain fixed /// at v3: raising the reader maximum must not silently enroll an existing writer /// in new semantics. -pub(crate) const SIDECAR_SCHEMA_VERSION: u32 = 4; +/// +/// v4 → v5: SchemaApply Phase-C confirmation. A v5 SchemaApply sidecar carries +/// the target schema identity and a durable manifest-published marker so an +/// empty table-pin set can distinguish pre-staging rollback from Phase-D delete +/// residue. This is a narrow bridge, not the future exact table-effect adapter. +/// +/// v5 → v6: EnsureIndices fixed rollback identity. Its physical effects still +/// use the legacy loose classifier, but a pre-minted rollback commit id and a +/// durably prepared audit payload make rollback re-entry outcome-exact. +pub(crate) const SIDECAR_SCHEMA_VERSION: u32 = 6; /// Schema version emitted by the legacy constructor. Fixed at v2 so merely /// teaching this binary to understand v3 does not silently enroll every writer @@ -224,6 +245,15 @@ pub(crate) const MUTATION_LOAD_SIDECAR_SCHEMA_VERSION: u32 = 3; /// Exact schema generation emitted by [`new_branch_merge_sidecar`]. pub(crate) const BRANCH_MERGE_SIDECAR_SCHEMA_VERSION: u32 = 4; +/// SchemaApply recovery generation with target identity + durable Phase-C +/// confirmation. Its table classification remains legacy loose-match. +pub(crate) const SCHEMA_APPLY_CONFIRMATION_SCHEMA_VERSION: u32 = 5; + +/// EnsureIndices generation with a fixed rollback outcome. This is narrower +/// than an exact effect adapter: it makes compensation idempotent without +/// claiming transaction-level ownership of the derived index commits. +pub(crate) const ENSURE_INDICES_ROLLBACK_SCHEMA_VERSION: u32 = 6; + /// Bound the cold-path transaction-history probes used by the v3/v4 exact /// recovery protocols. Normal v3 recovery reads one version and a v4 logical /// merge chain is currently at most three versions; a larger gap is derived @@ -598,6 +628,19 @@ pub(crate) struct RecoveryProtocolV4 { pub intended_delta: RecoveryManifestDelta, } +/// Schema-v6 EnsureIndices rollback identity. EnsureIndices remains a +/// loose-effect writer until its full RFC-022 adapter lands, but recovery must +/// still be able to prove that a previously published compensation was a +/// rollback rather than infer the outcome from aligned numeric table pins. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub(crate) struct RecoveryEnsureIndicesRollbackV6 { + pub rollback_graph_commit_id: String, + /// Bound before the first restore or rollback publish so a retry can replay + /// the original pre-compensation observations exactly. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub rollback_audit_outcomes: Option>, +} + /// In-memory representation of the on-disk JSON sidecar. #[derive(Debug, Clone, Serialize, Deserialize)] pub(crate) struct RecoverySidecar { @@ -632,12 +675,32 @@ pub(crate) struct RecoverySidecar { /// non-SchemaApply writers. #[serde(default, skip_serializing_if = "Vec::is_empty")] pub tombstones: Vec, + /// SchemaApply-only Phase-C confirmation. A metadata/table-set-only apply + /// may have zero table pins, so table classification alone cannot + /// distinguish a pre-staging crash from a completed apply whose Phase-D + /// sidecar delete failed. The writer flips this marker durably immediately + /// after its manifest commit and before final schema-file promotion. + /// + /// Optional/default-false for backward compatibility. The full exact + /// SchemaApply adapter will replace this narrow confirmation with its + /// complete authority and intended-delta protocol. + #[serde(default)] + pub schema_apply_manifest_published: bool, + /// Hash of the schema contract this SchemaApply intends to promote. Paired + /// with `schema_apply_manifest_published` so recovery can distinguish a + /// completed final rename from missing/corrupt staging after Phase C. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub schema_apply_target_schema_ir_hash: Option, /// RFC-022 exact-effect protocol. Absent for every v1/v2 sidecar. #[serde(default, skip_serializing_if = "Option::is_none")] pub protocol_v3: Option, /// RFC-022 BranchMerge protocol. Present only on schema-v4 sidecars. #[serde(default, skip_serializing_if = "Option::is_none")] pub protocol_v4: Option, + /// EnsureIndices-only fixed rollback identity. It does not make the + /// physical index effects exact; it only makes compensation retry-safe. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub ensure_indices_rollback_v6: Option, } /// Opaque handle returned by [`write_sidecar`] so the caller can delete @@ -789,6 +852,45 @@ pub(crate) async fn write_sidecar( }) } +/// Durably confirm that a legacy SchemaApply sidecar's Phase-C manifest commit +/// returned successfully. This marker is intentionally narrower than the exact +/// v3/v4 protocols: it resolves the zero-table Phase-D ambiguity while the full +/// SchemaApply adapter remains future work. +pub(crate) async fn confirm_schema_apply_manifest_published( + root_uri: &str, + storage: &dyn StorageAdapter, + sidecar: &mut RecoverySidecar, +) -> Result<()> { + if !matches!(sidecar.writer_kind, SidecarKind::SchemaApply) { + return Err(OmniError::manifest_internal(format!( + "schema publish confirmation requires SchemaApply sidecar, found {:?}", + sidecar.writer_kind, + ))); + } + if sidecar.schema_version != SCHEMA_APPLY_CONFIRMATION_SCHEMA_VERSION { + return Err(OmniError::manifest_internal(format!( + "SchemaApply sidecar '{}' uses schema-v{}, expected schema-v{} confirmation", + sidecar.operation_id, sidecar.schema_version, SCHEMA_APPLY_CONFIRMATION_SCHEMA_VERSION, + ))); + } + if sidecar.schema_apply_target_schema_ir_hash.is_none() { + return Err(OmniError::manifest_internal(format!( + "SchemaApply sidecar '{}' has no target schema identity to confirm", + sidecar.operation_id, + ))); + } + crate::failpoints::maybe_fail(crate::failpoints::names::RECOVERY_SIDECAR_CONFIRM)?; + sidecar.schema_apply_manifest_published = true; + let uri = sidecar_uri(root_uri, &sidecar.operation_id); + validate_sidecar_shape(&uri, sidecar)?; + let json = serde_json::to_string_pretty(sidecar).map_err(|error| { + OmniError::manifest_internal(format!( + "failed to serialize SchemaApply recovery confirmation: {error}" + )) + })?; + storage.write_text(&uri, &json).await +} + /// Phase-B confirmation: stamp each pin with the exact Lance HEAD its publish /// reached, then re-write the sidecar in place (same object). Called once, after /// the writer's whole multi-commit publish completed and before the manifest @@ -986,6 +1088,38 @@ fn validate_sidecar_shape(sidecar_uri: &str, sidecar: &RecoverySidecar) -> Resul )) }; + let carries_schema_apply_confirmation = sidecar.schema_apply_manifest_published + || sidecar.schema_apply_target_schema_ir_hash.is_some(); + if carries_schema_apply_confirmation + && sidecar.schema_version != SCHEMA_APPLY_CONFIRMATION_SCHEMA_VERSION + { + return Err(malformed(format!( + "SchemaApply confirmation fields require schema-v{}, found schema-v{}", + SCHEMA_APPLY_CONFIRMATION_SCHEMA_VERSION, sidecar.schema_version, + ))); + } + if carries_schema_apply_confirmation && !matches!(sidecar.writer_kind, SidecarKind::SchemaApply) + { + return Err(malformed( + "SchemaApply confirmation fields are present on another writer kind".to_string(), + )); + } + if sidecar.schema_apply_manifest_published + && sidecar.schema_apply_target_schema_ir_hash.is_none() + { + return Err(malformed( + "SchemaApply manifest confirmation has no target schema identity".to_string(), + )); + } + if sidecar.ensure_indices_rollback_v6.is_some() + && sidecar.schema_version != ENSURE_INDICES_ROLLBACK_SCHEMA_VERSION + { + return Err(malformed(format!( + "EnsureIndices fixed rollback requires schema-v{}, found schema-v{}", + ENSURE_INDICES_ROLLBACK_SCHEMA_VERSION, sidecar.schema_version, + ))); + } + if sidecar.schema_version < EXACT_EFFECT_IDENTITY_SCHEMA_VERSION { if sidecar.protocol_v3.is_some() || sidecar.protocol_v4.is_some() { return Err(malformed( @@ -999,6 +1133,34 @@ fn validate_sidecar_shape(sidecar_uri: &str, sidecar: &RecoverySidecar) -> Resul return validate_branch_merge_v4_shape(sidecar_uri, sidecar); } + if sidecar.schema_version == SCHEMA_APPLY_CONFIRMATION_SCHEMA_VERSION { + if !matches!(sidecar.writer_kind, SidecarKind::SchemaApply) { + return Err(malformed(format!( + "schema-v5 is reserved for SchemaApply, found {:?}", + sidecar.writer_kind, + ))); + } + if sidecar.branch.is_some() + || sidecar.protocol_v3.is_some() + || sidecar.protocol_v4.is_some() + { + return Err(malformed( + "schema-v5 SchemaApply must target main and cannot carry v3/v4 protocols" + .to_string(), + )); + } + if sidecar.schema_apply_target_schema_ir_hash.is_none() { + return Err(malformed( + "schema-v5 SchemaApply is missing its target schema identity".to_string(), + )); + } + return Ok(()); + } + + if sidecar.schema_version == ENSURE_INDICES_ROLLBACK_SCHEMA_VERSION { + return validate_ensure_indices_v6_shape(sidecar_uri, sidecar); + } + if sidecar.protocol_v4.is_some() { return Err(malformed( "protocol_v4 is present on a pre-v4 sidecar".to_string(), @@ -1172,6 +1334,62 @@ fn validate_sidecar_shape(sidecar_uri: &str, sidecar: &RecoverySidecar) -> Resul Ok(()) } +fn validate_ensure_indices_v6_shape(sidecar_uri: &str, sidecar: &RecoverySidecar) -> Result<()> { + let malformed = |reason: String| { + OmniError::manifest_internal(format!( + "recovery sidecar at '{}' has an invalid schema-v{} shape: {}", + sidecar_uri, sidecar.schema_version, reason + )) + }; + if !matches!(sidecar.writer_kind, SidecarKind::EnsureIndices) { + return Err(malformed(format!( + "schema-v6 is reserved for EnsureIndices, found {:?}", + sidecar.writer_kind, + ))); + } + if sidecar.protocol_v3.is_some() + || sidecar.protocol_v4.is_some() + || sidecar.merge_source_commit_id.is_some() + || !sidecar.additional_registrations.is_empty() + || !sidecar.tombstones.is_empty() + { + return Err(malformed( + "schema-v6 EnsureIndices cannot carry exact, merge, registration, or tombstone fields" + .to_string(), + )); + } + let protocol = sidecar.ensure_indices_rollback_v6.as_ref().ok_or_else(|| { + malformed("schema-v6 EnsureIndices is missing its fixed rollback payload".to_string()) + })?; + if protocol.rollback_graph_commit_id.is_empty() { + return Err(malformed( + "schema-v6 EnsureIndices has an empty rollback commit id".to_string(), + )); + } + let pin_keys: HashSet<&str> = sidecar + .tables + .iter() + .map(|pin| pin.table_key.as_str()) + .collect(); + if sidecar.tables.is_empty() || pin_keys.len() != sidecar.tables.len() { + return Err(malformed( + "schema-v6 EnsureIndices requires unique non-empty table pins".to_string(), + )); + } + if let Some(outcomes) = protocol.rollback_audit_outcomes.as_ref() { + let outcome_keys: HashSet<&str> = outcomes + .iter() + .map(|outcome| outcome.table_key.as_str()) + .collect(); + if outcome_keys.len() != outcomes.len() || !outcome_keys.is_subset(&pin_keys) { + return Err(malformed( + "rollback audit outcomes must name a unique subset of table pins".to_string(), + )); + } + } + Ok(()) +} + fn validate_branch_merge_v4_shape(sidecar_uri: &str, sidecar: &RecoverySidecar) -> Result<()> { let malformed = |reason: String| { OmniError::manifest_internal(format!( @@ -2041,10 +2259,11 @@ pub(crate) fn schema_apply_serial_queue_key() -> crate::db::write_queue::TableQu /// state), or abort (invariant violation). /// /// Idempotency: a crash mid-sweep leaves the sidecar (deletion is the final -/// step). v3 persists an exact rollback audit plan before restoring and uses a +/// step). v3/v4 persist an ownership-exact rollback audit plan before restoring; +/// schema-v6 EnsureIndices persists its loose-classifier plan. All three use a /// fixed rollback commit id, so re-entry resumes that outcome without another -/// restore or synthetic commit. Legacy sidecars re-classify and may append an -/// extra restore commit, which `omnigraph cleanup` reclaims. +/// restore or synthetic commit. Older legacy sidecars re-classify and may append +/// an extra restore commit, which `omnigraph cleanup` reclaims. /// /// Concurrency: a newly-opening handle is not yet published, but another handle /// for the same root may already be serving writes. Every handle obtains the @@ -2166,6 +2385,11 @@ async fn process_sidecar( return finalize_visible_v3_outcome(root_uri, storage.as_ref(), sidecar, outcome).await; } } + if sidecar.ensure_indices_rollback_v6.is_some() + && detect_visible_ensure_indices_rollback(root_uri, sidecar).await? + { + return finalize_visible_ensure_indices_rollback(root_uri, storage.as_ref(), sidecar).await; + } let mut states = Vec::with_capacity(sidecar.tables.len()); for pin in &sidecar.tables { let manifest_pinned = snapshot @@ -2179,7 +2403,8 @@ async fn process_sidecar( // target ref not existing yet: that is the crash-before-fork state, not // storage corruption. Once effects are confirmed, a missing ref is // impossible and remains a loud error. - let unpublished_fork = sidecar.protocol_v3.is_some() + let unpublished_fork = (sidecar.protocol_v3.is_some() + || matches!(sidecar.writer_kind, SidecarKind::EnsureIndices)) && pin .table_branch .as_deref() @@ -2190,10 +2415,11 @@ async fn process_sidecar( .map(|entry| entry.table_branch != pin.table_branch) .unwrap_or(true); let allow_missing_target_ref = unpublished_fork - && sidecar - .protocol_v3 - .as_ref() - .is_some_and(|protocol| protocol.effect_phase == RecoveryEffectPhase::Armed); + && (matches!(sidecar.writer_kind, SidecarKind::EnsureIndices) + || sidecar + .protocol_v3 + .as_ref() + .is_some_and(|protocol| protocol.effect_phase == RecoveryEffectPhase::Armed)); let planned_effect = sidecar.protocol_v3.as_ref().and_then(|protocol| { protocol .effects @@ -2465,7 +2691,59 @@ async fn process_sidecar( .iter() .zip(states.iter()) .any(|(pin, state)| state.manifest_pinned > pin.expected_version); + if matches!(sidecar.writer_kind, SidecarKind::EnsureIndices) + && all_no_movement + && !any_pin_advanced + { + if matches!(mode, RecoveryMode::RollForwardOnly) { + return Ok(false); + } + if matches!( + cleanup_unpublished_no_effect_forks( + root_uri, + storage.as_ref(), + sidecar, + &states, + ) + .await?, + NoEffectForkCleanup::DeferredPathChild { .. } + ) { + return Ok(false); + } + delete_sidecar_by_operation_id(root_uri, storage.as_ref(), &sidecar.operation_id) + .await?; + return Ok(true); + } if all_no_movement && any_pin_advanced { + if matches!(sidecar.writer_kind, SidecarKind::SchemaApply) + && sidecar.schema_version == SCHEMA_APPLY_CONFIRMATION_SCHEMA_VERSION + && !schema_apply_target_identity_is_live(root_uri, storage.as_ref(), sidecar) + .await? + { + if sidecar.schema_apply_manifest_published { + let message = format!( + "SchemaApply sidecar '{}' is manifest-aligned, but the live schema \ + identity does not match its confirmed target", + sidecar.operation_id, + ); + return match mode { + RecoveryMode::RollForwardOnly => { + warn!(operation_id = sidecar.operation_id.as_str(), "{message}"); + Ok(false) + } + RecoveryMode::Full => Err(OmniError::manifest_internal(message)), + }; + } + // A marker-false v5 sidecar whose live schema is still the + // old contract is a rollback (possibly recovery re-entry + // after its rollback publish), never stale roll-forward. + if matches!(mode, RecoveryMode::RollForwardOnly) { + return Ok(false); + } + return roll_back_sidecar(root_uri, storage.as_ref(), sidecar, &states) + .await + .map(|()| true); + } if sidecar.protocol_v3.is_some() { // Fixed outcome ids were checked before table // classification. Alignment without either id is not proof @@ -2502,6 +2780,7 @@ async fn process_sidecar( storage.as_ref(), sidecar, &states, + snapshot, ) .await .map(|()| true); @@ -2531,26 +2810,85 @@ async fn process_sidecar( if matches!(sidecar.writer_kind, SidecarKind::SchemaApply) && !schema_state_recovery.completed_schema_apply_sidecar_rename() { - return match mode { - RecoveryMode::Full => { + if sidecar.schema_version == SCHEMA_APPLY_CONFIRMATION_SCHEMA_VERSION { + let target_is_live = + schema_apply_target_identity_is_live(root_uri, storage.as_ref(), sidecar) + .await?; + if target_is_live && sidecar.schema_apply_manifest_published { + // Phase C and final schema promotion both landed; only + // audit/delete bookkeeping remains. warn!( operation_id = sidecar.operation_id.as_str(), - "recovery: rolling back SchemaApply sidecar because schema staging \ - files were not promoted in this recovery pass" + "recovery: cleaning up stale SchemaApply sidecar after its manifest \ + publish and schema promotion completed" ); - roll_back_sidecar(root_uri, storage.as_ref(), sidecar, &states) - .await - .map(|()| true) + return record_audit_recovery_rollforward( + root_uri, + storage.as_ref(), + sidecar, + &states, + snapshot, + ) + .await + .map(|()| true); } - RecoveryMode::RollForwardOnly => { - warn!( - operation_id = sidecar.operation_id.as_str(), - "recovery: deferring SchemaApply sidecar because schema staging files \ - were not promoted in this recovery pass" + if target_is_live { + // Recovery may have promoted staging and crashed before + // processing the sidecar. Marker=false proves original + // Phase C is not known to have landed, so fall through + // to normal roll_forward_all to publish registrations, + // tombstones, and table pins. + } else if sidecar.schema_apply_manifest_published { + let message = format!( + "SchemaApply sidecar '{}' confirms manifest publish, but the live \ + schema identity does not match its target; refusing to guess whether \ + schema promotion completed", + sidecar.operation_id, ); - Ok(false) + return match mode { + RecoveryMode::RollForwardOnly => { + warn!(operation_id = sidecar.operation_id.as_str(), "{message}"); + Ok(false) + } + RecoveryMode::Full => Err(OmniError::manifest_internal(message)), + }; + } else { + return match mode { + RecoveryMode::Full => { + warn!( + operation_id = sidecar.operation_id.as_str(), + "recovery: rolling back SchemaApply sidecar because schema \ + staging was not promoted" + ); + roll_back_sidecar(root_uri, storage.as_ref(), sidecar, &states) + .await + .map(|()| true) + } + RecoveryMode::RollForwardOnly => Ok(false), + }; } - }; + } else { + return match mode { + RecoveryMode::Full => { + warn!( + operation_id = sidecar.operation_id.as_str(), + "recovery: rolling back SchemaApply sidecar because schema staging \ + files were not promoted in this recovery pass" + ); + roll_back_sidecar(root_uri, storage.as_ref(), sidecar, &states) + .await + .map(|()| true) + } + RecoveryMode::RollForwardOnly => { + warn!( + operation_id = sidecar.operation_id.as_str(), + "recovery: deferring SchemaApply sidecar because schema staging \ + files were not promoted in this recovery pass" + ); + Ok(false) + } + }; + } } warn!( operation_id = sidecar.operation_id.as_str(), @@ -2646,6 +2984,24 @@ async fn process_sidecar( } } +async fn schema_apply_target_identity_is_live( + root_uri: &str, + storage: &dyn StorageAdapter, + sidecar: &RecoverySidecar, +) -> Result { + let target_hash = sidecar + .schema_apply_target_schema_ir_hash + .as_deref() + .ok_or_else(|| { + OmniError::manifest_internal(format!( + "SchemaApply sidecar '{}' has no target schema identity", + sidecar.operation_id, + )) + })?; + let live_schema = read_schema_state_identity(root_uri, storage).await?; + Ok(live_schema.schema_ir_hash == target_hash) +} + struct BranchMergeRefObservation { dataset: lance::Dataset, version: u64, @@ -3447,6 +3803,10 @@ fn has_exact_protocol(sidecar: &RecoverySidecar) -> bool { sidecar.protocol_v3.is_some() || sidecar.protocol_v4.is_some() } +fn has_fixed_rollback_identity(sidecar: &RecoverySidecar) -> bool { + has_exact_protocol(sidecar) || sidecar.ensure_indices_rollback_v6.is_some() +} + fn v4_effect_for<'a>( sidecar: &'a RecoverySidecar, table_key: &str, @@ -3466,7 +3826,8 @@ fn first_touch_fork_version(sidecar: &RecoverySidecar, pin: &SidecarTablePin) -> } /// Remove first-touch named-branch refs created by an Armed exact-protocol -/// attempt that never completed this table's planned effect. +/// attempt, or by the legacy EnsureIndices adapter, that never completed this +/// table's planned effect. /// /// The sidecar is durable before the ref is created, so it is the ownership /// record while the manifest still inherits the table from another branch. @@ -3483,7 +3844,7 @@ async fn cleanup_unpublished_no_effect_forks( sidecar: &RecoverySidecar, states: &[ClassifiedTable], ) -> Result { - if !has_exact_protocol(sidecar) { + if !has_exact_protocol(sidecar) && !matches!(sidecar.writer_kind, SidecarKind::EnsureIndices) { return Ok(NoEffectForkCleanup::Complete); } @@ -3492,6 +3853,16 @@ async fn cleanup_unpublished_no_effect_forks( if !state.unpublished_fork || state.effect_ownership != EffectOwnership::None { continue; } + // Legacy EnsureIndices has no transaction-identity ownership signal. + // Its loose classifier can still prove the no-effect case exactly: + // only `NoMovement` is an untouched fork. A moved HEAD must flow into + // the normal derived-state rollback path; treating it as a no-effect + // ref would try to delete a ref past its fork point and wedge recovery. + if matches!(sidecar.writer_kind, SidecarKind::EnsureIndices) + && !matches!(state.classification, TableClassification::NoMovement) + { + continue; + } let Some(target_branch) = pin .table_branch .as_deref() @@ -3640,12 +4011,32 @@ fn exact_rollback_audit_outcomes( .collect() } -/// Durably bind the exact audit payload for a v3/v4 rollback before the first -/// restore or rollback publish. A retry must replay the original observation, -/// not reconstruct it from post-restore HEADs. In particular, partial -/// multi-table attempts omit untouched pins, and a preflight-rebased v3 effect -/// records its actual `own_head -> current_manifest_pin` compensation. -async fn prepare_exact_rollback_audit_plan( +fn rollback_audit_outcomes_for_plan( + sidecar: &RecoverySidecar, + states: &[ClassifiedTable], +) -> Vec { + if has_exact_protocol(sidecar) { + return exact_rollback_audit_outcomes(sidecar, states); + } + sidecar + .tables + .iter() + .zip(states.iter()) + .filter(|(_, state)| table_requires_rollback_effect(state)) + .map(|(pin, state)| TableOutcome { + table_key: pin.table_key.clone(), + from_version: state.lance_head, + to_version: state.manifest_pinned, + }) + .collect() +} + +/// Durably bind the audit payload for every fixed-identity rollback before the +/// first restore or rollback publish. A retry must replay the original +/// observation, not reconstruct it from post-restore HEADs. v3/v4 filter by +/// exact ownership; schema-v6 EnsureIndices deliberately retains loose table +/// classification while making only the rollback outcome exact. +async fn prepare_fixed_rollback_audit_plan( root_uri: &str, storage: &dyn StorageAdapter, sidecar: &RecoverySidecar, @@ -3662,26 +4053,34 @@ async fn prepare_exact_rollback_audit_plan( .as_ref() .and_then(|protocol| protocol.rollback_audit_outcomes.as_ref()) }) + .or_else(|| { + prepared + .ensure_indices_rollback_v6 + .as_ref() + .and_then(|protocol| protocol.rollback_audit_outcomes.as_ref()) + }) .is_some(); if already_prepared { return Ok(prepared); } - let outcomes = exact_rollback_audit_outcomes(sidecar, states); + let outcomes = rollback_audit_outcomes_for_plan(sidecar, states); if let Some(protocol) = prepared.protocol_v3.as_mut() { protocol.rollback_audit_outcomes = Some(outcomes); } else if let Some(protocol) = prepared.protocol_v4.as_mut() { protocol.rollback_audit_outcomes = Some(outcomes); + } else if let Some(protocol) = prepared.ensure_indices_rollback_v6.as_mut() { + protocol.rollback_audit_outcomes = Some(outcomes); } else { return Err(OmniError::manifest_internal( - "prepare_exact_rollback_audit_plan called for a legacy sidecar", + "prepare_fixed_rollback_audit_plan called without a fixed rollback identity", )); } let uri = sidecar_uri(root_uri, &prepared.operation_id); validate_sidecar_shape(&uri, &prepared)?; let json = serde_json::to_string_pretty(&prepared).map_err(|error| { OmniError::manifest_internal(format!( - "failed to serialize exact rollback audit plan for sidecar '{}': {}", + "failed to serialize fixed rollback audit plan for sidecar '{}': {}", prepared.operation_id, error )) })?; @@ -3725,12 +4124,12 @@ async fn roll_back_sidecar( // longer has pre-restore table observations. Persist the exact audit plan // after fork cleanup and before the first restore so that path can replay it // without fabricating outcomes from pins. - let prepared_exact = if has_exact_protocol(sidecar) { - Some(prepare_exact_rollback_audit_plan(root_uri, storage, sidecar, states).await?) + let prepared_fixed = if has_fixed_rollback_identity(sidecar) { + Some(prepare_fixed_rollback_audit_plan(root_uri, storage, sidecar, states).await?) } else { None }; - let sidecar = prepared_exact.as_ref().unwrap_or(sidecar); + let sidecar = prepared_fixed.as_ref().unwrap_or(sidecar); // Restore every drifted table (RolledPastExpected / UnexpectedAtP1 / // UnexpectedMultistep) to its manifest-pinned content, then PUBLISH so @@ -3824,6 +4223,12 @@ async fn roll_back_sidecar( .as_ref() .and_then(|protocol| protocol.rollback_audit_outcomes.clone()) }) + .or_else(|| { + sidecar + .ensure_indices_rollback_v6 + .as_ref() + .and_then(|protocol| protocol.rollback_audit_outcomes.clone()) + }) .unwrap_or(outcomes); record_audit( root_uri, @@ -3856,8 +4261,9 @@ async fn record_audit_recovery_rollforward( storage: &dyn StorageAdapter, sidecar: &RecoverySidecar, states: &[ClassifiedTable], + snapshot: &Snapshot, ) -> Result<()> { - let outcomes: Vec = sidecar + let mut outcomes: Vec = sidecar .tables .iter() .zip(states.iter()) @@ -3867,6 +4273,16 @@ async fn record_audit_recovery_rollforward( to_version: state.manifest_pinned, }) .collect(); + for registration in &sidecar.additional_registrations { + outcomes.push(TableOutcome { + table_key: registration.table_key.clone(), + from_version: 0, + to_version: snapshot + .entry(®istration.table_key) + .map(|entry| entry.table_version) + .unwrap_or(0), + }); + } // The substrate is already in the post-roll-forward state (the prior pass's // table re-pin landed), so there are no table `updates` — but a recovery // commit is still recorded for this cleanup pass via a lineage-only publish @@ -3891,6 +4307,84 @@ async fn record_audit_recovery_rollforward( Ok(()) } +/// A schema-v6 EnsureIndices sidecar has no fixed original/forward identity, +/// but its rollback id is pre-minted before effects. Seeing that id in the +/// authoritative lineage is therefore exact proof that compensation already +/// published, even though the physical index effects remain loosely matched. +async fn detect_visible_ensure_indices_rollback( + root_uri: &str, + sidecar: &RecoverySidecar, +) -> Result { + let protocol = sidecar + .ensure_indices_rollback_v6 + .as_ref() + .expect("caller checked ensure_indices_rollback_v6"); + let (commits, _) = + ManifestCoordinator::read_graph_lineage_at(root_uri, sidecar.branch.as_deref()).await?; + let Some(commit) = commits + .iter() + .find(|commit| commit.graph_commit_id == protocol.rollback_graph_commit_id) + else { + return Ok(false); + }; + let expected_branch = sidecar.branch.as_deref().filter(|branch| *branch != "main"); + let expected_created_at = sidecar.started_at.parse::().map_err(|error| { + OmniError::manifest_internal(format!( + "EnsureIndices recovery sidecar '{}' has invalid started_at '{}': {}", + sidecar.operation_id, sidecar.started_at, error, + )) + })?; + if commit.manifest_branch.as_deref() != expected_branch + || commit.actor_id.as_deref() != Some(RECOVERY_ACTOR) + || commit.merged_parent_commit_id.is_some() + || commit.created_at != expected_created_at + { + return Err(OmniError::manifest_internal(format!( + "EnsureIndices recovery sidecar '{}' found rollback commit id '{}' with mismatched lineage", + sidecar.operation_id, protocol.rollback_graph_commit_id, + ))); + } + Ok(true) +} + +async fn finalize_visible_ensure_indices_rollback( + root_uri: &str, + storage: &dyn StorageAdapter, + sidecar: &RecoverySidecar, +) -> Result { + let protocol = sidecar + .ensure_indices_rollback_v6 + .as_ref() + .expect("caller checked ensure_indices_rollback_v6"); + let outcomes = protocol.rollback_audit_outcomes.clone().ok_or_else(|| { + OmniError::manifest_internal(format!( + "EnsureIndices recovery sidecar '{}' has visible fixed rollback commit '{}' but no durable rollback audit outcomes", + sidecar.operation_id, protocol.rollback_graph_commit_id, + )) + })?; + let mut audit = RecoveryAudit::open(root_uri).await?; + let already_recorded = audit.list().await?.iter().any(|record| { + record.operation_id == sidecar.operation_id + && record.recovery_kind == RecoveryKind::RolledBack + }); + if !already_recorded { + crate::failpoints::maybe_fail(crate::failpoints::names::RECOVERY_RECORD_AUDIT)?; + audit + .append(RecoveryAuditRecord { + graph_commit_id: protocol.rollback_graph_commit_id.clone(), + recovery_kind: RecoveryKind::RolledBack, + recovery_for_actor: sidecar.actor_id.clone(), + operation_id: sidecar.operation_id.clone(), + sidecar_writer_kind: format!("{:?}", sidecar.writer_kind), + per_table_outcomes: outcomes, + created_at: crate::db::now_micros()?, + }) + .await?; + } + delete_sidecar_by_operation_id(root_uri, storage, &sidecar.operation_id).await?; + Ok(true) +} + /// Finalize an exact-protocol sidecar whose tables are already manifest-aligned. /// /// Numeric table alignment alone is not evidence of WHICH outcome happened: @@ -4483,9 +4977,10 @@ async fn push_table_update( /// CAS as the table re-pin), so this only writes the `_graph_commit_recoveries` /// row, referencing that commit by `graph_commit_id`. A crash between the /// recovery publish and this audit append leaves a recovery commit with no audit -/// row. For v3, re-entry detects the fixed original/rollback commit and replays -/// its durable exact audit payload without minting another commit. Legacy -/// sidecars retain their stale-sidecar cleanup behavior. +/// row. v3/v4 re-entry detects the fixed original/rollback commit and replays +/// its durable exact audit payload; schema-v6 EnsureIndices does the same for +/// its fixed rollback outcome. Older legacy sidecars retain their stale-sidecar +/// cleanup behavior. async fn record_audit( root_uri: &str, sidecar: &RecoverySidecar, @@ -4703,11 +5198,31 @@ pub(crate) fn new_sidecar( merge_source_commit_id: None, additional_registrations: Vec::new(), tombstones: Vec::new(), + schema_apply_manifest_published: false, + schema_apply_target_schema_ir_hash: None, protocol_v3: None, protocol_v4: None, + ensure_indices_rollback_v6: None, } } +/// Arm the narrow schema-v6 EnsureIndices recovery bridge. Index effects still +/// use the legacy loose classifier, but compensation gets a stable lineage id +/// so a crash after rollback publish cannot be misread as roll-forward. +pub(crate) fn new_ensure_indices_sidecar( + branch: Option, + tables: Vec, +) -> Result { + let mut sidecar = new_sidecar(SidecarKind::EnsureIndices, branch, None, tables); + sidecar.schema_version = ENSURE_INDICES_ROLLBACK_SCHEMA_VERSION; + sidecar.ensure_indices_rollback_v6 = Some(RecoveryEnsureIndicesRollbackV6 { + rollback_graph_commit_id: ulid::Ulid::new().to_string(), + rollback_audit_outcomes: None, + }); + validate_sidecar_shape("", &sidecar)?; + Ok(sidecar) +} + /// Arm an RFC-022 mutation/load recovery sidecar. /// /// The returned v3 sidecar is still purely an intent (`Armed`): every staged @@ -4779,6 +5294,8 @@ pub(crate) fn new_occ_sidecar( merge_source_commit_id: None, additional_registrations: Vec::new(), tombstones: Vec::new(), + schema_apply_manifest_published: false, + schema_apply_target_schema_ir_hash: None, protocol_v3: Some(RecoveryProtocolV3 { authority, lineage, @@ -4793,6 +5310,7 @@ pub(crate) fn new_occ_sidecar( }, }), protocol_v4: None, + ensure_indices_rollback_v6: None, }; validate_sidecar_shape("", &sidecar)?; Ok(sidecar) @@ -4963,6 +5481,8 @@ pub(crate) fn new_branch_merge_sidecar( merge_source_commit_id: None, additional_registrations: Vec::new(), tombstones: Vec::new(), + schema_apply_manifest_published: false, + schema_apply_target_schema_ir_hash: None, protocol_v3: None, protocol_v4: Some(RecoveryProtocolV4 { authority, @@ -4973,6 +5493,7 @@ pub(crate) fn new_branch_merge_sidecar( effects, intended_delta, }), + ensure_indices_rollback_v6: None, }; validate_sidecar_shape("", &sidecar)?; Ok(sidecar) @@ -5316,6 +5837,63 @@ mod tests { assert_eq!(parsed.tables[0].table_key, "node:Person"); } + #[test] + fn schema_apply_confirmation_fields_are_version_gated() { + let legacy = new_sidecar(SidecarKind::SchemaApply, None, None, Vec::new()); + let legacy_json = serde_json::to_string(&legacy).unwrap(); + parse_sidecar("file:///tmp/__recovery/legacy-schema.json", &legacy_json) + .expect("plain schema-v2 SchemaApply remains readable"); + + let mut wrong_version = legacy.clone(); + wrong_version.schema_apply_target_schema_ir_hash = Some("target-hash".to_string()); + let wrong_json = serde_json::to_string(&wrong_version).unwrap(); + let error = parse_sidecar( + "file:///tmp/__recovery/wrong-version-schema.json", + &wrong_json, + ) + .expect_err("schema-v2 must not opt into schema-v5 confirmation semantics"); + assert!(error.to_string().contains("require schema-v5")); + + wrong_version.schema_version = SCHEMA_APPLY_CONFIRMATION_SCHEMA_VERSION; + let v5_json = serde_json::to_string(&wrong_version).unwrap(); + parse_sidecar("file:///tmp/__recovery/schema-v5.json", &v5_json) + .expect("schema-v5 accepts a target identity before Phase C confirmation"); + } + + #[test] + fn ensure_indices_fixed_rollback_is_version_gated() { + let pin = make_pin("node:Person", "file:///tmp/people.lance", 5, 6); + let legacy = new_sidecar( + SidecarKind::EnsureIndices, + Some("feature".to_string()), + None, + vec![pin.clone()], + ); + let legacy_json = serde_json::to_string(&legacy).unwrap(); + parse_sidecar("file:///tmp/__recovery/legacy-index.json", &legacy_json) + .expect("plain schema-v2 EnsureIndices remains readable"); + + let current = new_ensure_indices_sidecar(Some("feature".to_string()), vec![pin]).unwrap(); + assert_eq!( + current.schema_version, + ENSURE_INDICES_ROLLBACK_SCHEMA_VERSION + ); + assert!(current.ensure_indices_rollback_v6.is_some()); + let current_json = serde_json::to_string(¤t).unwrap(); + parse_sidecar("file:///tmp/__recovery/index-v6.json", ¤t_json) + .expect("schema-v6 accepts the fixed rollback payload"); + + let mut wrong_version = current; + wrong_version.schema_version = LEGACY_SIDECAR_SCHEMA_VERSION; + let wrong_json = serde_json::to_string(&wrong_version).unwrap(); + let error = parse_sidecar( + "file:///tmp/__recovery/wrong-version-index.json", + &wrong_json, + ) + .expect_err("schema-v2 must not opt into schema-v6 rollback semantics"); + assert!(error.to_string().contains("requires schema-v6")); + } + #[test] fn occ_sidecar_round_trips_armed_with_stable_ids_and_exact_effect() { let original = occ_sidecar(); diff --git a/crates/omnigraph/src/db/omnigraph.rs b/crates/omnigraph/src/db/omnigraph.rs index 132c7d8d..f59a1822 100644 --- a/crates/omnigraph/src/db/omnigraph.rs +++ b/crates/omnigraph/src/db/omnigraph.rs @@ -47,8 +47,7 @@ pub(crate) use table_ops::{DeferredTableFork, OpenedForMutation}; use super::commit_graph::GraphCommit; use super::manifest::{ - ManifestChange, Snapshot, SubTableEntry, TableRegistration, TableTombstone, - table_path_for_table_key, + ManifestChange, Snapshot, TableRegistration, TableTombstone, table_path_for_table_key, }; use super::schema_state::{ SCHEMA_SOURCE_FILENAME, load_or_bootstrap_schema_contract, load_validated_schema_contract, @@ -514,9 +513,7 @@ impl Omnigraph { ) .await?; } - crate::failpoints::maybe_fail( - crate::failpoints::names::OPEN_BEFORE_SCHEMA_CONTRACT_READ, - )?; + crate::failpoints::maybe_fail(crate::failpoints::names::OPEN_BEFORE_SCHEMA_CONTRACT_READ)?; // Read _schema.pg (post-recovery — may have just been renamed in). let schema_path = schema_source_uri(&root); let schema_source = storage.read_text(&schema_path).await?; @@ -684,9 +681,7 @@ impl Omnigraph { /// for planning and conservative table-gate enumeration. The caller MUST /// already hold `schema_apply_serial_queue_key`; this helper does not acquire /// it because the gate is a non-reentrant mutex. - pub(crate) async fn load_accepted_catalog_with_schema_gate_held( - &self, - ) -> Result> { + pub(crate) async fn load_accepted_catalog_with_schema_gate_held(&self) -> Result> { let (schema_ir, _) = load_validated_schema_contract(self.uri(), Arc::clone(&self.storage)).await?; let mut catalog = build_catalog_from_ir(&schema_ir)?; @@ -1346,10 +1341,7 @@ impl Omnigraph { /// schema -> target branch -> every accepted-catalog table gate before the /// ref mutation, which waits out any live in-process owner. SchemaApply /// remains graph-global and must still block deletion. - async fn heal_pending_recovery_sidecars_for_branch_delete( - &self, - branch: &str, - ) -> Result<()> { + async fn heal_pending_recovery_sidecars_for_branch_delete(&self, branch: &str) -> Result<()> { let outcome = self.heal_pending_recovery_sidecars_outcome().await?; if let Some(intent) = outcome .unresolved @@ -1383,14 +1375,11 @@ impl Omnigraph { let sidecars = crate::db::manifest::list_sidecars(&self.root_uri, self.storage.as_ref()).await?; let blocking = sidecars.iter().find(|sidecar| { - let sidecar_branch = sidecar - .branch - .as_deref() - .filter(|branch| *branch != "main"); + let sidecar_branch = sidecar.branch.as_deref().filter(|branch| *branch != "main"); sidecar.writer_kind == crate::db::manifest::SidecarKind::SchemaApply - || relevant_branches.iter().any(|branch| { - branch.filter(|name| *name != "main") == sidecar_branch - }) + || relevant_branches + .iter() + .any(|branch| branch.filter(|name| *name != "main") == sidecar_branch) }); if let Some(sidecar) = blocking { return Err(OmniError::recovery_required( @@ -1405,20 +1394,90 @@ impl Omnigraph { Ok(()) } + /// Final pre-arm ownership check for an existing physical table ref. + /// + /// Callers must already hold their complete schema -> branch -> table gate + /// envelope and must invoke this before writing their own recovery sidecar. + /// The manifest pin is the logical authority; a live Lance HEAD ahead of it + /// is either owned by an older recovery intent or is uncovered drift that + /// requires explicit operator repair. A new writer must never manufacture a + /// sidecar that retroactively claims that pre-existing physical effect. + /// + /// First-touch refs deliberately do not call this helper: their target ref + /// does not exist until after the writer's recovery intent is durable. + pub(crate) async fn ensure_existing_effect_baseline( + &self, + table_key: &str, + table_branch: Option<&str>, + expected_version: u64, + dataset: &SnapshotHandle, + ) -> Result<()> { + let head = dataset + .dataset() + .latest_version_id() + .await + .map_err(|error| OmniError::Lance(error.to_string()))?; + if head < expected_version { + return Err(OmniError::manifest_internal(format!( + "table '{}' Lance HEAD version {} is behind manifest version {}", + table_key, head, expected_version, + ))); + } + if head == expected_version { + return Ok(()); + } + + let normalized_branch = table_branch.filter(|branch| *branch != "main"); + let sidecars = + crate::db::manifest::list_sidecars(self.root_uri(), self.storage_adapter()).await; + match sidecars { + Ok(sidecars) => { + if let Some(owner) = sidecars.iter().find(|sidecar| { + sidecar.tables.iter().any(|pin| { + pin.table_key == table_key + && pin + .table_branch + .as_deref() + .filter(|branch| *branch != "main") + == normalized_branch + }) + }) { + return Err(OmniError::recovery_required( + owner.operation_id.clone(), + format!( + "table '{}' has Lance HEAD version {} ahead of manifest version {}; \ + the pending recovery operation owns this drift", + table_key, head, expected_version, + ), + )); + } + Err(OmniError::manifest_conflict(format!( + "table '{}' has Lance HEAD version {} ahead of manifest version {}; \ + run `omnigraph repair` before writing", + table_key, head, expected_version, + ))) + } + Err(list_error) => Err(OmniError::manifest_conflict(format!( + "table '{}' has Lance HEAD version {} ahead of manifest version {}; could not \ + classify the drift (sidecar listing failed: {}); run `omnigraph repair`, or \ + reopen the graph read-write if repair reports a pending recovery sidecar", + table_key, head, expected_version, list_error, + ))), + } + } + /// Final under-gate check for branch deletion. Target-branch sidecars are /// intentionally allowed: the held complete table envelope proves no live /// in-process owner can still be applying them, and deleting the branch /// makes their effects unreachable. Only graph-global schema recovery can /// still invalidate the operation. - async fn ensure_branch_delete_recovery_safe_under_gates( - &self, - branch: &str, - ) -> Result<()> { + async fn ensure_branch_delete_recovery_safe_under_gates(&self, branch: &str) -> Result<()> { let sidecars = crate::db::manifest::list_sidecars(&self.root_uri, self.storage.as_ref()).await?; - if let Some(sidecar) = sidecars.iter().find(|sidecar| { - sidecar.writer_kind == crate::db::manifest::SidecarKind::SchemaApply - }) { + if let Some(sidecar) = sidecars + .iter() + .find(|sidecar| sidecar.writer_kind == crate::db::manifest::SidecarKind::SchemaApply) + { return Err(OmniError::recovery_required( sidecar.operation_id.clone(), format!( @@ -1693,15 +1752,18 @@ impl Omnigraph { /// Ensure BTree scalar indices exist on key columns. /// Idempotent — Lance skips if index already exists. /// - /// Opens sub-tables at their latest version (not snapshot-pinned) because - /// indices must be created on the current head. Any version drift from the - /// snapshot is expected and logged. The resulting versions are committed - /// back to the manifest. + /// Plans from one manifest/schema token, then revalidates under the final + /// schema → branch → table gates. Existing target refs must still have live + /// Lance HEAD equal to their manifest pin; uncovered drift is refused with + /// explicit `omnigraph repair` guidance instead of being silently folded. + /// The verified handle is reused for index effects and the resulting + /// versions are committed back to the manifest. /// /// On named branches, indexing preserves lazy branching: /// unbranched subtables keep inheriting `main`, while subtables inherited - /// from an ancestor branch are first forked into the active branch before - /// their index metadata is updated. + /// from an ancestor branch remain inherited when no index work is needed. + /// When real index work exists they are forked into the active branch only + /// after the recovery sidecar is durable. /// Returns the declared indexes that could not be materialized on this /// pass (today: vector columns with no trainable vectors yet). They are /// deferred, not errors; a later `ensure_indices`/`optimize` builds them @@ -2025,9 +2087,7 @@ impl Omnigraph { .await; self.refresh_coordinator_only().await?; self.ensure_schema_apply_not_locked("branch_create").await?; - let control_catalog = self - .load_accepted_catalog_with_schema_gate_held() - .await?; + let control_catalog = self.load_accepted_catalog_with_schema_gate_held().await?; let table_queue_keys = self.table_queue_keys_for_branches( &[source.clone(), Some(target.clone())], &control_catalog, @@ -2117,20 +2177,14 @@ impl Omnigraph { self.refresh_coordinator_only().await?; self.ensure_schema_apply_not_locked("branch_create_from") .await?; - let control_catalog = self - .load_accepted_catalog_with_schema_gate_held() - .await?; - let table_queue_keys = self - .table_queue_keys_for_branches( - &[branch.clone(), Some(target_branch.clone())], - &control_catalog, - ); + let control_catalog = self.load_accepted_catalog_with_schema_gate_held().await?; + let table_queue_keys = self.table_queue_keys_for_branches( + &[branch.clone(), Some(target_branch.clone())], + &control_catalog, + ); let _table_guards = self.write_queue().acquire_many(&table_queue_keys).await; - self.ensure_no_pending_recovery_sidecars_under_gates( - &relevant, - "branch_create_from", - ) - .await?; + self.ensure_no_pending_recovery_sidecars_under_gates(&relevant, "branch_create_from") + .await?; self.refresh_coordinator_only().await?; self.ensure_schema_apply_not_locked("branch_create_from") .await?; @@ -2195,17 +2249,13 @@ impl Omnigraph { let _branch_guard = self.write_queue().acquire_branch(Some(&branch)).await; self.refresh_coordinator_only().await?; self.ensure_schema_apply_not_locked("branch_delete").await?; - let control_catalog = self - .load_accepted_catalog_with_schema_gate_held() - .await?; + let control_catalog = self.load_accepted_catalog_with_schema_gate_held().await?; let table_queue_keys = self.table_queue_keys_for_branches(&[Some(branch.clone())], &control_catalog); let _table_guards = self.write_queue().acquire_many(&table_queue_keys).await; self.ensure_branch_delete_recovery_safe_under_gates(&branch) .await?; - crate::failpoints::maybe_fail( - crate::failpoints::names::BRANCH_DELETE_POST_TABLE_GATES, - )?; + crate::failpoints::maybe_fail(crate::failpoints::names::BRANCH_DELETE_POST_TABLE_GATES)?; self.refresh_coordinator_only().await?; self.ensure_schema_apply_not_locked("branch_delete").await?; self.ensure_schema_state_valid().await?; diff --git a/crates/omnigraph/src/db/omnigraph/schema_apply.rs b/crates/omnigraph/src/db/omnigraph/schema_apply.rs index d860713d..a2bbb993 100644 --- a/crates/omnigraph/src/db/omnigraph/schema_apply.rs +++ b/crates/omnigraph/src/db/omnigraph/schema_apply.rs @@ -438,59 +438,106 @@ where table_tombstones.insert(dropped_table_key.clone(), tombstone_version); } - // Acquire per-(table_key, branch) queues for every existing table - // that schema_apply will rewrite or re-index. New tables (added or - // renamed targets) aren't acquired — they have no existing dataset - // to race against. Held across the per-table commit loop and the - // manifest publish via `commit_changes_with_actor` below. - // - // Schema-apply already holds the graph-wide `__schema_apply_lock__` - // sentinel branch, so these per-table acquisitions are uncontended in - // practice. They exist for symmetry with the recovery reconciler, which - // acquires the same queues before any `Dataset::restore` it issues for - // SchemaApply sidecars. - let schema_apply_queue_keys: Vec<(String, Option)> = recovery_pins - .iter() - .map(|pin| (pin.table_key.clone(), pin.table_branch.clone())) + // Complete effect envelope: the outer `apply_schema` already holds the + // graph-wide schema gate, so add main's branch gate and every live table + // gate in the shared schema -> branch -> sorted-table order. Schema apply + // is graph-global (including metadata-only changes and hard drops), so a + // rewrite-only subset is not a sufficient envelope. + let schema_apply_queue_keys: Vec<(String, Option)> = snapshot + .entries() + .map(|entry| (entry.table_key.clone(), entry.table_branch.clone())) .collect(); // The outer `apply_schema` holds the schema-control serialization key from // before sentinel creation through sentinel release. Per-table guards here // therefore cover only the concrete table effects; acquiring the schema key // again would deadlock because these queues are intentionally non-reentrant. - let writes_sidecar = !(recovery_pins.is_empty() - && sidecar_registrations.is_empty() - && sidecar_tombstones.is_empty()); + let _main_branch_guard = db.write_queue().acquire_branch(None).await; let _schema_apply_queue_guards = db .write_queue() .acquire_many(&schema_apply_queue_keys) .await; - let recovery_handle = if !writes_sidecar { - None - } else { - // `branch=None` because schema_apply publishes against main — - // the `__schema_apply_lock__` branch is purely a serialization - // sentinel (acquire_schema_apply_lock creates it; the manifest - // publish via coordinator.commit_changes_with_actor below targets - // the coordinator's active branch, which is the pre-lock branch). - // If the lock release fires before recovery, the lock branch is - // gone — the sidecar must not reference it. - let mut sidecar = crate::db::manifest::new_sidecar( - crate::db::manifest::SidecarKind::SchemaApply, - None, - // `apply_schema` doesn't currently take an actor (no `apply_schema_as` - // public API). The HTTP server's /schema/apply handler can pass actor - // context through a follow-up addition. For now, system-attributed. - None, - recovery_pins, - ); - sidecar.additional_registrations = sidecar_registrations; - sidecar.tombstones = sidecar_tombstones; - Some( - crate::db::manifest::write_sidecar(db.root_uri(), db.storage_adapter(), &sidecar) - .await?, + // The entry heal ran before planning, but another writer can arm recovery + // while this apply waits for its effect gates. List again under the complete + // envelope before this writer claims any physical state. + db.ensure_no_pending_recovery_sidecars_under_gates(&[None], "schema_apply") + .await?; + + // The snapshot was captured before the branch/table waits. Revalidate main + // now, while those gates are held, so a stale plan never writes a sidecar. + db.refresh_coordinator_only().await?; + let current_manifest_version = db.version().await; + if current_manifest_version != base_manifest_version { + return Err(OmniError::manifest_conflict(format!( + "schema apply lost its write lease: main advanced from v{} to v{} while schema apply was waiting for its effect gates", + base_manifest_version, current_manifest_version, + ))); + } + + // Prove every existing physical ref is still exactly at its manifest pin + // before arming the legacy SchemaApply sidecar. Retain the verified handles + // and reuse them below: reopening after this ownership check would create a + // needless second HEAD observation and blur the pre-arm boundary. + let mut existing_heads = HashMap::::new(); + for entry in snapshot.entries() { + let dataset_uri = db.storage().dataset_uri(&entry.table_path); + let head = db + .storage() + .open_dataset_head(&dataset_uri, entry.table_branch.as_deref()) + .await?; + db.ensure_existing_effect_baseline( + &entry.table_key, + entry.table_branch.as_deref(), + entry.table_version, + &head, ) - }; + .await?; + existing_heads.insert(entry.table_key.clone(), head); + } + + // Added types and rename targets have no manifest authority yet. A physical + // dataset already present at one of those paths is therefore an orphan or a + // foreign effect, never state this apply may retroactively claim through + // `additional_registrations` recovery. + let target_table_keys = added_tables + .iter() + .chain(renamed_tables.keys()) + .collect::>(); + for table_key in target_table_keys { + let table_path = table_path_for_table_key(table_key)?; + let dataset_uri = db.storage().dataset_uri(&table_path); + if db.storage_adapter().exists(&dataset_uri).await? { + return Err(OmniError::manifest_conflict(format!( + "schema apply target table '{}' has an existing dataset at '{}' but no live manifest entry; refusing to claim unowned physical state — inspect and remove the orphaned dataset before retrying", + table_key, dataset_uri, + ))); + } + } + + // Every non-empty migration writes a sidecar, including table-effect-free + // index/enum/metadata changes. Their empty table-pin set is intentional: + // schema staging is the durable threshold that lets recovery deterministically + // roll back a pre-staging crash or roll forward a post-staging crash. + // `branch=None` because schema_apply publishes against main — the + // `__schema_apply_lock__` branch is purely a serialization sentinel. + let mut sidecar = crate::db::manifest::new_sidecar( + crate::db::manifest::SidecarKind::SchemaApply, + None, + // `apply_schema` doesn't currently take an actor (no `apply_schema_as` + // public API). The HTTP server's /schema/apply handler can pass actor + // context through a follow-up addition. For now, system-attributed. + None, + recovery_pins, + ); + sidecar.schema_version = crate::db::manifest::SCHEMA_APPLY_CONFIRMATION_SCHEMA_VERSION; + sidecar.additional_registrations = sidecar_registrations; + sidecar.tombstones = sidecar_tombstones; + sidecar.schema_apply_target_schema_ir_hash = Some( + omnigraph_compiler::schema_ir_hash(&desired_ir) + .map_err(|error| OmniError::manifest_internal(error.to_string()))?, + ); + let recovery_handle = + crate::db::manifest::write_sidecar(db.root_uri(), db.storage_adapter(), &sidecar).await?; for table_key in &added_tables { let table_path = table_path_for_table_key(table_key)?; @@ -522,15 +569,16 @@ where source_table_key )) })?; - ensure_snapshot_entry_head_matches(db, source_entry).await?; - let source_ds = db - .storage() - .open_snapshot_at_table(&snapshot, source_table_key) - .await?; + let source_ds = existing_heads.get(source_table_key).ok_or_else(|| { + OmniError::manifest_internal(format!( + "missing preflighted source table '{}' for schema rename", + source_table_key + )) + })?; let current_catalog = db.catalog(); let batch = batch_for_schema_apply_rewrite( db, - &source_ds, + source_ds, source_table_key, ¤t_catalog, target_table_key, @@ -570,11 +618,12 @@ where table_key )) })?; - ensure_snapshot_entry_head_matches(db, entry).await?; - let source_ds = db - .storage() - .open_snapshot_at_table(&snapshot, table_key) - .await?; + let source_ds = existing_heads.remove(table_key).ok_or_else(|| { + OmniError::manifest_internal(format!( + "missing preflighted source table '{}' for schema apply", + table_key + )) + })?; let current_catalog = db.catalog(); let batch = batch_for_schema_apply_rewrite( db, @@ -593,10 +642,7 @@ where // non-main branches, so `entry.table_branch` is expected to be // `None`. But the defensive passthrough means a future relaxation // of the lock-check can't quietly open the wrong HEAD here. - let existing = db - .storage() - .open_dataset_head(&dataset_uri, entry.table_branch.as_deref()) - .await?; + let existing = source_ds; let staged = db.storage().stage_overwrite(&existing, batch).await?; let target_ds = db.storage().commit_staged(existing, staged).await?; // The rewrite drops the table's existing index coverage; it is @@ -677,6 +723,17 @@ where .commit_changes_with_actor(&manifest_changes, None) .await?; + // Metadata/table-set-only applies can have no table pins. Confirm Phase C + // before promoting staging files so a later Phase-D delete failure is + // distinguishable from a pre-staging crash without guessing from manifest + // version movement (a rollback recovery commit also moves that version). + crate::db::manifest::confirm_schema_apply_manifest_published( + db.root_uri(), + db.storage_adapter(), + &mut sidecar, + ) + .await?; + crate::failpoints::maybe_fail(crate::failpoints::names::SCHEMA_APPLY_AFTER_MANIFEST_COMMIT)?; db.storage @@ -710,14 +767,14 @@ where // artifact (recovery is a no-op and the sidecar is cleaned up). // Failing the schema_apply call would report failure for a migration // that already succeeded. - if let Some(handle) = recovery_handle { - if let Err(err) = crate::db::manifest::delete_sidecar(&handle, db.storage_adapter()).await { - tracing::warn!( - error = %err, - operation_id = handle.operation_id.as_str(), - "recovery sidecar cleanup failed; the next open's recovery sweep will resolve it" - ); - } + if let Err(err) = + crate::db::manifest::delete_sidecar(&recovery_handle, db.storage_adapter()).await + { + tracing::warn!( + error = %err, + operation_id = recovery_handle.operation_id.as_str(), + "recovery sidecar cleanup failed; the next open's recovery sweep will resolve it" + ); } // Hard-drop cleanup: run cleanup_old_versions on each dataset @@ -858,19 +915,6 @@ pub(super) async fn ensure_schema_apply_not_locked(db: &Omnigraph, operation: &s Ok(()) } -pub(super) async fn ensure_snapshot_entry_head_matches( - db: &Omnigraph, - entry: &SubTableEntry, -) -> Result<()> { - let dataset_uri = db.storage().dataset_uri(&entry.table_path); - let ds = db - .storage() - .open_dataset_head(&dataset_uri, entry.table_branch.as_deref()) - .await?; - db.storage() - .ensure_expected_version(&ds, &entry.table_key, entry.table_version) -} - pub(super) async fn batch_for_schema_apply_rewrite( db: &Omnigraph, source_ds: &SnapshotHandle, diff --git a/crates/omnigraph/src/db/omnigraph/table_ops.rs b/crates/omnigraph/src/db/omnigraph/table_ops.rs index 70cc9f1b..ba31ad51 100644 --- a/crates/omnigraph/src/db/omnigraph/table_ops.rs +++ b/crates/omnigraph/src/db/omnigraph/table_ops.rs @@ -84,25 +84,36 @@ pub(super) async fn ensure_indices_for_branch( db: &Omnigraph, branch: Option<&str>, ) -> Result> { - db.ensure_schema_state_valid().await?; db.ensure_schema_apply_idle("ensure_indices").await?; - let resolved = db.resolved_branch_target(branch).await?; - let snapshot = resolved.snapshot; + // Capture one coherent branch/schema authority for both index-work planning + // and the eventual effects. A long-lived handle's ArcSwap catalog can lag a + // schema apply performed through another handle; `WriteTxn::catalog` is built + // from the exact accepted IR whose identity is rechecked under the final + // gates below. + let txn = db.open_write_txn(branch).await?; + let snapshot = txn.base.clone(); let mut updates = Vec::new(); - let mut pending = Vec::new(); - let active_branch = resolved.branch; - let catalog = db.catalog(); + let mut pending_by_table = HashMap::>::new(); + let active_branch = txn.branch.clone(); + let catalog = Arc::clone(&txn.catalog); // Recovery sidecar: protect the per-table commit_staged loop in - // build_indices_on_dataset (one commit per index built). Only pins - // tables that ACTUALLY need index work — the classifier - // loose-matches for SidecarKind::EnsureIndices (the actual N - // depends on which indices are missing), but if a table needs zero - // commits and gets pinned, the all-or-nothing decision rule - // classifies it as `NoMovement` and rolls back legitimately- - // committed work on sibling tables. Steady-state runs (everything - // already indexed) skip the sidecar entirely. + // build_indices_on_dataset (one commit per index built). Only tables with + // actual index work enter the durable plan. In particular, a child branch + // does not first-touch/fork an inherited table merely to discover there is + // nothing to build: lazy inheritance remains the cheaper physical shape and + // avoids manufacturing a ref-only effect that the legacy loose sidecar + // cannot distinguish from an already-published pointer update. let mut recovery_pins: Vec = Vec::new(); + // Existing target refs are opened once during planning, verified against + // their live HEAD under the final gates, and then reused for the effect. + // Reopening after the pre-arm check would recreate a TOCTOU window. Deferred + // first-touch refs are intentionally absent here: their target ref may only + // be created after the sidecar is durable. + let mut existing_targets = + std::collections::HashMap::::new(); + let mut first_touch_sources = + std::collections::HashMap::::new(); for type_name in catalog.node_types.keys() { let table_key = format!("node:{}", type_name); let Some(entry) = snapshot.entry(&table_key) else { @@ -118,9 +129,22 @@ pub(super) async fn ensure_indices_for_branch( continue; } let full_path = format!("{}/{}", db.root_uri, entry.table_path); - if needs_index_work_node(db, type_name, &full_path, entry.table_branch.as_deref()).await? { + let first_touch = + active_branch.is_some() && entry.table_branch.as_deref() != active_branch.as_deref(); + let ds = if first_touch { + // The inherited owner's HEAD may advance independently after this + // graph branch was cut. Plan from the exact inherited snapshot, not + // from that owner's current HEAD. + db.storage().open_snapshot_at_entry(entry).await? + } else { + db.storage() + .open_dataset_head(&full_path, active_branch.as_deref()) + .await? + }; + let work = plan_index_work_node(db, &catalog, type_name, &table_key, &ds).await?; + if work.needs_commit { recovery_pins.push(crate::db::manifest::SidecarTablePin { - table_key, + table_key: table_key.clone(), table_path: full_path, expected_version: entry.table_version, post_commit_pin: entry.table_version + 1, @@ -135,6 +159,16 @@ pub(super) async fn ensure_indices_for_branch( // open_lance_head must check the same branch. table_branch: active_branch.clone(), }); + if !first_touch { + existing_targets.insert(table_key, ds); + } else { + first_touch_sources.insert(table_key, ds); + } + } else { + // Preserve the operator-facing pending status for an untrainable + // Vector column without invoking the mutating builder on a table + // that is absent from the durable effect plan. + pending_by_table.insert(table_key, work.pending); } } for edge_name in catalog.edge_types.keys() { @@ -146,9 +180,18 @@ pub(super) async fn ensure_indices_for_branch( continue; } let full_path = format!("{}/{}", db.root_uri, entry.table_path); - if needs_index_work_edge(db, &full_path, entry.table_branch.as_deref()).await? { + let first_touch = + active_branch.is_some() && entry.table_branch.as_deref() != active_branch.as_deref(); + let ds = if first_touch { + db.storage().open_snapshot_at_entry(entry).await? + } else { + db.storage() + .open_dataset_head(&full_path, active_branch.as_deref()) + .await? + }; + if needs_index_work_edge_on_dataset(db, &ds).await? { recovery_pins.push(crate::db::manifest::SidecarTablePin { - table_key, + table_key: table_key.clone(), table_path: full_path, expected_version: entry.table_version, post_commit_pin: entry.table_version + 1, @@ -163,6 +206,11 @@ pub(super) async fn ensure_indices_for_branch( // open_lance_head must check the same branch. table_branch: active_branch.clone(), }); + if !first_touch { + existing_targets.insert(table_key, ds); + } else { + first_touch_sources.insert(table_key, ds); + } } } // Acquire per-(table_key, active_branch) queues for every table @@ -175,82 +223,135 @@ pub(super) async fn ensure_indices_for_branch( .iter() .map(|pin| (pin.table_key.clone(), pin.table_branch.clone())) .collect(); + // Match the RFC-022/recovery total order: schema -> active branch -> sorted + // affected tables. Planning above is deliberately outside these gates; all + // authority and physical baselines are rechecked after waiting and before + // our sidecar or any target ref/table effect exists. + let _schema_guard = db + .write_queue() + .acquire(&crate::db::manifest::schema_apply_serial_queue_key()) + .await; + let _branch_guard = db + .write_queue() + .acquire_branch(active_branch.as_deref()) + .await; let _queue_guards = db.write_queue().acquire_many(&queue_keys).await; + db.ensure_no_pending_recovery_sidecars_under_gates( + &[active_branch.as_deref()], + "ensure_indices", + ) + .await?; + let live_snapshot = db.revalidate_write_txn(&txn).await?; + + for pin in &recovery_pins { + let prepared_entry = snapshot.entry(&pin.table_key).ok_or_else(|| { + OmniError::manifest_conflict(format!( + "table '{}' disappeared from the prepared index plan", + pin.table_key, + )) + })?; + let live_entry = live_snapshot.entry(&pin.table_key).ok_or_else(|| { + OmniError::manifest_read_set_changed( + format!("table_head:{}", pin.table_key), + Some(pin.expected_version.to_string()), + None, + ) + })?; + if live_entry.table_version != pin.expected_version + || live_entry.table_path != prepared_entry.table_path + || live_entry.table_branch != prepared_entry.table_branch + { + return Err(OmniError::manifest_read_set_changed( + format!("table_head:{}", pin.table_key), + Some(format!( + "{}:{}:{}", + prepared_entry.table_path, + prepared_entry.table_branch.as_deref().unwrap_or("main"), + pin.expected_version, + )), + Some(format!( + "{}:{}:{}", + live_entry.table_path, + live_entry.table_branch.as_deref().unwrap_or("main"), + live_entry.table_version, + )), + )); + } + + // First-touch target refs do not exist yet and remain governed by the + // existing sidecar-before-fork classifier below. Existing refs must be + // exactly at the logical manifest pin before this writer may claim any + // subsequent physical movement. + if let Some(ds) = existing_targets.get(&pin.table_key) { + db.ensure_existing_effect_baseline( + &pin.table_key, + pin.table_branch.as_deref(), + pin.expected_version, + ds, + ) + .await?; + } else if let Some(source) = first_touch_sources.get(&pin.table_key) { + let target_branch = active_branch.as_deref().ok_or_else(|| { + OmniError::manifest_internal(format!( + "first-touch index target '{}' has no active named branch", + pin.table_key, + )) + })?; + let branches = source + .dataset() + .list_branches() + .await + .map_err(|error| OmniError::Lance(error.to_string()))?; + if branches.contains_key(target_branch) { + return Err(OmniError::manifest_conflict(format!( + "index target ref '{}:{}' already exists while the graph manifest still \ + inherits the table from another branch; refusing to claim unowned \ + physical state — inspect and remove the orphaned ref before retrying", + pin.table_key, target_branch, + ))); + } + } + } + let recovery_handle = if recovery_pins.is_empty() { None } else { - let sidecar = crate::db::manifest::new_sidecar( - crate::db::manifest::SidecarKind::EnsureIndices, + // `ensure_indices` doesn't currently take an actor; recovery remains + // system-attributed. Schema-v6 pre-mints the rollback outcome so an + // interrupted compensation cannot later be mislabeled as roll-forward. + // Future: add `ensure_indices_as` to thread actor context. + let sidecar = crate::db::manifest::new_ensure_indices_sidecar( active_branch.clone(), - // `ensure_indices` doesn't currently take an actor; system-attributed. - // Future: add `ensure_indices_as` to thread actor context. - None, - recovery_pins, - ); + recovery_pins.clone(), + )?; Some( crate::db::manifest::write_sidecar(db.root_uri(), db.storage_adapter(), &sidecar) .await?, ) }; - for type_name in catalog.node_types.keys() { - let table_key = format!("node:{}", type_name); - let Some(entry) = snapshot.entry(&table_key) else { - continue; - }; - let full_path = format!("{}/{}", db.root_uri, entry.table_path); - let (mut ds, resolved_branch) = match active_branch.as_deref() { - Some(active_branch) => match entry.table_branch.as_deref() { - None => continue, - _ => { - open_owned_dataset_for_branch_write( - db, - &table_key, - &full_path, - entry.table_branch.as_deref(), - entry.table_version, - active_branch, - crate::db::MutationOpKind::SchemaRewrite, - false, - ) - .await? - } - }, - None => ( - db.storage().open_dataset_head(&full_path, None).await?, - None, - ), - }; - let row_count = db.storage().count_rows(&ds, None).await.unwrap_or(0); - if row_count > 0 { - pending.extend(build_indices_on_dataset(db, &table_key, &mut ds).await?); - } - - let state = db.storage().table_state(&full_path, &ds).await?; - if state.version != entry.table_version - || resolved_branch.as_deref() != entry.table_branch.as_deref() - { - updates.push(crate::db::SubTableUpdate { - table_key, - table_version: state.version, - table_branch: resolved_branch, - row_count: state.row_count, - version_metadata: state.version_metadata, - }); - } + if recovery_handle.is_some() && !first_touch_sources.is_empty() { + crate::failpoints::maybe_fail( + crate::failpoints::names::ENSURE_INDICES_POST_SIDECAR_PRE_FORK, + )?; } - for edge_name in catalog.edge_types.keys() { - let table_key = format!("edge:{}", edge_name); - let Some(entry) = snapshot.entry(&table_key) else { - continue; - }; - let full_path = format!("{}/{}", db.root_uri, entry.table_path); + // Execute exactly the keys named in the durable effect plan. Visiting every + // catalog table here would let a stale no-work observation become an + // unpinned commit if physical state changed after planning. + for pin in &recovery_pins { + let table_key = pin.table_key.clone(); + let entry = snapshot + .entry(&table_key) + .ok_or_else(|| OmniError::manifest(format!("no manifest entry for {}", table_key)))?; + let full_path = pin.table_path.clone(); let (mut ds, resolved_branch) = match active_branch.as_deref() { - Some(active_branch) => match entry.table_branch.as_deref() { - None => continue, - _ => { + Some(active_branch) => { + if let Some(ds) = existing_targets.remove(&table_key) { + (ds, Some(active_branch.to_string())) + } else { + first_touch_sources.remove(&table_key); open_owned_dataset_for_branch_write( db, &table_key, @@ -263,15 +364,22 @@ pub(super) async fn ensure_indices_for_branch( ) .await? } - }, + } None => ( - db.storage().open_dataset_head(&full_path, None).await?, + existing_targets.remove(&table_key).ok_or_else(|| { + OmniError::manifest_internal(format!( + "missing verified existing target for main table '{}'", + table_key, + )) + })?, None, ), }; - let row_count = db.storage().count_rows(&ds, None).await.unwrap_or(0); + let row_count = db.storage().count_rows(&ds, None).await?; if row_count > 0 { - pending.extend(build_indices_on_dataset(db, &table_key, &mut ds).await?); + let table_pending = + build_indices_on_dataset_for_catalog(db, &catalog, &table_key, &mut ds).await?; + pending_by_table.insert(table_key.clone(), table_pending); } let state = db.storage().table_state(&full_path, &ds).await?; @@ -286,6 +394,7 @@ pub(super) async fn ensure_indices_for_branch( version_metadata: state.version_metadata, }); } + crate::failpoints::maybe_fail(crate::failpoints::names::ENSURE_INDICES_POST_TABLE_EFFECT)?; } // Failpoint: pin the per-writer Phase B → Phase C residual for @@ -315,6 +424,14 @@ pub(super) async fn ensure_indices_for_branch( } } + // Preserve the historical, observable catalog order even though planning + // and physical effects now happen in separate phases. + let mut pending = Vec::new(); + for type_name in catalog.node_types.keys() { + if let Some(mut table_pending) = pending_by_table.remove(&format!("node:{type_name}")) { + pending.append(&mut table_pending); + } + } Ok(pending) } @@ -394,32 +511,33 @@ async fn vector_column_trainable( /// (DateTime/Date/numeric/Bool), FTS for free-text Strings, or a Vector index. /// Edges get BTree only (id, src, dst). This helper and the builder share /// `node_prop_index_kind` so they cannot drift — see its doc comment. -pub(super) async fn needs_index_work_node( +#[derive(Default)] +struct PlannedIndexWork { + needs_commit: bool, + pending: Vec, +} + +/// Classify index work against one already-selected dataset snapshot and one +/// operation-local accepted catalog. Keeping the opener outside this helper is +/// load-bearing for named-branch first touch: an inherited graph entry must be +/// checked at its exact pinned version, not at the inherited owner's newer HEAD. +async fn plan_index_work_node( db: &Omnigraph, + catalog: &Catalog, type_name: &str, - full_path: &str, - table_branch: Option<&str>, -) -> Result { - let ds = db - .storage() - .open_dataset_head(full_path, table_branch) - .await?; - // Empty tables are skipped by the ensure_indices loop, so they must - // not be pinned in the sidecar — pinning a table that produces zero - // commits classifies as NoMovement on recovery and forces all-or- - // nothing rollback of sibling tables' legitimate index work. - // Errors from count_rows are propagated: silently treating them as - // "0 rows" risks skipping a table that is actually about to be - // modified. - if db.storage().count_rows(&ds, None).await? == 0 { - return Ok(false); + table_key: &str, + ds: &SnapshotHandle, +) -> Result { + if db.storage().count_rows(ds, None).await? == 0 { + return Ok(PlannedIndexWork::default()); } - if !db.storage().has_btree_index(&ds, "id").await? { - return Ok(true); - } - let catalog = db.catalog(); + + let mut work = PlannedIndexWork { + needs_commit: !db.storage().has_btree_index(ds, "id").await?, + pending: Vec::new(), + }; let Some(node_type) = catalog.node_types.get(type_name) else { - return Ok(false); + return Ok(work); }; for index_cols in &node_type.indices { if index_cols.len() != 1 { @@ -431,31 +549,46 @@ pub(super) async fn needs_index_work_node( }; match node_prop_index_kind(prop_type) { Some(NodePropIndexKind::Fts) => { - if !db.storage().has_fts_index(&ds, prop_name).await? { - return Ok(true); - } + work.needs_commit |= !db.storage().has_fts_index(ds, prop_name).await?; } Some(NodePropIndexKind::Vector) => { - // Only count a missing vector index as buildable *work* when the - // column is trainable (>=1 non-null vector). An untrainable - // column would defer in the build and commit nothing; pinning it - // for recovery would be a zero-commit pin that classifies - // NoMovement and rolls back a sibling table's index work. - if !db.storage().has_vector_index(&ds, prop_name).await? - && vector_column_trainable(db, &ds, prop_name).await? - { - return Ok(true); + if !db.storage().has_vector_index(ds, prop_name).await? { + if vector_column_trainable(db, ds, prop_name).await? { + work.needs_commit = true; + } else { + work.pending.push(PendingIndex { + table_key: table_key.to_string(), + column: prop_name.clone(), + reason: "column has no non-null vectors to train on yet".to_string(), + }); + } } } Some(NodePropIndexKind::Btree) => { - if !db.storage().has_btree_index(&ds, prop_name).await? { - return Ok(true); - } + work.needs_commit |= !db.storage().has_btree_index(ds, prop_name).await?; } None => {} } } - Ok(false) + Ok(work) +} + +pub(super) async fn needs_index_work_node( + db: &Omnigraph, + type_name: &str, + full_path: &str, + table_branch: Option<&str>, +) -> Result { + let ds = db + .storage() + .open_dataset_head(full_path, table_branch) + .await?; + let catalog = db.catalog(); + Ok( + plan_index_work_node(db, &catalog, type_name, &format!("node:{type_name}"), &ds) + .await? + .needs_commit, + ) } /// Companion to `needs_index_work_node` for edge tables. @@ -477,6 +610,10 @@ pub(super) async fn needs_index_work_edge( .storage() .open_dataset_head(full_path, table_branch) .await?; + needs_index_work_edge_on_dataset(db, &ds).await +} + +async fn needs_index_work_edge_on_dataset(db: &Omnigraph, ds: &SnapshotHandle) -> Result { if db.storage().count_rows(&ds, None).await? == 0 { return Ok(false); } @@ -1455,8 +1592,7 @@ pub(super) async fn commit_updates( .await .current_branch() .map(str::to_string); - let prepared = - prepare_updates_for_commit(db, current_branch.as_deref(), updates, None).await?; + let prepared = prepare_updates_for_commit(db, current_branch.as_deref(), updates, None).await?; commit_prepared_updates(db, &prepared, None).await } diff --git a/crates/omnigraph/src/exec/merge.rs b/crates/omnigraph/src/exec/merge.rs index 01a3dc92..c67f5c15 100644 --- a/crates/omnigraph/src/exec/merge.rs +++ b/crates/omnigraph/src/exec/merge.rs @@ -25,6 +25,26 @@ enum CandidateTableState { RewriteMerged(StagedMergeResult), } +/// An existing target ref opened and verified against the target manifest pin +/// under branch merge's final schema -> branch -> table gate envelope. +/// +/// The handle is carried through Phase A and consumed by the physical effect so +/// the post-arm path cannot reopen a different HEAD. First-touch refs never +/// construct this value: their target ref is created only after the v4 recovery +/// sidecar is durable. +#[derive(Debug)] +struct PreparedExistingMergeTarget { + current: SnapshotHandle, + full_path: String, + table_branch: Option, +} + +impl PreparedExistingMergeTarget { + fn into_parts(self) -> (SnapshotHandle, String, Option) { + (self.current, self.full_path, self.table_branch) + } +} + #[derive(Debug)] struct StagedTable { _dir: TempDir, @@ -354,8 +374,7 @@ async fn compute_adopt_delta( let schema = schema_for_table_key(catalog, table_key)?; let mut append_writer = StagedTableWriter::new(&format!("{}_adopt_append", table_key), schema.clone())?; - let mut upsert_writer = - StagedTableWriter::new(&format!("{}_adopt_upsert", table_key), schema)?; + let mut upsert_writer = StagedTableWriter::new(&format!("{}_adopt_upsert", table_key), schema)?; let mut deleted_ids: Vec = Vec::new(); let mut base = OrderedTableCursor::from_snapshot_lazy(base_snapshot, table_key).await?; let mut source = OrderedTableCursor::from_snapshot_lazy(source_snapshot, table_key).await?; @@ -823,9 +842,7 @@ async fn classify_adopt( compute_adopt_delta(table_key, catalog, base_snapshot, source_snapshot).await?; match (advances_head, validation_delta) { (true, Some(delta)) => Ok(CandidateTableState::AdoptWithDelta(delta)), - (_, validation_delta) => { - Ok(CandidateTableState::AdoptSourceState { validation_delta }) - } + (_, validation_delta) => Ok(CandidateTableState::AdoptSourceState { validation_delta }), } } @@ -991,6 +1008,7 @@ async fn publish_rewritten_merge_table( target_db: &Omnigraph, table_key: &str, staged: &StagedMergeResult, + prepared_target: Option, planned_transactions: &[crate::table_store::StagedTransactionIdentity], ) -> Result { // Branch merge's source-rewrite path is Merge-shaped (upsert from @@ -998,13 +1016,16 @@ async fn publish_rewritten_merge_table( // (`stage_delete` + an exact commit) operates on rows the rewrite chose // to remove, not user-facing predicates, so Merge is the correct policy // here. - // `open_for_mutation` is the no-txn entry, so collapse #1's non-strict - // open-skip (gated on `txn.is_some()`) never fires here — the handle is - // always `Some`. - let (mut current_ds, full_path, table_branch) = target_db - .open_for_mutation(table_key, crate::db::MutationOpKind::Merge) - .await? - .require_handle("branch merge"); + // Existing refs consume the same handle verified before Phase A. Only a + // first-touch target reaches `open_for_mutation` here, after recovery owns + // the ref creation; its no-txn path always returns a handle. + let (mut current_ds, full_path, table_branch) = match prepared_target { + Some(prepared) => prepared.into_parts(), + None => target_db + .open_for_mutation(table_key, crate::db::MutationOpKind::Merge) + .await? + .require_handle("branch merge first touch"), + }; // Phase 1: merge_insert changed/new rows (preserves _row_created_at_version for // existing rows, bumps _row_last_updated_at_version only for actually-changed rows). @@ -1187,15 +1208,19 @@ async fn publish_adopted_delta( target_db: &Omnigraph, table_key: &str, delta: &AdoptDelta, + prepared_target: Option, planned_transactions: &[crate::table_store::StagedTransactionIdentity], ) -> Result { - // `open_for_mutation` is the no-txn entry, so collapse #1's non-strict - // open-skip (gated on `txn.is_some()`) never fires here — the handle is - // always `Some`. - let (mut current_ds, full_path, table_branch) = target_db - .open_for_mutation(table_key, crate::db::MutationOpKind::Merge) - .await? - .require_handle("branch merge"); + // Existing refs consume the same handle verified before Phase A. Only a + // first-touch target reaches `open_for_mutation` here, after recovery owns + // the ref creation; its no-txn path always returns a handle. + let (mut current_ds, full_path, table_branch) = match prepared_target { + Some(prepared) => prepared.into_parts(), + None => target_db + .open_for_mutation(table_key, crate::db::MutationOpKind::Merge) + .await? + .require_handle("branch merge first touch"), + }; // Phase 1a: append the NEW rows. `stage_append_stream` is a streaming // `Operation::Append` — no hash join — so it never buffers the delta and @@ -1717,6 +1742,7 @@ impl Omnigraph { let mut delta_slots = Vec::new(); let mut first_touch_effects = HashSet::new(); let mut planned_transactions_by_table = HashMap::new(); + let mut prepared_existing_targets = HashMap::new(); for table_key in &ordered_table_keys { let Some(candidate) = candidates.get(table_key) else { continue; @@ -1760,6 +1786,24 @@ impl Omnigraph { .map(|_| entry.table_version); if source_fork_version.is_some() { first_touch_effects.insert(table_key.clone()); + } else { + // Existing-ref effects must prove that the physical + // baseline still equals the captured manifest pin + // before this merge writes a sidecar that claims the + // next versions. Keep the opened handle for Phase B so + // the post-arm path cannot reopen a different HEAD. + let (current, full_path, table_branch) = self + .open_for_mutation(table_key, crate::db::MutationOpKind::Merge) + .await? + .require_handle("branch merge existing-effect preflight"); + prepared_existing_targets.insert( + table_key.clone(), + PreparedExistingMergeTarget { + current, + full_path, + table_branch, + }, + ); } let planned_transactions = plan_merge_transactions(table_key, candidate, expected_version)?; @@ -1813,6 +1857,31 @@ impl Omnigraph { } } + // Probe after every existing target handle has been collected, at the + // last pre-arm boundary under the complete gate envelope. First-touch + // targets are intentionally absent: their native ref does not exist + // until the sidecar below is durable. + for table_key in &ordered_table_keys { + let Some(prepared) = prepared_existing_targets.get(table_key) else { + continue; + }; + let expected_version = target_snapshot + .entry(table_key) + .ok_or_else(|| { + OmniError::manifest_internal(format!( + "prepared branch merge target '{table_key}' has no manifest entry" + )) + })? + .table_version; + self.ensure_existing_effect_baseline( + table_key, + prepared.table_branch.as_deref(), + expected_version, + &prepared.current, + ) + .await?; + } + // Keep the sidecar alongside its handle: after the whole physical // effect set completes, confirmation binds every output slot and every // first-touch ref identity before the one manifest CAS. @@ -1874,6 +1943,21 @@ impl Omnigraph { let Some(candidate_state) = candidates.get(table_key) else { continue; }; + let prepared_target = match candidate_state { + CandidateTableState::RewriteMerged(_) + | CandidateTableState::AdoptWithDelta(_) => { + if first_touch_effects.contains(table_key) { + None + } else { + Some(prepared_existing_targets.remove(table_key).ok_or_else(|| { + OmniError::manifest_internal(format!( + "branch merge table '{table_key}' lacks its verified existing target handle" + )) + })?) + } + } + CandidateTableState::AdoptSourceState { .. } => None, + }; let update = match candidate_state { CandidateTableState::AdoptSourceState { .. } => { publish_adopted_source_state( @@ -1890,7 +1974,8 @@ impl Omnigraph { "branch merge table '{table_key}' lacks its armed transaction chain" )) })?; - publish_adopted_delta(self, table_key, delta, planned).await? + publish_adopted_delta(self, table_key, delta, prepared_target, planned) + .await? } CandidateTableState::RewriteMerged(staged) => { let planned = planned_transactions_by_table.get(table_key).ok_or_else(|| { @@ -1898,7 +1983,14 @@ impl Omnigraph { "branch merge table '{table_key}' lacks its armed transaction chain" )) })?; - publish_rewritten_merge_table(self, table_key, staged, planned).await? + publish_rewritten_merge_table( + self, + table_key, + staged, + prepared_target, + planned, + ) + .await? } }; if first_touch_effects.contains(table_key) { diff --git a/crates/omnigraph/src/exec/staging.rs b/crates/omnigraph/src/exec/staging.rs index 126ff205..5e7fe811 100644 --- a/crates/omnigraph/src/exec/staging.rs +++ b/crates/omnigraph/src/exec/staging.rs @@ -396,21 +396,20 @@ impl MutationStaging { stage_inputs.push((table_key, table, path, expected)); } let concurrency = concurrency.min(stage_inputs.len()).max(1); - let mut staged_entries: Vec = futures::stream::iter( - stage_inputs.into_iter().map( + let mut staged_entries: Vec = + futures::stream::iter(stage_inputs.into_iter().map( |(table_key, table, path, expected)| async move { stage_pending_table(db, table_key, table, path, expected).await }, - ), - ) - .buffered(concurrency) - .collect::>>>() - .await - .into_iter() - .collect::>>()? - .into_iter() - .flatten() - .collect(); + )) + .buffered(concurrency) + .collect::>>>() + .await + .into_iter() + .collect::>>()? + .into_iter() + .flatten() + .collect(); // Second pass: stage deletes through the same staged path. D₂ // guarantees a delete-touched table carries no pending write batches, @@ -444,8 +443,7 @@ impl MutationStaging { .collect::>() .join(" OR ") }; - if let Some(entry) = - stage_delete_table(db, table_key, combined, path, expected).await? + if let Some(entry) = stage_delete_table(db, table_key, combined, path, expected).await? { staged_entries.push(entry); } @@ -474,7 +472,11 @@ async fn stage_pending_table( PendingMode::Overwrite => crate::db::MutationOpKind::SchemaRewrite, }; let ds = match path.deferred_fork.as_ref() { - Some(fork) => db.storage().open_snapshot_at_entry(&fork.source_entry).await?, + Some(fork) => { + db.storage() + .open_snapshot_at_entry(&fork.source_entry) + .await? + } None => { db.reopen_for_mutation( &table_key, @@ -567,7 +569,11 @@ async fn stage_delete_table( expected: u64, ) -> Result> { let ds = match path.deferred_fork.as_ref() { - Some(fork) => db.storage().open_snapshot_at_entry(&fork.source_entry).await?, + Some(fork) => { + db.storage() + .open_snapshot_at_entry(&fork.source_entry) + .await? + } None => { db.reopen_for_mutation( &table_key, @@ -761,8 +767,7 @@ impl StagedMutation { // are staged like every other write, so holding the queue from before // `commit_staged` through the publish keeps no Lance HEAD ahead of the // manifest on the happy path. - let mut queue_keys: Vec<(String, Option)> = - Vec::with_capacity(staged.len()); + let mut queue_keys: Vec<(String, Option)> = Vec::with_capacity(staged.len()); for entry in &staged { queue_keys.push((entry.table_key.clone(), entry.path.table_branch.clone())); } @@ -863,90 +868,19 @@ impl StagedMutation { } // Separate manifest-visible concurrency from uncovered Lance drift. - // The exact table pin already matched above; now verify that the - // live Lance HEAD also equals that pin. If an external raw Lance - // write or a pre-fix maintenance path moved HEAD without publishing - // `__manifest`, this write must not silently fold it. - // - // `latest_version_id` reads the latest manifest pointer off the - // already-open staged handle (the #2 staging open) WITHOUT a fresh - // `Dataset::open` — the same cheap live-HEAD probe - // `ManifestCoordinator::probe_latest_version` uses. This replaces a - // redundant `open_dataset_head` (RFC-013 step 3b, collapse - // #3): the drift comparison below is byte-identical; only how `head` - // is obtained changes (probe vs cold open). - let head = entry - .dataset - .dataset() - .latest_version_id() - .await - .map_err(|e| OmniError::Lance(e.to_string()))?; - if head < current { - return Err(OmniError::manifest_internal(format!( - "table '{}' Lance HEAD version {} is behind manifest version {}", - entry.table_key, head, current - ))); - } - if head > current { - // Error path only: attribute covered drift to the exact - // recovery operation that owns it. The entry-point heal ran - // before this attempt was prepared, so another writer can arm - // and confirm a sidecar while this attempt stages reclaimable - // files outside the gates. Calling the healer here would be a - // lock-order violation: recovery acquires the same - // schema -> branch -> table gates we currently hold. - // - // `list_sidecars` returns URI-sorted sidecars. Use the first - // matching table/ref claim so attribution stays deterministic - // even if malformed/concurrent state contains more than one - // claimant. Uncovered drift (external raw Lance write or a - // pre-fix maintenance path) remains a generic conflict routed - // through `omnigraph repair`. - let sidecars = crate::db::manifest::list_sidecars( - db.root_uri(), - db.storage_adapter(), - ) - .await; - match sidecars { - Ok(sidecars) => { - if let Some(owner) = sidecars.iter().find(|sidecar| { - sidecar.tables.iter().any(|pin| { - // Branch-aware: a sidecar pinning the same - // table on another branch does not own this - // branch's drift. - pin.table_key == entry.table_key - && pin.table_branch == entry.path.table_branch - }) - }) { - return Err(OmniError::recovery_required( - owner.operation_id.clone(), - format!( - "table '{}' has Lance HEAD version {} ahead of manifest \ - version {}; the pending recovery operation owns this drift", - entry.table_key, head, current, - ), - )); - } - return Err(OmniError::manifest_conflict(format!( - "table '{}' has Lance HEAD version {} ahead of manifest version {}; \ - run `omnigraph repair` before writing", - entry.table_key, head, current, - ))); - } - Err(list_err) => { - let action = format!( - "could not classify the drift (sidecar listing failed: {}); \ - run `omnigraph repair`, or reopen the graph read-write if \ - repair reports a pending recovery sidecar", - list_err - ); - return Err(OmniError::manifest_conflict(format!( - "table '{}' has Lance HEAD version {} ahead of manifest version {}; {}", - entry.table_key, head, current, action - ))); - } - } - } + // The shared pre-arm check probes this already-open staged handle, + // attributes covered drift to its exact recovery operation, and + // routes uncovered drift through explicit operator repair. Keeping + // the rule here and in control/maintenance adapters behind one + // helper prevents a future writer from weakening ownership by + // emitting its sidecar before checking the physical baseline. + db.ensure_existing_effect_baseline( + &entry.table_key, + entry.path.table_branch.as_deref(), + current, + &entry.dataset, + ) + .await?; } // An empty load/mutation has no independently durable table effect. // It may still publish its fixed lineage intent, but arming recovery @@ -969,10 +903,7 @@ impl StagedMutation { let mut pins: Vec = Vec::with_capacity(staged.len()); let mut planned_transactions = HashMap::with_capacity(staged.len()); for entry in &staged { - planned_transactions.insert( - entry.table_key.clone(), - entry.planned_transaction.clone(), - ); + planned_transactions.insert(entry.table_key.clone(), entry.planned_transaction.clone()); pins.push(SidecarTablePin { table_key: entry.table_key.clone(), table_path: entry.path.full_path.clone(), @@ -1010,17 +941,17 @@ impl StagedMutation { // sidecar nor a target fork exists yet. A concurrent winner may publish; // this attempt then discards/reprepares on the collision below. crate::failpoints::maybe_fail(crate::failpoints::names::FORK_BEFORE_CLASSIFY)?; - let sidecar_handle = Some( - write_sidecar(db.root_uri(), db.storage_adapter(), &sidecar).await?, - ); + let sidecar_handle = + Some(write_sidecar(db.root_uri(), db.storage_adapter(), &sidecar).await?); let operation_id = sidecar.operation_id.clone(); - if staged.iter().any(|entry| entry.path.deferred_fork.is_some()) { - crate::failpoints::maybe_fail( - crate::failpoints::names::MUTATION_POST_SIDECAR_PRE_FORK, - ) - .map_err(|error| { - OmniError::recovery_required(operation_id.clone(), error.to_string()) - })?; + if staged + .iter() + .any(|entry| entry.path.deferred_fork.is_some()) + { + crate::failpoints::maybe_fail(crate::failpoints::names::MUTATION_POST_SIDECAR_PRE_FORK) + .map_err(|error| { + OmniError::recovery_required(operation_id.clone(), error.to_string()) + })?; } // The v3 intent is now durable. Only now may a first-touch named-table @@ -1082,18 +1013,18 @@ impl StagedMutation { // when it can *prove* the ref was never changed. return Err(OmniError::recovery_required( operation_id, - format!("deferred table fork failed after recovery intent was armed: {error}"), + format!( + "deferred table fork failed after recovery intent was armed: {error}" + ), )); } } } if created_any_fork { - crate::failpoints::maybe_fail( - crate::failpoints::names::MUTATION_POST_FORK_PRE_COMMIT, - ) - .map_err(|error| { - OmniError::recovery_required(operation_id.clone(), error.to_string()) - })?; + crate::failpoints::maybe_fail(crate::failpoints::names::MUTATION_POST_FORK_PRE_COMMIT) + .map_err(|error| { + OmniError::recovery_required(operation_id.clone(), error.to_string()) + })?; } let mut updates: Vec = Vec::with_capacity(staged.len()); @@ -1136,10 +1067,8 @@ impl StagedMutation { ), )); } - committed_transactions.insert( - table_key.clone(), - outcome.committed_transaction().clone(), - ); + committed_transactions + .insert(table_key.clone(), outcome.committed_transaction().clone()); let new_ds = outcome.into_snapshot(); let state = db .storage() diff --git a/crates/omnigraph/src/failpoints.rs b/crates/omnigraph/src/failpoints.rs index f6557a80..3bee1d85 100644 --- a/crates/omnigraph/src/failpoints.rs +++ b/crates/omnigraph/src/failpoints.rs @@ -53,8 +53,7 @@ pub mod names { pub const BRANCH_DELETE_POST_TABLE_GATES: &str = "branch_delete.post_table_gates"; /// After native branch control completed its first recovery barrier, before /// it acquires schema -> branch -> table gates and performs the final check. - pub const BRANCH_CONTROL_POST_RECOVERY_BARRIER: &str = - "branch_control.post_recovery_barrier"; + pub const BRANCH_CONTROL_POST_RECOVERY_BARRIER: &str = "branch_control.post_recovery_barrier"; pub const BRANCH_MERGE_ADOPT_AFTER_APPEND_PRE_UPSERT: &str = "branch_merge.adopt_after_append_pre_upsert"; pub const BRANCH_MERGE_ADOPT_AFTER_UPSERT_PRE_DELETE: &str = @@ -62,18 +61,15 @@ pub mod names { /// Source/target heads and snapshots have been captured while the schema /// and both branch-incarnation gates are held, before merge planning or /// any durable table effect. - pub const BRANCH_MERGE_POST_AUTHORITY_CAPTURE: &str = - "branch_merge.post_authority_capture"; + pub const BRANCH_MERGE_POST_AUTHORITY_CAPTURE: &str = "branch_merge.post_authority_capture"; /// The v4 BranchMerge recovery intent is durable, before any first-touch /// target table ref is created. - pub const BRANCH_MERGE_POST_SIDECAR_PRE_FORK: &str = - "branch_merge.post_sidecar_pre_fork"; + pub const BRANCH_MERGE_POST_SIDECAR_PRE_FORK: &str = "branch_merge.post_sidecar_pre_fork"; pub const BRANCH_MERGE_POST_PHASE_B_PRE_MANIFEST_COMMIT: &str = "branch_merge.post_phase_b_pre_manifest_commit"; /// Every merge table effect is complete, but the sidecar is still in its /// pre-confirmation shape. - pub const BRANCH_MERGE_POST_EFFECTS_PRE_CONFIRM: &str = - "branch_merge.post_effects_pre_confirm"; + pub const BRANCH_MERGE_POST_EFFECTS_PRE_CONFIRM: &str = "branch_merge.post_effects_pre_confirm"; pub const BRANCH_MERGE_REWRITE_AFTER_DELETE_PRE_INDEX: &str = "branch_merge.rewrite_after_delete_pre_index"; pub const BRANCH_MERGE_REWRITE_AFTER_MERGE_PRE_DELETE: &str = @@ -82,12 +78,13 @@ pub mod names { pub const CLEANUP_RECONCILE_FORK: &str = "cleanup.reconcile_fork"; /// After cleanup's fast empty-sidecar probe, before it acquires the closed /// schema/branch/table GC gate set and performs the authoritative recheck. - pub const CLEANUP_POST_RECOVERY_CHECK_PRE_GATES: &str = - "cleanup.post_recovery_check_pre_gates"; + pub const CLEANUP_POST_RECOVERY_CHECK_PRE_GATES: &str = "cleanup.post_recovery_check_pre_gates"; pub const CLEANUP_RESOLVE_BRANCH_SNAPSHOT: &str = "cleanup.resolve_branch_snapshot"; pub const CLEANUP_TABLE_GC: &str = "cleanup.table_gc"; pub const ENSURE_INDICES_POST_PHASE_B_PRE_MANIFEST_COMMIT: &str = "ensure_indices.post_phase_b_pre_manifest_commit"; + pub const ENSURE_INDICES_POST_SIDECAR_PRE_FORK: &str = "ensure_indices.post_sidecar_pre_fork"; + pub const ENSURE_INDICES_POST_TABLE_EFFECT: &str = "ensure_indices.post_table_effect"; pub const ENSURE_INDICES_POST_STAGE_PRE_COMMIT_BTREE: &str = "ensure_indices.post_stage_pre_commit_btree"; pub const FORK_BEFORE_CLASSIFY: &str = "fork.before_classify"; @@ -151,8 +148,7 @@ pub mod names { pub const SCHEMA_APPLY_AFTER_STAGING_WRITE: &str = "schema_apply.after_staging_write"; pub const SCHEMA_APPLY_BEFORE_STAGING_WRITE: &str = "schema_apply.before_staging_write"; /// Reload owns the schema gate and is about to read/publish one contract view. - pub const SCHEMA_RELOAD_BEFORE_CONTRACT_READ: &str = - "schema_reload.before_contract_read"; + pub const SCHEMA_RELOAD_BEFORE_CONTRACT_READ: &str = "schema_reload.before_contract_read"; /// Injects a retryable `RowLevelCasContention` from `load_publish_state` so a /// test can prove the publisher's outer retry re-runs the load. pub const PUBLISH_LOAD_STATE_RETRYABLE_CONTENTION: &str = diff --git a/crates/omnigraph/tests/branching.rs b/crates/omnigraph/tests/branching.rs index 6a9ff5c4..ce652dce 100644 --- a/crates/omnigraph/tests/branching.rs +++ b/crates/omnigraph/tests/branching.rs @@ -1100,7 +1100,7 @@ async fn branch_created_from_non_main_inherits_branch_state() { } #[tokio::test] -async fn ensure_indices_on_child_branch_forks_inherited_table_ownership() { +async fn ensure_indices_on_child_branch_keeps_inherited_table_when_no_work_is_needed() { let dir = tempfile::tempdir().unwrap(); let uri = dir.path().to_str().unwrap(); let mut main = init_and_load(&dir).await; @@ -1141,7 +1141,8 @@ async fn ensure_indices_on_child_branch_forks_inherited_table_ownership() { .unwrap() .table_branch .as_deref(), - Some("experiment") + Some("feature"), + "index reconciliation must not manufacture a ref-only first-touch effect" ); assert_eq!( experiment_snap diff --git a/crates/omnigraph/tests/failpoints.rs b/crates/omnigraph/tests/failpoints.rs index 56297495..9b4ff58a 100644 --- a/crates/omnigraph/tests/failpoints.rs +++ b/crates/omnigraph/tests/failpoints.rs @@ -2856,6 +2856,186 @@ async fn recovery_rolls_forward_ensure_indices_on_feature_branch() { ); } +/// EnsureIndices first-touch ordering is sidecar-before-ref. A crash in that +/// window is a valid no-effect state: Full recovery must accept the missing +/// target ref, retire the intent, and leave the child inheriting its parent +/// until a clean retry performs real index work. +#[tokio::test] +#[serial] +async fn ensure_indices_first_touch_crash_before_ref_recovers_cleanly() { + use omnigraph::loader::{LoadMode, load_jsonl}; + + let _scenario = FailScenario::setup(); + let dir = tempfile::tempdir().unwrap(); + let uri = dir.path().to_str().unwrap().to_string(); + let mut db = Omnigraph::init(&uri, helpers::TEST_SCHEMA).await.unwrap(); + load_jsonl( + &mut db, + r#"{"type":"Person","data":{"name":"alice","age":30}} +"#, + LoadMode::Append, + ) + .await + .unwrap(); + db.branch_create("feature").await.unwrap(); + db.mutate( + "feature", + MUTATION_QUERIES, + "insert_person", + &mixed_params(&[("$name", "feature-row")], &[("$age", 31)]), + ) + .await + .unwrap(); + db.branch_create_from(omnigraph::db::ReadTarget::branch("feature"), "experiment") + .await + .unwrap(); + + // Pre-arm ownership fence: an unregistered target ref is foreign/orphaned + // state, never something a new loose sidecar may claim. + let person_uri = node_table_uri(&uri, "Person"); + let mut person = lance::Dataset::open(&person_uri).await.unwrap(); + let feature_head = person + .checkout_branch("feature") + .await + .unwrap() + .version() + .version; + person + .create_branch("experiment", ("feature", feature_head), None) + .await + .unwrap(); + let orphan_error = db + .ensure_indices_on("experiment") + .await + .expect_err("pre-existing target ref must be refused before recovery is armed"); + assert!( + orphan_error + .to_string() + .contains("refusing to claim unowned physical state"), + "unexpected orphan-ref refusal: {orphan_error}" + ); + assert!(helpers::recovery::sidecar_operation_ids(dir.path()).is_empty()); + person.delete_branch("experiment").await.unwrap(); + + { + let _failpoint = + ScopedFailPoint::new(names::ENSURE_INDICES_POST_SIDECAR_PRE_FORK, "return"); + db.ensure_indices_on("experiment") + .await + .expect_err("failpoint must fire after sidecar and before target ref creation"); + } + let operation_id = single_sidecar_operation_id(dir.path()); + drop(db); + + let recovered = Omnigraph::open(&uri) + .await + .expect("Full recovery must accept the missing first-touch target ref"); + assert!( + !dir.path() + .join("__recovery") + .join(format!("{operation_id}.json")) + .exists() + ); + let inherited = recovered + .snapshot_of(omnigraph::db::ReadTarget::branch("experiment")) + .await + .unwrap(); + assert_eq!( + inherited + .entry("node:Person") + .unwrap() + .table_branch + .as_deref(), + Some("feature") + ); + + recovered.ensure_indices_on("experiment").await.unwrap(); + let owned = recovered + .snapshot_of(omnigraph::db::ReadTarget::branch("experiment")) + .await + .unwrap(); + assert_eq!( + owned.entry("node:Person").unwrap().table_branch.as_deref(), + Some("experiment") + ); +} + +/// Mixed first-touch recovery must clean only untouched refs. A table whose +/// derived index HEAD moved is not a no-effect fork merely because the legacy +/// EnsureIndices payload lacks transaction identities; Full recovery restores +/// it through the normal loose-table rollback while accepting a missing sibling +/// ref. +#[tokio::test] +#[serial] +async fn ensure_indices_mixed_first_touch_rollback_does_not_delete_moved_ref() { + let _scenario = FailScenario::setup(); + let dir = tempfile::tempdir().unwrap(); + let uri = dir.path().to_str().unwrap().to_string(); + let db = Omnigraph::init(&uri, helpers::TEST_SCHEMA).await.unwrap(); + db.branch_create("feature").await.unwrap(); + db.load( + "feature", + "{\"type\":\"Person\",\"data\":{\"name\":\"alice\",\"age\":30}}\n\ + {\"type\":\"Company\",\"data\":{\"name\":\"acme\"}}", + LoadMode::Merge, + ) + .await + .unwrap(); + db.branch_create_from(omnigraph::db::ReadTarget::branch("feature"), "experiment") + .await + .unwrap(); + + { + let _failpoint = ScopedFailPoint::new(names::ENSURE_INDICES_POST_TABLE_EFFECT, "return"); + db.ensure_indices_on("experiment") + .await + .expect_err("failpoint must stop after the first first-touch table effect"); + } + let operation_id = single_sidecar_operation_id(dir.path()); + drop(db); + + { + let _failpoint = + ScopedFailPoint::new(names::RECOVERY_POST_ROLLBACK_PUBLISH_PRE_AUDIT, "return"); + let first_recovery = Omnigraph::open(&uri).await; + assert!( + first_recovery.is_err(), + "first recovery must stop after publishing its fixed rollback" + ); + } + assert!( + dir.path() + .join("__recovery") + .join(format!("{operation_id}.json")) + .exists(), + "interrupted rollback must retain its outcome-bearing sidecar" + ); + + let recovered = Omnigraph::open(&uri) + .await + .expect("mixed first-touch rollback retry must finalize its fixed outcome"); + assert!( + !dir.path() + .join("__recovery") + .join(format!("{operation_id}.json")) + .exists() + ); + assert_eq!( + helpers::count_rows_branch(&recovered, "experiment", "node:Person").await, + 1 + ); + assert_eq!( + helpers::count_rows_branch(&recovered, "experiment", "node:Company").await, + 1 + ); + drop(recovered); + assert_eq!( + recovery_audit_kinds(dir.path()).await, + vec!["RolledBack"], + "an interrupted index rollback must never be reclassified as RolledForward" + ); +} + /// Refresh-time recovery (Option B): the in-process `Omnigraph::refresh` /// runs roll-forward-only recovery, closing the long-running-server /// residual without restart. @@ -4683,6 +4863,328 @@ edge WorksAt: Person -> Company ); } +/// A metadata-only SchemaApply still changes the accepted schema contract, so +/// it needs a durable intent even though it has no table pins. A crash before +/// schema staging must roll that empty-table intent back unambiguously. +#[tokio::test] +#[serial] +async fn metadata_only_schema_apply_before_staging_rolls_back_on_next_open() { + let _scenario = FailScenario::setup(); + let dir = tempfile::tempdir().unwrap(); + let uri = dir.path().to_str().unwrap().to_string(); + let db = Omnigraph::init(&uri, helpers::TEST_SCHEMA).await.unwrap(); + let indexed_schema = helpers::TEST_SCHEMA.replace("age: I32?", "age: I32? @index"); + + let operation_id = { + let _failpoint = ScopedFailPoint::new(names::SCHEMA_APPLY_BEFORE_STAGING_WRITE, "return"); + let err = db.apply_schema(&indexed_schema).await.unwrap_err(); + assert!( + err.to_string() + .contains("injected failpoint triggered: schema_apply.before_staging_write"), + "unexpected error: {err}" + ); + single_sidecar_operation_id(dir.path()) + }; + drop(db); + + let recovered = Omnigraph::open(&uri) + .await + .expect("empty-table SchemaApply intent should roll back cleanly"); + drop(recovered); + assert_post_recovery_invariants( + dir.path(), + &operation_id, + RecoveryExpectation::RolledBack { tables: vec![] }, + ) + .await + .unwrap(); + assert_no_staging_files(dir.path()); + let live_schema = std::fs::read_to_string(dir.path().join("_schema.pg")).unwrap(); + assert!( + !live_schema.contains("age: I32? @index"), + "a pre-staging crash must retain the old accepted schema" + ); +} + +/// A recovery-owned rollback commit also advances the manifest. The durable +/// SchemaApply Phase-C marker—not numeric version movement—must therefore keep +/// a pre-staging crash classified as RolledBack when recovery itself is retried +/// after publishing rollback but before recording its audit. +#[tokio::test] +#[serial] +async fn metadata_only_schema_apply_rollback_retry_never_flips_forward() { + let _scenario = FailScenario::setup(); + let dir = tempfile::tempdir().unwrap(); + let uri = dir.path().to_str().unwrap().to_string(); + let db = Omnigraph::init(&uri, helpers::TEST_SCHEMA).await.unwrap(); + let indexed_schema = helpers::TEST_SCHEMA.replace("age: I32?", "age: I32? @index"); + + { + let _failpoint = ScopedFailPoint::new(names::SCHEMA_APPLY_BEFORE_STAGING_WRITE, "return"); + db.apply_schema(&indexed_schema).await.unwrap_err(); + } + let operation_id = single_sidecar_operation_id(dir.path()); + drop(db); + + { + let _failpoint = + ScopedFailPoint::new(names::RECOVERY_POST_ROLLBACK_PUBLISH_PRE_AUDIT, "return"); + let first_recovery = Omnigraph::open(&uri).await; + assert!( + first_recovery.is_err(), + "first recovery must stop after its rollback publish" + ); + } + assert!( + dir.path() + .join("__recovery") + .join(format!("{operation_id}.json")) + .exists(), + "failed audit phase must retain the original sidecar" + ); + + let recovered = Omnigraph::open(&uri) + .await + .expect("the next Full recovery must finish the rollback"); + drop(recovered); + assert_eq!( + recovery_audit_kinds(dir.path()).await, + vec!["RolledBack"], + "rollback retry must never be reclassified or additionally audited as RolledForward" + ); + let live_schema = std::fs::read_to_string(dir.path().join("_schema.pg")).unwrap(); + assert!(!live_schema.contains("age: I32? @index")); +} + +/// The same retry discriminator is required when SchemaApply moved a table +/// before staging. After recovery restores and publishes the table, numeric +/// pins look like a stale roll-forward; the old live schema hash proves the +/// outcome is still RolledBack. +#[tokio::test] +#[serial] +async fn pinned_schema_apply_rollback_retry_never_flips_forward() { + let _scenario = FailScenario::setup(); + let dir = tempfile::tempdir().unwrap(); + let uri = dir.path().to_str().unwrap().to_string(); + let db = helpers::init_and_load(&dir).await; + let desired = helpers::TEST_SCHEMA.replace( + " age: I32?\n}", + " age: I32?\n nickname: String?\n}", + ); + + { + let _failpoint = ScopedFailPoint::new(names::SCHEMA_APPLY_BEFORE_STAGING_WRITE, "return"); + db.apply_schema(&desired).await.unwrap_err(); + } + drop(db); + + { + let _failpoint = + ScopedFailPoint::new(names::RECOVERY_POST_ROLLBACK_PUBLISH_PRE_AUDIT, "return"); + let first_recovery = Omnigraph::open(&uri).await; + assert!(first_recovery.is_err()); + } + + let recovered = Omnigraph::open(&uri) + .await + .expect("pinned SchemaApply rollback retry must converge"); + assert_eq!(helpers::count_rows(&recovered, "node:Person").await, 4); + drop(recovered); + assert_eq!( + recovery_audit_kinds(dir.path()).await, + vec!["RolledBack"], + "restored table pins must not be mistaken for stale roll-forward" + ); + let live_schema = std::fs::read_to_string(dir.path().join("_schema.pg")).unwrap(); + assert!(!live_schema.contains("nickname: String?")); +} + +/// The other half of the zero-table protocol: once schema staging exists, the +/// same metadata-only intent must roll forward and accept the new contract. +#[tokio::test] +#[serial] +async fn metadata_only_schema_apply_after_staging_rolls_forward_on_next_open() { + let _scenario = FailScenario::setup(); + let dir = tempfile::tempdir().unwrap(); + let uri = dir.path().to_str().unwrap().to_string(); + let db = Omnigraph::init(&uri, helpers::TEST_SCHEMA).await.unwrap(); + let indexed_schema = helpers::TEST_SCHEMA.replace("age: I32?", "age: I32? @index"); + + let operation_id = { + let _failpoint = ScopedFailPoint::new(names::SCHEMA_APPLY_AFTER_STAGING_WRITE, "return"); + let err = db.apply_schema(&indexed_schema).await.unwrap_err(); + assert!( + err.to_string() + .contains("injected failpoint triggered: schema_apply.after_staging_write"), + "unexpected error: {err}" + ); + single_sidecar_operation_id(dir.path()) + }; + drop(db); + + let recovered = Omnigraph::open(&uri) + .await + .expect("empty-table SchemaApply intent should roll forward cleanly"); + drop(recovered); + assert_post_recovery_invariants( + dir.path(), + &operation_id, + RecoveryExpectation::RolledForward { tables: vec![] }, + ) + .await + .unwrap(); + assert_no_staging_files(dir.path()); + let live_schema = std::fs::read_to_string(dir.path().join("_schema.pg")).unwrap(); + assert!( + live_schema.contains("age: I32? @index"), + "an after-staging crash must promote the new accepted schema" + ); +} + +/// Schema-file recovery runs before sidecar processing. If it promotes staging +/// and the sweep then crashes, the next pass sees marker=false and no staging; +/// the live target hash must carry the decision forward into the normal +/// manifest roll-forward path. +#[tokio::test] +#[serial] +async fn metadata_only_schema_apply_recovers_after_promotion_prepass_crash() { + let _scenario = FailScenario::setup(); + let dir = tempfile::tempdir().unwrap(); + let uri = dir.path().to_str().unwrap().to_string(); + let db = Omnigraph::init(&uri, helpers::TEST_SCHEMA).await.unwrap(); + let indexed_schema = helpers::TEST_SCHEMA.replace("age: I32?", "age: I32? @index"); + + { + let _failpoint = ScopedFailPoint::new(names::SCHEMA_APPLY_AFTER_STAGING_WRITE, "return"); + db.apply_schema(&indexed_schema).await.unwrap_err(); + } + drop(db); + + { + let _failpoint = ScopedFailPoint::new(names::RECOVERY_POST_LIST_PRE_GATES, "return"); + let interrupted = Omnigraph::open(&uri).await; + assert!( + interrupted.is_err(), + "first recovery must stop after schema promotion and before sidecar processing" + ); + } + + let recovered = Omnigraph::open(&uri) + .await + .expect("live target identity must let the next pass finish roll-forward"); + drop(recovered); + assert_eq!( + recovery_audit_kinds(dir.path()).await, + vec!["RolledForward"] + ); + let live_schema = std::fs::read_to_string(dir.path().join("_schema.pg")).unwrap(); + assert!(live_schema.contains("age: I32? @index")); +} + +/// Phase D is outside SchemaApply's logical commit boundary too. With no table +/// pins, recovery must use the durable Phase-C marker plus target schema hash to +/// recognize that the metadata-only commit and promotion are already visible; +/// it must not mislabel the completed apply as RolledBack or wedge the next +/// write waiting for a full reopen. +#[tokio::test] +#[serial] +async fn metadata_only_schema_apply_delete_failure_heals_on_next_write() { + let _scenario = FailScenario::setup(); + let dir = tempfile::tempdir().unwrap(); + let uri = dir.path().to_str().unwrap().to_string(); + let db = Omnigraph::init(&uri, helpers::TEST_SCHEMA).await.unwrap(); + let indexed_schema = helpers::TEST_SCHEMA.replace("age: I32?", "age: I32? @index"); + + let operation_id = { + let _failpoint = ScopedFailPoint::new(names::RECOVERY_SIDECAR_DELETE, "return"); + db.apply_schema(&indexed_schema) + .await + .expect("Phase-D delete failure must not fail an already-visible schema apply"); + single_sidecar_operation_id(dir.path()) + }; + + db.mutate( + "main", + MUTATION_QUERIES, + "insert_person", + &mixed_params(&[("$name", "after-schema-heal")], &[("$age", 36)]), + ) + .await + .expect("the next write must heal an already-visible metadata-only apply in process"); + + assert!( + !dir.path() + .join("__recovery") + .join(format!("{operation_id}.json")) + .exists(), + "the next write's entry heal must retire the stale SchemaApply sidecar" + ); + assert_eq!( + recovery_audit_kinds(dir.path()).await, + vec!["RolledForward"], + "the completed SchemaApply must have exactly one forward audit and no rollback audit" + ); + let live_schema = std::fs::read_to_string(dir.path().join("_schema.pg")).unwrap(); + assert!( + live_schema.contains("age: I32? @index"), + "healing a stale Phase-D sidecar must retain the accepted schema" + ); + assert_eq!(helpers::count_rows(&db, "node:Person").await, 1); +} + +/// A failed AddType can leave its newly-created dataset behind even after the +/// legacy rollback retires the sidecar. A retry must not adopt that orphan as +/// if it were this attempt's effect; target absence is checked before arming a +/// new recovery intent. +#[tokio::test] +#[serial] +async fn schema_apply_retry_refuses_orphaned_add_type_target_before_sidecar() { + let _scenario = FailScenario::setup(); + let dir = tempfile::tempdir().unwrap(); + let uri = dir.path().to_str().unwrap().to_string(); + let db = Omnigraph::init(&uri, SCHEMA_V1).await.unwrap(); + + { + let _failpoint = ScopedFailPoint::new(names::SCHEMA_APPLY_BEFORE_STAGING_WRITE, "return"); + db.apply_schema(SCHEMA_V2_ADDED_TYPE) + .await + .expect_err("the pre-staging failpoint must leave the AddType intent pending"); + } + drop(db); + let recovered = Omnigraph::open(&uri) + .await + .expect("Full recovery should roll back the pre-staging AddType intent"); + assert!( + recovered + .snapshot_of(omnigraph::db::ReadTarget::branch("main")) + .await + .unwrap() + .entry("node:Company") + .is_none(), + "rollback must not register the orphan target" + ); + + let err = recovered + .apply_schema(SCHEMA_V2_ADDED_TYPE) + .await + .expect_err("retry must refuse the unregistered dataset left by the failed attempt"); + assert!( + err.to_string().contains("existing dataset") + && err + .to_string() + .contains("refusing to claim unowned physical state"), + "retry should explain the orphan-ownership refusal; got: {err}" + ); + assert!( + !dir.path().join("__recovery").exists() + || std::fs::read_dir(dir.path().join("__recovery")) + .unwrap() + .next() + .is_none(), + "target absence must fail before the retry writes a sidecar" + ); +} + #[tokio::test] #[serial] async fn schema_apply_phase_b_failure_recovered_on_next_open() { @@ -5545,9 +6047,8 @@ async fn branch_merge_post_effect_target_advance_requires_recovery_and_preserves names::BRANCH_MERGE_POST_PHASE_B_PRE_MANIFEST_COMMIT, ); let merge_handle = std::sync::Arc::clone(&merge_db); - let merge_task = tokio::spawn(async move { - merge_handle.branch_merge("source", "target").await - }); + let merge_task = + tokio::spawn(async move { merge_handle.branch_merge("source", "target").await }); merge_rv.wait_until_reached().await; // Publish a logically redundant Company pin directly on the target. The @@ -5643,9 +6144,8 @@ async fn branch_merge_post_effect_same_table_advance_fails_closed() { ); let merge_handle = std::sync::Arc::clone(&merge_db); - let merge_task = tokio::spawn(async move { - merge_handle.branch_merge("source", "target").await - }); + let merge_task = + tokio::spawn(async move { merge_handle.branch_merge("source", "target").await }); merge_rv.wait_until_reached().await; let person_uri = node_table_uri(&uri, "Person"); @@ -6125,9 +6625,8 @@ async fn branch_merge_confirmation_rejects_foreign_append_before_index_tail() { names::BRANCH_MERGE_REWRITE_AFTER_DELETE_PRE_INDEX, ); let merge_handle = std::sync::Arc::clone(&merge_db); - let merge_task = tokio::spawn(async move { - merge_handle.branch_merge("source", "target").await - }); + let merge_task = + tokio::spawn(async move { merge_handle.branch_merge("source", "target").await }); merge_rv.wait_until_reached().await; // The merge's logical data transaction is now at HEAD, but its index build @@ -7270,9 +7769,8 @@ async fn branch_merge_fences_target_delete_recreate_aba() { helpers::failpoint::Rendezvous::park_first(names::BRANCH_CONTROL_POST_RECOVERY_BARRIER); let merge_handle = std::sync::Arc::clone(&merge_db); - let merge_task = tokio::spawn(async move { - merge_handle.branch_merge("source", "target").await - }); + let merge_task = + tokio::spawn(async move { merge_handle.branch_merge("source", "target").await }); merge_rv.wait_until_reached().await; let control_handle = std::sync::Arc::clone(&control_db); @@ -7294,12 +7792,10 @@ async fn branch_merge_fences_target_delete_recreate_aba() { // The control task is known to be immediately before its gate acquisition. // It must remain blocked while merge holds the target-incarnation gate. - let control_blocked = tokio::time::timeout( - std::time::Duration::from_millis(250), - &mut control_task, - ) - .await - .is_err(); + let control_blocked = + tokio::time::timeout(std::time::Duration::from_millis(250), &mut control_task) + .await + .is_err(); let target_unchanged_while_parked = lance::Dataset::open(&person_uri) .await .unwrap() @@ -7382,19 +7878,15 @@ async fn branch_merge_fences_concurrent_sync_on_same_handle() { helpers::failpoint::Rendezvous::park_first(names::BRANCH_MERGE_POST_AUTHORITY_CAPTURE); let merge_handle = std::sync::Arc::clone(&db); - let merge_task = tokio::spawn(async move { - merge_handle.branch_merge("source", "target").await - }); + let merge_task = + tokio::spawn(async move { merge_handle.branch_merge("source", "target").await }); merge_rv.wait_until_reached().await; let sync_handle = std::sync::Arc::clone(&db); let mut sync_task = tokio::spawn(async move { sync_handle.sync_branch("other").await }); - let sync_blocked = tokio::time::timeout( - std::time::Duration::from_millis(250), - &mut sync_task, - ) - .await - .is_err(); + let sync_blocked = tokio::time::timeout(std::time::Duration::from_millis(250), &mut sync_task) + .await + .is_err(); merge_rv.release(); assert!( sync_blocked, @@ -7429,9 +7921,8 @@ async fn branch_merge_rejects_fresh_target_manifest_change_before_effects() { helpers::failpoint::Rendezvous::park_first(names::BRANCH_MERGE_POST_AUTHORITY_CAPTURE); let merge_handle = std::sync::Arc::clone(&merge_db); - let merge_task = tokio::spawn(async move { - merge_handle.branch_merge("source", "target").await - }); + let merge_task = + tokio::spawn(async move { merge_handle.branch_merge("source", "target").await }); merge_rv.wait_until_reached().await; let before = helpers::version_branch(&merge_db, "target").await.unwrap(); @@ -7485,6 +7976,90 @@ async fn branch_merge_rejects_fresh_target_manifest_change_before_effects() { ); } +/// The final merge fence covers physical state as well as manifest authority. +/// Drift injected after authority capture but before the final table gates must +/// be rejected before BranchMerge writes a recovery sidecar that could falsely +/// claim the pre-existing Lance commit. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +#[serial] +async fn branch_merge_rejects_late_uncovered_target_drift_before_sidecar() { + let _scenario = FailScenario::setup(); + let dir = tempfile::tempdir().unwrap(); + let uri = dir.path().to_str().unwrap().to_string(); + let db = helpers::init_and_load(&dir).await; + db.branch_create("source").await.unwrap(); + db.mutate( + "source", + MUTATION_QUERIES, + "insert_person", + &mixed_params(&[("$name", "source-only")], &[("$age", 34)]), + ) + .await + .unwrap(); + db.mutate( + "main", + MUTATION_QUERIES, + "insert_person", + &mixed_params(&[("$name", "target-only")], &[("$age", 35)]), + ) + .await + .unwrap(); + let manifest_before = db + .snapshot_of(omnigraph::db::ReadTarget::branch("main")) + .await + .unwrap() + .entry("node:Person") + .unwrap() + .table_version; + let db = std::sync::Arc::new(db); + let merge_rv = + helpers::failpoint::Rendezvous::park_first(names::BRANCH_MERGE_POST_AUTHORITY_CAPTURE); + let merge_handle = std::sync::Arc::clone(&db); + let merge_task = tokio::spawn(async move { merge_handle.branch_merge("source", "main").await }); + merge_rv.wait_until_reached().await; + + let person_uri = node_table_uri(&uri, "Person"); + let mut raw_main = lance::Dataset::open(&person_uri).await.unwrap(); + helpers::lance_delete_inline(&mut raw_main, "1 = 2").await; + let raw_head = raw_main.version().version; + assert!( + raw_head > manifest_before, + "fixture must create uncovered drift" + ); + merge_rv.release(); + + let err = merge_task + .await + .unwrap() + .expect_err("merge must reject physical drift that predates its recovery intent"); + assert!( + err.to_string().contains("omnigraph repair"), + "error should direct the operator to repair; got: {err}" + ); + assert!( + !dir.path().join("__recovery").exists() + || std::fs::read_dir(dir.path().join("__recovery")) + .unwrap() + .next() + .is_none(), + "the refusal must happen before BranchMerge arms recovery" + ); + let manifest_after = db + .snapshot_of(omnigraph::db::ReadTarget::branch("main")) + .await + .unwrap() + .entry("node:Person") + .unwrap() + .table_version; + let head_after = lance::Dataset::open(&person_uri) + .await + .unwrap() + .version() + .version; + assert_eq!(manifest_after, manifest_before); + assert_eq!(head_after, raw_head); +} + /// Source is a captured immutable input, not part of the target publisher's /// atomic read set. A later commit on the same source incarnation must not make /// the merge substitute the newer source head (or reject an otherwise-valid @@ -7502,9 +8077,8 @@ async fn branch_merge_source_advance_keeps_captured_source_parent() { helpers::failpoint::Rendezvous::park_first(names::BRANCH_MERGE_POST_AUTHORITY_CAPTURE); let merge_handle = std::sync::Arc::clone(&merge_db); - let merge_task = tokio::spawn(async move { - merge_handle.branch_merge("source", "target").await - }); + let merge_task = + tokio::spawn(async move { merge_handle.branch_merge("source", "target").await }); merge_rv.wait_until_reached().await; // Model a foreign source writer without changing row content: advance the @@ -7663,9 +8237,8 @@ async fn branch_merge_rechecks_late_sidecar_after_table_gates() { helpers::failpoint::Rendezvous::park_first(names::BRANCH_MERGE_POST_AUTHORITY_CAPTURE); let merge_handle = std::sync::Arc::clone(&merge_db); - let merge_task = tokio::spawn(async move { - merge_handle.branch_merge("source", "target").await - }); + let merge_task = + tokio::spawn(async move { merge_handle.branch_merge("source", "target").await }); merge_rv.wait_until_reached().await; const OPERATION_ID: &str = "01H000000000000000000LATE"; @@ -7716,9 +8289,8 @@ async fn cleanup_rechecks_sidecars_under_gc_gates() { let uri = dir.path().to_str().unwrap().to_string(); drop(helpers::init_and_load(&dir).await); - let cleanup_rv = helpers::failpoint::Rendezvous::park_first( - names::CLEANUP_POST_RECOVERY_CHECK_PRE_GATES, - ); + let cleanup_rv = + helpers::failpoint::Rendezvous::park_first(names::CLEANUP_POST_RECOVERY_CHECK_PRE_GATES); let cleanup_uri = uri.clone(); let cleanup_task = tokio::spawn(async move { let mut db = Omnigraph::open(&cleanup_uri).await.unwrap(); @@ -7832,7 +8404,9 @@ async fn full_recovery_rereads_sidecar_body_after_discovery() { .as_array_mut() .expect("branch-merge sidecar tables must be an array"); assert!( - tables.iter().any(|table| !table["confirmed_version"].is_null()), + tables + .iter() + .any(|table| !table["confirmed_version"].is_null()), "fixture must begin with a confirmed BranchMerge residual" ); for table in tables { diff --git a/crates/omnigraph/tests/maintenance.rs b/crates/omnigraph/tests/maintenance.rs index 258bb16d..50af9a50 100644 --- a/crates/omnigraph/tests/maintenance.rs +++ b/crates/omnigraph/tests/maintenance.rs @@ -111,7 +111,10 @@ async fn optimize_on_empty_graph_returns_stats_per_table_with_no_changes() { .iter() .find(|s| s.table_key == "__manifest") .expect("optimize stats missing internal table __manifest"); - assert!(!s.committed, "__manifest should be a no-op on an empty graph"); + assert!( + !s.committed, + "__manifest should be a no-op on an empty graph" + ); } #[tokio::test] @@ -267,7 +270,9 @@ async fn optimize_clears_stale_auto_cleanup_and_preserves_versions() { let ds = Dataset::open(&manifest_uri).await.unwrap(); // (a) the stale auto_cleanup config was cleared (non-destructive by construction). assert!( - !ds.config().keys().any(|k| k.starts_with("lance.auto_cleanup.")), + !ds.config() + .keys() + .any(|k| k.starts_with("lance.auto_cleanup.")), "optimize must clear stale auto_cleanup config; config = {:?}", ds.config() ); @@ -290,7 +295,12 @@ async fn optimize_clears_stale_auto_cleanup_and_preserves_versions() { #[tokio::test] async fn optimize_clears_stale_auto_cleanup_on_data_tables_too() { let dir = tempfile::tempdir().unwrap(); - let root = dir.path().to_str().unwrap().trim_end_matches('/').to_string(); + let root = dir + .path() + .to_str() + .unwrap() + .trim_end_matches('/') + .to_string(); let mut db = init_and_load(&dir).await; add_person_fragments(&mut db).await; // multiple fragments → will_compact @@ -330,7 +340,9 @@ async fn optimize_clears_stale_auto_cleanup_on_data_tables_too() { let ds = Dataset::open(&person_full).await.unwrap(); // (a) the stale auto_cleanup config was cleared (non-destructive by construction). assert!( - !ds.config().keys().any(|k| k.starts_with("lance.auto_cleanup.")), + !ds.config() + .keys() + .any(|k| k.starts_with("lance.auto_cleanup.")), "optimize must clear stale auto_cleanup config on data tables; config = {:?}", ds.config() ); @@ -856,6 +868,132 @@ async fn delete_only_mutation_refuses_uncovered_drift_before_inline_commit() { ); } +fn recovery_sidecar_count(dir: &tempfile::TempDir) -> usize { + let recovery = dir.path().join("__recovery"); + if !recovery.exists() { + return 0; + } + std::fs::read_dir(recovery).unwrap().count() +} + +#[tokio::test] +async fn schema_apply_refuses_uncovered_drift_before_arming_recovery() { + let dir = tempfile::tempdir().unwrap(); + let root = dir + .path() + .to_str() + .unwrap() + .trim_end_matches('/') + .to_string(); + let mut db = init_and_load(&dir).await; + let (manifest_before, head_before, _) = forge_person_compaction_drift(&mut db, &root).await; + let desired = TEST_SCHEMA.replace( + " age: I32?\n}", + " age: I32?\n nickname: String?\n}", + ); + + let err = db + .apply_schema(&desired) + .await + .expect_err("schema apply must not claim or fold uncovered table drift"); + assert!( + err.to_string().contains("omnigraph repair"), + "error should direct the operator to repair; got: {err}" + ); + assert_eq!( + recovery_sidecar_count(&dir), + 0, + "a pre-existing effect must be rejected before SchemaApply writes its sidecar" + ); + + let (manifest_after, head_after, _) = person_manifest_and_head(&db, &root).await; + assert_eq!(manifest_after, manifest_before); + assert_eq!(head_after, head_before); +} + +#[tokio::test] +async fn ensure_indices_refuses_uncovered_drift_before_arming_recovery() { + let dir = tempfile::tempdir().unwrap(); + let root = dir + .path() + .to_str() + .unwrap() + .trim_end_matches('/') + .to_string(); + let uri = dir.path().to_str().unwrap(); + let mut db = Omnigraph::init(uri, TEST_SCHEMA).await.unwrap(); + load_jsonl(&mut db, TEST_DATA, LoadMode::Overwrite) + .await + .unwrap(); + let (manifest_before, head_before, _) = forge_person_delete_drift(&db, &root).await; + + let err = db + .ensure_indices() + .await + .expect_err("index reconciliation must not claim or fold uncovered table drift"); + assert!( + err.to_string().contains("omnigraph repair"), + "error should direct the operator to repair; got: {err}" + ); + assert_eq!( + recovery_sidecar_count(&dir), + 0, + "a pre-existing effect must be rejected before EnsureIndices writes its sidecar" + ); + let (manifest_after, head_after, _) = person_manifest_and_head(&db, &root).await; + assert_eq!(manifest_after, manifest_before); + assert_eq!(head_after, head_before); + + db.repair(RepairOptions { + confirm: true, + force: true, + }) + .await + .expect("explicit repair should remain available after the refusal"); + db.ensure_indices() + .await + .expect("index reconciliation should succeed once repair establishes ownership"); +} + +#[tokio::test] +async fn branch_merge_refuses_uncovered_target_drift_before_arming_recovery() { + let dir = tempfile::tempdir().unwrap(); + let root = dir + .path() + .to_str() + .unwrap() + .trim_end_matches('/') + .to_string(); + let mut db = init_and_load(&dir).await; + db.branch_create("feature").await.unwrap(); + db.mutate( + "feature", + MUTATION_QUERIES, + "insert_person", + &mixed_params(&[("$name", "feature-only")], &[("$age", 39)]), + ) + .await + .unwrap(); + let (manifest_before, head_before, _) = forge_person_compaction_drift(&mut db, &root).await; + + let err = db + .branch_merge("feature", "main") + .await + .expect_err("branch merge must not claim or fold uncovered target drift"); + assert!( + err.to_string().contains("omnigraph repair"), + "error should direct the operator to repair; got: {err}" + ); + assert_eq!( + recovery_sidecar_count(&dir), + 0, + "a pre-existing effect must be rejected before BranchMerge writes its sidecar" + ); + let (manifest_after, head_after, _) = person_manifest_and_head(&db, &root).await; + assert_eq!(manifest_after, manifest_before); + assert_eq!(head_after, head_before); +} + // Regression: `optimize` must REFUSE when an unresolved recovery sidecar is // pending. Operating on an unrecovered graph could publish a partial write that // the all-or-nothing recovery sweep would roll back; the operator must reopen diff --git a/crates/omnigraph/tests/schema_apply.rs b/crates/omnigraph/tests/schema_apply.rs index 0f70c01f..47d04e17 100644 --- a/crates/omnigraph/tests/schema_apply.rs +++ b/crates/omnigraph/tests/schema_apply.rs @@ -104,6 +104,28 @@ query insert_project($name: String) { ) .await .expect("source write must use the token-bound post-apply catalog"); + let project_before_indices = stale_handle + .snapshot_of(ReadTarget::branch("source")) + .await + .unwrap() + .entry("node:Project") + .unwrap() + .table_version; + stale_handle + .ensure_indices_on("source") + .await + .expect("index planning must use the same token-bound post-apply catalog"); + let project_after_indices = stale_handle + .snapshot_of(ReadTarget::branch("source")) + .await + .unwrap() + .entry("node:Project") + .unwrap() + .table_version; + assert!( + project_after_indices > project_before_indices, + "the stale handle must discover and build Project's declared key index" + ); assert_eq!( stale_handle .branch_merge("source", "target") @@ -151,13 +173,12 @@ async fn stale_handle_branch_delete_gates_tables_added_by_schema_apply() { delete_rv.wait_until_reached().await; let index_handle = Arc::clone(&index_reconciler); - let mut index_task = tokio::spawn(async move { index_handle.ensure_indices_on("target").await }); - let index_blocked = tokio::time::timeout( - std::time::Duration::from_millis(250), - &mut index_task, - ) - .await - .is_err(); + let mut index_task = + tokio::spawn(async move { index_handle.ensure_indices_on("target").await }); + let index_blocked = + tokio::time::timeout(std::time::Duration::from_millis(250), &mut index_task) + .await + .is_err(); delete_rv.release(); assert!( index_blocked, @@ -1099,9 +1120,10 @@ async fn apply_schema_defers_vector_index_on_empty_table() { body: String? @index\n \ embedding: Vector(8) @index\n\ }\n"; - let result = db.apply_schema(v2).await.expect( - "schema apply must succeed: an empty-table vector @index is deferred, not fatal", - ); + let result = db + .apply_schema(v2) + .await + .expect("schema apply must succeed: an empty-table vector @index is deferred, not fatal"); assert!(result.applied, "the scalar @index change must apply"); // The deferred declarations are not dropped: after data arrives, the @@ -1150,7 +1172,10 @@ async fn index_only_constraint_apply_touches_no_table_data() { // Add an @index on the existing `n` column. let v2 = "node Doc {\n slug: String @key\n n: I64 @index\n}\n"; - let result = db.apply_schema(v2).await.expect("index-only apply must succeed"); + let result = db + .apply_schema(v2) + .await + .expect("index-only apply must succeed"); assert!(result.applied, "the @index addition must apply"); let after = db diff --git a/docs/dev/invariants.md b/docs/dev/invariants.md index 3e602e7a..1b121fcc 100644 --- a/docs/dev/invariants.md +++ b/docs/dev/invariants.md @@ -84,6 +84,11 @@ converge the physical state. 5. **Recovery is part of the commit protocol.** Writers that can advance Lance HEAD before manifest publish must write `__recovery/{ulid}.json` sidecars. + Under their final schema → branch → table gates, they must first prove every + existing physical target still equals its manifest pin; a new sidecar must + never claim an older writer's effect or uncovered drift. First-touch refs use + sidecar-before-ref ordering. A non-noop SchemaApply writes a sidecar even + with zero table pins because schema-contract staging is durable state. `Omnigraph::open` in read-write mode runs the all-or-nothing sweep; the write entry points (`load_as`, `mutate_as`, `apply_schema_as`, `branch_merge_as`) and `refresh` run roll-forward-only recovery in-process, diff --git a/docs/dev/writes.md b/docs/dev/writes.md index baf7bbcd..bfe1de1f 100644 --- a/docs/dev/writes.md +++ b/docs/dev/writes.md @@ -54,8 +54,11 @@ Mutation and load use a closed prepare → effect → publish attempt: identity instead: Lance branch-local files cannot be staged until its target ref exists; 3. acquire the schema gate, branch gate, then sorted table queues; re-check for - a relevant sidecar armed since step 1, then revalidate the token. Any - unresolved relevant intent returns typed `RecoveryRequired` before effects; + a relevant sidecar armed since step 1, then revalidate the token and require + every existing physical target's live Lance HEAD to equal its manifest pin. + Any unresolved relevant intent returns typed `RecoveryRequired`; uncovered + HEAD drift points to `omnigraph repair`. Both fail before this attempt arms + recovery; 4. on a pre-effect mismatch, discard the complete attempt. Append/Insert/Merge reprepare with a bounded retry; strict Update/Delete/Overwrite return typed `ReadSetChanged`; @@ -89,9 +92,13 @@ RFC-022 adapter contract: the immutable base/source/target snapshots outside table gates; 3. acquire the conservative all-catalog source/target table envelope, re-list recovery intent, revalidate the complete target token, and revalidate the - source incarnation. A target change returns typed `ReadSetChanged` before - effects. A later source-head advance is allowed: the contract is "merge the - captured source commit," never "substitute whatever source is latest"; + source incarnation. Before arming, every existing target ref that will receive + a physical effect must also have live Lance HEAD equal to its captured target + manifest pin; the verified handle is carried into the effect instead of being + reopened. First-touch refs remain absent until after the sidecar. A target + change returns typed `ReadSetChanged` before effects. A later source-head + advance is allowed: the contract is "merge the captured source commit," never + "substitute whatever source is latest"; 4. pre-mint the merge lineage and each table's ordered Lance data-transaction chain, then arm a schema-v4 BranchMerge sidecar before the first HEAD advance or first-touch table ref. Logical data steps commit with those exact @@ -341,6 +348,14 @@ are left at `Lance HEAD = manifest_pinned + 1`. `branch_merge_on_current_target`, `ensure_indices_for_branch`, `optimize_all_tables`): +Before Phase A, under the writer's final schema → branch → table gates, existing +physical targets must still match their manifest pins. Ahead drift is never folded +or claimed by manufacturing a new sidecar; it is attributed to an existing recovery +intent or refused with explicit `omnigraph repair` guidance. First-touch targets use +the separate sidecar-before-ref protocol. SchemaApply also verifies that AddType and +RenameType target dataset paths are absent, so recovery cannot register an orphan or +foreign dataset as if this apply created it. + 1. **Phase A**: writer writes a sidecar JSON to `__recovery/{ulid}.json` BEFORE its first independently durable physical effect (including a first-touch Lance branch ref) or HEAD-advancing commit @@ -355,7 +370,18 @@ are left at `Lance HEAD = manifest_pinned + 1`. carries its pre-minted transaction identity. Branch merge uses schema v4: it distinguishes multi-commit HEAD effects from ref-only forks, records each multi-commit effect's ordered exact transaction chain, and records the - complete intended manifest delta, including pointer-only slots. + complete intended manifest delta, including pointer-only slots. SchemaApply + uses schema v5 as a narrow bridge: every non-noop apply records its target + schema hash even when the table-pin set is empty, then durably marks the + sidecar manifest-published immediately after Phase C. This distinguishes a + pre-staging rollback from completed schema promotion whose Phase-D delete + failed; table effects remain on the legacy loose classifier until the full + exact SchemaApply adapter lands. EnsureIndices uses schema v6 as a second + narrow bridge: table effects are still loosely classified, but the sidecar + pre-mints its rollback commit id. Before the first restore, recovery durably + binds the original per-table rollback audit plan, so an interruption after + rollback publish retries as that same `RolledBack` outcome instead of being + mistaken for a stale roll-forward. 2. **Phase B**: writer's per-table `commit_staged` loop runs. - **Phase-B confirmation:** a schema-v4 `BranchMerge` writer advances each table's HEAD by *several* exact commits (append → upsert → @@ -440,8 +466,9 @@ recovery sweep in `crates/omnigraph/src/db/manifest/recovery.rs`: `recovery_for_actor` (the original sidecar's actor), `operation_id`, and exact per-table outcomes. Schema-v3 Mutation/Load and schema-v4 BranchMerge roll-forward publish the interrupted writer's fixed lineage intent, - including its original actor; rollback and legacy recovery commits use - `actor_id = "omnigraph:recovery"`. Ordinary + including its original actor. Schema-v6 EnsureIndices rollback reuses its + pre-minted recovery commit id and durable audit plan. Other rollback and + legacy recovery commits use `actor_id = "omnigraph:recovery"`. Ordinary commit history is therefore not a complete recovery enumeration, and the CLI currently has no public query for the recovery-audit table. - Sidecar deleted as the final step. @@ -459,10 +486,15 @@ restart and without an explicit refresh. The heal lists `__recovery/` (one `list_dir`; empty in the steady state) and, per sidecar, acquires schema → branch → sorted-table gates that overlap the writer's guarded sidecar lifetime. RFC-022 mutation/load writers hold the complete order. Branch -merge now holds schema plus source/target branch authority for its whole attempt -and then the all-catalog source/target table envelope; other legacy adapters -serialize through their existing table or schema gate until their own adapter -slices land. The +merge holds schema plus source/target branch authority for its whole attempt and +then the all-catalog source/target table envelope. SchemaApply holds schema → main +branch → every live table; EnsureIndices holds schema → target branch → every table +in its durable work plan. SchemaApply's schema-v5 confirmation closes its zero-pin +outcome ambiguity while retaining loose table classification. EnsureIndices' +schema-v6 payload also retains loose table-effect classification, but gives rollback +a fixed commit id and a durable pre-restore audit plan so recovery re-entry cannot +flip the outcome. Both close the pre-arm ownership boundary while their full exact +adapters remain future work. Optimize retains its legacy adapter. The manager is shared by every `Omnigraph` handle for one canonical local root identity (relative, absolute, and symlink aliases converge; object-store/custom schemes stay opaque), so this @@ -471,9 +503,9 @@ in-flight sidecar forward from under it (a sidecar whose queues can be acquired belongs to a writer that finished or died; an existence re-check after the wait skips the finished case). Lock order is schema → branch → sorted tables → coordinator, matching the writer effect path. -Enrolled mutation/load attempts and branch merge perform one additional +Mutation/load, branch merge, SchemaApply, and EnsureIndices perform one additional `list_dir` after acquiring their authority gates; that final check closes the -pre-gate recovery TOCTOU without moving mutation/load validation or staged-file +pre-gate recovery TOCTOU without moving validation or reclaimable staged-file construction under the gate. Pinned by the four `tests/failpoints.rs::*_after_finalize_publisher_failure_heals_without_reopen` diff --git a/docs/rfcs/rfc-022-unified-write-path.md b/docs/rfcs/rfc-022-unified-write-path.md index d1031286..3a098a6a 100644 --- a/docs/rfcs/rfc-022-unified-write-path.md +++ b/docs/rfcs/rfc-022-unified-write-path.md @@ -322,6 +322,15 @@ cross-process lock and must not substitute for a target read set. With all gates held, the coordinator loads fresh authority state and compares every member of `ReadSet`. +Revalidation also proves the physical pre-state that the new recovery intent would +claim. For every existing table ref in the effect envelope, live Lance `HEAD` must +equal the prepared manifest pin. `HEAD < pin` is an internal invariant failure; +`HEAD > pin` belongs either to an already-durable recovery intent or to uncovered +drift that requires explicit repair. The attempt must reject that state **before** +writing its own sidecar; a new sidecar may never retroactively claim a pre-existing +physical effect. A first-touch ref is the deliberate exception: it does not exist at +this point and follows the sidecar-before-ref protocol in §4.3 instead. + - If all members match, the attempt may arm recovery. - If any member differs, no physical effect may run. The attempt releases its gates, discards staged state, and restarts from Prepare. @@ -471,6 +480,15 @@ able to enumerate every adapter and every entry point that invokes it. - Include accepted schema identity and every affected table in `ReadSet`. - Cover schema staging-file promotion, data-table schema/field-metadata commits, registrations, tombstones, and final schema identity with one recovery intent. +- A non-noop schema change still needs that recovery intent when it has no table + effects: schema-contract staging and promotion are independently durable state, + so an empty table-pin set means “metadata-only SchemaApply,” not “effect-free.” +- Until SchemaApply receives the full exact adapter, its schema-v5 bridge records + the target schema identity at arm time and durably confirms Phase C before final + schema-file promotion. Recovery may classify an empty-pin Phase-D residue as + completed only when both that confirmation and the live target identity match; + numeric manifest movement alone is not an outcome discriminator because a + rollback recovery commit also advances the manifest. - Write the sidecar before the first table HEAD advance, including unenforced-PK metadata backfill or other inline metadata commits. - A branch-wide or graph-wide migration must enumerate every physical manifest/data @@ -482,6 +500,14 @@ able to enumerate every adapter and every entry point that invokes it. - It records the exact achieved version rather than assuming one version of movement. - If the new data-table version is selected through `__manifest`, publishing that pointer is a graph-visible commit and uses this protocol. +- Until EnsureIndices receives its full exact effect adapter, its schema-v6 bridge + retains legacy loose table classification but pre-mints a stable rollback commit + id. Recovery persists the observed rollback audit plan before the first restore; + a retry after rollback publish therefore finalizes the same `RolledBack` outcome + instead of inferring direction from aligned numeric pins. +- Schema-v6 does not prove ownership of an index commit and does not add a fixed + forward outcome. Exact transaction chains, fixed original lineage, and exact + first-touch ref identity remain requirements of the full adapter. - Logical operations never fail because a derived index is absent or behind. - Physical-only internal-table maintenance remains the exception in Section 8. diff --git a/docs/user/search/indexes.md b/docs/user/search/indexes.md index c861ec22..473ae772 100644 --- a/docs/user/search/indexes.md +++ b/docs/user/search/indexes.md @@ -28,7 +28,7 @@ list/`Blob` columns → none. ## L2 — OmniGraph orchestration - **`@index`/`@key` declares intent; the physical index is derived state.** A migration records the declaration in the catalog/IR and never fails on it — `schema apply` builds **no** indexes. Mutation/load likewise publish only their exact data effects; they do not widen the recovery plan with index commits. Reads stay correct while an index is missing or partially covered by falling back to scans (vector search to brute-force). A later `ensure_indices`/`optimize` materializes every buildable declaration; an untrainable Vector column remains pending rather than failing the logical write. -- `ensure_indices()` / `ensure_indices_on(branch)` — idempotent build of BTREE + inverted + vector indexes for the current head; safe to re-run; returns the columns it had to defer as pending. `optimize` runs it after compaction, so the maintenance cron is the convergence path for deferred indexes. +- `ensure_indices()` / `ensure_indices_on(branch)` — idempotent build of BTREE + inverted + vector indexes for the current manifest head; safe to re-run; returns the columns it had to defer as pending. Existing Lance HEAD drift is refused with `omnigraph repair` guidance rather than silently adopted. On a lazy child branch, an ancestor-owned table stays inherited when there is no index work; real index work first verifies the target ref is absent, then creates it only after recovery intent is durable. `optimize` runs the reconciler after compaction, so the maintenance cron is the convergence path for deferred indexes. - Indexes are built on the *branch head* (not on a snapshot), so reads always see the current index state. - **Lazy branch forking for indexes**: a branch that hasn't mutated a sub-table doesn't need its own index — the main lineage's index is reused until the first write triggers a copy-on-write fork. - Vector index parameters (metric, nlist, nprobe, etc.) are not exposed in the schema; they default at the Lance layer and are picked up automatically when an index is asked for on a Vector column.