apunkt/omnigraph
1
0
Fork 0
mirror of https://github.com/ModernRelay/omnigraph.git synced 2026-07-12 03:12:11 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

Harden RFC-022 pre-arm recovery ownership (#346)

Browse source
This commit is contained in:
Andrew Altshuler 2026-07-11 23:52:53 +03:00 • committed by GitHub
parent e0e145aa92
commit 905a27c4bd
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
17 changed files with 2166 additions and 538 deletions
Download patch file Download diff file Expand all files Collapse all files

10
crates/omnigraph/src/db/manifest.rs
View file

@ -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)]

670
crates/omnigraph/src/db/manifest/recovery.rs
View file

@ -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<Vec<TableOutcome>>,
}
/// 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<SidecarTombstone>,
/// 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<String>,
/// RFC-022 exact-effect protocol. Absent for every v1/v2 sidecar.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub protocol_v3: Option<RecoveryProtocolV3>,
/// RFC-022 BranchMerge protocol. Present only on schema-v4 sidecars.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub protocol_v4: Option<RecoveryProtocolV4>,
/// 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<RecoveryEnsureIndicesRollbackV6>,
}
/// 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<bool> {
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<NoEffectForkCleanup> {
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<TableOutcome> {
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<TableOutcome> = sidecar
let mut outcomes: Vec<TableOutcome> = 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(&registration.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<bool> {
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::<i64>().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<bool> {
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<String>,
tables: Vec<SidecarTablePin>,
) -> Result<RecoverySidecar> {
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("<new-ensure-indices-sidecar>", &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("<new-occ-sidecar>", &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("<new-branch-merge-sidecar>", &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(&current).unwrap();
parse_sidecar("file:///tmp/__recovery/index-v6.json", &current_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();

158
crates/omnigraph/src/db/omnigraph.rs
View file

@ -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<Arc<Catalog>> {
pub(crate) async fn load_accepted_catalog_with_schema_gate_held(&self) -> Result<Arc<Catalog>> {
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?;

200
crates/omnigraph/src/db/omnigraph/schema_apply.rs
View file

@ -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<String>)> = 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<String>)> = 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::<String, SnapshotHandle>::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::<BTreeSet<_>>();
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,
&current_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,

382
crates/omnigraph/src/db/omnigraph/table_ops.rs
View file

@ -84,25 +84,36 @@ pub(super) async fn ensure_indices_for_branch(
db: &Omnigraph,
branch: Option<&str>,
) -> Result<Vec<PendingIndex>> {
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::<String, Vec<PendingIndex>>::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<crate::db::manifest::SidecarTablePin> = 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::<String, crate::storage_layer::SnapshotHandle>::new();
let mut first_touch_sources =
std::collections::HashMap::<String, crate::storage_layer::SnapshotHandle>::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<PendingIndex>,
}
/// 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<bool> {
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<PlannedIndexWork> {
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<bool> {
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<bool> {
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
}

134
crates/omnigraph/src/exec/merge.rs
View file

@ -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<String>,
}
impl PreparedExistingMergeTarget {
fn into_parts(self) -> (SnapshotHandle, String, Option<String>) {
(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<String> = 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<PreparedExistingMergeTarget>,
planned_transactions: &[crate::table_store::StagedTransactionIdentity],
) -> Result<crate::db::SubTableUpdate> {
// 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<PreparedExistingMergeTarget>,
planned_transactions: &[crate::table_store::StagedTransactionIdentity],
) -> Result<crate::db::SubTableUpdate> {
// `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) {

183
crates/omnigraph/src/exec/staging.rs
View file

@ -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<StagedTableEntry> = futures::stream::iter(
stage_inputs.into_iter().map(
let mut staged_entries: Vec<StagedTableEntry> =
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::<Vec<Result<Option<StagedTableEntry>>>>()
.await
.into_iter()
.collect::<Result<Vec<_>>>()?
.into_iter()
.flatten()
.collect();
))
.buffered(concurrency)
.collect::<Vec<Result<Option<StagedTableEntry>>>>()
.await
.into_iter()
.collect::<Result<Vec<_>>>()?
.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::<Vec<_>>()
.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<Option<StagedTableEntry>> {
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<String>)> =
Vec::with_capacity(staged.len());
let mut queue_keys: Vec<(String, Option<String>)> = 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<SidecarTablePin> = 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<SubTableUpdate> = 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()

20
crates/omnigraph/src/failpoints.rs
View file

@ -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 =

5
crates/omnigraph/tests/branching.rs
View file

@ -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

654
crates/omnigraph/tests/failpoints.rs
View file

@ -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 {

146
crates/omnigraph/tests/maintenance.rs
View file

@ -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

47
crates/omnigraph/tests/schema_apply.rs
View file

@ -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