From f66e04565992d4fca23d09c7331c4200ad75b7ab Mon Sep 17 00:00:00 2001 From: OceanLi Date: Thu, 10 Sep 2026 19:42:34 -0400 Subject: [PATCH] fix(runtime): cap ordered fences before writer admission --- crates/khive-pack-kg/src/handler_defs.rs | 8 +- .../src/handlers/stream_fence_batch_tests.rs | 115 ++++++++++++++++++ crates/khive-runtime/src/note_write.rs | 19 ++- crates/khive-runtime/src/note_write_tests.rs | 47 +++++++ ...ADR-172-versioned-notes-compare-and-set.md | 14 +++ 5 files changed, 198 insertions(+), 5 deletions(-) diff --git a/crates/khive-pack-kg/src/handler_defs.rs b/crates/khive-pack-kg/src/handler_defs.rs index 566ee338d..1c1219f53 100644 --- a/crates/khive-pack-kg/src/handler_defs.rs +++ b/crates/khive-pack-kg/src/handler_defs.rs @@ -24,7 +24,7 @@ pub(crate) static KG_HANDLERS: [HandlerDef; 24] = [ ParamDef { name: "stream", param_type: "string", required: true, description: "Namespace-scoped stream name, at most 512 UTF-8 bytes and no U+0000.", resolution_mode: IdResolutionMode::NotApplicable }, ParamDef { name: "record", param_type: "JSON value", required: true, description: "Required JSON value, including scalar or null; stored as note content.", resolution_mode: IdResolutionMode::NotApplicable }, ParamDef { name: "expected_seq", param_type: "integer", required: false, description: "Append only if this entry would receive this sequence; conflict details include next_seq.", resolution_mode: IdResolutionMode::NotApplicable }, - ParamDef { name: "fence", param_type: "object or array of object", required: false, description: "A fence object {key, kind, expected_version} or a non-empty list of distinct (kind, key) objects, checked in order in the same writer transaction. Missing or stale returns fence_conflict; list refusals also carry string index (zero-based). Explicit null is invalid.", resolution_mode: IdResolutionMode::NotApplicable }, + ParamDef { name: "fence", param_type: "object or array of object", required: false, description: "A fence object {key, kind, expected_version} or a non-empty list of at most 100 distinct (kind, key) objects, checked in order in the same writer transaction. Oversized lists return invalid_input naming the cap and count sent before entry interpretation or writer admission. Missing or stale returns fence_conflict; list refusals also carry string index (zero-based). Explicit null is invalid.", resolution_mode: IdResolutionMode::NotApplicable }, ParamDef { name: "note_kind", param_type: "string", required: false, description: "Registered note kind; defaults to observation.", resolution_mode: IdResolutionMode::NotApplicable }, ParamDef { name: "embed", param_type: "boolean", required: false, description: "Defaults false: no embedding rows or vector-index work, while lexical indexing and stream reads remain available. True embeds under every registered model unless embedding_model selects one.", resolution_mode: IdResolutionMode::NotApplicable }, ParamDef { name: "embedding_model", param_type: "string", required: false, description: "Select one registered embedding model. Requires embed=true; otherwise invalid_input before any write.", resolution_mode: IdResolutionMode::NotApplicable }, @@ -60,7 +60,7 @@ pub(crate) static KG_HANDLERS: [HandlerDef; 24] = [ visibility: Visibility::Verb, category: VerbCategory::Commissive, params: &[ - ParamDef { name: "ops", param_type: "array of object", required: true, description: "Members in order. {op: \"append\", stream, record, expected_seq?, fence?, embed?, embedding_model?}: record is any JSON value, including null. Append embed defaults false; true embeds under all registered models or the one embedding_model selects. embedding_model requires embed=true or the entire list is invalid_input before any write; atomic errors name details.member. Each append fence is an object or non-empty ordered list of {key, kind, expected_version}; all append-member fences are checked in order before any member writes in their transaction. A stale or missing list entry reports a string index; atomic refusal also names the string member position. Null, empty lists and duplicate (kind,key) entries are invalid before any member writes. {op: \"write\", key, kind, doc, tags?, embed?, expected_version?}: doc is any JSON value; omitted/null expected_version creates only if absent, while a positive version updates only an existing key at that version. Duplicate write (kind, key) pairs refuse the whole batch. An unknown op is that member's unknown_op refusal, placed by the mode.", resolution_mode: IdResolutionMode::NotApplicable }, + ParamDef { name: "ops", param_type: "array of object", required: true, description: "Members in order. {op: \"append\", stream, record, expected_seq?, fence?, embed?, embedding_model?}: record is any JSON value, including null. Append embed defaults false; true embeds under all registered models or the one embedding_model selects. embedding_model requires embed=true or the entire list is invalid_input before any write; atomic errors name details.member. Each append fence is an object or non-empty ordered list of at most 100 {key, kind, expected_version} entries; oversized lists return invalid_input naming the cap and count sent before fence entry interpretation or any writer admission, in both modes. All append-member fences are checked in order before any member writes in their transaction. A stale or missing list entry reports a string index; atomic refusal also names the string member position. Null, empty lists and duplicate (kind,key) entries are invalid before any member writes. {op: \"write\", key, kind, doc, tags?, embed?, expected_version?}: doc is any JSON value; omitted/null expected_version creates only if absent, while a positive version updates only an existing key at that version. Duplicate write (kind, key) pairs refuse the whole batch. An unknown op is that member's unknown_op refusal, placed by the mode.", resolution_mode: IdResolutionMode::NotApplicable }, ParamDef { name: "fence", param_type: "object", required: false, description: "{key, kind, expected_version}: a keyed note whose positive version must match, checked once inside the atomic transaction before the first write. A missing or stale note refuses the whole batch with fence_conflict. Selects atomic mode; atomic=false beside a fence is refused. List-valued fences are not supported.", resolution_mode: IdResolutionMode::NotApplicable }, ParamDef { name: "observed", param_type: "array of object", required: false, description: "Requires atomic mode; supplying observed alone does not select it. Set atomic=true or supply fence. [{key, kind, version, id?, live_until?}], checked inside the transaction before the first write. A positive version asserts the live key's exact version; explicit null asserts no live holder. A mismatch refuses the batch with version_conflict naming the key and the entry index. The version field is required. Optional live_until is a dotted document path (for example lease.expires_at), requires a positive version, and must hold an RFC 3339 timestamp strictly later than one writer-transaction clock reading. Expired or unreadable fields refuse with expired or live_until_unreadable; an expired refusal names the deadline it read and the clock it compared, an unreadable one names value_type (absent, string, number, boolean, array, object, null) and never the value itself. Optional id pins the note UUID and requires a positive version; a replacement refuses with identity_conflict before the version check. The refusal includes current_id only when the caller may learn the holder, under the same disclosure rule as key_conflict. Without id, the entry asserts version equality on the note holding the key at commit time. Refused with invalid_input in per-member mode.", resolution_mode: IdResolutionMode::NotApplicable }, ParamDef { name: "atomic", param_type: "boolean", required: false, description: "true: one transaction, all or nothing. false: one transaction per member, refusals as member values. Defaults to whether fence is present. observed alone does not select atomic mode; observed without fence requires atomic=true, otherwise invalid_input.", resolution_mode: IdResolutionMode::NotApplicable }, @@ -76,7 +76,7 @@ pub(crate) static KG_HANDLERS: [HandlerDef; 24] = [ params: &[ ParamDef { name: "key", param_type: "string", required: false, description: "Singleton notes only: immutable live namespace/kind identity, at most 512 UTF-8 bytes without U+0000. An occupied key fails with key_conflict; existing_id is disclosed only when list is allowed.", resolution_mode: IdResolutionMode::NotApplicable }, ParamDef { name: "embed", param_type: "bool", required: false, description: "Singleton notes only: defaults false for head and true otherwise. False skips inference and vector insertion while retaining lexical indexing.", resolution_mode: IdResolutionMode::NotApplicable }, - ParamDef { name: "fence", param_type: "object or array of object", required: false, description: "Singleton notes only. A fence object {key, kind, expected_version} or a non-empty list of distinct (kind, key) objects, checked in order in the same writer transaction. Missing or stale returns fence_conflict; list refusals also carry string index (zero-based). Explicit null is invalid.", resolution_mode: IdResolutionMode::NotApplicable }, + ParamDef { name: "fence", param_type: "object or array of object", required: false, description: "Singleton notes only. A fence object {key, kind, expected_version} or a non-empty list of at most 100 distinct (kind, key) objects, checked in order in the same writer transaction. Oversized lists return invalid_input naming the cap and count sent before entry interpretation or writer admission. Missing or stale returns fence_conflict; list refusals also carry string index (zero-based). Explicit null is invalid.", resolution_mode: IdResolutionMode::NotApplicable }, ParamDef { name: "kind", param_type: "string", @@ -462,7 +462,7 @@ pub(crate) static KG_HANDLERS: [HandlerDef; 24] = [ category: VerbCategory::Declaration, params: &[ ParamDef { name: "expected_version", param_type: "integer", required: false, description: "Notes only: positive persisted version required inside the writer transaction. A stale version fails without mutation, with reason=version_conflict and expected_version/current_version details. Omission preserves unconditional caller semantics.", resolution_mode: IdResolutionMode::NotApplicable }, - ParamDef { name: "fence", param_type: "object or array of object", required: false, description: "Singleton notes only. A fence object {key, kind, expected_version} or a non-empty list of distinct (kind, key) objects, checked in order in the same writer transaction. Missing or stale returns fence_conflict; list refusals also carry string index (zero-based). Explicit null is invalid.", resolution_mode: IdResolutionMode::NotApplicable }, + ParamDef { name: "fence", param_type: "object or array of object", required: false, description: "Singleton notes only. A fence object {key, kind, expected_version} or a non-empty list of at most 100 distinct (kind, key) objects, checked in order in the same writer transaction. Oversized lists return invalid_input naming the cap and count sent before entry interpretation or writer admission. Missing or stale returns fence_conflict; list refusals also carry string index (zero-based). Explicit null is invalid.", resolution_mode: IdResolutionMode::NotApplicable }, ParamDef { name: "embed", param_type: "bool", required: false, description: "Notes only: omission retains embedding state. True enables reindexing; false performs no inference, removes existing vector rows transactionally and retains lexical indexing. Delayed reindex work cannot restore a stale revision.", resolution_mode: IdResolutionMode::NotApplicable }, ParamDef { name: "id", diff --git a/crates/khive-pack-kg/src/handlers/stream_fence_batch_tests.rs b/crates/khive-pack-kg/src/handlers/stream_fence_batch_tests.rs index e02b8e973..1211c8db2 100644 --- a/crates/khive-pack-kg/src/handlers/stream_fence_batch_tests.rs +++ b/crates/khive-pack-kg/src/handlers/stream_fence_batch_tests.rs @@ -193,3 +193,118 @@ async fn ordered_fences_batch_validates_all_members_before_writing() { } } } + +#[tokio::test] +async fn ordered_fences_cap_accepts_exactly_100_on_every_write_surface() { + let (_, registry) = batch_surface(); + let mut fences = Vec::new(); + for index in 0..100 { + let key = format!("fence-cap/{index}"); + lease(®istry, &key).await; + fences.push(fence(&key, 1)); + } + let target = registry + .dispatch( + "create", + json!({"kind":"head","content":"{}","fence":fences}), + ) + .await + .unwrap(); + registry + .dispatch( + "update", + json!({"id":target["id"],"content":"{\"updated\":true}","fence":fences}), + ) + .await + .unwrap(); + registry + .dispatch( + "stream.append", + json!({"stream":"fence-cap","record":1,"fence":fences}), + ) + .await + .unwrap(); + for atomic in [true, false] { + let result = registry + .dispatch( + "stream.batch", + json!({"atomic":atomic,"ops":[ + {"op":"append","stream":"fence-cap","record":2,"fence":fences} + ]}), + ) + .await + .unwrap(); + assert_eq!(result["committed"], true); + } + assert_eq!(heads(®istry, &["fence-cap"]).await, vec![3]); + let target = registry + .dispatch("get", json!({"id":target["id"]})) + .await + .unwrap(); + assert_eq!(target["version"], 2); +} + +#[tokio::test] +async fn ordered_fences_cap_refuses_before_entry_interpretation_and_writer_admission() { + let (rt, registry) = batch_surface(); + let target = lease(®istry, "target").await; + let distinct: Vec = (0..101) + .map(|index| fence(&format!("fence-cap/{index}"), 1)) + .collect(); + for fences in [ + json!(distinct), + json!(vec![Value::Null; 101]), + json!(vec![fence("duplicate", 0); 101]), + ] { + let before = population(&rt).await; + let before_writers = rt.backend().pool().writer_acquisition_snapshot(); + for (verb, args) in [ + ( + "create", + json!({"kind":"head","content":"{}","fence":fences}), + ), + ( + "update", + json!({"id":target["id"],"content":"{}","fence":fences}), + ), + ( + "stream.append", + json!({"stream":"fence-cap","record":1,"fence":fences}), + ), + ( + "stream.batch", + json!({"atomic":true,"ops":[ + {"op":"append","stream":"fence-cap","record":1}, + {"op":"append","stream":"fence-cap","record":2,"fence":fences} + ]}), + ), + ( + "stream.batch", + json!({"atomic":false,"ops":[ + {"op":"append","stream":"fence-cap","record":1}, + {"op":"append","stream":"fence-cap","record":2,"fence":fences} + ]}), + ), + ] { + let error = registry.dispatch(verb, args).await.unwrap_err(); + let RuntimeError::InvalidInput(message) = error else { + panic!("{verb} must refuse invalid_input: {error}"); + }; + assert!(message.contains("at most 100 entries"), "{verb}: {message}"); + assert!(message.contains("sent 101"), "{verb}: {message}"); + assert_eq!(population(&rt).await, before, "{verb}"); + assert_eq!( + rt.backend().pool().writer_acquisition_snapshot(), + before_writers, + "{verb} must refuse before writer admission" + ); + } + } + assert_eq!(heads(®istry, &["fence-cap"]).await, vec![0]); + let after = registry + .dispatch("get", json!({"id":target["id"]})) + .await + .unwrap(); + assert_eq!(after["version"], target["version"]); + assert_eq!(after["content"], target["content"]); +} diff --git a/crates/khive-runtime/src/note_write.rs b/crates/khive-runtime/src/note_write.rs index 7976d0d89..df1ab9a82 100644 --- a/crates/khive-runtime/src/note_write.rs +++ b/crates/khive-runtime/src/note_write.rs @@ -7,6 +7,9 @@ use uuid::Uuid; use crate::{KhiveRuntime, NamespaceToken, RuntimeError, RuntimeResult}; +/// Each fence adds a keyed read while holding the writer, like a batch observation. +pub const MAX_NOTE_FENCES: usize = 100; + #[derive(Clone, Debug, Deserialize, Serialize)] #[serde(deny_unknown_fields)] pub struct NoteFence { @@ -42,6 +45,15 @@ impl From for NoteFences { } impl NoteFences { + fn validate_count(count: usize) -> RuntimeResult<()> { + if count > MAX_NOTE_FENCES { + return Err(RuntimeError::InvalidInput(format!( + "fence list admits at most {MAX_NOTE_FENCES} entries; this call sent {count}: each fence is a read taken while holding the writer" + ))); + } + Ok(()) + } + pub fn entries(&self) -> &[NoteFence] { match self { Self::One(fence) => std::slice::from_ref(fence), @@ -50,6 +62,7 @@ impl NoteFences { } pub fn validate(&self) -> RuntimeResult<()> { + Self::validate_count(self.entries().len())?; if self.entries().is_empty() { return Err(RuntimeError::InvalidInput( "fence list must not be empty".into(), @@ -76,7 +89,11 @@ impl<'de> Deserialize<'de> for NoteFences { One(NoteFence), Many(Vec), } - let fences = match Shape::deserialize(deserializer)? { + let value = serde_json::Value::deserialize(deserializer)?; + if let Some(entries) = value.as_array() { + Self::validate_count(entries.len()).map_err(serde::de::Error::custom)?; + } + let fences = match Shape::deserialize(value).map_err(serde::de::Error::custom)? { Shape::One(fence) => Self::One(fence), Shape::Many(fences) => Self::Many(fences), }; diff --git a/crates/khive-runtime/src/note_write_tests.rs b/crates/khive-runtime/src/note_write_tests.rs index 3d4da2abc..92769c3bd 100644 --- a/crates/khive-runtime/src/note_write_tests.rs +++ b/crates/khive-runtime/src/note_write_tests.rs @@ -18,6 +18,53 @@ use crate::{KhiveRuntime, NamespaceToken, RuntimeError, RuntimeResult}; const MODEL: &str = "note-version-test"; +#[test] +fn ordered_fences_cap_precedes_entry_validation_for_json_and_typed_inputs() { + use crate::note_write::NoteFences; + + let entries: Vec = (0..100) + .map(|index| NoteFence { + key: format!("fence-cap/{index}"), + kind: "head".into(), + expected_version: 1, + }) + .collect(); + let at_cap = NoteFences::Many(entries.clone()); + at_cap.validate().unwrap(); + serde_json::from_value::(serde_json::to_value(&at_cap).unwrap()).unwrap(); + + let mut over_cap = entries; + over_cap.push(NoteFence { + key: "fence-cap/100".into(), + kind: "head".into(), + expected_version: 1, + }); + for malformed in [false, true] { + let mut entries = over_cap.clone(); + if malformed { + entries[0].expected_version = 0; + } + let fences = NoteFences::Many(entries); + let error = fences.validate().unwrap_err(); + assert!(matches!(error, RuntimeError::InvalidInput(_)), "{error}"); + for message in [ + error.to_string(), + serde_json::from_value::(serde_json::to_value(&fences).unwrap()) + .unwrap_err() + .to_string(), + ] { + assert!(message.contains("at most 100 entries"), "{message}"); + assert!(message.contains("sent 101"), "{message}"); + } + } + let malformed = json!(vec![serde_json::Value::Null; 101]); + let message = serde_json::from_value::(malformed) + .unwrap_err() + .to_string(); + assert!(message.contains("at most 100 entries"), "{message}"); + assert!(message.contains("sent 101"), "{message}"); +} + #[tokio::test] async fn ordered_fences_observe_prior_write_in_same_transaction() { let (runtime, token, _) = fixture(); diff --git a/docs/adr/ADR-172-versioned-notes-compare-and-set.md b/docs/adr/ADR-172-versioned-notes-compare-and-set.md index 236589ed0..388968d10 100644 --- a/docs/adr/ADR-172-versioned-notes-compare-and-set.md +++ b/docs/adr/ADR-172-versioned-notes-compare-and-set.md @@ -469,6 +469,16 @@ Clarification (2026-09-10): the once-per-`(kind, key)` target rule applies indep fence list and to the keyed-write members of [ADR-174 A1.1](ADR-174-ordered-streams-append.md); it does not expand the batch-wide `fence`, which deliberately remains object-only. +Cardinality bound (2026-09-10, #2507): an ordered fence list admits at most **100 entries**. +An oversized list is `invalid_input`, naming both the cap and the count sent in its message, +before interpreting any fence entry or requesting a writer. The same bound applies to +singleton note `create`, note `update`, `stream.append`, and each `stream.batch` append +member, in both atomic and per-member modes. Every append member's shape is validated before +any member transaction starts, so an oversized later member cannot leave earlier writes. +Exactly 100 entries remains valid subject to the existing shape and fence checks. The cap +matches ADR-174's 100 observed-entry allowance: both lists add keyed reads while holding the +writer. This is a per-list limit, not a new aggregate limit across batch members. + The Python client accepts a dictionary or list of dictionaries and preserves entry order. Its generic `stream.append` builder preserves explicitly supplied null so the server can reject it. @@ -478,6 +488,10 @@ refusal, unchanged lease rows, and malformed-input domain-population controls. T must distinguish a check before transaction admission from one inside the admitted writer transaction, including another process renewing a lease at that boundary. Checking only the first entry or omitting `index` must each fail their corresponding list control. +The cardinality controls include exactly 100 distinct entries, 101 entries, an oversized list +with a malformed first entry, unchanged writer-acquisition counters on refusal, and a later +oversized append member in both batch modes. Disabling the cap or moving it below entry +interpretation must make the corresponding over-limit control fail. Append members of `stream.batch` accept the same optional object or non-empty list in their own `fence` field. In atomic mode, every member fence is checked in