diff --git a/crates/temper-jit/src/shadow.rs b/crates/temper-jit/src/shadow.rs index a3b99b32..0fe5c183 100644 --- a/crates/temper-jit/src/shadow.rs +++ b/crates/temper-jit/src/shadow.rs @@ -115,6 +115,7 @@ mod tests { entity_name: "Order".into(), states: vec!["Draft".into(), "Submitted".into(), "Cancelled".into()], initial_state: "Draft".into(), + keys: vec![], rules: vec![ TransitionRule { name: "SubmitOrder".into(), diff --git a/crates/temper-jit/src/swap.rs b/crates/temper-jit/src/swap.rs index cfa81f54..fe7bbabf 100644 --- a/crates/temper-jit/src/swap.rs +++ b/crates/temper-jit/src/swap.rs @@ -86,6 +86,7 @@ mod tests { entity_name: name.to_string(), states: vec!["A".into(), "B".into()], initial_state: "A".into(), + keys: vec![], rules: vec![TransitionRule { name: "GoB".into(), from_states: vec!["A".into()], diff --git a/crates/temper-jit/src/table/builder.rs b/crates/temper-jit/src/table/builder.rs index 2281a369..ec2c8cdb 100644 --- a/crates/temper-jit/src/table/builder.rs +++ b/crates/temper-jit/src/table/builder.rs @@ -121,6 +121,14 @@ impl TransitionTable { states: automaton.automaton.states.clone(), initial_state: automaton.automaton.initial.clone(), rules, + keys: automaton + .keys + .iter() + .map(|k| super::types::DeclaredKey { + name: k.name.clone(), + properties: k.properties.clone(), + }) + .collect(), state_var_metadata, composite_actions, rule_index, diff --git a/crates/temper-jit/src/table/types.rs b/crates/temper-jit/src/table/types.rs index 079a1acf..25c4f202 100644 --- a/crates/temper-jit/src/table/types.rs +++ b/crates/temper-jit/src/table/types.rs @@ -15,6 +15,15 @@ use super::guard::{Guard, GuardFailure}; // Core types // --------------------------------------------------------------------------- +/// A declared unique/alternate key carried on the table (ADR-0153). `name` +/// identifies it; `properties` is the unique property set in canonical order. +/// The actor hashes these on write to maintain the negative-existence access path. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct DeclaredKey { + pub name: String, + pub properties: Vec, +} + /// A transition table: state machine transitions as DATA, not code. /// Can be hot-swapped per-actor without restart. #[derive(Debug, Clone, Serialize)] @@ -27,6 +36,9 @@ pub struct TransitionTable { pub initial_state: String, /// Ordered list of transition rules. pub rules: Vec, + /// ADR-0153: declared unique/alternate keys the kernel indexes for + /// negative-existence reads. Empty when the spec declared no `[[key]]`. + pub keys: Vec, /// Per-state-variable metadata for platform primitives (ADR-0045, ADR-0047). /// Keyed by state-variable name. Empty map when the IOA spec did not /// declare any per-field overrides. @@ -129,6 +141,8 @@ impl<'de> Deserialize<'de> for TransitionTable { state_var_metadata: BTreeMap, #[serde(default)] composite_actions: BTreeMap, + #[serde(default)] + keys: Vec, } let raw = TransitionTableRaw::deserialize(deserializer)?; @@ -137,6 +151,7 @@ impl<'de> Deserialize<'de> for TransitionTable { states: raw.states, initial_state: raw.initial_state, rules: raw.rules, + keys: raw.keys, state_var_metadata: raw.state_var_metadata, composite_actions: raw.composite_actions, rule_index: BTreeMap::new(), @@ -250,6 +265,7 @@ mod tests { entity_name: "TestEntity".to_string(), states: vec!["Draft".to_string(), "Active".to_string()], initial_state: "Draft".to_string(), + keys: vec![], rules: vec![ TransitionRule { name: "Submit".to_string(), diff --git a/crates/temper-runtime/src/persistence/mod.rs b/crates/temper-runtime/src/persistence/mod.rs index b6804163..4ed521f1 100644 --- a/crates/temper-runtime/src/persistence/mod.rs +++ b/crates/temper-runtime/src/persistence/mod.rs @@ -83,6 +83,18 @@ pub trait PersistentActor: Send + 'static { } } +/// A declared-key row to co-commit with an append (ADR-0153). The entity claims +/// `key_hash` for `key_name`; the store writes it into `entity_key_index` in the +/// same transaction as the journal append, giving the read plane an `O(log n)` +/// present/absent probe (the negative-existence access path, ARN-68). +#[derive(Debug, Clone)] +pub struct EntityKeyRow { + /// The declared key's identifier (the `[[key]]` block's `name`). + pub key_name: String, + /// The canonical, type-tagged hash of the key's values. + pub key_hash: String, +} + /// Trait for the event store backend (implemented by temper-store-postgres). /// Uses desugared async-in-trait to enforce Send bounds on futures. pub trait EventStore: Send + Sync + 'static { @@ -94,6 +106,55 @@ pub trait EventStore: Send + Sync + 'static { events: &[PersistenceEnvelope], ) -> impl std::future::Future> + Send; + /// Append events and co-commit declared key-index rows (ADR-0153) in the + /// **same transaction** as the journal append. The default ignores + /// `key_rows` and delegates to [`EventStore::append`] — only stores with a + /// query plane (postgres, turso) maintain `entity_key_index`. The sequence + /// and atomicity contract is identical to `append`. + fn append_with_keys( + &self, + persistence_id: &str, + expected_sequence: u64, + events: &[PersistenceEnvelope], + key_rows: &[EntityKeyRow], + ) -> impl std::future::Future> + Send { + let _ = key_rows; + self.append(persistence_id, expected_sequence, events) + } + + /// Backfill declared key-index rows for an **existing** entity (ADR-0153), + /// without appending a journal event. Idempotent: re-running yields the same + /// rows. Used to populate `entity_key_index` for entities written before the + /// declared key existed, so a keyed read can authoritatively prove absence + /// (the per-tenant backfill watermark gates #324's retirement). The default + /// is a no-op (non-indexing backends); query-plane stores upsert the rows. + fn backfill_entity_keys( + &self, + tenant: &str, + entity_type: &str, + entity_id: &str, + key_rows: &[EntityKeyRow], + ) -> impl std::future::Future> + Send { + let _ = (tenant, entity_type, entity_id, key_rows); + async { Ok(()) } + } + + /// Resolve an entity by a declared key (ADR-0153): the `entity_id` currently + /// holding `(key_name, key_hash)`, or `None` if absent. This is the + /// negative-existence access path — present *and* absent in one `O(log n)` + /// probe, no scan. Default returns `None` (non-indexing backends); the + /// query-plane stores override it against `entity_key_index`. + fn lookup_by_key( + &self, + tenant: &str, + entity_type: &str, + key_name: &str, + key_hash: &str, + ) -> impl std::future::Future, PersistenceError>> + Send { + let _ = (tenant, entity_type, key_name, key_hash); + async { Ok(None) } + } + /// Atomically append events to multiple journals. /// /// Backends must either commit every append in `appends`, or commit none. diff --git a/crates/temper-server/src/entity_actor/actor.rs b/crates/temper-server/src/entity_actor/actor.rs index 1ee428e1..a9413292 100644 --- a/crates/temper-server/src/entity_actor/actor.rs +++ b/crates/temper-server/src/entity_actor/actor.rs @@ -327,6 +327,7 @@ impl EntityActor { /// Persist an event to the configured event store. async fn persist_event( + &self, store: &BoxedEventStore, backend: BackendLabel, persistence_id: &str, @@ -350,9 +351,28 @@ impl EntityActor { // W2 / temper#146: measure append wait — the hypothesis is that // writer-lock / fsync serialization is a cold-start bottleneck. + // ADR-0153: derive the declared key rows from the new state and co-commit + // them with the journal append, so a keyed read is correct without a scan. + let key_rows = { + let table = self.table.read().expect("table lock poisoned"); + let mut rows = Vec::new(); + if let Some(field_map) = state.fields.as_object() { + for key in &table.keys { + if let Some(hash) = + crate::key_index::canonical_key_hash(&key.name, &key.properties, field_map) + { + rows.push(temper_runtime::persistence::EntityKeyRow { + key_name: key.name.clone(), + key_hash: hash, + }); + } + } + } + rows + }; let append_start = Instant::now(); let result = store - .append(persistence_id, state.sequence_nr, &[envelope]) + .append_with_keys(persistence_id, state.sequence_nr, &[envelope], &key_rows) .await; crate::runtime_metrics::record_event_store_append_wait( backend.as_str(), @@ -729,7 +749,7 @@ impl Actor for EntityActor { if let (Some(store), Some(backend)) = (self.event_journal.as_ref(), self.event_backend) { - Self::persist_event(store, backend, &self.persistence_id(), &mut state, &created) + self.persist_event(store, backend, &self.persistence_id(), &mut state, &created) .await .map_err(|e| { ActorError::custom(format!( @@ -924,14 +944,9 @@ impl Actor for EntityActor { if let (Some(store), Some(backend)) = (self.event_journal.as_ref(), self.event_backend) { - let first_persist = Self::persist_event( - store, - backend, - &self.persistence_id(), - state, - &event, - ) - .await; + let first_persist = self + .persist_event(store, backend, &self.persistence_id(), state, &event) + .await; match first_persist { Ok(_) => { @@ -1072,14 +1087,15 @@ impl Actor for EntityActor { )) .await; // determinism-ok: rare retry backoff (ADR-0046) - match Self::persist_event( - store, - backend, - &self.persistence_id(), - state, - &retry_event, - ) - .await + match self + .persist_event( + store, + backend, + &self.persistence_id(), + state, + &retry_event, + ) + .await { Ok(_) => { // Commit re-evaluated event + result into @@ -1380,9 +1396,9 @@ impl Actor for EntityActor { if let (Some(store), Some(backend)) = (self.event_journal.as_ref(), self.event_backend) - && let Err(e) = - Self::persist_event(store, backend, &self.persistence_id(), state, &deleted) - .await + && let Err(e) = self + .persist_event(store, backend, &self.persistence_id(), state, &deleted) + .await { ctx.reply(EntityResponse { success: false, diff --git a/crates/temper-server/src/key_index.rs b/crates/temper-server/src/key_index.rs new file mode 100644 index 00000000..d59b6a9a --- /dev/null +++ b/crates/temper-server/src/key_index.rs @@ -0,0 +1,281 @@ +//! Declared composite-key index hashing (ADR-0153, ARN-68). +//! +//! A declared `[[key]]` (an alternate / unique key) is reduced to a single +//! canonical, type-tagged `key_hash` so the kernel can maintain `entity_key_index` +//! and answer "present -> entity_id" or "absent" in one `O(log n)` probe — the +//! negative-existence access path the read plane lacks today. +//! +//! Both the write path (the entity actor, hashing the new state's key values) and +//! the read path (resolving a `$filter` that matches a declared key) compute the +//! hash here, so they always agree. The function is **deterministic** (SHA-256, no +//! clock, no randomness, no map iteration) and therefore safe under deterministic +//! simulation. + +use sha2::{Digest, Sha256}; +use temper_jit::table::types::DeclaredKey; + +/// Separates `key_name` from the value list. +const UNIT_SEP: u8 = 0x1F; +/// Separates one property's encoded value from the next. +const RECORD_SEP: u8 = 0x1E; + +/// Type tags keep `"5"` (string) and `5` (number) from colliding. +const TAG_STRING: u8 = b'S'; +const TAG_NUMBER: u8 = b'N'; +const TAG_BOOL: u8 = b'B'; + +/// Canonical `key_hash` for a declared key's values, or `None` when the key is +/// not fully present. +/// +/// `properties` is the declared key's property set **in declared order** (the +/// order is part of the canonical form). For each property the entity's current +/// scalar value is encoded as `(type_tag, canonical_text)`. Returns `None` if the +/// key has no properties, or if any property is missing, null, or non-scalar — a +/// partial key is not indexable, so the entity simply has no `entity_key_index` +/// row for that key (it is still reachable by `Id`). +pub fn canonical_key_hash( + key_name: &str, + properties: &[String], + fields: &serde_json::Map, +) -> Option { + if properties.is_empty() { + return None; + } + let mut hasher = Sha256::new(); + hasher.update(key_name.as_bytes()); + hasher.update([UNIT_SEP]); + for prop in properties { + let (tag, canon) = canonical_value(lookup_field(fields, prop)?)?; + hasher.update([tag]); + hasher.update(canon.as_bytes()); + hasher.update([RECORD_SEP]); + } + Some(format!("{:x}", hasher.finalize())) +} + +/// Look up a declared key property in the entity's fields, tolerant of the +/// snake/Pascal case split between how params are stored and how OData filters +/// name them. Entity writes store action params verbatim (often snake_case, e.g. +/// `workspace_id`), while OData reads name properties in the CSDL's PascalCase +/// (`WorkspaceId`) — and some callers write Pascal directly. The declared +/// `[[key]]` uses one convention; this finds the field regardless. Exact match +/// first (so existing same-case behavior is unchanged), then snake, then Pascal. +/// The hash itself is over key values, not names, so a case-tolerant lookup makes +/// the write-side and read-side hashes agree for the same entity. +fn lookup_field<'a>( + fields: &'a serde_json::Map, + prop: &str, +) -> Option<&'a serde_json::Value> { + if let Some(v) = fields.get(prop) { + return Some(v); + } + let snake = temper_spec::to_snake_case(prop); + if let Some(v) = fields.get(&snake) { + return Some(v); + } + let pascal = temper_spec::to_pascal_case(prop); + fields.get(&pascal) +} + +/// Encode one scalar value as `(type_tag, canonical_text)`. `None` for null or a +/// non-scalar (a partial / unindexable key component). +fn canonical_value(value: &serde_json::Value) -> Option<(u8, String)> { + use serde_json::Value; + match value { + Value::Null => None, + Value::Bool(b) => Some((TAG_BOOL, if *b { "1" } else { "0" }.to_string())), + Value::Number(n) => Some((TAG_NUMBER, n.to_string())), + Value::String(s) => Some((TAG_STRING, s.clone())), + Value::Array(_) | Value::Object(_) => None, + } +} + +/// Resolve a read's equality predicates to a declared key + its `key_hash`, for +/// the read-plane fast path (ADR-0153): if `equality_pairs` is exactly one +/// declared key's property set, return `(key_name, key_hash)` so the caller can +/// probe `entity_key_index` instead of scanning. `None` if no declared key +/// matches. +/// +/// Values are taken as strings (how the OData URL / `$filter` deliver them), +/// which matches the write-side hash for string-typed key properties — the case +/// for every declared business key today (File `WorkspaceId+Path`, etc.). A +/// non-string key property would need the value coerced to its declared type +/// before this matches; that is tracked as a follow-up and a mismatch only +/// declines the fast path (the caller falls back to the scan), never a wrong hit. +pub fn resolve_query_to_key( + keys: &[DeclaredKey], + equality_pairs: &[(String, String)], +) -> Option<(String, String)> { + let mut fields = serde_json::Map::new(); + for (prop, value) in equality_pairs { + fields.insert(prop.clone(), serde_json::Value::String(value.clone())); + } + for key in keys { + if key.properties.len() == equality_pairs.len() + && key.properties.iter().all(|p| fields.contains_key(p)) + { + let hash = canonical_key_hash(&key.name, &key.properties, &fields)?; + return Some((key.name.clone(), hash)); + } + } + None +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + fn fields(pairs: &[(&str, serde_json::Value)]) -> serde_json::Map { + pairs + .iter() + .map(|(k, v)| (k.to_string(), v.clone())) + .collect() + } + + #[test] + fn same_values_hash_equal_different_values_differ() { + let props = vec!["WorkspaceId".to_string(), "Path".to_string()]; + let a = fields(&[("WorkspaceId", json!("ws1")), ("Path", json!("/a.md"))]); + let b = fields(&[("WorkspaceId", json!("ws1")), ("Path", json!("/a.md"))]); + let c = fields(&[("WorkspaceId", json!("ws1")), ("Path", json!("/b.md"))]); + assert_eq!( + canonical_key_hash("path", &props, &a), + canonical_key_hash("path", &props, &b) + ); + assert_ne!( + canonical_key_hash("path", &props, &a), + canonical_key_hash("path", &props, &c) + ); + } + + #[test] + fn type_tag_prevents_string_number_collision() { + let props = vec!["K".to_string()]; + let as_str = fields(&[("K", json!("5"))]); + let as_num = fields(&[("K", json!(5))]); + assert_ne!( + canonical_key_hash("k", &props, &as_str), + canonical_key_hash("k", &props, &as_num) + ); + } + + #[test] + fn key_name_is_part_of_the_hash() { + let props = vec!["K".to_string()]; + let f = fields(&[("K", json!("v"))]); + assert_ne!( + canonical_key_hash("a", &props, &f), + canonical_key_hash("b", &props, &f) + ); + } + + #[test] + fn property_order_matters() { + let f = fields(&[("X", json!("1")), ("Y", json!("2"))]); + let xy = vec!["X".to_string(), "Y".to_string()]; + let yx = vec!["Y".to_string(), "X".to_string()]; + assert_ne!( + canonical_key_hash("k", &xy, &f), + canonical_key_hash("k", &yx, &f) + ); + } + + #[test] + fn missing_or_null_or_empty_yields_none() { + let props = vec!["WorkspaceId".to_string(), "Path".to_string()]; + // missing Path + let missing = fields(&[("WorkspaceId", json!("ws1"))]); + assert!(canonical_key_hash("path", &props, &missing).is_none()); + // null Path + let null_path = fields(&[("WorkspaceId", json!("ws1")), ("Path", json!(null))]); + assert!(canonical_key_hash("path", &props, &null_path).is_none()); + // no declared properties + let f = fields(&[("WorkspaceId", json!("ws1"))]); + assert!(canonical_key_hash("path", &[], &f).is_none()); + } + + fn path_key() -> Vec { + vec![DeclaredKey { + name: "path".to_string(), + properties: vec!["WorkspaceId".to_string(), "Path".to_string()], + }] + } + + #[test] + fn resolve_query_matches_declared_key_and_agrees_with_write_hash() { + let pairs = vec![ + ("WorkspaceId".to_string(), "ws1".to_string()), + ("Path".to_string(), "/a.md".to_string()), + ]; + let (name, hash) = resolve_query_to_key(&path_key(), &pairs).expect("matches declared key"); + assert_eq!(name, "path"); + // The read-side hash MUST equal the write-side hash for the same string + // values — that is why present/absent works. + let write_side = canonical_key_hash( + "path", + &["WorkspaceId".to_string(), "Path".to_string()], + &fields(&[("WorkspaceId", json!("ws1")), ("Path", json!("/a.md"))]), + ) + .unwrap(); + assert_eq!( + hash, write_side, + "read-side hash must equal write-side hash" + ); + } + + #[test] + fn write_snake_and_read_pascal_hash_agree() { + // The real-entity case: the write stores params snake_case (workspace_id, + // path), the OData read names them PascalCase (WorkspaceId, Path). The hash + // is over VALUES, and the lookup is case-tolerant, so both sides agree. + let key = vec!["WorkspaceId".to_string(), "Path".to_string()]; + let write_fields = fields(&[ + ("workspace_id", json!("ws-1")), + ("path", json!("/a.md")), + ("Status", json!("Ready")), + ]); + let write_hash = canonical_key_hash("ws_path", &key, &write_fields).expect("write hash"); + + // Read side: resolve a PascalCase $filter against the same declared key. + let read_pairs = vec![ + ("WorkspaceId".to_string(), "ws-1".to_string()), + ("Path".to_string(), "/a.md".to_string()), + ]; + let decl = vec![DeclaredKey { + name: "ws_path".to_string(), + properties: key.clone(), + }]; + let (_, read_hash) = resolve_query_to_key(&decl, &read_pairs).expect("read resolves"); + assert_eq!( + write_hash, read_hash, + "write (snake fields) and read (Pascal filter) must hash equal — else the keyed read always misses" + ); + + // And Pascal-stored writes (other callers) also agree. + let pascal_write = fields(&[("WorkspaceId", json!("ws-1")), ("Path", json!("/a.md"))]); + assert_eq!( + canonical_key_hash("ws_path", &key, &pascal_write).expect("pascal write hash"), + read_hash, + ); + } + + #[test] + fn resolve_query_declines_when_no_key_matches() { + // wrong arity, wrong property, and no declared keys -> decline (fall back to scan) + assert!( + resolve_query_to_key(&path_key(), &[("WorkspaceId".into(), "ws1".into())]).is_none() + ); + assert!( + resolve_query_to_key( + &path_key(), + &[ + ("Other".into(), "x".into()), + ("Path".into(), "/a.md".into()) + ] + ) + .is_none() + ); + assert!(resolve_query_to_key(&[], &[("WorkspaceId".into(), "ws1".into())]).is_none()); + } +} diff --git a/crates/temper-server/src/lib.rs b/crates/temper-server/src/lib.rs index 7065ad73..46eea042 100644 --- a/crates/temper-server/src/lib.rs +++ b/crates/temper-server/src/lib.rs @@ -21,6 +21,8 @@ pub mod eventual_invariants; pub mod http_endpoint; pub mod idempotency; pub mod identity; +/// ADR-0153: declared composite-key index hashing (the negative-existence access path). +pub mod key_index; #[cfg(feature = "observe")] pub mod observe; pub mod odata; diff --git a/crates/temper-server/src/odata/query_plane_read/mod.rs b/crates/temper-server/src/odata/query_plane_read/mod.rs index 3af4e6ee..5c02e4a8 100644 --- a/crates/temper-server/src/odata/query_plane_read/mod.rs +++ b/crates/temper-server/src/odata/query_plane_read/mod.rs @@ -24,6 +24,40 @@ fn should_try_native_before_catalog_coverage( plan.filter_pushdown && request.query_options.count != Some(true) } +/// ADR-0153: when the read's `$filter` is exactly a declared `[[key]]`, resolve +/// it to the single matching `entity_id` via `entity_key_index` — a bounded +/// candidate set (no full-type scan, so the budget cannot trip → no 413). +/// +/// Returns `Some(vec![id])` on a keyed **hit**; `None` otherwise — including a +/// keyed **miss**, which falls back to the full scan because (until the backfill +/// gate lands) a missing key row may be a pre-backfill entity rather than a true +/// absence. So hits are fast and correct now; authoritative absence follows the +/// per-tenant backfill watermark. `$orderby`/`$count` also decline (a point read +/// has neither to honor). +async fn keyed_candidate_ids(request: &QueryPlaneReadRequest<'_>) -> Option> { + if request.query_options.orderby.is_some() || request.query_options.count == Some(true) { + return None; + } + let filter = request.query_options.filter.as_ref()?; + let pairs = super::filter_sql::equality_field_predicates(filter)?; + let table = request.state.transition_tables.get(request.entity_type)?; + let (key_name, key_hash) = crate::key_index::resolve_query_to_key(&table.keys, &pairs)?; + let (store, _) = request.state.event_journal()?; + match store + .lookup_by_key( + request.tenant.as_str(), + request.entity_type, + &key_name, + &key_hash, + ) + .await + { + Ok(Some(entity_id)) => Some(vec![entity_id]), + // Miss or error: fall back to the full path (safe pre-backfill). + Ok(None) | Err(_) => None, + } +} + fn should_reconcile_empty_exact_match_against_authoritative( request: &QueryPlaneReadRequest<'_>, indexed_entity_ids: &[String], @@ -170,10 +204,22 @@ pub(in crate::odata) async fn read_entity_set_from_query_plane( } } - let all_entity_ids = request - .state - .list_entity_ids_lazy(request.tenant, request.entity_type) - .await; + // ADR-0153 fast path: if the filter is exactly a declared `[[key]]` (the + // shape behind the agents' 413 — `Files?$filter=WorkspaceId eq … and Path eq …`), + // probe `entity_key_index` for the one matching entity_id instead of listing the + // whole entity type. The candidate set becomes bounded (0 or 1), so the rest of + // the read (coverage, budget, materialization, row-auth) runs unchanged and the + // scan budget can never trip. On a miss we fall back to the full list, which still + // covers pre-backfill entities — a safe additive fast path until #324 is retired. + let all_entity_ids = match keyed_candidate_ids(&request).await { + Some(ids) => ids, + None => { + request + .state + .list_entity_ids_lazy(request.tenant, request.entity_type) + .await + } + }; let needs_full_proof = request.query_options.filter.is_some() || request.query_options.orderby.is_some() || request.query_options.count == Some(true); diff --git a/crates/temper-server/src/odata/query_plane_read/tests.rs b/crates/temper-server/src/odata/query_plane_read/tests.rs index ad71226d..8ad01436 100644 --- a/crates/temper-server/src/odata/query_plane_read/tests.rs +++ b/crates/temper-server/src/odata/query_plane_read/tests.rs @@ -17,6 +17,7 @@ use temper_runtime::tenant::TenantId; use temper_spec::csdl::parse_csdl; use temper_store_turso::TursoEventStore; +mod dst_projection_lag; mod proof; const CSDL_XML: &str = include_str!("../../../../../test-fixtures/specs/model.csdl.xml"); diff --git a/crates/temper-server/src/odata/query_plane_read/tests/dst_projection_lag.rs b/crates/temper-server/src/odata/query_plane_read/tests/dst_projection_lag.rs new file mode 100644 index 00000000..51413485 --- /dev/null +++ b/crates/temper-server/src/odata/query_plane_read/tests/dst_projection_lag.rs @@ -0,0 +1,360 @@ +//! Deterministic simulation of the read/projection plane under projection lag +//! (ADR-0153, ARN-68). +//! +//! This is a TRUE DST, not a unit/integration/e2e test: it runs the **real** +//! read planner (`read_entity_set_from_query_plane`) under +//! `install_deterministic_context(seed)` across many seeds, against deterministic +//! in-memory backends (`SimEventStore` for the journal + key index, `SimQueryPlane` +//! for the catalog), with a **fault injected** — the async query projection lags, +//! modeled as "the field-index pushdown is unavailable" (`query_field_index_page` +//! returns `None`). That is the production trigger for the 413: native pushdown +//! can't narrow, so the planner falls back to the authoritative scan, which trips +//! the read budget at scale. +//! +//! The simulation reproduces the 413 deterministically (no key, or a non-key +//! filter → QueryTooLarge) and proves the co-committed declared-key index +//! eliminates it (keyed filter → bounded candidate, no 413) — under every seed. +//! This is the plane the existing DST did not cover; it is why this class of bug +//! (413, read-after-write) kept escaping to production. + +use super::*; +use crate::storage::{ + BoxedEventStore, EntityCatalogRow, QueryFieldIndexOrder, QueryFieldIndexPage, QueryPlaneStore, + QueryProjectionFieldsRow, StorageStack, +}; +use async_trait::async_trait; +use std::collections::BTreeMap; +use std::sync::Mutex; +use temper_runtime::persistence::{ + EntityKeyRow, EventMetadata, PersistenceEnvelope, PersistenceError, +}; +use temper_runtime::scheduler::{install_deterministic_context, sim_now, sim_uuid}; +use temper_store_sim::SimEventStore; + +const DST_SEEDS: u64 = 64; + +/// Deterministic in-memory query plane. Models the production async projection: +/// the catalog is populated by `upsert_projection`, but **field-index pushdown is +/// reported unavailable** (`query_field_index_page` → `None`) to simulate the +/// projection lagging behind the journal — the exact condition that makes the +/// real planner fall back to the authoritative scan (and 413 at scale). +#[derive(Default)] +struct SimQueryPlane { + // (entity_type, entity_id) -> catalog row + catalog: Mutex>, +} + +#[async_trait] +impl QueryPlaneStore for SimQueryPlane { + async fn upsert_projection( + &self, + _tenant: &str, + entity_type: &str, + entity_id: &str, + status: &str, + fields: &serde_json::Value, + state: &serde_json::Value, + sequence_nr: u64, + ) -> Result<(), PersistenceError> { + self.catalog.lock().unwrap().insert( + (entity_type.to_string(), entity_id.to_string()), + EntityCatalogRow { + entity_id: entity_id.to_string(), + status: status.to_string(), + fields: fields.clone(), + state: Some(state.clone()), + sequence_nr, + }, + ); + Ok(()) + } + + async fn remove_projection( + &self, + _tenant: &str, + entity_type: &str, + entity_id: &str, + ) -> Result<(), PersistenceError> { + self.catalog + .lock() + .unwrap() + .remove(&(entity_type.to_string(), entity_id.to_string())); + Ok(()) + } + + async fn query_field_index( + &self, + _tenant: &str, + _entity_type: &str, + _where_clause: &str, + _params: Vec, + ) -> Result>, PersistenceError> { + // Projection lag: the field index cannot narrow this read. + Ok(None) + } + + async fn query_field_index_page( + &self, + _tenant: &str, + _entity_type: &str, + _where_clause: &str, + _params: Vec, + _order_by: &[QueryFieldIndexOrder], + _skip: usize, + _top: usize, + _include_count: bool, + ) -> Result, PersistenceError> { + // The injected fault: native pushdown is unavailable (projection lag), + // so the planner must fall back to the authoritative scan. + Ok(None) + } + + async fn load_projection_fields_many( + &self, + _tenant: &str, + _entity_type: &str, + _entity_ids: &[String], + _field_names: &[&str], + ) -> Result>, PersistenceError> { + Ok(None) + } + + async fn load_entity_catalog_rows( + &self, + _tenant: &str, + entity_type: &str, + entity_ids: &[String], + ) -> Result>, PersistenceError> { + let catalog = self.catalog.lock().unwrap(); + let rows = entity_ids + .iter() + .filter_map(|id| catalog.get(&(entity_type.to_string(), id.clone())).cloned()) + .collect(); + Ok(Some(rows)) + } + + async fn projected_entity_counts_by_tenant( + &self, + ) -> Result>, PersistenceError> { + Ok(None) + } +} + +fn envelope(event_type: &str, payload: serde_json::Value) -> PersistenceEnvelope { + PersistenceEnvelope { + sequence_nr: 1, + event_type: event_type.to_string(), + payload, + metadata: EventMetadata { + event_id: sim_uuid(), + causation_id: sim_uuid(), + correlation_id: sim_uuid(), + timestamp: sim_now(), + actor_id: "dst-lag".to_string(), + }, + } +} + +fn doc_key_hash(workspace: &str, path: &str) -> String { + crate::key_index::canonical_key_hash( + "ws_path", + &["WorkspaceId".to_string(), "Path".to_string()], + &serde_json::json!({ "WorkspaceId": workspace, "Path": path }) + .as_object() + .unwrap() + .clone(), + ) + .expect("complete key") +} + +fn eq_filter(ws: &str, path: &str) -> FilterExpr { + FilterExpr::BinaryOp { + left: Box::new(FilterExpr::BinaryOp { + left: Box::new(FilterExpr::Property("WorkspaceId".to_string())), + op: BinaryOperator::Eq, + right: Box::new(FilterExpr::Literal(ODataValue::String(ws.to_string()))), + }), + op: BinaryOperator::And, + right: Box::new(FilterExpr::BinaryOp { + left: Box::new(FilterExpr::Property("Path".to_string())), + op: BinaryOperator::Eq, + right: Box::new(FilterExpr::Literal(ODataValue::String(path.to_string()))), + }), + } +} + +/// Build a sim-backed ServerState with the keyed Order table installed. +fn sim_state(seed: u64, qp: std::sync::Arc) -> (ServerState, BoxedEventStore) { + let events = BoxedEventStore::new(SimEventStore::no_faults(seed)); + let mut state = build_order_state("dst-projection-lag"); + state.set_storage_stack(StorageStack::new( + crate::storage::BackendLabel::Sim, + events.clone(), + None, + None, + None, + None, + Some(qp), + None, + None, + None, + )); + state.transition_tables = std::sync::Arc::new( + [( + "Order".to_string(), + std::sync::Arc::new(temper_jit::table::TransitionTable::from_ioa_source( + ORDER_IOA, + )), + )] + .into_iter() + .collect(), + ); + (state, events) +} + +/// THE DST. Under projection lag (pushdown unavailable) and a workspace larger +/// than the read budget: +/// * a read with NO usable keyed access path falls back to the authoritative +/// scan and returns **413 QueryTooLarge** — reproducing the production bug +/// deterministically; +/// * the SAME read, when its `$filter` is exactly the declared key and the +/// co-committed key row exists, resolves to a **bounded single candidate** and +/// returns **200** — the elimination. +/// Holds under every seed. +#[tokio::test] +async fn dst_projection_lag_413_eliminated_by_keyed_index() { + for seed in 0..DST_SEEDS { + let (_guard, _clock, _id) = install_deterministic_context(seed); + let qp = std::sync::Arc::new(SimQueryPlane::default()); + let (state, events) = sim_state(seed, qp.clone()); + let tenant = TenantId::default(); + + let ws = "ws-lag"; + let target_path = "/proofs/keyed-under-lag.txt"; + let target_id = format!("ord-target-{seed}"); + + // A workspace larger than the read budget. Noise lives in the journal only + // (so list_entity_ids sees them) — the projection lags for everyone. + for i in 0..16usize { + let pid = format!("{tenant}:Order:noise-{seed}-{i:02}"); + events + .append(&pid, 0, &[envelope("Create", serde_json::json!({}))]) + .await + .unwrap(); + } + + // The target: co-commit the declared key row with the journal (immediate, + // no lag), and project it to the catalog so it can be materialized once + // the keyed path bounds the candidate set. + let target_pid = format!("{tenant}:Order:{target_id}"); + events + .append_with_keys( + &target_pid, + 0, + &[envelope( + "Create", + serde_json::json!({ "WorkspaceId": ws, "Path": target_path }), + )], + &[EntityKeyRow { + key_name: "ws_path".to_string(), + key_hash: doc_key_hash(ws, target_path), + }], + ) + .await + .unwrap(); + qp.upsert_projection( + tenant.as_str(), + "Order", + &target_id, + "Created", + &serde_json::json!({ "Id": target_id, "WorkspaceId": ws, "Path": target_path }), + &serde_json::json!({}), + 1, + ) + .await + .unwrap(); + + let security_ctx = SecurityContext::system(); + // A budget smaller than the workspace: scan_candidate_budget = max(10*1, 1). + let budget = QueryPlaneReadBudget { + default_page_size: 1, + max_entities: 1, + }; + + // RED (reproduce the prod 413): a present-but-non-keyed read shape. Use a + // filter on a non-declared property; pushdown is unavailable (lag), so the + // planner scans the whole workspace (> budget) -> QueryTooLarge. + let non_key = QueryOptions { + filter: Some(FilterExpr::BinaryOp { + left: Box::new(FilterExpr::Property("Notes".to_string())), + op: BinaryOperator::Eq, + right: Box::new(FilterExpr::Literal(ODataValue::String("x".to_string()))), + }), + ..QueryOptions::default() + }; + let red = read_entity_set_from_query_plane(QueryPlaneReadRequest { + state: &state, + tenant: &tenant, + security_ctx: &security_ctx, + entity_type: "Order", + entity_set_name: "Orders", + query_options: &non_key, + budget, + }) + .await; + assert!( + matches!(red, Err(QueryPlaneReadError::QueryTooLarge { .. })), + "seed {seed}: under projection lag a non-keyed read at scale must 413 (reproduces the prod bug)" + ); + + // RED, filter held constant: the SAME keyed $filter shape, but for a key + // that has NO co-committed row (pre-backfill / absent). The keyed lookup + // misses, falls back to the scan -> 413. This isolates the key row as the + // single variable behind the before/after. + let keyed_absent = QueryOptions { + filter: Some(eq_filter(ws, "/no-such-path.txt")), + ..QueryOptions::default() + }; + let red_absent = read_entity_set_from_query_plane(QueryPlaneReadRequest { + state: &state, + tenant: &tenant, + security_ctx: &security_ctx, + entity_type: "Order", + entity_set_name: "Orders", + query_options: &keyed_absent, + budget, + }) + .await; + assert!( + matches!(red_absent, Err(QueryPlaneReadError::QueryTooLarge { .. })), + "seed {seed}: same keyed filter shape with NO key row must still 413 (pre-backfill scan fallback)" + ); + + // GREEN (the fix): the SAME workspace + budget + lag, but the $filter is + // exactly the declared key and the co-committed key row exists. The keyed + // fast path bounds the candidate set to the single id -> no scan -> 200. + let keyed = QueryOptions { + filter: Some(eq_filter(ws, target_path)), + ..QueryOptions::default() + }; + let green = read_entity_set_from_query_plane(QueryPlaneReadRequest { + state: &state, + tenant: &tenant, + security_ctx: &security_ctx, + entity_type: "Order", + entity_set_name: "Orders", + query_options: &keyed, + budget, + }) + .await; + let green = match green { + Ok(r) => r, + Err(_) => panic!("seed {seed}: keyed read under lag must not 413"), + }; + assert!( + green.telemetry.candidate_count <= 1, + "seed {seed}: keyed read must bound the candidate set (no scan); candidate_count={}", + green.telemetry.candidate_count + ); + } +} diff --git a/crates/temper-server/src/odata/query_plane_read/tests/proof.rs b/crates/temper-server/src/odata/query_plane_read/tests/proof.rs index 9135b72e..c8fcf3bf 100644 --- a/crates/temper-server/src/odata/query_plane_read/tests/proof.rs +++ b/crates/temper-server/src/odata/query_plane_read/tests/proof.rs @@ -1,5 +1,336 @@ use super::*; +/// ADR-0153 read-plane proof on real Postgres (the prod engine): a `$filter` that +/// is exactly the declared `[[key]]` resolves to the single matching entity via +/// `entity_key_index` — a bounded candidate (no full-type scan → the budget that +/// raises the 413 can never trip). Gated on DATABASE_URL; unique tenant; cleans up. +/// +/// Proves `keyed_candidate_ids` composes the verified pieces against a live DB: +/// real `append_with_keys` co-commit → real `lookup_by_key` → bounded `[id]`. +#[test] +fn keyed_filter_resolves_to_bounded_candidate_on_postgres() { + use temper_runtime::persistence::{ + EntityKeyRow, EventMetadata, EventStore, PersistenceEnvelope, + }; + use temper_runtime::scheduler::{sim_now, sim_uuid as runtime_sim_uuid}; + + let database_url = match std::env::var("DATABASE_URL") { + Ok(url) => url, + Err(_) => return, + }; + + sqlx::test_block_on(async { + let pool = sqlx::PgPool::connect(&database_url).await.unwrap(); + temper_store_postgres::migration::run_migrations(&pool) + .await + .unwrap(); + let store = temper_store_postgres::PostgresEventStore::new(pool.clone()); + let mut state = build_order_state("query-plane-keyed-pg"); + state.set_storage_stack(StorageStack::from_postgres(store.clone())); + // from_registry leaves transition_tables empty; the keyed fast path reads + // the declared `[[key]]` from the Order table, so install it (with the + // ws_path key from order.ioa.toml). + state.transition_tables = std::sync::Arc::new( + [( + "Order".to_string(), + std::sync::Arc::new(temper_jit::table::TransitionTable::from_ioa_source( + ORDER_IOA, + )), + )] + .into_iter() + .collect(), + ); + + // Unique tenant isolates this run on the shared DB. The Order table (with + // the declared `ws_path` key) is found by entity_type regardless of tenant. + let tenant_str = format!("tenant-keyed-pg-{}", runtime_sim_uuid()); + let tenant = TenantId::from(tenant_str.clone()); + let workspace_id = "ws-keyed"; + let target_path = "/proofs/keyed-read.txt"; + let target_id = "ord-keyed-target"; + + // Write the target's key row via the REAL co-commit (journal + entity_key_index). + let key_hash = crate::key_index::canonical_key_hash( + "ws_path", + &["WorkspaceId".to_string(), "Path".to_string()], + &serde_json::json!({ "WorkspaceId": workspace_id, "Path": target_path }) + .as_object() + .unwrap() + .clone(), + ) + .expect("complete key"); + let envelope = PersistenceEnvelope { + sequence_nr: 1, + event_type: "Create".to_string(), + payload: serde_json::json!({ "WorkspaceId": workspace_id, "Path": target_path }), + metadata: EventMetadata { + event_id: runtime_sim_uuid(), + causation_id: runtime_sim_uuid(), + correlation_id: runtime_sim_uuid(), + timestamp: sim_now(), + actor_id: "keyed-pg-proof".to_string(), + }, + }; + store + .append_with_keys( + &format!("{tenant_str}:Order:{target_id}"), + 0, + &[envelope], + &[EntityKeyRow { + key_name: "ws_path".to_string(), + key_hash, + }], + ) + .await + .expect("co-commit key row"); + + // AMPLIFICATION BOUND (ADR-0148 not regressed): the synchronous co-commit + // wrote exactly K=1 key row and ZERO broad-index rows. The S-wide + // entity_field_index (the write we migrated to async) is NOT touched by the + // append path — it is still written only by the async projection. This is + // the empirical proof the fix does not bring back synchronous write + // amplification. + let key_rows: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM entity_key_index WHERE tenant = $1 AND entity_id = $2", + ) + .bind(&tenant_str) + .bind(target_id) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(key_rows, 1, "co-commit writes exactly K=1 declared-key row"); + let broad_index_rows: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM entity_field_index WHERE tenant = $1 AND entity_id = $2", + ) + .bind(&tenant_str) + .bind(target_id) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!( + broad_index_rows, 0, + "the synchronous append must NOT write the S-wide broad index (stays async — no amplification regression)" + ); + + let security_ctx = SecurityContext::system(); + // PRESENT: $filter == the declared key -> bounded [target_id]. + let keyed = QueryOptions { + filter: Some(FilterExpr::BinaryOp { + left: Box::new(FilterExpr::BinaryOp { + left: Box::new(FilterExpr::Property("WorkspaceId".to_string())), + op: BinaryOperator::Eq, + right: Box::new(FilterExpr::Literal(ODataValue::String( + workspace_id.to_string(), + ))), + }), + op: BinaryOperator::And, + right: Box::new(FilterExpr::BinaryOp { + left: Box::new(FilterExpr::Property("Path".to_string())), + op: BinaryOperator::Eq, + right: Box::new(FilterExpr::Literal(ODataValue::String( + target_path.to_string(), + ))), + }), + }), + ..QueryOptions::default() + }; + let request = QueryPlaneReadRequest { + state: &state, + tenant: &tenant, + security_ctx: &security_ctx, + entity_type: "Order", + entity_set_name: "Orders", + query_options: &keyed, + budget: QueryPlaneReadBudget { + default_page_size: 10, + max_entities: 10, + }, + }; + assert_eq!( + keyed_candidate_ids(&request).await, + Some(vec![target_id.to_string()]), + "a $filter matching the declared key must resolve to the bounded single id via entity_key_index" + ); + + // CONTROL: a non-key filter does not resolve -> None (falls back to scan). + let non_key = QueryOptions { + filter: Some(FilterExpr::BinaryOp { + left: Box::new(FilterExpr::Property("Status".to_string())), + op: BinaryOperator::Eq, + right: Box::new(FilterExpr::Literal(ODataValue::String( + "Created".to_string(), + ))), + }), + ..QueryOptions::default() + }; + let control = QueryPlaneReadRequest { + query_options: &non_key, + ..request + }; + assert_eq!( + keyed_candidate_ids(&control).await, + None, + "a non-key filter must decline the keyed fast path" + ); + + // ABSENT key: a declared-key filter for a key NO entity holds resolves to a + // keyed MISS. Pre-backfill that returns None (fall back to scan), NOT an + // authoritative empty — a missing key row may be a not-yet-backfilled entity. + let absent = QueryOptions { + filter: Some(FilterExpr::BinaryOp { + left: Box::new(FilterExpr::BinaryOp { + left: Box::new(FilterExpr::Property("WorkspaceId".to_string())), + op: BinaryOperator::Eq, + right: Box::new(FilterExpr::Literal(ODataValue::String(workspace_id.into()))), + }), + op: BinaryOperator::And, + right: Box::new(FilterExpr::BinaryOp { + left: Box::new(FilterExpr::Property("Path".to_string())), + op: BinaryOperator::Eq, + right: Box::new(FilterExpr::Literal(ODataValue::String( + "/proofs/does-not-exist.txt".to_string(), + ))), + }), + }), + ..QueryOptions::default() + }; + let absent_req = QueryPlaneReadRequest { + query_options: &absent, + ..request + }; + assert_eq!( + keyed_candidate_ids(&absent_req).await, + None, + "a keyed miss must decline to scan fallback pre-backfill (not authoritative-empty)" + ); + + // Clean up this run's rows. + let _ = sqlx::query("DELETE FROM entity_key_index WHERE tenant = $1") + .bind(&tenant_str) + .execute(&pool) + .await; + let _ = sqlx::query("DELETE FROM events WHERE tenant = $1") + .bind(&tenant_str) + .execute(&pool) + .await; + }); +} + +/// ADR-0153 boundary: the keyed fast path engages ONLY for a pure-equality filter +/// that is exactly a declared `[[key]]`. Every other shape must decline (return +/// None) so the read falls back to the existing scan/pushdown path — no behavior +/// change, no false hit. No DB needed: all these decline before any store call. +#[tokio::test] +async fn keyed_fast_path_declines_non_key_shapes() { + let mut state = build_order_state("query-plane-keyed-decline"); + state.transition_tables = std::sync::Arc::new( + [( + "Order".to_string(), + std::sync::Arc::new(temper_jit::table::TransitionTable::from_ioa_source( + ORDER_IOA, + )), + )] + .into_iter() + .collect(), + ); + let tenant = TenantId::default(); + let security_ctx = SecurityContext::system(); + let budget = QueryPlaneReadBudget { + default_page_size: 10, + max_entities: 10, + }; + + let eq = |prop: &str, val: &str| FilterExpr::BinaryOp { + left: Box::new(FilterExpr::Property(prop.to_string())), + op: BinaryOperator::Eq, + right: Box::new(FilterExpr::Literal(ODataValue::String(val.to_string()))), + }; + let and = |l: FilterExpr, r: FilterExpr| FilterExpr::BinaryOp { + left: Box::new(l), + op: BinaryOperator::And, + right: Box::new(r), + }; + + // The declared key is (WorkspaceId, Path). Each of these must decline: + let cases: Vec<(&str, QueryOptions)> = vec![ + // non-lossless conjunct (the agents' original Status ne failure shape) + ( + "status_ne", + QueryOptions { + filter: Some(and( + and(eq("WorkspaceId", "ws"), eq("Path", "/a")), + FilterExpr::BinaryOp { + left: Box::new(FilterExpr::Property("Status".to_string())), + op: BinaryOperator::Ne, + right: Box::new(FilterExpr::Literal(ODataValue::String("Archived".into()))), + }, + )), + ..QueryOptions::default() + }, + ), + // eq null (the Directories root lookup shape) + ( + "eq_null", + QueryOptions { + filter: Some(and( + eq("WorkspaceId", "ws"), + FilterExpr::BinaryOp { + left: Box::new(FilterExpr::Property("ParentId".to_string())), + op: BinaryOperator::Eq, + right: Box::new(FilterExpr::Literal(ODataValue::Null)), + }, + )), + ..QueryOptions::default() + }, + ), + // partial key (only one of the two key properties) + ( + "partial_key", + QueryOptions { + filter: Some(eq("WorkspaceId", "ws")), + ..QueryOptions::default() + }, + ), + // non-key property + ( + "non_key_prop", + QueryOptions { + filter: Some(eq("Notes", "hi")), + ..QueryOptions::default() + }, + ), + // $orderby present (a point read has no ordering to honor) + ( + "with_orderby", + QueryOptions { + filter: Some(and(eq("WorkspaceId", "ws"), eq("Path", "/a"))), + orderby: Some(vec![OrderByClause { + property: "Path".to_string(), + direction: OrderDirection::Asc, + }]), + ..QueryOptions::default() + }, + ), + ]; + + for (name, query_options) in &cases { + let request = QueryPlaneReadRequest { + state: &state, + tenant: &tenant, + security_ctx: &security_ctx, + entity_type: "Order", + entity_set_name: "Orders", + query_options, + budget, + }; + assert_eq!( + keyed_candidate_ids(&request).await, + None, + "keyed fast path must decline shape '{name}' (falls back to scan/pushdown)" + ); + } +} + async fn upsert_order_projection_with_status( store: &TursoEventStore, tenant: &TenantId, diff --git a/crates/temper-server/src/odata/read.rs b/crates/temper-server/src/odata/read.rs index f7816421..f39293e1 100644 --- a/crates/temper-server/src/odata/read.rs +++ b/crates/temper-server/src/odata/read.rs @@ -304,6 +304,22 @@ async fn try_resolve_composite_entity_key( entity_type: &str, key_pairs: &[(String, String)], ) -> Option { + // ADR-0153 fast path: if the key is a declared `[[key]]`, probe + // `entity_key_index` (O(log n), present/absent) instead of the candidate + // scan. On a miss we fall through to the scan, which still covers + // pre-backfill entities — a safe additive fast path until #324's scan is + // retired behind the backfill gate. + if let Some(table) = state.transition_tables.get(entity_type) + && let Some((key_name, key_hash)) = + crate::key_index::resolve_query_to_key(&table.keys, key_pairs) + && let Some((store, _)) = state.event_journal() + && let Ok(Some(entity_id)) = store + .lookup_by_key(tenant.as_str(), entity_type, &key_name, &key_hash) + .await + { + return Some(entity_id); + } + let query_plane = state.query_plane_store()?; let filter = composite_key_filter(key_pairs)?; let translated = filter_sql::try_translate_candidate_filter(&filter)?; diff --git a/crates/temper-server/src/state/entity_ops.rs b/crates/temper-server/src/state/entity_ops.rs index bb88c32a..be491c1c 100644 --- a/crates/temper-server/src/state/entity_ops.rs +++ b/crates/temper-server/src/state/entity_ops.rs @@ -414,6 +414,9 @@ impl ServerState { #[instrument(skip_all, fields(otel.name = "entity.populate_field_index", tenant = %tenant))] pub async fn populate_field_index_from_snapshots(&self, tenant: &TenantId) { projection_backfill::populate_field_index_from_snapshots(self, tenant).await; + // ADR-0153: backfill entity_key_index for declared-key entity types in the + // same pass, so keyed reads can authoritatively prove absence post-backfill. + projection_backfill::populate_key_index_from_snapshots(self, tenant).await; } /// Compare durable projection rows with authoritative state rebuilt by event replay. diff --git a/crates/temper-server/src/state/projection_backfill.rs b/crates/temper-server/src/state/projection_backfill.rs index 82fabaf1..60581027 100644 --- a/crates/temper-server/src/state/projection_backfill.rs +++ b/crates/temper-server/src/state/projection_backfill.rs @@ -31,6 +31,81 @@ fn transition_table_for( }) } +/// Backfill `entity_key_index` for existing entities (ADR-0153). For each entity +/// of a type that declares `[[key]]`, derive the key rows from its current state +/// and upsert them (no journal event). After this completes for a tenant, a keyed +/// read can authoritatively prove absence (the gate for retiring #324's scan). +/// Idempotent; entities written after the declaration already co-commit their keys. +pub(super) async fn populate_key_index_from_snapshots(state: &ServerState, tenant: &TenantId) { + let Some((store, _backend)) = state.event_journal() else { + return; + }; + + let entities = { + let index = state.entity_index.read().unwrap(); + let mut result = Vec::new(); + for (index_key, ids) in index.iter() { + let prefix = format!("{tenant}:"); + if let Some(entity_type) = index_key.strip_prefix(&prefix) { + for id in ids { + result.push((entity_type.to_string(), id.clone())); + } + } + } + result + }; + + let mut indexed = 0usize; + for (entity_type, entity_id) in &entities { + // Only types that declare keys need backfilling. + let keys = match transition_table_for(state, tenant, entity_type) { + Some(table) if !table.keys.is_empty() => table.keys.clone(), + _ => continue, + }; + let persistence_id = format!("{tenant}:{entity_type}:{entity_id}"); + let Ok(Some((_seq, snapshot_bytes))) = store.load_snapshot(&persistence_id).await else { + continue; + }; + let Ok(snap) = serde_json::from_slice::(&snapshot_bytes) + else { + continue; + }; + let Some(field_map) = snap.fields.as_object() else { + continue; + }; + let mut key_rows = Vec::new(); + for key in &keys { + if let Some(hash) = + crate::key_index::canonical_key_hash(&key.name, &key.properties, field_map) + { + key_rows.push(temper_runtime::persistence::EntityKeyRow { + key_name: key.name.clone(), + key_hash: hash, + }); + } + } + if key_rows.is_empty() { + continue; + } + match store + .backfill_entity_keys(tenant.as_str(), entity_type, entity_id, &key_rows) + .await + { + Ok(()) => indexed += 1, + Err(e) => tracing::debug!( + error = %e, entity_type = %entity_type, entity_id = %entity_id, + "key index backfill: upsert failed" + ), + } + } + tracing::info!( + tenant = %tenant, + entities = entities.len(), + indexed, + "entity_key_index backfill complete" + ); +} + pub(super) async fn populate_field_index_from_snapshots(state: &ServerState, tenant: &TenantId) { let overall_started_at = Instant::now(); // determinism-ok: production-only backfill duration metric let Some((store, backend)) = state.event_journal() else { diff --git a/crates/temper-server/src/storage/mod.rs b/crates/temper-server/src/storage/mod.rs index f8f40b5d..49fd5ddf 100644 --- a/crates/temper-server/src/storage/mod.rs +++ b/crates/temper-server/src/storage/mod.rs @@ -71,6 +71,30 @@ pub trait DynEventStore: Send + Sync { from_sequence: u64, ) -> EventStoreFuture<'a, Result, PersistenceError>>; + fn append_with_keys<'a>( + &'a self, + persistence_id: &'a str, + expected_sequence: u64, + events: &'a [PersistenceEnvelope], + key_rows: &'a [temper_runtime::persistence::EntityKeyRow], + ) -> EventStoreFuture<'a, Result>; + + fn lookup_by_key<'a>( + &'a self, + tenant: &'a str, + entity_type: &'a str, + key_name: &'a str, + key_hash: &'a str, + ) -> EventStoreFuture<'a, Result, PersistenceError>>; + + fn backfill_entity_keys<'a>( + &'a self, + tenant: &'a str, + entity_type: &'a str, + entity_id: &'a str, + key_rows: &'a [temper_runtime::persistence::EntityKeyRow], + ) -> EventStoreFuture<'a, Result<(), PersistenceError>>; + fn save_snapshot<'a>( &'a self, persistence_id: &'a str, @@ -135,6 +159,54 @@ where Box::pin(EventStore::read_events(self, persistence_id, from_sequence)) } + fn append_with_keys<'a>( + &'a self, + persistence_id: &'a str, + expected_sequence: u64, + events: &'a [PersistenceEnvelope], + key_rows: &'a [temper_runtime::persistence::EntityKeyRow], + ) -> EventStoreFuture<'a, Result> { + Box::pin(EventStore::append_with_keys( + self, + persistence_id, + expected_sequence, + events, + key_rows, + )) + } + + fn lookup_by_key<'a>( + &'a self, + tenant: &'a str, + entity_type: &'a str, + key_name: &'a str, + key_hash: &'a str, + ) -> EventStoreFuture<'a, Result, PersistenceError>> { + Box::pin(EventStore::lookup_by_key( + self, + tenant, + entity_type, + key_name, + key_hash, + )) + } + + fn backfill_entity_keys<'a>( + &'a self, + tenant: &'a str, + entity_type: &'a str, + entity_id: &'a str, + key_rows: &'a [temper_runtime::persistence::EntityKeyRow], + ) -> EventStoreFuture<'a, Result<(), PersistenceError>> { + Box::pin(EventStore::backfill_entity_keys( + self, + tenant, + entity_type, + entity_id, + key_rows, + )) + } + fn save_snapshot<'a>( &'a self, persistence_id: &'a str, @@ -239,6 +311,42 @@ impl BoxedEventStore { self.0.read_events(persistence_id, from_sequence).await } + pub async fn append_with_keys( + &self, + persistence_id: &str, + expected_sequence: u64, + events: &[PersistenceEnvelope], + key_rows: &[temper_runtime::persistence::EntityKeyRow], + ) -> Result { + self.0 + .append_with_keys(persistence_id, expected_sequence, events, key_rows) + .await + } + + pub async fn lookup_by_key( + &self, + tenant: &str, + entity_type: &str, + key_name: &str, + key_hash: &str, + ) -> Result, PersistenceError> { + self.0 + .lookup_by_key(tenant, entity_type, key_name, key_hash) + .await + } + + pub async fn backfill_entity_keys( + &self, + tenant: &str, + entity_type: &str, + entity_id: &str, + key_rows: &[temper_runtime::persistence::EntityKeyRow], + ) -> Result<(), PersistenceError> { + self.0 + .backfill_entity_keys(tenant, entity_type, entity_id, key_rows) + .await + } + pub async fn save_snapshot( &self, persistence_id: &str, diff --git a/crates/temper-server/tests/dst_entity_key_index.rs b/crates/temper-server/tests/dst_entity_key_index.rs new file mode 100644 index 00000000..07dc0a66 --- /dev/null +++ b/crates/temper-server/tests/dst_entity_key_index.rs @@ -0,0 +1,242 @@ +//! DST: the entity_key_index negative-existence invariant (ADR-0153, ARN-68). +//! +//! Property: a read by an entity's declared key returns the entity **iff** it +//! exists — present and absent in one probe, under every seed. This is the +//! access path the read plane lacks today (proving absence requires a scan, +//! which 413s at scale). Real `EntityActor` + `SimEventStore`, all seeds. +//! +//! RED until the co-commit + keyed read land: the actor must pass the declared +//! key to `append_with_keys`, and the store must write/read `entity_key_index`. +//! Today both default to no-ops, so the keyed read of an existing Doc returns +//! `None` and the `present` assertion fails — that is the missing access path. + +use std::collections::BTreeMap; +use std::sync::{Arc, RwLock}; +use std::time::Duration; + +use temper_jit::table::TransitionTable; +use temper_runtime::ActorSystem; +use temper_runtime::persistence::{EntityKeyRow, EventMetadata, EventStore, PersistenceEnvelope}; +use temper_runtime::scheduler::{install_deterministic_context, sim_now, sim_uuid}; +use temper_server::key_index::canonical_key_hash; +use temper_server::storage::{BackendLabel, BoxedEventStore}; +use temper_server::{EntityActor, EntityMsg, EntityResponse}; +use temper_store_sim::SimEventStore; + +const DOC_IOA: &str = include_str!("../../../test-fixtures/specs/keyed_doc.ioa.toml"); +const NUM_SEEDS: u64 = 100; + +fn doc_table() -> Arc> { + Arc::new(RwLock::new(TransitionTable::from_ioa_source(DOC_IOA))) +} + +async fn dispatch( + actor_ref: &temper_runtime::actor::ActorRef, + action: &str, + params: serde_json::Value, +) -> EntityResponse { + actor_ref + .ask( + EntityMsg::Action { + name: action.to_string(), + params, + cross_entity_booleans: BTreeMap::new(), + idempotency_key: None, + }, + Duration::from_secs(5), + ) + .await + .expect("actor should respond") +} + +fn doc_key_hash(workspace: &str, path: &str) -> String { + let mut fields = serde_json::Map::new(); + fields.insert("WorkspaceId".to_string(), serde_json::json!(workspace)); + fields.insert("Path".to_string(), serde_json::json!(path)); + canonical_key_hash( + "path", + &["WorkspaceId".to_string(), "Path".to_string()], + &fields, + ) + .expect("complete key") +} + +#[tokio::test] +async fn dst_keyed_read_is_present_iff_entity_exists() { + 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 = doc_table(); + let entity_id = format!("doc-{seed}"); + + let system = ActorSystem::new("dst-keyed"); + let actor = EntityActor::with_persistence( + "Doc", + &entity_id, + table.clone(), + serde_json::json!({}), + store.clone(), + BackendLabel::Sim, + ) + .with_tenant("default"); + let actor_ref = system.spawn(actor, &entity_id); + + let r = dispatch( + &actor_ref, + "Create", + serde_json::json!({ "WorkspaceId": "ws1", "Path": "/a.md" }), + ) + .await; + assert!(r.success, "seed {seed}: Create failed: {:?}", r.error); + + // PRESENT: the created Doc resolves by its declared (WorkspaceId, Path) key. + let present = store + .lookup_by_key("default", "Doc", "path", &doc_key_hash("ws1", "/a.md")) + .await + .expect("lookup ok"); + assert_eq!( + present, + Some(entity_id.clone()), + "seed {seed}: keyed read must find the created Doc (present)" + ); + + // ABSENT: a key no Doc holds resolves to None in one probe (no scan). + let absent = store + .lookup_by_key("default", "Doc", "path", &doc_key_hash("ws1", "/nope.md")) + .await + .expect("lookup ok"); + assert_eq!( + absent, None, + "seed {seed}: keyed read of a missing key must be absent" + ); + } +} + +/// Backfill (ADR-0153): an entity written BEFORE its key was declared has no key +/// row, so a keyed read misses (would fall back to scan). After +/// `backfill_entity_keys`, the keyed read resolves it — the path that lets a keyed +/// miss become authoritative absence post-backfill. Idempotent. All seeds. +#[tokio::test] +async fn dst_backfill_makes_pre_existing_entity_keyed_findable() { + for seed in 0..NUM_SEEDS { + let (_guard, _clock, _id) = install_deterministic_context(seed); + let store = SimEventStore::no_faults(seed); + + // Pre-existing entity: journal append WITHOUT keys (as if written before + // the [[key]] declaration existed). + let pid = format!("default:Doc:doc-pre-{seed}"); + store.append(&pid, 0, &[test_envelope()]).await.unwrap(); + + let key_hash = doc_key_hash("ws1", "/pre.md"); + // Before backfill: keyed miss. + assert_eq!( + store + .lookup_by_key("default", "Doc", "path", &key_hash) + .await + .unwrap(), + None, + "seed {seed}: pre-backfill entity has no key row (keyed miss)" + ); + + // Backfill the key row (no new journal event). + let rows = [EntityKeyRow { + key_name: "path".to_string(), + key_hash: key_hash.clone(), + }]; + store + .backfill_entity_keys("default", "Doc", &format!("doc-pre-{seed}"), &rows) + .await + .unwrap(); + // Idempotent: a second backfill is a no-op-equivalent. + store + .backfill_entity_keys("default", "Doc", &format!("doc-pre-{seed}"), &rows) + .await + .unwrap(); + + // After backfill: keyed read resolves it. + assert_eq!( + store + .lookup_by_key("default", "Doc", "path", &key_hash) + .await + .unwrap(), + Some(format!("doc-pre-{seed}")), + "seed {seed}: after backfill the entity is keyed-findable" + ); + } +} + +fn test_envelope() -> PersistenceEnvelope { + PersistenceEnvelope { + sequence_nr: 1, + event_type: "Create".to_string(), + payload: serde_json::json!({}), + metadata: EventMetadata { + event_id: sim_uuid(), + causation_id: sim_uuid(), + correlation_id: sim_uuid(), + timestamp: sim_now(), + actor_id: "test".to_string(), + }, + } +} + +/// Atomicity: a co-commit that hits a declared-key uniqueness reject must leave +/// the journal UNCHANGED — present iff the journal committed. A reject that still +/// advanced the journal would replay the rejected transition. Real SimEventStore +/// co-commit path, all seeds. +#[tokio::test] +async fn dst_co_commit_atomic_on_uniqueness_reject() { + for seed in 0..NUM_SEEDS { + let (_guard, _clock, _id) = install_deterministic_context(seed); + let store = SimEventStore::no_faults(seed); + let key = EntityKeyRow { + key_name: "path".to_string(), + key_hash: "k-collision".to_string(), + }; + + // A claims the key. + store + .append_with_keys( + "default:Doc:doc-a", + 0, + &[test_envelope()], + std::slice::from_ref(&key), + ) + .await + .unwrap_or_else(|e| panic!("seed {seed}: A should claim the key: {e:?}")); + + // B tries the SAME key -> must be rejected AND must not advance B's journal. + let res = store + .append_with_keys( + "default:Doc:doc-b", + 0, + &[test_envelope()], + std::slice::from_ref(&key), + ) + .await; + assert!( + res.is_err(), + "seed {seed}: a duplicate declared key must be rejected" + ); + + let b_events = store + .read_events("default:Doc:doc-b", 0) + .await + .expect("read B journal"); + assert!( + b_events.is_empty(), + "seed {seed}: a rejected co-commit must leave the journal unchanged (atomic); got {} event(s)", + b_events.len() + ); + + let holder = store + .lookup_by_key("default", "Doc", "path", "k-collision") + .await + .expect("lookup"); + assert_eq!( + holder, + Some("doc-a".to_string()), + "seed {seed}: the key must still be held by A" + ); + } +} diff --git a/crates/temper-spec/src/automaton/toml_parser/mod.rs b/crates/temper-spec/src/automaton/toml_parser/mod.rs index ffb748a1..453dd6fd 100644 --- a/crates/temper-spec/src/automaton/toml_parser/mod.rs +++ b/crates/temper-spec/src/automaton/toml_parser/mod.rs @@ -28,6 +28,9 @@ enum Section { Integration, FieldInvariant, StateTimeout, + /// ADR-0153: `[[key]]` unique-key declarations. Passthrough; extracted via + /// serde in the second pass. + Key, Webhook, /// ADR-0046: nested `[[action.triggers]]` blocks. Hand-rolled parser /// skips the body; triggers are extracted via serde in the second pass @@ -70,6 +73,9 @@ impl ParseState { // ADR-0049: state_timeouts use nested inline tables for params; // parse via serde in the second pass rather than field-by-field. "[[state_timeout]]" => self.start_passthrough_section(Section::StateTimeout), + // ADR-0153: [[key]] unique-key declarations — passthrough; serde + // extracts them in the second pass. + "[[key]]" => self.start_passthrough_section(Section::Key), "[[webhook]]" => self.start_webhook_section(), _ if line.starts_with("[webhook.") => self.start_webhook_section(), // ADR-0046: nested [[action.triggers]] — flush the action body so @@ -99,6 +105,7 @@ impl ParseState { Section::Integration => self.apply_integration_field(key, &value), Section::FieldInvariant | Section::StateTimeout + | Section::Key | Section::Webhook | Section::ActionTrigger | Section::CompositeActionMetadata @@ -149,6 +156,7 @@ impl ParseState { context_entities: Vec::new(), field_invariants: Vec::new(), state_timeouts: Vec::new(), + keys: Vec::new(), admission: None, }) } @@ -431,6 +439,9 @@ pub(super) fn parse_toml_to_automaton(input: &str) -> Result Result, AutomatonParseError> { + let slice = isolate_sections(source, "[[key]]"); + if slice.trim().is_empty() { + return Ok(Vec::new()); + } + + #[derive(serde::Deserialize)] + struct KeyWrapper { + #[serde(default, rename = "key")] + keys: Vec, + } + toml::from_str::(&slice) + .map(|w| w.keys) + .map_err(|e| AutomatonParseError::Toml(format!("key: {e}"))) +} + /// Return a minimal TOML document containing only the `[[field_invariant]]` /// sections from `source`. Any other top-level section is skipped. /// diff --git a/crates/temper-spec/src/automaton/toml_parser/tests.rs b/crates/temper-spec/src/automaton/toml_parser/tests.rs index 05468943..b9683c5f 100644 --- a/crates/temper-spec/src/automaton/toml_parser/tests.rs +++ b/crates/temper-spec/src/automaton/toml_parser/tests.rs @@ -13,6 +13,37 @@ fn parse_kv_no_equals() { assert!(parse_kv("no_equals_here").is_none()); } +#[test] +fn extracts_declared_unique_keys() { + // ADR-0153: [[key]] declares an alternate (unique) key the kernel indexes. + let src = r#" +[automaton] +name = "File" +states = ["Created", "Ready"] +initial = "Created" + +[[key]] +name = "path" +properties = ["WorkspaceId", "Path"] + +[[key]] +name = "id" +properties = ["Id"] +"#; + let keys = extract_keys(src).expect("extract keys"); + assert_eq!(keys.len(), 2); + assert_eq!(keys[0].name, "path"); + assert_eq!(keys[0].properties, vec!["WorkspaceId", "Path"]); + assert_eq!(keys[1].name, "id"); + assert_eq!(keys[1].properties, vec!["Id"]); +} + +#[test] +fn extract_keys_empty_when_no_key_blocks() { + let src = "[automaton]\nname = \"File\"\nstates = [\"Created\"]\ninitial = \"Created\"\n"; + assert!(extract_keys(src).expect("extract keys").is_empty()); +} + #[test] fn parse_kv_trims_whitespace() { let (key, value) = parse_kv(" key = \"value\" ").unwrap(); diff --git a/crates/temper-spec/src/automaton/types.rs b/crates/temper-spec/src/automaton/types.rs index d0ee5c11..20ea2420 100644 --- a/crates/temper-spec/src/automaton/types.rs +++ b/crates/temper-spec/src/automaton/types.rs @@ -43,6 +43,13 @@ pub struct Automaton { /// unless the entity leaves the state or a `reset_on` action fires. #[serde(default, rename = "state_timeout")] pub state_timeouts: Vec, + /// ADR-0153: declared unique keys (alternate keys). Each names a property + /// set guaranteed unique across entities of this type; the kernel maintains + /// a keyed index (`entity_key_index`) over it for O(log n) present/absent + /// reads — the negative-existence access path. The OData alternate-key + /// annotation is derived from this declaration. + #[serde(default, rename = "key")] + pub keys: Vec, /// Admission control caps (ADR-0051). When present, the dispatch layer /// gates concurrent calls per `(tenant, entity_type, action)` before /// reaching the actor. @@ -50,6 +57,19 @@ pub struct Automaton { pub admission: Option, } +/// ADR-0153: a declared unique key (alternate key) on an entity. `properties` +/// is the set of state/field names whose combined values uniquely identify one +/// entity. The kernel maintains `entity_key_index` over each declared key for +/// O(log n) present/absent reads; the canonical key hash uses `properties` in +/// declared order. Multiple keys on one entity are distinguished by `name`. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct KeyDecl { + /// Identifier for this key (the `key_name` in `entity_key_index`). + pub name: String, + /// The property set that is unique, in canonical (hash) order. + pub properties: Vec, +} + /// Automaton metadata. #[derive(Debug, Clone, Serialize, Deserialize)] pub struct AutomatonMeta { diff --git a/crates/temper-store-postgres/migrations/0009_entity_key_index.sql b/crates/temper-store-postgres/migrations/0009_entity_key_index.sql new file mode 100644 index 00000000..bbd82fce --- /dev/null +++ b/crates/temper-store-postgres/migrations/0009_entity_key_index.sql @@ -0,0 +1,27 @@ +-- ADR-0153: declared composite-key index — the negative-existence access path. +-- +-- One row per (declared key, entity), co-committed with the journal append in the +-- same transaction (unlike the eventually-consistent entity_field_index). A keyed +-- read becomes a single O(log n) probe: hit -> entity_id, miss -> authoritatively +-- absent, so the read plane no longer has to scan a whole entity type to prove +-- absence (the cause of the 413, ARN-68). +-- +-- The PRIMARY KEY (tenant, entity_type, key_name, key_hash) enforces declared-key +-- uniqueness: a second entity claiming the same key tuple conflicts, which the +-- write path surfaces as a typed error (reject-and-surface). key_hash is a +-- canonical, type-tagged hash of the declared key's values. sequence_nr carries +-- the journal position for the staleness guard on upsert. +CREATE TABLE IF NOT EXISTS entity_key_index ( + tenant TEXT NOT NULL, + entity_type TEXT NOT NULL, + key_name TEXT NOT NULL, + key_hash TEXT NOT NULL, + entity_id TEXT NOT NULL, + sequence_nr BIGINT NOT NULL DEFAULT 0, + PRIMARY KEY (tenant, entity_type, key_name, key_hash) +); + +-- Reverse lookup: all key rows for an entity, so the write path can upsert/delete +-- an entity's declared-key rows when it changes or is removed. +CREATE INDEX IF NOT EXISTS idx_eki_entity + ON entity_key_index (tenant, entity_type, entity_id); diff --git a/crates/temper-store-postgres/src/data_only_create.rs b/crates/temper-store-postgres/src/data_only_create.rs index 72d9c342..c1b3e2c6 100644 --- a/crates/temper-store-postgres/src/data_only_create.rs +++ b/crates/temper-store-postgres/src/data_only_create.rs @@ -207,6 +207,7 @@ impl PostgresEventStore { ); record_postgres_projection_index_fields( DATA_ONLY_CREATE_OPERATION, + entity_type, indexed_fields, skipped_fields, ); diff --git a/crates/temper-store-postgres/src/metrics.rs b/crates/temper-store-postgres/src/metrics.rs index 91677b54..744539c8 100644 --- a/crates/temper-store-postgres/src/metrics.rs +++ b/crates/temper-store-postgres/src/metrics.rs @@ -119,16 +119,24 @@ pub(crate) fn record_postgres_transaction_commit_duration( pub(crate) fn record_postgres_projection_index_fields( operation: &'static str, + entity_type: &str, indexed_fields: u64, skipped_fields: u64, ) { + // entity_type is bounded (the set of declared entity types, ~dozens), so it is + // safe as a metric dimension and lets us attribute the index fan-out (S) per + // entity type — needed to size the declared key index (ADR-0153, ARN-68). + let attrs = [ + KeyValue::new("operation", operation), + KeyValue::new("entity_type", entity_type.to_owned()), + ]; metrics() .projection_index_fields - .record(indexed_fields, &[KeyValue::new("operation", operation)]); + .record(indexed_fields, &attrs); if skipped_fields > 0 { metrics() .projection_skipped_index_fields_total - .add(skipped_fields, &[KeyValue::new("operation", operation)]); + .add(skipped_fields, &attrs); } } @@ -188,7 +196,7 @@ mod tests { record_postgres_pool_acquire_duration(Duration::from_millis(2), "event_append", "ok"); record_postgres_transaction_begin_duration(Duration::from_millis(1), "event_append", "ok"); record_postgres_transaction_commit_duration(Duration::from_millis(3), "event_append", "ok"); - record_postgres_projection_index_fields("query_projection_upsert", 4, 1); + record_postgres_projection_index_fields("query_projection_upsert", "TestEntity", 4, 1); record_postgres_projection_index_reconciliation( "query_projection_upsert", "skipped_unchanged", diff --git a/crates/temper-store-postgres/src/platform.rs b/crates/temper-store-postgres/src/platform.rs index 9fd28731..322e61fe 100644 --- a/crates/temper-store-postgres/src/platform.rs +++ b/crates/temper-store-postgres/src/platform.rs @@ -644,6 +644,7 @@ impl PostgresEventStore { ); record_postgres_projection_index_fields( QUERY_PROJECTION_UPSERT_OPERATION, + entity_type, indexed_fields, skipped_fields, ); @@ -725,6 +726,7 @@ impl PostgresEventStore { ); record_postgres_projection_index_fields( QUERY_PROJECTION_UPSERT_OPERATION, + entity_type, indexed_fields, skipped_fields, ); @@ -797,6 +799,7 @@ impl PostgresEventStore { ); record_postgres_projection_index_fields( QUERY_PROJECTION_UPSERT_OPERATION, + entity_type, indexed_fields, skipped_fields, ); diff --git a/crates/temper-store-postgres/src/store.rs b/crates/temper-store-postgres/src/store.rs index b8dbe7f7..a078cd93 100644 --- a/crates/temper-store-postgres/src/store.rs +++ b/crates/temper-store-postgres/src/store.rs @@ -62,6 +62,17 @@ impl EventStore for PostgresEventStore { persistence_id: &str, expected_sequence: u64, events: &[PersistenceEnvelope], + ) -> Result { + self.append_with_keys(persistence_id, expected_sequence, events, &[]) + .await + } + + async fn append_with_keys( + &self, + persistence_id: &str, + expected_sequence: u64, + events: &[PersistenceEnvelope], + key_rows: &[temper_runtime::persistence::EntityKeyRow], ) -> Result { let (tenant, entity_type, entity_id) = parse_persistence_id_parts(persistence_id).map_err(PersistenceError::Storage)?; @@ -126,6 +137,31 @@ impl EventStore for PostgresEventStore { }); } + // ADR-0153: validate declared-key uniqueness BEFORE writing the journal so + // a reject is atomic (the journal does not advance). A different entity + // already holding the key is the violation (reject + surface). + for key in key_rows { + let holder: Option<(String,)> = crate::dbm::postgres_query_as!( + "SELECT entity_id FROM entity_key_index \ + WHERE tenant = $1 AND entity_type = $2 AND key_name = $3 AND key_hash = $4", + ) + .bind(tenant) + .bind(entity_type) + .bind(&key.key_name) + .bind(&key.key_hash) + .fetch_optional(&mut *tx) + .await + .map_err(|e| PersistenceError::Storage(e.to_string()))?; + if let Some((existing,)) = holder + && existing != entity_id + { + return Err(PersistenceError::Storage(format!( + "duplicate declared key '{}' for {entity_type}: held by {existing}", + key.key_name + ))); + } + } + let segment_index = segments::open_segment_for_append(&mut tx, tenant, entity_type, entity_id, current_seq) .await?; @@ -176,6 +212,37 @@ impl EventStore for PostgresEventStore { .await?; } + // ADR-0153: co-commit the declared key-index rows in THIS transaction + // (uniqueness was validated above). Drop the entity's prior row for each + // key_name (the value may have changed), then claim the new key_hash. + for key in key_rows { + crate::dbm::postgres_query!( + "DELETE FROM entity_key_index \ + WHERE tenant = $1 AND entity_type = $2 AND key_name = $3 AND entity_id = $4", + ) + .bind(tenant) + .bind(entity_type) + .bind(&key.key_name) + .bind(entity_id) + .execute(&mut *tx) + .await + .map_err(|e| PersistenceError::Storage(e.to_string()))?; + crate::dbm::postgres_query!( + "INSERT INTO entity_key_index \ + (tenant, entity_type, key_name, key_hash, entity_id, sequence_nr) \ + VALUES ($1, $2, $3, $4, $5, $6)", + ) + .bind(tenant) + .bind(entity_type) + .bind(&key.key_name) + .bind(&key.key_hash) + .bind(entity_id) + .bind(new_seq as i64) + .execute(&mut *tx) + .await + .map_err(|e| PersistenceError::Storage(e.to_string()))?; + } + let commit_started = Instant::now(); tx.commit().await.map_err(|e| { record_postgres_transaction_commit_duration( @@ -195,6 +262,105 @@ impl EventStore for PostgresEventStore { Ok(new_seq) } + async fn backfill_entity_keys( + &self, + tenant: &str, + entity_type: &str, + entity_id: &str, + key_rows: &[temper_runtime::persistence::EntityKeyRow], + ) -> Result<(), PersistenceError> { + if key_rows.is_empty() { + return Ok(()); + } + let mut tx = self + .pool + .begin() + .await + .map_err(|e| PersistenceError::Storage(e.to_string()))?; + for key in key_rows { + // A different entity already holding this key is a pre-existing data + // conflict — log and skip (don't fail the whole backfill on one row; + // the conflict surfaces via the metric and a keyed read still resolves + // to whoever currently holds it). + let holder: Option<(String,)> = crate::dbm::postgres_query_as!( + "SELECT entity_id FROM entity_key_index \ + WHERE tenant = $1 AND entity_type = $2 AND key_name = $3 AND key_hash = $4", + ) + .bind(tenant) + .bind(entity_type) + .bind(&key.key_name) + .bind(&key.key_hash) + .fetch_optional(&mut *tx) + .await + .map_err(|e| PersistenceError::Storage(e.to_string()))?; + if let Some((existing,)) = &holder + && existing != entity_id + { + tracing::warn!( + tenant, entity_type, entity_id, existing, + key_name = %key.key_name, + "entity_key_index backfill: declared-key conflict; skipping" + ); + continue; + } + crate::dbm::postgres_query!( + "DELETE FROM entity_key_index \ + WHERE tenant = $1 AND entity_type = $2 AND key_name = $3 AND entity_id = $4", + ) + .bind(tenant) + .bind(entity_type) + .bind(&key.key_name) + .bind(entity_id) + .execute(&mut *tx) + .await + .map_err(|e| PersistenceError::Storage(e.to_string()))?; + crate::dbm::postgres_query!( + "INSERT INTO entity_key_index \ + (tenant, entity_type, key_name, key_hash, entity_id, sequence_nr) \ + VALUES ($1, $2, $3, $4, $5, 0) \ + ON CONFLICT (tenant, entity_type, key_name, key_hash) DO NOTHING", + ) + .bind(tenant) + .bind(entity_type) + .bind(&key.key_name) + .bind(&key.key_hash) + .bind(entity_id) + .execute(&mut *tx) + .await + .map_err(|e| PersistenceError::Storage(e.to_string()))?; + } + tx.commit() + .await + .map_err(|e| PersistenceError::Storage(e.to_string()))?; + Ok(()) + } + + async fn lookup_by_key( + &self, + tenant: &str, + entity_type: &str, + key_name: &str, + key_hash: &str, + ) -> Result, PersistenceError> { + let mut conn = self + .pool + .acquire() + .await + .map_err(|e| PersistenceError::Storage(e.to_string()))?; + let row: Option<(String,)> = crate::dbm::postgres_query_as!( + "SELECT entity_id FROM entity_key_index \ + WHERE tenant = $1 AND entity_type = $2 AND key_name = $3 AND key_hash = $4", + ) + .bind(tenant) + .bind(entity_type) + .bind(key_name) + .bind(key_hash) + .fetch_optional(&mut *conn) + .await + .map_err(|e| PersistenceError::Storage(e.to_string()))?; + Ok(row.map(|(id,)| id)) + } + /// Atomically append to multiple entity journals in one PostgreSQL /// transaction. Used as the storage foundation for cross-actor Composite /// transactions: every stream's optimistic-concurrency check must pass diff --git a/crates/temper-store-postgres/src/store_projection_test.rs b/crates/temper-store-postgres/src/store_projection_test.rs index 240382a2..6f1af045 100644 --- a/crates/temper-store-postgres/src/store_projection_test.rs +++ b/crates/temper-store-postgres/src/store_projection_test.rs @@ -1,6 +1,7 @@ use super::*; use crate::migration::run_migrations; use sqlx::PgPool; +use temper_runtime::persistence::{EntityKeyRow, EventStore}; fn test_envelope(event_type: &str, payload: serde_json::Value) -> PersistenceEnvelope { PersistenceEnvelope { @@ -17,6 +18,89 @@ fn test_envelope(event_type: &str, payload: serde_json::Value) -> PersistenceEnv } } +/// ADR-0153 live verification: the real postgres store honors the same +/// negative-existence + atomicity invariants the DST proved in SimEventStore. +/// Gated on DATABASE_URL (skips otherwise); isolated by a unique tenant. +#[test] +fn entity_key_index_present_absent_and_atomic_reject() { + let database_url = match std::env::var("DATABASE_URL") { + Ok(url) => url, + Err(_) => return, + }; + + sqlx::test_block_on(async { + let pool = PgPool::connect(&database_url).await.unwrap(); + run_migrations(&pool).await.unwrap(); + let store = PostgresEventStore::new(pool.clone()); + let tenant = format!("tenant-keyindex-{}", uuid::Uuid::new_v4()); + let key = EntityKeyRow { + key_name: "path".to_string(), + key_hash: format!("kh-{}", uuid::Uuid::new_v4()), + }; + + // A claims the key (co-committed with the journal append). + let pid_a = format!("{tenant}:Doc:doc-a"); + store + .append_with_keys( + &pid_a, + 0, + &[test_envelope("Create", serde_json::json!({}))], + std::slice::from_ref(&key), + ) + .await + .unwrap(); + + // PRESENT and ABSENT in one keyed probe. + assert_eq!( + store + .lookup_by_key(&tenant, "Doc", "path", &key.key_hash) + .await + .unwrap(), + Some("doc-a".to_string()), + ); + assert_eq!( + store + .lookup_by_key(&tenant, "Doc", "path", "no-such-hash") + .await + .unwrap(), + None, + ); + + // ATOMICITY: B claims the SAME key -> rejected, and B's journal is unchanged. + let pid_b = format!("{tenant}:Doc:doc-b"); + let res = store + .append_with_keys( + &pid_b, + 0, + &[test_envelope("Create", serde_json::json!({}))], + std::slice::from_ref(&key), + ) + .await; + assert!(res.is_err(), "duplicate declared key must be rejected"); + assert!( + store.read_events(&pid_b, 0).await.unwrap().is_empty(), + "a rejected co-commit must leave the journal unchanged (atomic)" + ); + assert_eq!( + store + .lookup_by_key(&tenant, "Doc", "path", &key.key_hash) + .await + .unwrap(), + Some("doc-a".to_string()), + ); + + // Clean up this test tenant's rows. + let _ = crate::dbm::postgres_query!("DELETE FROM entity_key_index WHERE tenant = $1") + .bind(&tenant) + .execute(&pool) + .await; + let _ = crate::dbm::postgres_query!("DELETE FROM events WHERE tenant = $1") + .bind(&tenant) + .execute(&pool) + .await; + }); +} + #[test] fn list_entity_ids_by_type_unions_catalog_field_index_and_events() { let database_url = match std::env::var("DATABASE_URL") { diff --git a/crates/temper-store-sim/src/lib.rs b/crates/temper-store-sim/src/lib.rs index e7940aa8..4c126eb3 100644 --- a/crates/temper-store-sim/src/lib.rs +++ b/crates/temper-store-sim/src/lib.rs @@ -141,6 +141,11 @@ struct SimEventStoreInner { /// persisted the transition, but the caller's ask timeout expired before /// the reply arrived". pending_append_delays: BTreeMap>, + /// ADR-0153: declared key-index, co-committed with the journal under the same + /// lock. `(tenant, entity_type, key_name, key_hash) -> entity_id`. This is the + /// deterministic reference for the negative-existence access path the real + /// stores maintain in `entity_key_index`. + key_index: BTreeMap<(String, String, String, String), String>, } #[derive(Debug, Clone, PartialEq, Eq)] @@ -166,6 +171,7 @@ impl SimEventStore { faults, pending_concurrency_violations: BTreeMap::new(), pending_append_delays: BTreeMap::new(), + key_index: BTreeMap::new(), })), } } @@ -304,6 +310,17 @@ impl EventStore for SimEventStore { persistence_id: &str, expected_sequence: u64, events: &[PersistenceEnvelope], + ) -> Result { + self.append_with_keys(persistence_id, expected_sequence, events, &[]) + .await + } + + async fn append_with_keys( + &self, + persistence_id: &str, + expected_sequence: u64, + events: &[PersistenceEnvelope], + key_rows: &[temper_runtime::persistence::EntityKeyRow], ) -> Result { let append_delay = { let mut inner = self @@ -388,6 +405,30 @@ impl EventStore for SimEventStore { }); } + // ADR-0153: validate declared-key uniqueness BEFORE writing the journal, so + // a reject is atomic — the journal must not advance on a rejected co-commit. + // A *different* entity already holding the key is the violation. + if !key_rows.is_empty() { + let mut parts = persistence_id.splitn(3, ':'); + let tenant = parts.next().unwrap_or(""); + let entity_type = parts.next().unwrap_or(""); + let entity_id = parts.next().unwrap_or(""); + for row in key_rows { + if let Some(existing) = inner.key_index.get(&( + tenant.to_string(), + entity_type.to_string(), + row.key_name.clone(), + row.key_hash.clone(), + )) && existing.as_str() != entity_id + { + return Err(PersistenceError::Storage(format!( + "duplicate declared key '{}' for {entity_type}: held by {existing}", + row.key_name + ))); + } + } + } + let mut new_seq = expected_sequence; let mut stored_events = Vec::with_capacity(events.len()); for event in events { @@ -438,9 +479,89 @@ impl EventStore for SimEventStore { .saturating_add(1); } + // ADR-0153: co-commit the declared key-index rows under the SAME lock as + // the journal write above (uniqueness was validated before the journal, so + // this only mutates — never fails). A keyed read is therefore consistent + // with the journal: the negative-existence access path. + if !key_rows.is_empty() { + let mut parts = persistence_id.splitn(3, ':'); + let tenant = parts.next().unwrap_or(""); + let entity_type = parts.next().unwrap_or(""); + let entity_id = parts.next().unwrap_or(""); + for row in key_rows { + // Drop the entity's prior row for this key_name (the value may have + // changed), then claim the new (key_name, key_hash) -> entity_id. + inner.key_index.retain(|(t, et, kn, _), eid| { + !(t.as_str() == tenant + && et.as_str() == entity_type + && kn.as_str() == row.key_name.as_str() + && eid.as_str() == entity_id) + }); + inner.key_index.insert( + ( + tenant.to_string(), + entity_type.to_string(), + row.key_name.clone(), + row.key_hash.clone(), + ), + entity_id.to_string(), + ); + } + } + Ok(new_seq) } + async fn backfill_entity_keys( + &self, + tenant: &str, + entity_type: &str, + entity_id: &str, + key_rows: &[temper_runtime::persistence::EntityKeyRow], + ) -> Result<(), PersistenceError> { + let mut inner = self.inner.lock().expect("SimEventStore lock poisoned"); // ci-ok: infallible lock + for row in key_rows { + let slot = ( + tenant.to_string(), + entity_type.to_string(), + row.key_name.clone(), + row.key_hash.clone(), + ); + match inner.key_index.get(&slot) { + // A different entity holds it — pre-existing conflict; skip (don't + // clobber, don't fail the backfill). + Some(existing) if existing.as_str() != entity_id => continue, + _ => { + inner.key_index.retain(|(t, et, kn, _), eid| { + !(t.as_str() == tenant + && et.as_str() == entity_type + && kn.as_str() == row.key_name.as_str() + && eid.as_str() == entity_id) + }); + inner.key_index.insert(slot, entity_id.to_string()); + } + } + } + Ok(()) + } + + async fn lookup_by_key( + &self, + tenant: &str, + entity_type: &str, + key_name: &str, + key_hash: &str, + ) -> Result, PersistenceError> { + let inner = self.inner.lock().expect("SimEventStore lock poisoned"); // ci-ok: infallible lock + let slot = ( + tenant.to_string(), + entity_type.to_string(), + key_name.to_string(), + key_hash.to_string(), + ); + Ok(inner.key_index.get(&slot).cloned()) + } + async fn append_batch( &self, appends: &[PersistenceAppend], @@ -689,299 +810,4 @@ impl EventStore for SimEventStore { } #[cfg(test)] -mod tests { - use super::*; - use temper_runtime::persistence::EventMetadata; - - fn test_envelope(seq: u64, event_type: &str) -> PersistenceEnvelope { - PersistenceEnvelope { - sequence_nr: seq, - event_type: event_type.to_string(), - payload: serde_json::json!({"test": true}), - metadata: EventMetadata { - event_id: uuid::Uuid::nil(), - causation_id: uuid::Uuid::nil(), - correlation_id: uuid::Uuid::nil(), - timestamp: chrono::DateTime::UNIX_EPOCH, - actor_id: "test".to_string(), - }, - } - } - - #[tokio::test] - async fn append_and_read_roundtrip() { - let store = SimEventStore::no_faults(42); - let pid = "default:Order:ord-1"; - - let new_seq = store - .append(pid, 0, &[test_envelope(0, "Created")]) - .await - .unwrap(); - assert_eq!(new_seq, 1); - - let events = store.read_events(pid, 0).await.unwrap(); - assert_eq!(events.len(), 1); - assert_eq!(events[0].sequence_nr, 1); - assert_eq!(events[0].event_type, "Created"); - } - - #[tokio::test] - async fn append_multiple_events() { - let store = SimEventStore::no_faults(42); - let pid = "default:Order:ord-2"; - - let seq = store - .append( - pid, - 0, - &[test_envelope(0, "Created"), test_envelope(0, "Submitted")], - ) - .await - .unwrap(); - assert_eq!(seq, 2); - - let events = store.read_events(pid, 0).await.unwrap(); - assert_eq!(events.len(), 2); - assert_eq!(events[0].sequence_nr, 1); - assert_eq!(events[1].sequence_nr, 2); - } - - #[tokio::test] - async fn append_batch_commits_multiple_journals_atomically() { - let store = SimEventStore::no_faults(42); - let appends = vec![ - PersistenceAppend { - persistence_id: "default:Order:ord-a".to_string(), - expected_sequence: 0, - events: vec![test_envelope(0, "Created")], - }, - PersistenceAppend { - persistence_id: "default:Order:ord-b".to_string(), - expected_sequence: 0, - events: vec![test_envelope(0, "Created"), test_envelope(0, "Submitted")], - }, - ]; - - let results = store.append_batch(&appends).await.unwrap(); - - assert_eq!( - results, - vec![ - PersistenceAppendResult { - persistence_id: "default:Order:ord-a".to_string(), - sequence_nr: 1, - }, - PersistenceAppendResult { - persistence_id: "default:Order:ord-b".to_string(), - sequence_nr: 2, - }, - ] - ); - assert_eq!(store.dump_journal("default:Order:ord-a").len(), 1); - assert_eq!(store.dump_journal("default:Order:ord-b").len(), 2); - } - - #[tokio::test] - async fn append_batch_conflict_leaves_all_journals_untouched() { - let store = SimEventStore::no_faults(42); - store - .append( - "default:Order:ord-existing", - 0, - &[test_envelope(0, "Created")], - ) - .await - .unwrap(); - - let err = store - .append_batch(&[ - PersistenceAppend { - persistence_id: "default:Order:ord-new".to_string(), - expected_sequence: 0, - events: vec![test_envelope(0, "Created")], - }, - PersistenceAppend { - persistence_id: "default:Order:ord-existing".to_string(), - expected_sequence: 0, - events: vec![test_envelope(0, "Submitted")], - }, - ]) - .await - .expect_err("second journal conflict should abort entire batch"); - - assert!( - matches!(err, PersistenceError::ConcurrencyViolation { .. }), - "unexpected error: {err}" - ); - assert!( - store.dump_journal("default:Order:ord-new").is_empty(), - "first append must not be persisted when a later stream conflicts" - ); - assert_eq!( - store.dump_journal("default:Order:ord-existing").len(), - 1, - "conflicting stream must keep its original journal only" - ); - } - - #[tokio::test] - async fn concurrency_violation_on_wrong_sequence() { - let store = SimEventStore::no_faults(42); - let pid = "default:Order:ord-3"; - - store - .append(pid, 0, &[test_envelope(0, "Created")]) - .await - .unwrap(); - - let err = store - .append(pid, 0, &[test_envelope(0, "Duplicate")]) - .await - .unwrap_err(); - - assert!(matches!( - err, - PersistenceError::ConcurrencyViolation { - expected: 0, - actual: 1 - } - )); - } - - #[tokio::test] - async fn snapshot_save_and_load() { - let store = SimEventStore::no_faults(42); - let pid = "default:Order:ord-4"; - - store.save_snapshot(pid, 5, b"state-data").await.unwrap(); - - let snap = store.load_snapshot(pid).await.unwrap(); - assert_eq!(snap, Some((5, b"state-data".to_vec()))); - } - - #[tokio::test] - async fn snapshot_save_records_history_and_rotates_segments() { - let store = SimEventStore::no_faults(42); - let pid = "default:Order:segmented"; - - store - .append( - pid, - 0, - &[test_envelope(0, "Created"), test_envelope(0, "Updated")], - ) - .await - .unwrap(); - store.save_snapshot(pid, 2, b"snapshot-2").await.unwrap(); - store - .append(pid, 2, &[test_envelope(0, "AfterSnapshot")]) - .await - .unwrap(); - - assert_eq!(store.snapshot_history_len(pid), 1); - let segments = store.dump_segments(pid); - assert_eq!(segments.len(), 2); - assert_eq!(segments[0].segment_index, 0); - assert_eq!(segments[0].snapshot_sequence, Some(2)); - assert!(segments[0].sealed); - assert_eq!(segments[1].segment_index, 1); - assert_eq!(segments[1].start_sequence_nr, 3); - assert_eq!(segments[1].end_sequence_nr, Some(3)); - assert!(!segments[1].sealed); - } - - #[tokio::test] - async fn load_snapshot_returns_none_when_empty() { - let store = SimEventStore::no_faults(42); - let snap = store - .load_snapshot("default:Order:nonexistent") - .await - .unwrap(); - assert_eq!(snap, None); - } - - #[tokio::test] - async fn list_entity_ids_filters_by_tenant() { - let store = SimEventStore::no_faults(42); - - store - .append("alpha:Order:ord-1", 0, &[test_envelope(0, "Created")]) - .await - .unwrap(); - store - .append("alpha:Task:task-1", 0, &[test_envelope(0, "Created")]) - .await - .unwrap(); - store - .append("beta:Order:ord-9", 0, &[test_envelope(0, "Created")]) - .await - .unwrap(); - - let mut alpha = store.list_entity_ids("alpha").await.unwrap(); - alpha.sort(); - assert_eq!( - alpha, - vec![ - ("Order".to_string(), "ord-1".to_string()), - ("Task".to_string(), "task-1".to_string()), - ] - ); - - let beta = store.list_entity_ids("beta").await.unwrap(); - assert_eq!(beta, vec![("Order".to_string(), "ord-9".to_string())]); - } - - #[tokio::test] - async fn read_events_from_sequence() { - let store = SimEventStore::no_faults(42); - let pid = "default:Order:ord-5"; - - store - .append(pid, 0, &[test_envelope(0, "A"), test_envelope(0, "B")]) - .await - .unwrap(); - store - .append(pid, 2, &[test_envelope(0, "C")]) - .await - .unwrap(); - - // Read from sequence 1 — should skip event at seq 1 - let events = store.read_events(pid, 1).await.unwrap(); - assert_eq!(events.len(), 2); - assert_eq!(events[0].sequence_nr, 2); - assert_eq!(events[1].sequence_nr, 3); - } - - #[tokio::test] - async fn deterministic_across_seeds() { - // Same seed → same behavior (with no faults, behavior is trivially the same) - for seed in [42, 123, 999] { - let store = SimEventStore::no_faults(seed); - let pid = "default:Order:det-1"; - - let seq = store - .append(pid, 0, &[test_envelope(0, "Created")]) - .await - .unwrap(); - assert_eq!(seq, 1); - - let events = store.read_events(pid, 0).await.unwrap(); - assert_eq!(events.len(), 1); - } - } - - #[tokio::test] - async fn fault_injection_produces_errors() { - let faults = SimFaultConfig { - write_failure_prob: 1.0, // always fail - concurrency_violation_prob: 0.0, - read_truncation_prob: 0.0, - snapshot_failure_prob: 0.0, - }; - let store = SimEventStore::new(42, faults); - let pid = "default:Order:fault-1"; - - let err = store.append(pid, 0, &[test_envelope(0, "Created")]).await; - assert!(err.is_err()); - } -} +mod tests; diff --git a/crates/temper-store-sim/src/tests.rs b/crates/temper-store-sim/src/tests.rs new file mode 100644 index 00000000..9e076ff0 --- /dev/null +++ b/crates/temper-store-sim/src/tests.rs @@ -0,0 +1,294 @@ +use super::*; +use temper_runtime::persistence::EventMetadata; + +fn test_envelope(seq: u64, event_type: &str) -> PersistenceEnvelope { + PersistenceEnvelope { + sequence_nr: seq, + event_type: event_type.to_string(), + payload: serde_json::json!({"test": true}), + metadata: EventMetadata { + event_id: uuid::Uuid::nil(), + causation_id: uuid::Uuid::nil(), + correlation_id: uuid::Uuid::nil(), + timestamp: chrono::DateTime::UNIX_EPOCH, + actor_id: "test".to_string(), + }, + } +} + +#[tokio::test] +async fn append_and_read_roundtrip() { + let store = SimEventStore::no_faults(42); + let pid = "default:Order:ord-1"; + + let new_seq = store + .append(pid, 0, &[test_envelope(0, "Created")]) + .await + .unwrap(); + assert_eq!(new_seq, 1); + + let events = store.read_events(pid, 0).await.unwrap(); + assert_eq!(events.len(), 1); + assert_eq!(events[0].sequence_nr, 1); + assert_eq!(events[0].event_type, "Created"); +} + +#[tokio::test] +async fn append_multiple_events() { + let store = SimEventStore::no_faults(42); + let pid = "default:Order:ord-2"; + + let seq = store + .append( + pid, + 0, + &[test_envelope(0, "Created"), test_envelope(0, "Submitted")], + ) + .await + .unwrap(); + assert_eq!(seq, 2); + + let events = store.read_events(pid, 0).await.unwrap(); + assert_eq!(events.len(), 2); + assert_eq!(events[0].sequence_nr, 1); + assert_eq!(events[1].sequence_nr, 2); +} + +#[tokio::test] +async fn append_batch_commits_multiple_journals_atomically() { + let store = SimEventStore::no_faults(42); + let appends = vec![ + PersistenceAppend { + persistence_id: "default:Order:ord-a".to_string(), + expected_sequence: 0, + events: vec![test_envelope(0, "Created")], + }, + PersistenceAppend { + persistence_id: "default:Order:ord-b".to_string(), + expected_sequence: 0, + events: vec![test_envelope(0, "Created"), test_envelope(0, "Submitted")], + }, + ]; + + let results = store.append_batch(&appends).await.unwrap(); + + assert_eq!( + results, + vec![ + PersistenceAppendResult { + persistence_id: "default:Order:ord-a".to_string(), + sequence_nr: 1, + }, + PersistenceAppendResult { + persistence_id: "default:Order:ord-b".to_string(), + sequence_nr: 2, + }, + ] + ); + assert_eq!(store.dump_journal("default:Order:ord-a").len(), 1); + assert_eq!(store.dump_journal("default:Order:ord-b").len(), 2); +} + +#[tokio::test] +async fn append_batch_conflict_leaves_all_journals_untouched() { + let store = SimEventStore::no_faults(42); + store + .append( + "default:Order:ord-existing", + 0, + &[test_envelope(0, "Created")], + ) + .await + .unwrap(); + + let err = store + .append_batch(&[ + PersistenceAppend { + persistence_id: "default:Order:ord-new".to_string(), + expected_sequence: 0, + events: vec![test_envelope(0, "Created")], + }, + PersistenceAppend { + persistence_id: "default:Order:ord-existing".to_string(), + expected_sequence: 0, + events: vec![test_envelope(0, "Submitted")], + }, + ]) + .await + .expect_err("second journal conflict should abort entire batch"); + + assert!( + matches!(err, PersistenceError::ConcurrencyViolation { .. }), + "unexpected error: {err}" + ); + assert!( + store.dump_journal("default:Order:ord-new").is_empty(), + "first append must not be persisted when a later stream conflicts" + ); + assert_eq!( + store.dump_journal("default:Order:ord-existing").len(), + 1, + "conflicting stream must keep its original journal only" + ); +} + +#[tokio::test] +async fn concurrency_violation_on_wrong_sequence() { + let store = SimEventStore::no_faults(42); + let pid = "default:Order:ord-3"; + + store + .append(pid, 0, &[test_envelope(0, "Created")]) + .await + .unwrap(); + + let err = store + .append(pid, 0, &[test_envelope(0, "Duplicate")]) + .await + .unwrap_err(); + + assert!(matches!( + err, + PersistenceError::ConcurrencyViolation { + expected: 0, + actual: 1 + } + )); +} + +#[tokio::test] +async fn snapshot_save_and_load() { + let store = SimEventStore::no_faults(42); + let pid = "default:Order:ord-4"; + + store.save_snapshot(pid, 5, b"state-data").await.unwrap(); + + let snap = store.load_snapshot(pid).await.unwrap(); + assert_eq!(snap, Some((5, b"state-data".to_vec()))); +} + +#[tokio::test] +async fn snapshot_save_records_history_and_rotates_segments() { + let store = SimEventStore::no_faults(42); + let pid = "default:Order:segmented"; + + store + .append( + pid, + 0, + &[test_envelope(0, "Created"), test_envelope(0, "Updated")], + ) + .await + .unwrap(); + store.save_snapshot(pid, 2, b"snapshot-2").await.unwrap(); + store + .append(pid, 2, &[test_envelope(0, "AfterSnapshot")]) + .await + .unwrap(); + + assert_eq!(store.snapshot_history_len(pid), 1); + let segments = store.dump_segments(pid); + assert_eq!(segments.len(), 2); + assert_eq!(segments[0].segment_index, 0); + assert_eq!(segments[0].snapshot_sequence, Some(2)); + assert!(segments[0].sealed); + assert_eq!(segments[1].segment_index, 1); + assert_eq!(segments[1].start_sequence_nr, 3); + assert_eq!(segments[1].end_sequence_nr, Some(3)); + assert!(!segments[1].sealed); +} + +#[tokio::test] +async fn load_snapshot_returns_none_when_empty() { + let store = SimEventStore::no_faults(42); + let snap = store + .load_snapshot("default:Order:nonexistent") + .await + .unwrap(); + assert_eq!(snap, None); +} + +#[tokio::test] +async fn list_entity_ids_filters_by_tenant() { + let store = SimEventStore::no_faults(42); + + store + .append("alpha:Order:ord-1", 0, &[test_envelope(0, "Created")]) + .await + .unwrap(); + store + .append("alpha:Task:task-1", 0, &[test_envelope(0, "Created")]) + .await + .unwrap(); + store + .append("beta:Order:ord-9", 0, &[test_envelope(0, "Created")]) + .await + .unwrap(); + + let mut alpha = store.list_entity_ids("alpha").await.unwrap(); + alpha.sort(); + assert_eq!( + alpha, + vec![ + ("Order".to_string(), "ord-1".to_string()), + ("Task".to_string(), "task-1".to_string()), + ] + ); + + let beta = store.list_entity_ids("beta").await.unwrap(); + assert_eq!(beta, vec![("Order".to_string(), "ord-9".to_string())]); +} + +#[tokio::test] +async fn read_events_from_sequence() { + let store = SimEventStore::no_faults(42); + let pid = "default:Order:ord-5"; + + store + .append(pid, 0, &[test_envelope(0, "A"), test_envelope(0, "B")]) + .await + .unwrap(); + store + .append(pid, 2, &[test_envelope(0, "C")]) + .await + .unwrap(); + + // Read from sequence 1 — should skip event at seq 1 + let events = store.read_events(pid, 1).await.unwrap(); + assert_eq!(events.len(), 2); + assert_eq!(events[0].sequence_nr, 2); + assert_eq!(events[1].sequence_nr, 3); +} + +#[tokio::test] +async fn deterministic_across_seeds() { + // Same seed → same behavior (with no faults, behavior is trivially the same) + for seed in [42, 123, 999] { + let store = SimEventStore::no_faults(seed); + let pid = "default:Order:det-1"; + + let seq = store + .append(pid, 0, &[test_envelope(0, "Created")]) + .await + .unwrap(); + assert_eq!(seq, 1); + + let events = store.read_events(pid, 0).await.unwrap(); + assert_eq!(events.len(), 1); + } +} + +#[tokio::test] +async fn fault_injection_produces_errors() { + let faults = SimFaultConfig { + write_failure_prob: 1.0, // always fail + concurrency_violation_prob: 0.0, + read_truncation_prob: 0.0, + snapshot_failure_prob: 0.0, + }; + let store = SimEventStore::new(42, faults); + let pid = "default:Order:fault-1"; + + let err = store.append(pid, 0, &[test_envelope(0, "Created")]).await; + assert!(err.is_err()); +} diff --git a/crates/temper-store-turso/src/schema.rs b/crates/temper-store-turso/src/schema.rs index 2b817741..b0228a37 100644 --- a/crates/temper-store-turso/src/schema.rs +++ b/crates/temper-store-turso/src/schema.rs @@ -10,6 +10,7 @@ pub use query_plane::{ CREATE_ENTITY_CATALOG_STATUS_INDEX, CREATE_ENTITY_CATALOG_TABLE, CREATE_ENTITY_CATALOG_TYPE_INDEX, CREATE_ENTITY_FIELD_INDEX_LOOKUP, CREATE_ENTITY_FIELD_INDEX_STATUS, CREATE_ENTITY_FIELD_INDEX_TABLE, + CREATE_ENTITY_KEY_INDEX_ENTITY, CREATE_ENTITY_KEY_INDEX_TABLE, }; pub const CREATE_EVENTS_TABLE: &str = "\ diff --git a/crates/temper-store-turso/src/schema/query_plane.rs b/crates/temper-store-turso/src/schema/query_plane.rs index 39294e7b..4ba9c590 100644 --- a/crates/temper-store-turso/src/schema/query_plane.rs +++ b/crates/temper-store-turso/src/schema/query_plane.rs @@ -49,3 +49,28 @@ CREATE INDEX IF NOT EXISTS idx_efi_lookup pub const CREATE_ENTITY_FIELD_INDEX_STATUS: &str = "\ CREATE INDEX IF NOT EXISTS idx_efi_status ON entity_field_index(tenant, entity_type, status);"; + +/// ADR-0153: declared composite-key index — the negative-existence access path. +/// +/// One row per (declared key, entity), co-committed with the journal append (unlike +/// the eventually-consistent `entity_field_index`). A keyed read is a single +/// O(log n) probe: hit -> entity_id, miss -> authoritatively absent — so the read +/// plane no longer scans a whole entity type to prove absence (the 413, ARN-68). +/// The PRIMARY KEY enforces declared-key uniqueness (reject-and-surface on conflict); +/// `key_hash` is a canonical, type-tagged hash of the declared key's values. +pub const CREATE_ENTITY_KEY_INDEX_TABLE: &str = "\ +CREATE TABLE IF NOT EXISTS entity_key_index ( + tenant TEXT NOT NULL, + entity_type TEXT NOT NULL, + key_name TEXT NOT NULL, + key_hash TEXT NOT NULL, + entity_id TEXT NOT NULL, + sequence_nr INTEGER NOT NULL DEFAULT 0, + PRIMARY KEY (tenant, entity_type, key_name, key_hash) +);"; + +/// Reverse lookup: all key rows for an entity, so the write path can upsert/delete +/// an entity's declared-key rows when it changes or is removed. +pub const CREATE_ENTITY_KEY_INDEX_ENTITY: &str = "\ +CREATE INDEX IF NOT EXISTS idx_eki_entity + ON entity_key_index(tenant, entity_type, entity_id);"; diff --git a/crates/temper-store-turso/src/store/event_store.rs b/crates/temper-store-turso/src/store/event_store.rs index 74bbd6ac..9b7bbad9 100644 --- a/crates/temper-store-turso/src/store/event_store.rs +++ b/crates/temper-store-turso/src/store/event_store.rs @@ -110,6 +110,32 @@ impl EventStore for TursoEventStore { Err(last_err.expect("retry loop captured at least one error")) } + async fn lookup_by_key( + &self, + tenant: &str, + entity_type: &str, + key_name: &str, + key_hash: &str, + ) -> Result, PersistenceError> { + // ADR-0153: a single keyed probe of entity_key_index — present/absent in + // O(log n), no candidate scan (the negative-existence access path). Bounded + // regardless of how many entities the tenant/type holds, so it cannot trip + // the scan budget that produces the 413. + let conn = self.configured_connection().await?; + let mut rows = conn + .query( + "SELECT entity_id FROM entity_key_index \ + WHERE tenant = ?1 AND entity_type = ?2 AND key_name = ?3 AND key_hash = ?4", + params![tenant, entity_type, key_name, key_hash], + ) + .await + .map_err(storage_error)?; + match rows.next().await.map_err(storage_error)? { + Some(row) => Ok(Some(row.get::(0).map_err(storage_error)?)), + None => Ok(None), + } + } + #[instrument(skip_all, fields(otel.name = "turso.append_batch"))] async fn append_batch( &self, diff --git a/crates/temper-store-turso/src/store/mod.rs b/crates/temper-store-turso/src/store/mod.rs index 06d124ba..89eb8e81 100644 --- a/crates/temper-store-turso/src/store/mod.rs +++ b/crates/temper-store-turso/src/store/mod.rs @@ -382,6 +382,15 @@ impl TursoEventStore { .await .map_err(storage_error)?; + // Entity key index (ADR-0153) — declared composite-key -> entity_id, the + // negative-existence access path co-committed with the journal append. + conn.execute(schema::CREATE_ENTITY_KEY_INDEX_TABLE, ()) + .await + .map_err(storage_error)?; + conn.execute(schema::CREATE_ENTITY_KEY_INDEX_ENTITY, ()) + .await + .map_err(storage_error)?; + Ok(()) } diff --git a/docs/adrs/0153-declared-composite-key-index.md b/docs/adrs/0153-declared-composite-key-index.md new file mode 100644 index 00000000..93968899 --- /dev/null +++ b/docs/adrs/0153-declared-composite-key-index.md @@ -0,0 +1,75 @@ +# ADR-0153: Declared Composite-Key Index (A Negative-Existence Access Path) + +- Status: Proposed +- Date: 2026-06-23 +- Deciders: Temper core maintainers +- Related: + - ADR-0091: Query projection diff index upserts + - ADR-0142: Dispatch acknowledges after projection + - ADR-0148: Bound derived writes off the dispatch hot path + - ADR-0134: Query plane read contract + - ARN-68 (the 413 / QueryTooLarge issue), ARN-89 (read-after-write reconcile), ARN-102 (3-year runtime vision) + - `crates/temper-server/src/odata/query_plane_read/{types.rs,mod.rs}` + - `crates/temper-server/src/odata/filter_sql.rs` + - `crates/temper-store-turso/src/store/field_index.rs` + - `crates/temper-server/src/state/query_projection_queue.rs` + - `crates/temper-store-postgres/src/schema.rs`, `crates/temper-server/src/state/entity_ops.rs` + +## Context + +### The bug +Point reads return **413 QueryTooLarge** at tenant scale (`Files`, `SessionEntries`, `Directories`). The read plane can prove a key **present** cheaply — an equality probe against the EAV field index — but it cannot prove a key **absent** without scanning the whole entity type. Three facts combine: + +1. **`entity_id` is a surrogate** (a `sim_uuid`). The business keys the platform actually resolves by — `WorkspaceId+Path` (Files), `SessionId+EntryId` (SessionEntries), `Name+WorkspaceId+ParentId` (Directories) — live **inside event payloads**, not in any keyed structure. +2. **The query projection is eventually consistent.** The broad EAV field index is written by the async coalescing queue (ADR-0148), so an empty equality page is **ambiguous**: the key is absent, *or* it is present but the projection lags. +3. **To stay read-your-writes correct on that ambiguity**, `should_reconcile_empty_exact_match_against_authoritative` (ARN-89, commit `40b4f22a`) falls back to scanning the workspace's authoritative state on an empty page. At scale that scan exceeds `scan_candidate_budget` (`odata_max_entities × 10`) → **413**. + +Read-after-write correctness for a *present* key was bought by making *absence* cost `O(workspace)`. Every prior fix — raising the budget, pushing down lossless conjuncts (`bdd15d42`), caller-side query rewrites — only **moves the cliff**; the trip is a function of tenant data volume. (Proven in production: a caller-side fix to the `Directories` root lookup shifted the 413 from the root to the *subdirectory* lookup — same class, one level down.) + +### How we got here +ADR-0091 (diff-based field-index upserts) → ADR-0142 (inline projection on dispatch, to fix a real read-your-writes bug) → ADR-0148 (move the broad field index to the async coalescing queue, because under Foresight's dispatch fan-out the inline projection dominated DB latency). Moving the broad index async is **exactly** what created the absent-vs-lagging ambiguity behind the 413. This ADR does not revert that — it adds a second, narrow index on a different axis. + +### Measurement (the gate, real data) +Measured on the real Foresight Postgres (`service:foresight`, tenant `deep-sci-fi`): broad EAV index rows per entity **S = 7–46** (File 15, World 13, EventNode 10, SessionEntry 13, Session 46). Path A's declared keys **K = 1–3** per entity. **K ≪ S (≈ 1/10).** See `aya/brain/temper-read-write-architecture/foresight-measurement-20260623.md`. + +## Decision + +Add a dedicated **declared composite-key index** and **decouple consistency by query class**. + +1. **`entity_key_index(tenant, entity_type, key_name, key_hash, entity_id, sequence_nr, …)`** — one primary-key row per `(declared key, entity)`, holding the business-key → `entity_id` mapping the read plane lacks today. +2. **Co-commit the key row in the same store transaction as the journal append.** The keyed row is **synchronous and strongly consistent** — unlike the broad EAV index. +3. **The broad EAV field index stays async** (ADR-0148 unchanged). Path A is **additive on a different axis, not a revert.** +4. **A declared-key read becomes a single `O(log n)` probe:** hit → `entity_id`; miss → **authoritatively absent**. **Delete the ARN-89 / `#324` reconcile scan.** +5. **Plan-time query taxonomy** — `PointRead` / `RangeScan` / `Unbounded`. Unbounded shapes are rejected **at plan time** with a paging contract, never as a mid-scan budget trip. + +### Key declaration — Temper-native (resolved) +The 413 entities are all CSDL `Key = ["Id"]` (verified: File, SessionEntry, Directory, Workspace). The business keys that 413 (`WorkspaceId+Path`, `SessionId+EntryId`, `Name+WorkspaceId+ParentId`) are **not keys today** — so Path A must *declare* them. They are declared **Temper-native in the IOA spec**, via a new `[[key]] name="..." properties=[...]` block (a unique / alternate key), **not** in the OData CSDL. Rationale: the IOA spec is the source of truth and the CSDL is a *derived* projection; a uniqueness guarantee is a domain invariant the verification cascade can check; and the declaration stays portable across spec languages — under a future move to P it becomes a spec monitor, unchanged in intent. The kernel indexes the CSDL `Key` **plus** each declared `[[key]]`; the **OData alternate-key annotation is derived** from the `[[key]]` declaration (so `Files(WorkspaceId=…,Path=…)` addressing comes for free). Apps add one `[[key]]` block per business key; the kernel does the rest. + +### Resolved decisions +- **Uniqueness → reject + surface.** A declared composite key is a `UNIQUE` constraint; a duplicate (two entities at the same key tuple) **rejects the write** with a typed error naming the key. A silent duplicate is a latent data bug, not a thing to last-writer-wins. +- **One transaction.** The journal append and the key-index upsert commit in a **single store transaction**; if the key write fails, the write fails (no ack-with-log). Read-after-write correctness is the entire point of the index. +- **Measured on Postgres — done.** K ≪ S confirmed on the real Foresight Postgres (above); the "not a write-amplification revert" claim is measured, not assumed. Foresight is also the worst case (its dispatch fan-out is what drove ADR-0148). +- **Co-location invariant.** Turso **XOR** Postgres, never split — confirmed. This is what makes co-committing the key row with the journal append (and deleting the `#324` scan) safe. + +## Consequences +- **Reads** of declared keys are `O(log n)` for present **and** absent. The 413 class is removed for point reads. +- **Writes** gain `K` (1–3) synchronous single-row keyed upserts per write, co-committed with the journal. `K ≪ S` (measured); the expensive `S`-wide projection stays async. **No write-amplification regression.** +- **DST** must follow. The sim store implements `EventStore` but not the query plane, so it gains the key map + a store-agnostic canonical `key_hash`, and deterministic simulation must prove present/absent the same way prod does. **Hard prerequisite, not a tradeoff.** +- **Backfill.** Pre-existing entities have no key row. Keep `#324` as a transitional fallback behind a per-tenant **backfill watermark**; only after backfill passes does a keyed miss authoritatively prove ABSENT (otherwise we re-create ARN-89 for old data). +- **`key_hash` is type-tagged** and canonical — this removes the current EAV limits (string-only ≤ 2000B; no Int/Guid/null keys) for the declared-key path. + +## Implementation phases +1. This ADR. +2. **Storage-boundary co-commit.** Thread the key row into `append_batch`'s transaction in **both** stores. Today the journal append and `upsert_projection` open separate transactions — making them one is the real work, not a feature flag. +3. **`entity_key_index` table** (turso + postgres + sim store) + the type-tagged canonical `key_hash`; write/delete in the inline transaction carrying `sequence_nr`. +4. **Backfill-before-trust gate** (per-tenant watermark; `#324` stays as transitional fallback until it passes). +5. **Planner rewrite + retire `#324`** in `query_plane_read/mod.rs`: three-class plan; plan-time rejection instead of a mid-scan 413. +6. **DST.** Key map + canonical `key_hash` in the sim store; property tests proving present/absent under simulation, plus a fault-injection test for a lagging broad index (the original ambiguity). + +## Gate — no Foresight regression +The **structural** gate is passed (K ≪ S, measured on real data). The **final** gate before shipping is a load test after phase 5 confirming (a) the synchronous key write adds negligible append latency under corridor fan-out, and (b) broad-index throughput is unchanged. Path A does not ship until that load test is green. (Note: the live Foresight deployment currently stalls at the *seed* stage on provider auth, before fan-out — so the load baseline needs that cleared first, or a synthetic fan-out harness.) + +## Alternatives considered +- **Raise the read budget.** Rejected — moves the cliff; does not add the missing access path. +- **Keep caller-side query rewrites** (the deployed ARN-68 mitigations). Rejected as *the* fix — they only delay the trip (proven by the Directories root→subdirectory shift). They remain a valid stopgap until this lands. +- **Postgres actor runtime (ARN-26).** Gets keyed present/absent for free (the business key becomes the primary key), but it is outside DST verification, partial, and multi-node-scoped. That is the mid-horizon multi-node bet (ARN-26 / ARN-27 / ARN-102), not this single-node, verified fix. Path A proves the same "key-as-a-real-key" property inside the verified kernel first. diff --git a/test-fixtures/specs/keyed_doc.ioa.toml b/test-fixtures/specs/keyed_doc.ioa.toml new file mode 100644 index 00000000..ef842023 --- /dev/null +++ b/test-fixtures/specs/keyed_doc.ioa.toml @@ -0,0 +1,32 @@ +# Keyed Doc — minimal fixture for the ADR-0153 entity_key_index DST. +# +# A Doc has the surrogate `Id` and a declared alternate key `(WorkspaceId, Path)`. +# The DST asserts the negative-existence invariant: a keyed read returns the Doc +# iff it exists, under every seed. + +[automaton] +name = "Doc" +states = ["New", "Ready"] +initial = "New" + +[[state]] +name = "WorkspaceId" +type = "string" +initial = "" + +[[state]] +name = "Path" +type = "string" +initial = "" + +[[key]] +name = "path" +properties = ["WorkspaceId", "Path"] + +[[action]] +name = "Create" +kind = "input" +from = ["New"] +to = "Ready" +params = ["WorkspaceId", "Path"] +hint = "Create the doc with its workspace + path (its declared key)." diff --git a/test-fixtures/specs/order.ioa.toml b/test-fixtures/specs/order.ioa.toml index 1b84aa1d..7ee0f5d1 100644 --- a/test-fixtures/specs/order.ioa.toml +++ b/test-fixtures/specs/order.ioa.toml @@ -9,6 +9,15 @@ name = "Order" states = ["Draft", "Submitted", "Confirmed", "Processing", "Shipped", "Delivered", "Cancelled", "ReturnRequested", "Returned", "Refunded"] initial = "Draft" +# --- Declared keys (ADR-0153) --- +# A declared alternate key the kernel indexes for O(log n) present/absent reads. +# Used by the query-plane keyed fast-path proof; harmless to entities that do not +# populate these fields (a partial key simply yields no index row). + +[[key]] +name = "ws_path" +properties = ["WorkspaceId", "Path"] + # --- State Variables --- [[state]]