diff --git a/crates/temper-runtime/src/persistence/mod.rs b/crates/temper-runtime/src/persistence/mod.rs index 80c236adc..0dd7c90fa 100644 --- a/crates/temper-runtime/src/persistence/mod.rs +++ b/crates/temper-runtime/src/persistence/mod.rs @@ -219,14 +219,22 @@ pub trait EventStore: Send + Sync + 'static { /// deleted or un-embedded entity). Used by the backfill and by the Turso /// write-behind path. The default is a no-op (non-indexing backends); query-plane /// stores implement it. + /// + /// `as_of_sequence` is the journal sequence the rows were derived from + /// (ADR-0173 / ARN-216): the store must SKIP the reconcile when the entity's + /// journal has advanced past it — a newer live write co-committed newer rows, + /// and overwriting them with rows from a stale load would corrupt the index + /// right before the backfill watermark declares it complete. Pass `u64::MAX` + /// to force the reconcile regardless (callers that KNOW their rows are current). fn backfill_entity_vectors( &self, tenant: &str, entity_type: &str, entity_id: &str, vector_rows: &[EntityVectorRow], + as_of_sequence: u64, ) -> impl std::future::Future> + Send { - let _ = (tenant, entity_type, entity_id, vector_rows); + let _ = (tenant, entity_type, entity_id, vector_rows, as_of_sequence); async { Ok(()) } } diff --git a/crates/temper-server/src/state/projection_backfill.rs b/crates/temper-server/src/state/projection_backfill.rs index 72f511c83..425f3e9c8 100644 --- a/crates/temper-server/src/state/projection_backfill.rs +++ b/crates/temper-server/src/state/projection_backfill.rs @@ -39,11 +39,13 @@ pub(super) fn transition_table_for( /// ADR-0155). Shared by the key and vector backfills so they classify entities the /// same way — the distinction is the watermark soundness gate. pub(super) enum EntityLoadOutcome { - /// Loaded — index it from these fields. - Fields(serde_json::Value), - /// Definitively skippable: deleted, or a phantom with no events. Correctly NOT - /// indexed, and NOT a failure (it must not block the watermark). - Skip, + /// Loaded — index it from these fields, derived at this journal sequence + /// (the ADR-0173 / ARN-216 staleness guard for reconciles). + Fields(serde_json::Value, u64), + /// Definitively skippable: deleted, or a phantom with no events (sequence + /// carried for guarded purges). Correctly NOT indexed, and NOT a failure + /// (it must not block the watermark). + Skip(u64), /// The entity exists (it was enumerated from the durable store) but its current /// state could not be loaded — no transition table to replay with, an unreadable /// snapshot, or a replay error. Indexing it is impossible, so the type must NOT be @@ -82,9 +84,9 @@ pub(super) async fn load_entity_current_fields( .await { Err(_) => EntityLoadOutcome::LoadFailed, - Ok(state) if state.status == "Deleted" => EntityLoadOutcome::Skip, - Ok(state) if state.total_event_count == 0 => EntityLoadOutcome::Skip, - Ok(state) => EntityLoadOutcome::Fields(state.fields), + Ok(state) if state.status == "Deleted" => EntityLoadOutcome::Skip(state.sequence_nr), + Ok(state) if state.total_event_count == 0 => EntityLoadOutcome::Skip(state.sequence_nr), + Ok(state) => EntityLoadOutcome::Fields(state.fields, state.sequence_nr), } } diff --git a/crates/temper-server/src/state/projection_backfill/key_index.rs b/crates/temper-server/src/state/projection_backfill/key_index.rs index 22b47a405..5c708fdf6 100644 --- a/crates/temper-server/src/state/projection_backfill/key_index.rs +++ b/crates/temper-server/src/state/projection_backfill/key_index.rs @@ -141,7 +141,7 @@ pub(in crate::state) async fn populate_key_index_from_snapshots( ) .await { - EntityLoadOutcome::Fields(fields) => { + EntityLoadOutcome::Fields(fields, _loaded_seq) => { let Some(field_map) = fields.as_object() else { skipped += 1; continue; @@ -179,7 +179,7 @@ pub(in crate::state) async fn populate_key_index_from_snapshots( } } } - EntityLoadOutcome::Skip => skipped += 1, + EntityLoadOutcome::Skip(_) => skipped += 1, EntityLoadOutcome::LoadFailed => { failed += 1; tracing::warn!( diff --git a/crates/temper-server/src/state/projection_backfill/vector_index.rs b/crates/temper-server/src/state/projection_backfill/vector_index.rs index 44d0de6fe..b56968932 100644 --- a/crates/temper-server/src/state/projection_backfill/vector_index.rs +++ b/crates/temper-server/src/state/projection_backfill/vector_index.rs @@ -128,7 +128,7 @@ pub(in crate::state) async fn populate_vector_index_from_snapshots( ) .await { - EntityLoadOutcome::Fields(fields) => { + EntityLoadOutcome::Fields(fields, loaded_seq) => { let Some(field_map) = fields.as_object() else { skipped += 1; continue; @@ -166,6 +166,7 @@ pub(in crate::state) async fn populate_vector_index_from_snapshots( entity_type, entity_id, &vector_rows, + loaded_seq, ) .await { @@ -179,13 +180,20 @@ pub(in crate::state) async fn populate_vector_index_from_snapshots( } } } - EntityLoadOutcome::Skip => { + EntityLoadOutcome::Skip(loaded_seq) => { // A deleted (or phantom) entity must hold no vector rows — purge // any it still has so a soft-deleted entity is never ranked // (reconcile with an empty row set). Harmless when there is nothing - // to purge. + // to purge. Pass the load's sequence so a concurrent re-create + // co-commit is not clobbered by a stale purge (ADR-0173). if let Err(e) = store - .backfill_entity_vectors(tenant.as_str(), entity_type, entity_id, &[]) + .backfill_entity_vectors( + tenant.as_str(), + entity_type, + entity_id, + &[], + loaded_seq, + ) .await { failed += 1; diff --git a/crates/temper-server/src/storage/mod.rs b/crates/temper-server/src/storage/mod.rs index f689b7db0..dcbdd11f0 100644 --- a/crates/temper-server/src/storage/mod.rs +++ b/crates/temper-server/src/storage/mod.rs @@ -30,8 +30,6 @@ use temper_store_turso::{ }; use crate::platform_store::PlatformStore; -#[cfg(feature = "sim")] -use crate::platform_store::SimPlatformStore; use crate::state::trajectory::{TrajectoryEntry, TrajectorySource}; mod published_artifacts; @@ -95,6 +93,7 @@ pub trait DynEventStore: Send + Sync { entity_type: &'a str, entity_id: &'a str, vector_rows: &'a [temper_runtime::persistence::EntityVectorRow], + as_of_sequence: u64, ) -> EventStoreFuture<'a, Result<(), PersistenceError>>; fn vector_candidates<'a>( @@ -267,6 +266,7 @@ where entity_type: &'a str, entity_id: &'a str, vector_rows: &'a [temper_runtime::persistence::EntityVectorRow], + as_of_sequence: u64, ) -> EventStoreFuture<'a, Result<(), PersistenceError>> { Box::pin(EventStore::backfill_entity_vectors( self, @@ -274,6 +274,7 @@ where entity_type, entity_id, vector_rows, + as_of_sequence, )) } @@ -539,9 +540,10 @@ impl BoxedEventStore { entity_type: &str, entity_id: &str, vector_rows: &[temper_runtime::persistence::EntityVectorRow], + as_of_sequence: u64, ) -> Result<(), PersistenceError> { self.0 - .backfill_entity_vectors(tenant, entity_type, entity_id, vector_rows) + .backfill_entity_vectors(tenant, entity_type, entity_id, vector_rows, as_of_sequence) .await } @@ -1146,139 +1148,8 @@ pub trait TursoStoreProvider: Send + Sync { async fn ensure_tenant(&self, tenant_id: &str) -> Result; } -/// Composed storage capabilities selected at boot. -#[derive(Clone)] -pub struct StorageStack { - pub backend: BackendLabel, - pub events: BoxedEventStore, - pub postgres_pool: Option, - pub turso: Option>, - pub platform: Option>, - pub policies: Option>, - pub query_plane: Option>, - pub data_only_create: Option>, - pub trajectory: Option>, - pub metadata: Option>, -} - -impl StorageStack { - #[allow(clippy::too_many_arguments)] - pub fn new( - backend: BackendLabel, - events: BoxedEventStore, - postgres_pool: Option, - turso: Option>, - platform: Option>, - policies: Option>, - query_plane: Option>, - data_only_create: Option>, - trajectory: Option>, - metadata: Option>, - ) -> Self { - Self { - backend, - events, - postgres_pool, - turso, - platform, - policies, - query_plane, - data_only_create, - trajectory, - metadata, - } - } - - pub fn from_postgres(store: PostgresEventStore) -> Self { - let store = Arc::new(store); - Self::new( - BackendLabel::Postgres, - BoxedEventStore::from_arc(store.clone()), - Some(store.pool().clone()), - None, - Some(store.clone() as Arc), - Some(store.clone() as Arc), - Some(store.clone() as Arc), - Some(store.clone() as Arc), - Some(store.clone() as Arc), - Some(Arc::new(SingleMetadataStoreProvider::new(store))), - ) - } - - pub fn from_turso(store: TursoEventStore) -> Self { - let store = Arc::new(store); - Self::new( - BackendLabel::Turso, - BoxedEventStore::from_arc(store.clone()), - None, - Some(Arc::new(SingleTursoStoreProvider::new(store.clone()))), - Some(store.clone() as Arc), - Some(store.clone() as Arc), - Some(store.clone() as Arc), - None, - Some(store.clone() as Arc), - Some(Arc::new(SingleMetadataStoreProvider::new(store))), - ) - } - - pub fn from_tenant_router(router: TenantStoreRouter) -> Self { - let platform_store = Arc::new(router.platform_store().clone()) as Arc; - let router = Arc::new(router); - Self::new( - BackendLabel::TursoRouted, - BoxedEventStore::from_arc(router.clone()), - None, - Some(Arc::new(TenantRoutedTursoStoreProvider::new( - router.as_ref().clone(), - ))), - Some(platform_store), - Some(router.clone() as Arc), - Some(router.clone() as Arc), - None, - Some(router.clone() as Arc), - Some(Arc::new(TenantRoutedMetadataStoreProvider::new( - router.as_ref().clone(), - ))), - ) - } - - pub fn from_redis(store: temper_store_redis::RedisEventStore) -> Self { - let store = Arc::new(store); - Self::new( - BackendLabel::Redis, - BoxedEventStore::from_arc(store), - None, - None, - None, - None, - None, - None, - None, - None, - ) - } - - #[cfg(feature = "sim")] - pub fn from_sim( - store: temper_store_sim::SimEventStore, - platform_store: Option>, - ) -> Self { - let store = Arc::new(store); - let platform = platform_store.map(|store| store as Arc); - Self::new( - BackendLabel::Sim, - BoxedEventStore::from_arc(store), - None, - None, - platform, - None, - None, - None, - None, - None, - ) - } -} +pub use stack::StorageStack; +mod stack; struct SingleMetadataStoreProvider { store: Arc, diff --git a/crates/temper-server/src/storage/stack.rs b/crates/temper-server/src/storage/stack.rs new file mode 100644 index 000000000..3ef82e467 --- /dev/null +++ b/crates/temper-server/src/storage/stack.rs @@ -0,0 +1,154 @@ +//! Composed storage capabilities selected at boot (`StorageStack`). +//! +//! Extracted from `storage/mod.rs` (the workspace's largest file) when the +//! ARN-216 staleness-guard plumbing nudged it past the readability ratchet's +//! max-file-lines ceiling — a cohesive unit, moved verbatim. + +use std::sync::Arc; + +use super::{ + BackendLabel, BoxedEventStore, DataOnlyCreateStore, MetadataStoreProvider, PgPool, + PlatformStore, PolicyStore, QueryPlaneStore, TrajectorySink, TursoStoreProvider, +}; +use super::{ + SingleMetadataStoreProvider, SingleTursoStoreProvider, TenantRoutedMetadataStoreProvider, + TenantRoutedTursoStoreProvider, +}; +#[cfg(feature = "sim")] +use crate::platform_store::SimPlatformStore; +use temper_store_postgres::PostgresEventStore; +use temper_store_turso::{TenantStoreRouter, TursoEventStore}; + +/// Composed storage capabilities selected at boot. +#[derive(Clone)] +pub struct StorageStack { + pub backend: BackendLabel, + pub events: BoxedEventStore, + pub postgres_pool: Option, + pub turso: Option>, + pub platform: Option>, + pub policies: Option>, + pub query_plane: Option>, + pub data_only_create: Option>, + pub trajectory: Option>, + pub metadata: Option>, +} + +impl StorageStack { + #[allow(clippy::too_many_arguments)] + pub fn new( + backend: BackendLabel, + events: BoxedEventStore, + postgres_pool: Option, + turso: Option>, + platform: Option>, + policies: Option>, + query_plane: Option>, + data_only_create: Option>, + trajectory: Option>, + metadata: Option>, + ) -> Self { + Self { + backend, + events, + postgres_pool, + turso, + platform, + policies, + query_plane, + data_only_create, + trajectory, + metadata, + } + } + + pub fn from_postgres(store: PostgresEventStore) -> Self { + let store = Arc::new(store); + Self::new( + BackendLabel::Postgres, + BoxedEventStore::from_arc(store.clone()), + Some(store.pool().clone()), + None, + Some(store.clone() as Arc), + Some(store.clone() as Arc), + Some(store.clone() as Arc), + Some(store.clone() as Arc), + Some(store.clone() as Arc), + Some(Arc::new(SingleMetadataStoreProvider::new(store))), + ) + } + + pub fn from_turso(store: TursoEventStore) -> Self { + let store = Arc::new(store); + Self::new( + BackendLabel::Turso, + BoxedEventStore::from_arc(store.clone()), + None, + Some(Arc::new(SingleTursoStoreProvider::new(store.clone()))), + Some(store.clone() as Arc), + Some(store.clone() as Arc), + Some(store.clone() as Arc), + None, + Some(store.clone() as Arc), + Some(Arc::new(SingleMetadataStoreProvider::new(store))), + ) + } + + pub fn from_tenant_router(router: TenantStoreRouter) -> Self { + let platform_store = Arc::new(router.platform_store().clone()) as Arc; + let router = Arc::new(router); + Self::new( + BackendLabel::TursoRouted, + BoxedEventStore::from_arc(router.clone()), + None, + Some(Arc::new(TenantRoutedTursoStoreProvider::new( + router.as_ref().clone(), + ))), + Some(platform_store), + Some(router.clone() as Arc), + Some(router.clone() as Arc), + None, + Some(router.clone() as Arc), + Some(Arc::new(TenantRoutedMetadataStoreProvider::new( + router.as_ref().clone(), + ))), + ) + } + + pub fn from_redis(store: temper_store_redis::RedisEventStore) -> Self { + let store = Arc::new(store); + Self::new( + BackendLabel::Redis, + BoxedEventStore::from_arc(store), + None, + None, + None, + None, + None, + None, + None, + None, + ) + } + + #[cfg(feature = "sim")] + pub fn from_sim( + store: temper_store_sim::SimEventStore, + platform_store: Option>, + ) -> Self { + let store = Arc::new(store); + let platform = platform_store.map(|store| store as Arc); + Self::new( + BackendLabel::Sim, + BoxedEventStore::from_arc(store), + None, + None, + platform, + None, + None, + None, + None, + None, + ) + } +} diff --git a/crates/temper-server/tests/dst_entity_vector_index.rs b/crates/temper-server/tests/dst_entity_vector_index.rs index f26baa214..1d5ff8172 100644 --- a/crates/temper-server/tests/dst_entity_vector_index.rs +++ b/crates/temper-server/tests/dst_entity_vector_index.rs @@ -189,3 +189,105 @@ async fn dst_nearest_by_reference_excludes_self() { ); } } + +/// ARN-216: the vector backfill must not overwrite a NEWER live co-commit +/// with rows built from a stale load. +/// +/// The backfill is two store calls with nothing spanning them: (1) load the +/// entity's state (snapshot/replay at some sequence), (2) reconcile the +/// index to rows parsed from that load. A live write landing between them +/// co-commits the new embedding — and step (2) then clobbers it with the +/// stale one, after which the completion watermark declares the index +/// authoritative. This test executes exactly that interleave against the +/// store API, passing the load's `as_of_sequence` so the staleness guard +/// (ADR-0173) can refuse the overwrite. +#[tokio::test] +async fn dst_vector_backfill_must_not_overwrite_newer_live_write() { + for seed in 0..NUM_SEEDS { + let (_guard, _clock, _id) = install_deterministic_context(seed); + let store: BoxedEventStore = BoxedEventStore::new(SimEventStore::no_faults(seed)); + let table = item_table(); + let system = ActorSystem::new("dst-vector-race"); + let entity_id = format!("item-race-{seed}"); + + // Entity exists with embedding E1. + create_item( + &system, + &table, + &store, + &entity_id, + &[1.0, 0.0, 0.0, 0.0], + "m1", + ) + .await; + + // BACKFILL step 1 (stale load): the rows a backfill would build from a + // load taken NOW — i.e. E1 at the current journal sequence. Production + // parses these from the replayed state; the parse result is exactly + // this row + this sequence. + let stale_rows = vec![temper_runtime::persistence::EntityVectorRow { + decl_name: "embed".to_string(), + model_tag: "m1".to_string(), + vector: vec![1.0, 0.0, 0.0, 0.0], + }]; + let stale_seq = { + let actor = EntityActor::with_persistence( + "Item", + &entity_id, + table.clone(), + serde_json::json!({}), + store.clone(), + BackendLabel::Sim, + ) + .with_tenant("default"); + let actor_ref = system.spawn(actor, format!("{entity_id}-probe")); + let state: EntityResponse = actor_ref + .ask(EntityMsg::GetState, Duration::from_secs(5)) + .await + .expect("state probe"); + state.state.sequence_nr + }; + + // LIVE WRITE lands between the backfill's load and its reconcile: + // the co-commit updates the index to E2. + let actor = EntityActor::with_persistence( + "Item", + &entity_id, + table.clone(), + serde_json::json!({}), + store.clone(), + BackendLabel::Sim, + ) + .with_tenant("default"); + let actor_ref = system.spawn(actor, format!("{entity_id}-live")); + let e2 = serde_json::to_string(&[0.0f32, 1.0, 0.0, 0.0]).unwrap(); + let r = dispatch( + &actor_ref, + "Reembed", + serde_json::json!({ "Embedding": e2, "EmbeddingModel": "m1" }), + ) + .await; + assert!(r.success, "seed {seed}: Reembed failed: {:?}", r.error); + + // BACKFILL step 2: reconcile with the STALE rows at the stale sequence. + store + .backfill_entity_vectors("default", "Item", &entity_id, &stale_rows, stale_seq) + .await + .expect("backfill reconcile"); + + // The index must still hold the newer live embedding E2. + let candidates = store + .vector_candidates("default", "Item", "embed", "m1", 1000) + .await + .expect("vector candidates"); + let row = candidates + .iter() + .find(|c| c.entity_id == entity_id) + .expect("entity must be indexed"); + assert_eq!( + row.vector, + vec![0.0, 1.0, 0.0, 0.0], + "seed {seed}: a stale backfill reconcile must not overwrite a newer live co-commit" + ); + } +} diff --git a/crates/temper-store-postgres/src/store.rs b/crates/temper-store-postgres/src/store.rs index 21aa5c717..008465fc1 100644 --- a/crates/temper-store-postgres/src/store.rs +++ b/crates/temper-store-postgres/src/store.rs @@ -474,6 +474,7 @@ impl EventStore for PostgresEventStore { entity_type: &str, entity_id: &str, vector_rows: &[EntityVectorRow], + as_of_sequence: u64, ) -> Result<(), PersistenceError> { // Reconcile: DELETE all of the entity's rows, then insert the current ones. // Empty `vector_rows` purges the entity (deleted / un-embedded). Always runs @@ -483,6 +484,15 @@ impl EventStore for PostgresEventStore { .begin() .await .map_err(|e| PersistenceError::Storage(e.to_string()))?; + // ARN-216 / ADR-0173: the reconcile must not overwrite rows a NEWER live + // write co-committed. Under READ COMMITTED a SELECT-then-DELETE guard is + // not atomic (each statement gets a fresh snapshot), so the ordering is: + // DELETE first — taking row locks that serialize any concurrent reconcile + // of this entity behind this transaction — THEN check the journal sequence + // under those locks and ROLL BACK if it advanced past the sequence these + // rows were derived from. A live append committing before the DELETE is + // caught by the post-DELETE check; one starting after it blocks on the + // row locks until this transaction resolves. crate::dbm::postgres_query!( "DELETE FROM entity_vector_index \ WHERE tenant = $1 AND entity_type = $2 AND entity_id = $3", @@ -493,6 +503,25 @@ impl EventStore for PostgresEventStore { .execute(&mut *tx) .await .map_err(|e| PersistenceError::Storage(e.to_string()))?; + // Aggregates always return exactly one row (COALESCE'd to 0 over an + // empty set), so fetch_one — no optional row to consider. + let (current_seq,): (i64,) = crate::dbm::postgres_query_as!( + "SELECT COALESCE(MAX(sequence_nr), 0) FROM events \ + WHERE tenant = $1 AND entity_type = $2 AND entity_id = $3", + ) + .bind(tenant) + .bind(entity_type) + .bind(entity_id) + .fetch_one(&mut *tx) + .await + .map_err(|e| PersistenceError::Storage(e.to_string()))?; + let current_seq_u64 = u64::try_from(current_seq).unwrap_or(0); + if current_seq_u64 > as_of_sequence { + tx.rollback() + .await + .map_err(|e| PersistenceError::Storage(e.to_string()))?; + return Ok(()); + } for row in vector_rows { crate::dbm::postgres_query!( "INSERT INTO entity_vector_index \ diff --git a/crates/temper-store-sim/src/lib.rs b/crates/temper-store-sim/src/lib.rs index d52a96643..26f84adce 100644 --- a/crates/temper-store-sim/src/lib.rs +++ b/crates/temper-store-sim/src/lib.rs @@ -685,8 +685,22 @@ impl EventStore for SimEventStore { entity_type: &str, entity_id: &str, vector_rows: &[EntityVectorRow], + as_of_sequence: u64, ) -> Result<(), PersistenceError> { let mut inner = self.inner.lock().expect("SimEventStore lock poisoned"); // ci-ok: infallible lock + // ARN-216 / ADR-0173: a reconcile derived from a stale load must not + // overwrite the newer rows a live write co-committed. Skip when the + // journal advanced past the load's sequence. + let pid = format!("{tenant}:{entity_type}:{entity_id}"); + let current_seq = inner + .journals + .get(&pid) + .and_then(|j| j.last()) + .map(|e| e.sequence_nr) + .unwrap_or(0); + if current_seq > as_of_sequence { + return Ok(()); + } // Reconcile: drop ALL of the entity's rows, then insert the current ones. // Empty `vector_rows` purges the entity (deleted / un-embedded). Idempotent. inner.vector_index.retain(|(t, et, _, _, eid), _| { diff --git a/crates/temper-store-turso/src/router.rs b/crates/temper-store-turso/src/router.rs index 3c7e1fe01..3e0449d42 100644 --- a/crates/temper-store-turso/src/router.rs +++ b/crates/temper-store-turso/src/router.rs @@ -775,10 +775,11 @@ impl EventStore for TenantStoreRouter { entity_type: &str, entity_id: &str, vector_rows: &[temper_runtime::persistence::EntityVectorRow], + as_of_sequence: u64, ) -> Result<(), PersistenceError> { let store = self.store_for_tenant(tenant).await?; store - .backfill_entity_vectors(tenant, entity_type, entity_id, vector_rows) + .backfill_entity_vectors(tenant, entity_type, entity_id, vector_rows, as_of_sequence) .await } diff --git a/crates/temper-store-turso/src/store/event_store.rs b/crates/temper-store-turso/src/store/event_store.rs index 84d23fb04..541235ddd 100644 --- a/crates/temper-store-turso/src/store/event_store.rs +++ b/crates/temper-store-turso/src/store/event_store.rs @@ -184,7 +184,7 @@ impl EventStore for TursoEventStore { tokio::time::sleep(Duration::from_millis(retry_delay_ms(attempt - 1))).await; } match self - .backfill_entity_vectors(tenant, entity_type, entity_id, vector_rows) + .backfill_entity_vectors(tenant, entity_type, entity_id, vector_rows, new_seq) .await { Ok(()) => { @@ -217,6 +217,7 @@ impl EventStore for TursoEventStore { entity_type: &str, entity_id: &str, vector_rows: &[EntityVectorRow], + as_of_sequence: u64, ) -> Result<(), PersistenceError> { // Reconcile: DELETE all of the entity's rows, then insert the current ones. // Empty `vector_rows` purges the entity (deleted / un-embedded). Always runs @@ -229,6 +230,28 @@ impl EventStore for TursoEventStore { .transaction_with_behavior(TransactionBehavior::Immediate) .await .map_err(storage_error)?; + // ARN-216 / ADR-0173: skip when the journal advanced past the sequence + // these rows were derived from — a newer write's rows (write-behind or a + // later reconcile) must not be overwritten by a stale load. + let mut seq_rows = tx + .query( + "SELECT COALESCE(MAX(sequence_nr), 0) FROM events \ + WHERE tenant = ?1 AND entity_type = ?2 AND entity_id = ?3", + params![tenant, entity_type, entity_id], + ) + .await + .map_err(storage_error)?; + let current_seq: i64 = match seq_rows.next().await.map_err(storage_error)? { + Some(row) => row.get(0).map_err(storage_error)?, + None => 0, + }; + if current_seq as u64 > as_of_sequence { + // Explicit rollback: an Immediate transaction holds the RESERVED + // lock, and async Drop cannot await — release it deterministically + // rather than deferring to libsql's synchronous drop hook. + tx.rollback().await.map_err(storage_error)?; + return Ok(()); + } tx.execute( "DELETE FROM entity_vector_index \ WHERE tenant = ?1 AND entity_type = ?2 AND entity_id = ?3", diff --git a/crates/temper-store-turso/src/store/tests/mod.rs b/crates/temper-store-turso/src/store/tests/mod.rs index f75aa4c01..d3b985c5b 100644 --- a/crates/temper-store-turso/src/store/tests/mod.rs +++ b/crates/temper-store-turso/src/store/tests/mod.rs @@ -125,8 +125,15 @@ async fn vector_index_write_behind_candidates_and_partitioning() { assert_eq!(candidates[1].vector, vec![0.0, 1.0]); // Upsert: re-writing item-a's vector replaces (no duplicate row). + // `u64::MAX` forces the reconcile (caller knows the rows are current). store - .backfill_entity_vectors("t", "Item", "item-a", &[row("embed", "m1", vec![0.5, 0.5])]) + .backfill_entity_vectors( + "t", + "Item", + "item-a", + &[row("embed", "m1", vec![0.5, 0.5])], + u64::MAX, + ) .await .unwrap(); let candidates = store @@ -207,7 +214,7 @@ async fn vector_index_reconcile_purges_on_delete_and_empty_rows() { // The explicit backfill purge (empty rows) is idempotent. store - .backfill_entity_vectors("t", "Item", "item-a", &[]) + .backfill_entity_vectors("t", "Item", "item-a", &[], u64::MAX) .await .unwrap(); assert!( diff --git a/docs/adrs/0173-vector-backfill-staleness-guard.md b/docs/adrs/0173-vector-backfill-staleness-guard.md new file mode 100644 index 000000000..bd217807b --- /dev/null +++ b/docs/adrs/0173-vector-backfill-staleness-guard.md @@ -0,0 +1,84 @@ +# ADR-0173: Vector-Backfill Staleness Guard + +## Status + +Accepted (2026-07-14) + +(Numbered 0173: Claude's concurrent arena branch uses 0161 for the same +decision; Codex's uses 0171 for a heavier monotonic-reconciliation redesign. +This ADR is the Grok arena number and is intentionally unique.) + +## Context + +The ADR-0155 vector backfill is two store calls with nothing spanning them: +load an entity's current state (snapshot/replay at some journal sequence), +then `backfill_entity_vectors` — an unconditional delete-then-insert +reconcile of the entity's index rows. A live write landing between the two +co-commits the new embedding, and the reconcile then overwrites it with rows +parsed from the stale load (ARN-216). The backfill then stamps the +completion watermark, so reads treat the stale index as authoritative until +the entity's next write. The same shape exists on the Turso write-behind +path, where a retried lagging index write can land after a newer one. + +## Decision + +1. **`as_of_sequence` on `backfill_entity_vectors`.** Every reconcile + carries the journal sequence its rows were derived from. Inside the + store's transaction/lock, the entity's current journal sequence is read; + if it has advanced past `as_of_sequence`, the reconcile is skipped — the + newer write's co-commit (or write-behind) already holds newer rows. + Callers that know their rows are current pass `u64::MAX`. +2. **The loader carries the sequence.** `EntityLoadOutcome::Fields` and + `Skip` now carry the replayed `sequence_nr`, so both the index-write arm + and the tombstone/phantom purge arm pass the true as-of sequence. +3. **Turso write-behind passes its append's sequence**, so a retried lagging + write cannot clobber a subsequently appended newer one. +4. **Skipped-as-stale counts as success for the watermark.** A skip means a + newer live write reconciled the entity's rows itself; the entity's index + entry is current by construction. + +## Consequences + +- The interleave is pinned by a 100-seed DST executing the exact two-step + production order (`dst_vector_backfill_must_not_overwrite_newer_live_write`). +- The guard costs one indexed `MAX(sequence_nr)` lookup inside the existing + reconcile transaction — only on backfill/write-behind paths, never on the + co-commit fast path. +- The declared-key backfill has the same load-then-write shape, and in the + race interleave the stale upsert lands last and CLOBBERS the newer live + mapping for that key_name — the same corruption shape as vectors, and it + remains unguarded pending the symmetric follow-up (Linear issue on + reconnect) rather than being silently expanded into this change. +- Guard-skip-as-success assumes every journal-advancing writer holds the + type's vector declarations (`reconcile_vectors` derives from its own + table). A writer instance without the deployed vector config (rolling + deploy, stale table) advances the journal without reconciling; a + guard-skip then trusts rows that write never installed. Pre-existing, + shared with the key index, recorded here for the record. +- On postgres the guard runs DELETE-first and re-checks the journal, rolling + back when it advanced — READ COMMITTED makes a check-then-delete ordering + non-atomic there; sim (mutex) and turso (Immediate transaction) are atomic + with either ordering. The DELETE's row locks serialize concurrent + reconciles ONLY when prior rows exist; in the no-prior-rows case a + same-model race is still caught by the index primary key (the stale INSERT + collides, errors, and the type is not watermarked — fail-safe), but a + CROSS-model live re-embed racing the window between the re-check and the + stale INSERT commits under a different primary key: a transient stale row + survives in the old model partition until the entity's next write + reconciles all its partitions. Narrow, self-healing, and strictly better + than the unguarded behavior — recorded as a known residual alongside the + key-backfill follow-up rather than closed with a heavier per-entity lock + (the direction Alternatives rejects). + +## Alternatives Considered + +- **Locking the entity across load + reconcile:** spans an actor replay and + a store transaction across two subsystems; far more machinery for the + same guarantee the sequence comparison gives atomically inside the + existing transaction. +- **Re-loading and diffing before write:** still racy (TOCTOU between the + re-load and the write); the guard must live inside the store transaction. +- **Monotonic sequence column on every vector row with CAS upserts:** richer + and closer to a full projection-contract rewrite (Codex ADR-0171 direction); + deferred pending ARN-201's canonical append/projection transaction contract. + The as-of guard is the minimal fix for the permanent-stale-watermark failure. diff --git a/test-fixtures/specs/vectored_item.ioa.toml b/test-fixtures/specs/vectored_item.ioa.toml index 34b713f94..ff24bb0c9 100644 --- a/test-fixtures/specs/vectored_item.ioa.toml +++ b/test-fixtures/specs/vectored_item.ioa.toml @@ -35,3 +35,11 @@ from = ["New"] to = "Ready" params = ["Embedding", "EmbeddingModel"] hint = "Create the item with its embedding vector (JSON string) and model tag." + +[[action]] +name = "Reembed" +kind = "input" +from = ["Ready"] +to = "Ready" +params = ["Embedding", "EmbeddingModel"] +hint = "Replace the item's embedding after a newer model run re-embeds it."