Skip to content

fix(stores): guard vector backfill against stale overwrites (ARN-216) #395

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
rita-aga wants to merge 3 commits into
base: main
Choose a base branch
from
Draft

fix(stores): guard vector backfill against stale overwrites (ARN-216) #395

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
163787e
test(server): failing DST for vector-backfill stale-overwrite race (A…
rita-aga Jul 14, 2026
b36c898
fix(stores): guard vector backfill against stale overwrites (ARN-216)
rita-aga Jul 14, 2026
d6d9e82
fix(store-postgres): non-wrapping i64 to u64 seq cast in vector guard…
rita-aga Jul 15, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 9 additions & 1 deletion crates/temper-runtime/src/persistence/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<Output = Result<(), PersistenceError>> + Send {
let _ = (tenant, entity_type, entity_id, vector_rows);
let _ = (tenant, entity_type, entity_id, vector_rows, as_of_sequence);
async { Ok(()) }
}

Expand Down
18 changes: 10 additions & 8 deletions crates/temper-server/src/state/projection_backfill.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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),
}
}

Expand Down
4 changes: 2 additions & 2 deletions crates/temper-server/src/state/projection_backfill/key_index.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -141,7 +141,7 @@ pub(in crate::state) async fn populate_key_index_from_snapshots(
)
.await
{
EntityLoadOutcome::Fields(fields) => {
EntityLoadOutcome::Fields(fields, _loaded_seq) => {

@greptile-apps greptile-apps Bot Jul 14, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Key backfill staleness guard not wired up

_loaded_seq is intentionally discarded here, leaving the key-index backfill with the exact same load-then-write race this PR fixes for vectors. The ADR (section "Consequences") documents this as a known residual: a stale upsert landing after a concurrent live key write will clobber the newer mapping, and the completion watermark will then treat the stale entry as authoritative. The impact is worse than the vector case — a wrong key mapping can route entity lookups to the wrong entity (or produce a false-absence hit) until the entity's next write corrects it. The ADR says a "Linear issue on reconnect" is tracked for the symmetric follow-up; confirming that issue exists and is prioritized before watermarking any type with key declarations would reduce the window this is exploitable.

Prompt To Fix With AI 
This is a comment left during a code review.
Path: crates/temper-server/src/state/projection_backfill/key_index.rs
Line: 144

Comment:
**Key backfill staleness guard not wired up**

`_loaded_seq` is intentionally discarded here, leaving the key-index backfill with the exact same load-then-write race this PR fixes for vectors. The ADR (section "Consequences") documents this as a known residual: a stale upsert landing after a concurrent live key write will clobber the newer mapping, and the completion watermark will then treat the stale entry as authoritative. The impact is worse than the vector case — a wrong key mapping can route entity lookups to the wrong entity (or produce a false-absence hit) until the entity's next write corrects it. The ADR says a "Linear issue on reconnect" is tracked for the symmetric follow-up; confirming that issue exists and is prioritized before watermarking any type with key declarations would reduce the window this is exploitable.

How can I resolve this? If you propose a fix, please make it concise.

Fix in Claude Code Fix in Codex Fix in Cursor

let Some(field_map) = fields.as_object() else {
skipped += 1;
continue;
Expand Down Expand Up @@ -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!(
Expand Down
16 changes: 12 additions & 4 deletions crates/temper-server/src/state/projection_backfill/vector_index.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -166,6 +166,7 @@ pub(in crate::state) async fn populate_vector_index_from_snapshots(
entity_type,
entity_id,
&vector_rows,
loaded_seq,
)
.await
{
Expand All @@ -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;
Expand Down
143 changes: 7 additions & 136 deletions crates/temper-server/src/storage/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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>(
Expand Down Expand Up @@ -267,13 +266,15 @@ 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,
tenant,
entity_type,
entity_id,
vector_rows,
as_of_sequence,
))
}

Expand Down Expand Up @@ -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
}

Expand Down Expand Up @@ -1146,139 +1148,8 @@ pub trait TursoStoreProvider: Send + Sync {
async fn ensure_tenant(&self, tenant_id: &str) -> Result<bool, PersistenceError>;
}

/// Composed storage capabilities selected at boot.
#[derive(Clone)]
pub struct StorageStack {
pub backend: BackendLabel,
pub events: BoxedEventStore,
pub postgres_pool: Option<PgPool>,
pub turso: Option<Arc<dyn TursoStoreProvider>>,
pub platform: Option<Arc<dyn PlatformStore>>,
pub policies: Option<Arc<dyn PolicyStore>>,
pub query_plane: Option<Arc<dyn QueryPlaneStore>>,
pub data_only_create: Option<Arc<dyn DataOnlyCreateStore>>,
pub trajectory: Option<Arc<dyn TrajectorySink>>,
pub metadata: Option<Arc<dyn MetadataStoreProvider>>,
}

impl StorageStack {
#[allow(clippy::too_many_arguments)]
pub fn new(
backend: BackendLabel,
events: BoxedEventStore,
postgres_pool: Option<PgPool>,
turso: Option<Arc<dyn TursoStoreProvider>>,
platform: Option<Arc<dyn PlatformStore>>,
policies: Option<Arc<dyn PolicyStore>>,
query_plane: Option<Arc<dyn QueryPlaneStore>>,
data_only_create: Option<Arc<dyn DataOnlyCreateStore>>,
trajectory: Option<Arc<dyn TrajectorySink>>,
metadata: Option<Arc<dyn MetadataStoreProvider>>,
) -> 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<dyn PlatformStore>),
Some(store.clone() as Arc<dyn PolicyStore>),
Some(store.clone() as Arc<dyn QueryPlaneStore>),
Some(store.clone() as Arc<dyn DataOnlyCreateStore>),
Some(store.clone() as Arc<dyn TrajectorySink>),
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<dyn PlatformStore>),
Some(store.clone() as Arc<dyn PolicyStore>),
Some(store.clone() as Arc<dyn QueryPlaneStore>),
None,
Some(store.clone() as Arc<dyn TrajectorySink>),
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<dyn PlatformStore>;
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<dyn PolicyStore>),
Some(router.clone() as Arc<dyn QueryPlaneStore>),
None,
Some(router.clone() as Arc<dyn TrajectorySink>),
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<Arc<SimPlatformStore>>,
) -> Self {
let store = Arc::new(store);
let platform = platform_store.map(|store| store as Arc<dyn PlatformStore>);
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<dyn MetadataStore>,
Expand Down
Loading
Loading