diff --git a/CHANGELOG.md b/CHANGELOG.md index 119360e7c..f23b13de2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,10 @@ ### Fixed +- Outgoing-edge bucket iterators expose their exact remaining length without + inspecting edges, so bounded readers can precharge the entire source bucket + before filtering by type. + - The native pure interpreter evaluates bounded UTF-8 byte predicates with complete aperture charging and typed Boolean values. Native provider judgments independently validate matching helper/read guards and their bounded scan work. @@ -137,6 +141,15 @@ ### Added +- Outgoing-edge range claims retain source WARP, source node and edge type in + footprint union and scheduler reservation, including observed empty ranges. + Range-bearing operation footprint identities bind both range axes separately; + footprints without ranges keep their V1 identities. Range slots survive patch + identity, retained provenance and native Tick input/output metadata. Historical + host range readings replay a selected prefix and preserve edge relabeling, + movement and deletion. This does not implement the authored operation effect + bridge. + - Native Edict provider judgments accept the distinct bounded source-closure artifact, preserve its identity and refuse mixed or relabelled source forms. An explicit generation variant carries its owning schema binding; generic diff --git a/crates/echo-dry-tests/src/demo_rules.rs b/crates/echo-dry-tests/src/demo_rules.rs index f1b77de20..2e01e1849 100644 --- a/crates/echo-dry-tests/src/demo_rules.rs +++ b/crates/echo-dry-tests/src/demo_rules.rs @@ -170,6 +170,8 @@ fn base_scope_footprint(view: GraphView<'_>, scope: &NodeId) -> BaseScopeFootpri fn compute_motion_footprint(view: GraphView<'_>, scope: &NodeId) -> Footprint { let base = base_scope_footprint(view, scope); Footprint { + r_read: warp_core::OutgoingEdgeRangeSet::default(), + r_write: warp_core::OutgoingEdgeRangeSet::default(), n_read: base.n_read, n_write: NodeSet::default(), e_read: EdgeSet::default(), @@ -285,6 +287,8 @@ fn compute_port_footprint(view: GraphView<'_>, scope: &NodeId) -> Footprint { b_in.insert(base.warp_id, pack_port_key(scope, 0, true)); } Footprint { + r_read: warp_core::OutgoingEdgeRangeSet::default(), + r_write: warp_core::OutgoingEdgeRangeSet::default(), n_read: base.n_read, n_write, e_read: EdgeSet::default(), diff --git a/crates/echo-dry-tests/src/footprint.rs b/crates/echo-dry-tests/src/footprint.rs index 1fad7c390..ec10e8d06 100644 --- a/crates/echo-dry-tests/src/footprint.rs +++ b/crates/echo-dry-tests/src/footprint.rs @@ -323,6 +323,8 @@ impl FootprintBuilder { /// Consumes the builder and returns the constructed [`Footprint`]. pub fn build(self) -> Footprint { Footprint { + r_read: warp_core::OutgoingEdgeRangeSet::default(), + r_write: warp_core::OutgoingEdgeRangeSet::default(), n_read: self.n_read, n_write: self.n_write, e_read: self.e_read, diff --git a/crates/warp-core/src/admission.rs b/crates/warp-core/src/admission.rs index 48354356f..a3c01f80c 100644 --- a/crates/warp-core/src/admission.rs +++ b/crates/warp-core/src/admission.rs @@ -15,7 +15,7 @@ //! without changing the single-tick law that all admission still happens under //! `super_tick()`. -use crate::footprint::{AttachmentSet, EdgeSet, Footprint, NodeSet, PortSet}; +use crate::footprint::{AttachmentSet, EdgeSet, Footprint, NodeSet, OutgoingEdgeRangeSet, PortSet}; use crate::ident::Hash; /// Engine-defined policy identity for one admission act. @@ -42,6 +42,8 @@ pub struct AffectedRegion { pub edges: EdgeSet, /// Attachments locally affected by the claim. pub attachments: AttachmentSet, + /// Outgoing-edge ranges affected by the claim, including empty ranges. + pub outgoing_edge_ranges: OutgoingEdgeRangeSet, } impl AffectedRegion { @@ -52,6 +54,7 @@ impl AffectedRegion { nodes: footprint.n_write.clone(), edges: footprint.e_write.clone(), attachments: footprint.a_write.clone(), + outgoing_edge_ranges: footprint.r_write.clone(), } } } diff --git a/crates/warp-core/src/cmd.rs b/crates/warp-core/src/cmd.rs index df4f82fe5..7213090e1 100644 --- a/crates/warp-core/src/cmd.rs +++ b/crates/warp-core/src/cmd.rs @@ -163,6 +163,8 @@ fn import_suffix_intent_footprint(view: crate::GraphView<'_>, scope: &NodeId) -> })); Footprint { + r_read: crate::footprint::OutgoingEdgeRangeSet::default(), + r_write: crate::footprint::OutgoingEdgeRangeSet::default(), n_read, n_write, e_read: EdgeSet::default(), diff --git a/crates/warp-core/src/contract_host.rs b/crates/warp-core/src/contract_host.rs index b4e23c073..b3cf5cf1e 100644 --- a/crates/warp-core/src/contract_host.rs +++ b/crates/warp-core/src/contract_host.rs @@ -99,6 +99,8 @@ pub fn runtime_ingress_eint_read_footprint(view: GraphView<'_>, scope: &NodeId) local_id: *scope, })); Footprint { + r_read: crate::footprint::OutgoingEdgeRangeSet::default(), + r_write: crate::footprint::OutgoingEdgeRangeSet::default(), n_read, n_write: NodeSet::default(), e_read: EdgeSet::default(), diff --git a/crates/warp-core/src/echo_operation.rs b/crates/warp-core/src/echo_operation.rs index 9eaf8a7b1..9eda59feb 100644 --- a/crates/warp-core/src/echo_operation.rs +++ b/crates/warp-core/src/echo_operation.rs @@ -124,6 +124,7 @@ const CREATE_APPLICATION_BASIS_DOMAIN: &[u8] = const CREATE_RESULT_ABSENCE_PROPOSITION_DOMAIN: &[u8] = b"echo:operation-result-anchored-node-alpha-absence-proposition:v1\0"; const FOOTPRINT_DIGEST_DOMAIN: &[u8] = b"echo:operation-footprint:v1\0"; +const RANGE_FOOTPRINT_DIGEST_DOMAIN: &[u8] = b"echo:operation-footprint:v2\0"; const RECEIPT_DIGEST_DOMAIN: &[u8] = b"echo:operation-receipt:v1\0"; const COMPOSITION_DIGEST_DOMAIN: &[u8] = b"echo:operation-singleton-composition:v1\0"; const PLAN_DIGEST_DOMAIN: &[u8] = b"echo:operation-plan:v1\0"; @@ -7509,7 +7510,12 @@ fn record_node_read(footprint: &mut Footprint, node: NodeKey) { fn footprint_digest(footprint: &Footprint) -> Hash { let mut hasher = Hasher::new(); - hasher.update(FOOTPRINT_DIGEST_DOMAIN); + let has_ranges = !footprint.r_read.is_empty() || !footprint.r_write.is_empty(); + hasher.update(if has_ranges { + RANGE_FOOTPRINT_DIGEST_DOMAIN + } else { + FOOTPRINT_DIGEST_DOMAIN + }); hash_node_set( &mut hasher, b"nr", @@ -7559,6 +7565,17 @@ fn footprint_digest(footprint: &Footprint) -> Hash { footprint.b_out.iter(), ); hasher.update(&footprint.factor_mask.to_le_bytes()); + if has_ranges { + for (label, ranges) in [(b"rr", &footprint.r_read), (b"rw", &footprint.r_write)] { + hasher.update(label); + hasher.update(&(ranges.len() as u64).to_le_bytes()); + for range in ranges.iter() { + hasher.update(range.source.warp_id.as_bytes()); + hasher.update(range.source.local_id.as_bytes()); + hasher.update(range.edge_type.as_bytes()); + } + } + } hasher.finalize().into() } @@ -8689,6 +8706,17 @@ mod tests { debug_assert_action_member_alignment_v1(2, 1); } + #[test] + fn footprint_without_ranges_retains_its_v1_identity() { + assert_eq!( + footprint_digest(&Footprint::default()), + [ + 133, 20, 8, 211, 91, 122, 210, 37, 17, 71, 156, 145, 21, 30, 109, 175, 39, 57, 176, + 34, 41, 209, 18, 169, 48, 224, 15, 83, 138, 167, 145, 2 + ] + ); + } + #[test] fn footprint_digest_is_prefix_free_across_adjacent_sets() { let mut read_node_bytes = [0; 64]; @@ -8749,6 +8777,40 @@ mod tests { let mut footprint = Footprint::default(); footprint.b_out.insert(node.warp_id, 9); footprints.push(footprint); + let range = crate::footprint::OutgoingEdgeRange { + source: node, + edge_type: crate::TypeId(digest(10)), + }; + let mut footprint = Footprint::default(); + footprint.r_read.insert(range); + footprints.push(footprint); + let mut footprint = Footprint::default(); + footprint.r_write.insert(range); + footprints.push(footprint); + for changed in [ + crate::footprint::OutgoingEdgeRange { + source: NodeKey { + warp_id: crate::WarpId(digest(11)), + ..node + }, + ..range + }, + crate::footprint::OutgoingEdgeRange { + source: NodeKey { + local_id: crate::NodeId(digest(11)), + ..node + }, + ..range + }, + crate::footprint::OutgoingEdgeRange { + edge_type: crate::TypeId(digest(11)), + ..range + }, + ] { + let mut footprint = Footprint::default(); + footprint.r_read.insert(changed); + footprints.push(footprint); + } footprints.push(Footprint { factor_mask: 1, ..Footprint::default() diff --git a/crates/warp-core/src/engine_impl.rs b/crates/warp-core/src/engine_impl.rs index 9a6e27b33..7b8891c3b 100644 --- a/crates/warp-core/src/engine_impl.rs +++ b/crates/warp-core/src/engine_impl.rs @@ -51,6 +51,10 @@ use crate::tx::TxId; use crate::warp_state::{WarpInstance, WarpState}; use crate::worldline_state::WorldlineState; use crate::ContractInverseHandler; + +#[cfg(test)] +#[path = "engine_impl/range_tests.rs"] +mod range_tests; use std::sync::Arc; const RUNTIME_INGRESS_EVENT_TYPE: &str = "runtime/ingress/event"; @@ -3230,6 +3234,12 @@ pub(crate) fn footprints_conflict( { return true; } + if a.r_write.intersects(&b.r_write) + || a.r_write.intersects(&b.r_read) + || b.r_write.intersects(&a.r_read) + { + return true; + } if a.a_write.intersects(&b.a_write) || a.a_write.intersects(&b.a_read) || b.a_write.intersects(&a.a_read) @@ -3349,6 +3359,13 @@ fn extend_slots_from_footprint( in_slots.insert(SlotId::Edge(*key)); out_slots.insert(SlotId::Edge(*key)); } + for key in fp.r_read.iter() { + in_slots.insert(SlotId::OutgoingEdges(*key)); + } + for key in fp.r_write.iter() { + in_slots.insert(SlotId::OutgoingEdges(*key)); + out_slots.insert(SlotId::OutgoingEdges(*key)); + } // Attachments (already warp-scoped via AttachmentKey) for key in fp.a_read.iter() { @@ -3526,6 +3543,8 @@ mod tests { a_write.insert(key); } crate::Footprint { + r_read: crate::footprint::OutgoingEdgeRangeSet::default(), + r_write: crate::footprint::OutgoingEdgeRangeSet::default(), n_read: crate::NodeSet::default(), n_write: crate::NodeSet::default(), e_read: crate::EdgeSet::default(), @@ -3566,6 +3585,8 @@ mod tests { a_write.insert(key); } crate::Footprint { + r_read: crate::footprint::OutgoingEdgeRangeSet::default(), + r_write: crate::footprint::OutgoingEdgeRangeSet::default(), n_read: crate::NodeSet::default(), n_write: crate::NodeSet::default(), e_read: crate::EdgeSet::default(), @@ -3660,6 +3681,8 @@ mod tests { local_id: *scope, })); crate::Footprint { + r_read: crate::footprint::OutgoingEdgeRangeSet::default(), + r_write: crate::footprint::OutgoingEdgeRangeSet::default(), n_read, n_write: crate::NodeSet::default(), e_read: crate::EdgeSet::default(), diff --git a/crates/warp-core/src/engine_impl/range_tests.rs b/crates/warp-core/src/engine_impl/range_tests.rs new file mode 100644 index 000000000..94c7b4bcd --- /dev/null +++ b/crates/warp-core/src/engine_impl/range_tests.rs @@ -0,0 +1,148 @@ +// SPDX-License-Identifier: Apache-2.0 +// © James Ross Ω FLYING•ROBOTS +#![allow(clippy::expect_used)] // Test assertions may abort the witness. +//! Native scheduler witness for declared range claims, not an Edict bridge. +use super::*; +use crate::{EdgeKey, EdgeRecord, Footprint, OutgoingEdgeRange, PatternGraph, TickDelta}; + +const OBSERVE: &str = "test/range-observe"; +const INSERT: &str = "test/range-insert"; + +fn range(view: GraphView<'_>, scope: &NodeId) -> OutgoingEdgeRange { + OutgoingEdgeRange { + source: NodeKey { + warp_id: view.warp_id(), + local_id: *scope, + }, + edge_type: make_type_id("test/range-type"), + } +} + +fn observe_footprint(view: GraphView<'_>, scope: &NodeId) -> Footprint { + let mut footprint = Footprint::default(); + footprint.r_read.insert(range(view, scope)); + footprint +} + +fn insert_footprint(view: GraphView<'_>, scope: &NodeId) -> Footprint { + let mut footprint = Footprint::default(); + footprint.r_write.insert(range(view, scope)); + footprint.n_write.insert(NodeKey { + warp_id: view.warp_id(), + local_id: *scope, + }); + footprint.e_write.insert(EdgeKey { + warp_id: view.warp_id(), + local_id: make_edge_id("test/range-edge"), + }); + footprint +} + +// The native fixture supplies a conservative range claim. It does not test +// interpreted read instrumentation or the broader node-granted adjacency API. +fn observe(_: GraphView<'_>, _: &NodeId, _: &mut TickDelta) {} + +fn insert(view: GraphView<'_>, scope: &NodeId, delta: &mut TickDelta) { + delta.push(WarpOp::UpsertEdge { + warp_id: view.warp_id(), + record: EdgeRecord { + id: make_edge_id("test/range-edge"), + from: *scope, + to: make_node_id("test/range-target"), + ty: make_type_id("test/range-type"), + }, + }); +} + +fn rule( + id: Hash, + name: &'static str, + footprint: fn(GraphView<'_>, &NodeId) -> Footprint, + executor: fn(GraphView<'_>, &NodeId, &mut TickDelta), +) -> RewriteRule { + RewriteRule { + id, + name, + left: PatternGraph { nodes: vec![] }, + matcher: |_, _| true, + executor: RuleExecutor::legacy(executor), + compute_footprint: footprint, + factor_mask: 0, + conflict_policy: ConflictPolicy::Abort, + join_fn: None, + } +} + +#[test] +fn a_real_tick_rejects_insertion_against_an_empty_range_and_retains_both_roles() { + let source = make_node_id("test/range-source"); + let target = make_node_id("test/range-target"); + let mut store = GraphStore::default(); + for node in [source, target] { + store.insert_node( + node, + NodeRecord { + ty: make_type_id("test/range-node"), + }, + ); + } + assert_eq!(store.edges_from(&source).count(), 0); + let mut engine = Engine::new(store, source); + let key = OutgoingEdgeRange { + source: engine.current_root, + edge_type: make_type_id("test/range-type"), + }; + let mut ids = [[1; 32], [2; 32]]; + ids.sort_by_key(|id| scope_hash(id, &key.source)); + engine + .register_rule(rule(ids[0], OBSERVE, observe_footprint, observe)) + .expect("reader"); + engine + .register_rule(rule(ids[1], INSERT, insert_footprint, insert)) + .expect("writer"); + let initial = engine.state.clone(); + let tx = engine.begin(); + assert!(matches!( + engine.apply(tx, INSERT, &source), + Ok(ApplyResult::Applied) + )); + assert!(matches!( + engine.apply(tx, OBSERVE, &source), + Ok(ApplyResult::Applied) + )); + let (_, receipt, first) = engine.commit_with_receipt(tx).expect("first Tick"); + assert_eq!(receipt.entries().len(), 2); + assert_eq!(receipt.entries()[0].rule_id, ids[0]); + assert_eq!( + receipt.entries()[0].disposition, + TickReceiptDisposition::Applied + ); + assert_eq!( + receipt.entries()[1].disposition, + TickReceiptDisposition::Rejected(TickReceiptRejection::FootprintConflict) + ); + assert_eq!(receipt.blocked_by(1), &[0]); + let slot = SlotId::OutgoingEdges(key); + assert!(first.in_slots().contains(&slot)); + assert!(!first.out_slots().contains(&slot)); + assert!(first.ops().is_empty()); + let tx = engine.begin(); + assert!(matches!( + engine.apply(tx, INSERT, &source), + Ok(ApplyResult::Applied) + )); + let (_, receipt, second) = engine.commit_with_receipt(tx).expect("second Tick"); + assert_eq!( + receipt.entries()[0].disposition, + TickReceiptDisposition::Applied + ); + assert!(second.in_slots().contains(&slot)); + assert!(second.out_slots().contains(&slot)); + let mut replay = initial; + first.apply_to_state(&mut replay).expect("first replay"); + second.apply_to_state(&mut replay).expect("second replay"); + assert_eq!( + compute_state_root(&replay, &key.source), + compute_state_root(&engine.state, &key.source) + ); +} diff --git a/crates/warp-core/src/footprint.rs b/crates/warp-core/src/footprint.rs index a9d7ef8c6..b2a3e28b4 100644 --- a/crates/warp-core/src/footprint.rs +++ b/crates/warp-core/src/footprint.rs @@ -28,7 +28,54 @@ use std::collections::BTreeSet; use crate::attachment::AttachmentKey; -use crate::ident::{EdgeId, EdgeKey, NodeId, NodeKey, WarpId}; +use crate::ident::{EdgeId, EdgeKey, NodeId, NodeKey, TypeId, WarpId}; + +#[cfg(test)] +#[path = "footprint/range_tests.rs"] +mod range_tests; + +/// One outgoing-edge range, including an observed empty range. +/// +/// A fresh edge ID cannot evade a claim on its source and edge type. The source +/// includes its WARP identity; the type is the edge type, not either node type. +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct OutgoingEdgeRange { + /// Source of the outgoing edges. + pub source: NodeKey, + /// Type selecting the outgoing edges. + pub edge_type: TypeId, +} + +/// Canonically ordered set of outgoing-edge ranges. +#[derive(Debug, Clone, PartialEq, Eq, Default)] +pub struct OutgoingEdgeRangeSet(BTreeSet); + +impl OutgoingEdgeRangeSet { + /// Claims a range, even if it currently contains no edge. + pub fn insert(&mut self, key: OutgoingEdgeRange) { + self.0.insert(key); + } + + /// Iterates in source-WARP, source-node, edge-type order. + pub fn iter(&self) -> impl Iterator { + self.0.iter() + } + + /// Whether the two sets share any range key. + pub fn intersects(&self, other: &Self) -> bool { + intersects_btree(&self.0, &other.0) + } + + /// Whether no range is claimed. + pub fn is_empty(&self) -> bool { + self.0.is_empty() + } + + /// Number of distinct claimed ranges. + pub fn len(&self) -> usize { + self.0.len() + } +} // ============================================================================= // Packed Port Key @@ -261,6 +308,10 @@ pub struct Footprint { pub e_read: EdgeSet, /// Edges written/created/deleted by the rewrite (warp-scoped). pub e_write: EdgeSet, + /// Outgoing-edge ranges observed, including empty ranges. + pub r_read: OutgoingEdgeRangeSet, + /// Outgoing-edge ranges changed by edge insertion, removal or retargeting. + pub r_write: OutgoingEdgeRangeSet, /// Attachment slots read by the rewrite. pub a_read: AttachmentSet, /// Attachment slots written by the rewrite. @@ -299,6 +350,8 @@ impl Footprint { self.n_write.0.extend(other.n_write.0.iter().copied()); self.e_read.0.extend(other.e_read.0.iter().copied()); self.e_write.0.extend(other.e_write.0.iter().copied()); + self.r_read.0.extend(other.r_read.0.iter().copied()); + self.r_write.0.extend(other.r_write.0.iter().copied()); self.a_read.0.extend(other.a_read.0.iter().copied()); self.a_write.0.extend(other.a_write.0.iter().copied()); self.b_in.0.extend(other.b_in.0.iter().copied()); @@ -312,6 +365,8 @@ impl Footprint { && self.n_write.is_empty() && self.e_read.is_empty() && self.e_write.is_empty() + && self.r_read.is_empty() + && self.r_write.is_empty() && self.a_read.is_empty() && self.a_write.is_empty() && self.b_in.is_empty() @@ -361,6 +416,12 @@ impl Footprint { { return false; } + if self.r_write.intersects(&other.r_write) + || self.r_write.intersects(&other.r_read) + || other.r_write.intersects(&self.r_read) + { + return false; + } if self.a_write.intersects(&other.a_write) || self.a_write.intersects(&other.a_read) || other.a_write.intersects(&self.a_read) diff --git a/crates/warp-core/src/footprint/range_tests.rs b/crates/warp-core/src/footprint/range_tests.rs new file mode 100644 index 000000000..586394714 --- /dev/null +++ b/crates/warp-core/src/footprint/range_tests.rs @@ -0,0 +1,79 @@ +// SPDX-License-Identifier: Apache-2.0 +// © James Ross Ω FLYING•ROBOTS +use super::*; +use crate::TypeId; + +fn range(warp: u8, source: u8, ty: u8) -> OutgoingEdgeRange { + OutgoingEdgeRange { + source: NodeKey { + warp_id: WarpId([warp; 32]), + local_id: NodeId([source; 32]), + }, + edge_type: TypeId([ty; 32]), + } +} + +#[test] +fn an_empty_range_observation_conflicts_with_a_fresh_edge_write() { + let key = range(1, 2, 3); + let mut observation = Footprint::default(); + observation.r_read.insert(key); + let mut insertion = Footprint::default(); + insertion.r_write.insert(key); + insertion.e_write.insert(EdgeKey { + warp_id: key.source.warp_id, + local_id: EdgeId([99; 32]), + }); + assert!(!observation.independent(&insertion)); + assert!(!insertion.independent(&observation)); +} + +#[test] +fn range_claims_keep_read_sharing_and_full_key_separation() { + let mut left = Footprint::default(); + left.r_read.insert(range(1, 2, 3)); + assert!(left.independent(&left)); + for key in [range(9, 2, 3), range(1, 9, 3), range(1, 2, 9)] { + let mut right = Footprint::default(); + right.r_write.insert(key); + assert!(left.independent(&right)); + assert!(right.independent(&left)); + } + left.r_write.insert(range(1, 2, 3)); + assert!(!left.independent(&left)); +} + +#[test] +fn union_preserves_range_claims_and_the_unpopulated_mask() { + let mut ranges = Footprint::default(); + ranges.r_read.insert(range(1, 2, 3)); + ranges.r_write.insert(range(1, 4, 3)); + let original = ranges.clone(); + let mut populated = Footprint { + factor_mask: 1, + ..Footprint::default() + }; + populated.n_read.insert(range(1, 2, 3).source); + ranges.union_assign(&populated); + assert_eq!(ranges.factor_mask, 0); + assert_eq!(ranges.r_read, original.r_read); + assert_eq!(ranges.r_write, original.r_write); + populated.union_assign(&original); + assert_eq!(populated.factor_mask, 0); + assert_eq!(populated.r_read, original.r_read); + assert_eq!(populated.r_write, original.r_write); + assert_eq!(ranges.bounded_site().claim_footprint, ranges); +} + +#[test] +fn admission_region_preserves_the_range_write_wake() { + let mut footprint = Footprint::default(); + footprint.r_write.insert(range(1, 2, 3)); + assert_eq!( + footprint + .bounded_site() + .affected_region + .outgoing_edge_ranges, + footprint.r_write + ); +} diff --git a/crates/warp-core/src/graph.rs b/crates/warp-core/src/graph.rs index 5b07fad0e..0607f9ab9 100644 --- a/crates/warp-core/src/graph.rs +++ b/crates/warp-core/src/graph.rs @@ -427,8 +427,14 @@ impl GraphStore { /// /// Edges are yielded in insertion order. For deterministic traversal /// (e.g., snapshot hashing), callers must sort by `EdgeId`. - pub fn edges_from(&self, id: &NodeId) -> impl Iterator { - self.edges_from.get(id).into_iter().flatten() + /// The remaining bucket length is available in constant time, without + /// advancing or inspecting an edge. Bounded range readers can therefore + /// charge the entire bucket before filtering by edge type. + pub fn edges_from(&self, id: &NodeId) -> impl ExactSizeIterator { + self.edges_from + .get(id) + .map_or(&[][..], Vec::as_slice) + .iter() } /// Returns a mutable reference to a node when it exists. @@ -777,6 +783,55 @@ mod tests { use super::*; use crate::ident::{make_edge_id, make_node_id, make_type_id}; + // A range budget must count nonmatching edges before advancing the iterator. + #[test] + fn outgoing_bucket_length_is_available_before_inspection_and_tracks_moves() { + let mut store = GraphStore::default(); + let a = make_node_id("range/a"); + let b = make_node_id("range/b"); + let ty = make_type_id("range/matching"); + let other = make_type_id("range/nonmatching"); + for node in [a, b] { + store.insert_node(node, NodeRecord { ty }); + } + assert_eq!(store.edges_from(&a).len(), 0); + for (name, edge_type) in [("first", ty), ("second", other), ("third", other)] { + store.insert_edge( + a, + EdgeRecord { + id: make_edge_id(name), + from: a, + to: b, + ty: edge_type, + }, + ); + } + let mut bucket = store.edges_from(&a); + assert_eq!(bucket.len(), 3); + assert_eq!( + bucket.next().map(|edge| edge.id), + Some(make_edge_id("first")) + ); + assert_eq!(bucket.len(), 2); + drop(bucket); + store.insert_edge( + b, + EdgeRecord { + id: make_edge_id("second"), + from: b, + to: a, + ty: other, + }, + ); + assert_eq!(store.edges_from(&a).len(), 2); + assert_eq!(store.edges_from(&b).len(), 1); + assert!(store.delete_edge_exact(a, make_edge_id("first"))); + assert_eq!(store.edges_from(&a).len(), 1); + assert!(store.delete_node_cascade(b)); + assert_eq!(store.edges_from(&a).len(), 0); + assert_eq!(store.edges_from(&b).len(), 0); + } + #[test] fn delete_node_cascade_updates_reverse_indexes() { let mut store = GraphStore::default(); diff --git a/crates/warp-core/src/inbox.rs b/crates/warp-core/src/inbox.rs index c2668e5a3..e99bb90cb 100644 --- a/crates/warp-core/src/inbox.rs +++ b/crates/warp-core/src/inbox.rs @@ -184,6 +184,8 @@ fn inbox_footprint(view: GraphView<'_>, scope: &NodeId) -> Footprint { } Footprint { + r_read: crate::footprint::OutgoingEdgeRangeSet::default(), + r_write: crate::footprint::OutgoingEdgeRangeSet::default(), n_read, n_write, e_read, @@ -245,6 +247,8 @@ fn ack_pending_footprint(view: GraphView<'_>, scope: &NodeId) -> Footprint { n_write.insert_with_warp(warp_id, inbox_id); Footprint { + r_read: crate::footprint::OutgoingEdgeRangeSet::default(), + r_write: crate::footprint::OutgoingEdgeRangeSet::default(), n_read, n_write, e_read, diff --git a/crates/warp-core/src/lib.rs b/crates/warp-core/src/lib.rs index 2279dab0c..1dd517bae 100644 --- a/crates/warp-core/src/lib.rs +++ b/crates/warp-core/src/lib.rs @@ -291,7 +291,8 @@ pub use engine_impl::{ pub use execution_evidence::{ExecutionEvidenceKey, ExecutionFootprintEvidence}; pub use execution_graph_view::ExecutionGraphView; pub use footprint::{ - pack_port_key, AttachmentSet, EdgeSet, Footprint, NodeSet, PortKey, PortSet, WarpScopedPortKey, + pack_port_key, AttachmentSet, EdgeSet, Footprint, NodeSet, OutgoingEdgeRange, + OutgoingEdgeRangeSet, PortKey, PortSet, WarpScopedPortKey, }; pub use footprint_guard::{FootprintViolation, FootprintViolationWithPanic, ViolationKind}; pub use graph::{DeleteNodeError, GraphStore}; diff --git a/crates/warp-core/src/observation.rs b/crates/warp-core/src/observation.rs index de08d230d..8230f21f6 100644 --- a/crates/warp-core/src/observation.rs +++ b/crates/warp-core/src/observation.rs @@ -2546,6 +2546,12 @@ fn hash_slot(hasher: &mut Hasher, slot: &SlotId) { hasher.update(warp_id.as_bytes()); hasher.update(&port_key.to_le_bytes()); } + SlotId::OutgoingEdges(range) => { + hasher.update(&[5]); + hasher.update(range.source.warp_id.as_bytes()); + hasher.update(range.source.local_id.as_bytes()); + hasher.update(range.edge_type.as_bytes()); + } } } diff --git a/crates/warp-core/src/provenance_codec.rs b/crates/warp-core/src/provenance_codec.rs index 6995d8498..a1b44b6b2 100644 --- a/crates/warp-core/src/provenance_codec.rs +++ b/crates/warp-core/src/provenance_codec.rs @@ -5,6 +5,10 @@ use bytes::Bytes; use thiserror::Error; +#[cfg(test)] +#[path = "provenance_codec/range_tests.rs"] +mod range_tests; + use crate::{ attachment::{AtomPayload, AttachmentKey, AttachmentOwner, AttachmentPlane, AttachmentValue}, clock::{GlobalTick, WorldlineTick}, @@ -455,6 +459,11 @@ fn push_slot(out: &mut Vec, slot: SlotId) { push_hash(out, warp_id.as_bytes()); push_u64(out, port); } + SlotId::OutgoingEdges(range) => { + out.push(5); + push_node_key(out, range.source); + push_hash(out, range.edge_type.as_bytes()); + } } } @@ -833,6 +842,10 @@ impl<'a> RetainedProvenanceCursor<'a> { 2 => Ok(SlotId::Edge(self.read_edge_key()?)), 3 => Ok(SlotId::Attachment(self.read_attachment_key()?)), 4 => Ok(SlotId::Port((WarpId(self.read_hash()?), self.read_u64()?))), + 5 => Ok(SlotId::OutgoingEdges(crate::footprint::OutgoingEdgeRange { + source: self.read_node_key()?, + edge_type: TypeId(self.read_hash()?), + })), tag => Err(unknown_tag("slot", tag)), } } diff --git a/crates/warp-core/src/provenance_codec/range_tests.rs b/crates/warp-core/src/provenance_codec/range_tests.rs new file mode 100644 index 000000000..089e98d54 --- /dev/null +++ b/crates/warp-core/src/provenance_codec/range_tests.rs @@ -0,0 +1,129 @@ +// SPDX-License-Identifier: Apache-2.0 +// © James Ross Ω FLYING•ROBOTS +#![allow(clippy::expect_used)] // Test assertions may abort the witness. +use super::*; +use crate::OutgoingEdgeRange; + +fn range() -> OutgoingEdgeRange { + OutgoingEdgeRange { + source: NodeKey { + warp_id: WarpId([1; 32]), + local_id: NodeId([2; 32]), + }, + edge_type: TypeId([3; 32]), + } +} + +#[test] +fn retained_range_slot_has_exact_bytes_and_refuses_every_truncated_prefix() { + let slot = SlotId::OutgoingEdges(range()); + let mut bytes = Vec::new(); + push_slot(&mut bytes, slot); + let expected = [vec![5], vec![1; 32], vec![2; 32], vec![3; 32]].concat(); + assert_eq!(bytes, expected); + let mut cursor = RetainedProvenanceCursor::new(&bytes); + assert_eq!(cursor.read_slot(), Ok(slot)); + assert_eq!(cursor.finish(), Ok(())); + for end in 0..bytes.len() { + let mut cursor = RetainedProvenanceCursor::new(&bytes[..end]); + assert_eq!( + cursor.read_slot(), + Err(RetainedProvenanceError::UnexpectedEof) + ); + } +} + +#[test] +fn legacy_point_slot_bytes_remain_unchanged() { + let mut bytes = Vec::new(); + push_slot(&mut bytes, SlotId::Node(range().source)); + assert_eq!(bytes, [vec![1], vec![1; 32], vec![2; 32]].concat()); + bytes.clear(); + push_slot(&mut bytes, SlotId::Port((range().source.warp_id, 7))); + assert_eq!( + bytes, + [vec![4], vec![1; 32], 7_u64.to_le_bytes().to_vec()].concat() + ); +} + +#[test] +fn patch_identity_binds_each_range_component_and_read_write_role() { + let patch = |read, key| { + let slot = SlotId::OutgoingEdges(key); + WarpTickPatchV1::new( + 1, + [0; 32], + TickCommitStatus::Committed, + if read { vec![slot] } else { vec![] }, + if read { vec![] } else { vec![slot] }, + vec![], + ) + }; + let base = patch(true, range()); + for key in [ + OutgoingEdgeRange { + source: NodeKey { + warp_id: WarpId([4; 32]), + ..range().source + }, + ..range() + }, + OutgoingEdgeRange { + source: NodeKey { + local_id: NodeId([4; 32]), + ..range().source + }, + ..range() + }, + OutgoingEdgeRange { + edge_type: TypeId([4; 32]), + ..range() + }, + ] { + assert_ne!(base.digest(), patch(true, key).digest()); + } + assert_ne!(base.digest(), patch(false, range()).digest()); +} + +#[test] +fn retained_worldline_patch_preserves_mixed_point_and_range_slots() { + let slot = SlotId::OutgoingEdges(range()); + let patch = WarpTickPatchV1::new( + 1, + [7; 32], + TickCommitStatus::Committed, + vec![slot, SlotId::Node(range().source)], + vec![slot], + vec![], + ); + let retained = WorldlineTickPatchV1 { + header: WorldlineTickHeaderV1 { + commit_global_tick: GlobalTick::from_raw(1), + policy_id: 1, + rule_pack_id: [7; 32], + plan_digest: [8; 32], + decision_digest: [9; 32], + rewrites_digest: [10; 32], + }, + warp_id: range().source.warp_id, + ops: patch.ops().to_vec(), + in_slots: patch.in_slots().to_vec(), + out_slots: patch.out_slots().to_vec(), + patch_digest: patch.digest(), + }; + let mut bytes = Vec::new(); + push_worldline_patch(&mut bytes, &retained); + let mut cursor = RetainedProvenanceCursor::new(&bytes); + let decoded = cursor.read_worldline_patch().expect("retained patch"); + assert_eq!(cursor.finish(), Ok(())); + assert_eq!(decoded, retained); + let rebuilt = WarpTickPatchV1::new( + decoded.header.policy_id, + decoded.header.rule_pack_id, + TickCommitStatus::Committed, + decoded.in_slots, + decoded.out_slots, + decoded.ops, + ); + assert_eq!(rebuilt.digest(), retained.patch_digest); +} diff --git a/crates/warp-core/src/scheduler.rs b/crates/warp-core/src/scheduler.rs index 3f94c8c10..41724ab72 100644 --- a/crates/warp-core/src/scheduler.rs +++ b/crates/warp-core/src/scheduler.rs @@ -14,12 +14,16 @@ use std::collections::BTreeMap; use std::sync::Arc; -use crate::footprint::{Footprint, WarpScopedPortKey}; +use crate::footprint::{Footprint, OutgoingEdgeRange, WarpScopedPortKey}; use crate::ident::{CompactRuleId, EdgeKey, Hash, NodeKey}; use crate::telemetry::TelemetrySink; use crate::tick_delta::OpOrigin; use crate::tx::TxId; +#[cfg(test)] +#[path = "scheduler/range_tests.rs"] +mod range_tests; + /// Active footprint tracking for one transaction, using per-transaction /// [`GenSet`]s (`BTreeMap`-backed, O(log n) per lookup). /// @@ -36,6 +40,10 @@ pub(crate) struct ActiveFootprints { edges_written: GenSet, /// Edges read by reserved rewrites (warp-scoped) edges_read: GenSet, + /// Outgoing-edge ranges changed by reserved rewrites. + ranges_written: GenSet, + /// Outgoing-edge ranges observed by reserved rewrites, empty included. + ranges_read: GenSet, /// Attachments written by reserved rewrites (already warp-scoped via NodeKey/EdgeKey) attachments_written: GenSet, /// Attachments read by reserved rewrites (already warp-scoped via NodeKey/EdgeKey) @@ -51,6 +59,8 @@ impl ActiveFootprints { nodes_read: GenSet::new(), edges_written: GenSet::new(), edges_read: GenSet::new(), + ranges_written: GenSet::new(), + ranges_read: GenSet::new(), attachments_written: GenSet::new(), attachments_read: GenSet::new(), ports: GenSet::new(), @@ -194,6 +204,18 @@ impl RadixScheduler { } } + // Range writes conflict with prior writes OR reads, including absence. + for key in pr.footprint.r_write.iter() { + if active.ranges_written.contains(*key) || active.ranges_read.contains(*key) { + return true; + } + } + for key in pr.footprint.r_read.iter() { + if active.ranges_written.contains(*key) { + return true; + } + } + // Attachment writes conflict with prior writes OR reads. for key in pr.footprint.a_write.iter() { if active.attachments_written.contains(*key) || active.attachments_read.contains(*key) { @@ -241,6 +263,12 @@ impl RadixScheduler { for key in pr.footprint.e_read.iter() { active.edges_read.mark(*key); } + for key in pr.footprint.r_write.iter() { + active.ranges_written.mark(*key); + } + for key in pr.footprint.r_read.iter() { + active.ranges_read.mark(*key); + } for key in pr.footprint.a_write.iter() { active.attachments_written.mark(*key); } diff --git a/crates/warp-core/src/scheduler/range_tests.rs b/crates/warp-core/src/scheduler/range_tests.rs new file mode 100644 index 000000000..fd9b7e1b3 --- /dev/null +++ b/crates/warp-core/src/scheduler/range_tests.rs @@ -0,0 +1,69 @@ +// SPDX-License-Identifier: Apache-2.0 +// © James Ross Ω FLYING•ROBOTS +use super::*; +use crate::{footprint::OutgoingEdgeRange, NodeId, TypeId, WarpId}; + +fn key(ty: u8) -> OutgoingEdgeRange { + OutgoingEdgeRange { + source: NodeKey { + warp_id: WarpId([1; 32]), + local_id: NodeId([2; 32]), + }, + edge_type: TypeId([ty; 32]), + } +} + +fn candidate(reads: &[OutgoingEdgeRange], writes: &[OutgoingEdgeRange]) -> PendingRewrite { + let mut footprint = Footprint::default(); + for range in reads { + footprint.r_read.insert(*range); + } + for range in writes { + footprint.r_write.insert(*range); + } + PendingRewrite { + rule_id: [1; 32], + compact_rule: CompactRuleId(1), + scope_hash: [2; 32], + scope: key(1).source, + footprint, + phase: RewritePhase::Matched, + origin: OpOrigin::default(), + } +} + +#[test] +fn range_reservations_conflict_in_both_orders_and_allow_shared_reads() { + let tx = TxId::from_raw(1); + for (left_writes, right_writes, expected) in [ + (false, false, true), + (false, true, false), + (true, false, false), + (true, true, false), + ] { + let mut scheduler = RadixScheduler::default(); + let make = |writes| { + if writes { + candidate(&[], &[key(3)]) + } else { + candidate(&[key(3)], &[]) + } + }; + assert!(scheduler.reserve(tx, &mut make(left_writes))); + let mut right = make(right_writes); + assert_eq!(scheduler.reserve(tx, &mut right), expected); + assert_eq!(right.phase == RewritePhase::Reserved, expected); + assert!(scheduler.reserve(tx, &mut candidate(&[], &[key(4)]))); + } +} + +#[test] +fn rejected_range_claims_do_not_leak_and_finalization_releases_them() { + let tx = TxId::from_raw(1); + let mut scheduler = RadixScheduler::default(); + assert!(scheduler.reserve(tx, &mut candidate(&[key(3)], &[]))); + assert!(!scheduler.reserve(tx, &mut candidate(&[], &[key(3), key(4)]))); + assert!(scheduler.reserve(tx, &mut candidate(&[key(4)], &[]))); + scheduler.finalize_tx(tx); + assert!(scheduler.reserve(tx, &mut candidate(&[], &[key(3), key(4)]))); +} diff --git a/crates/warp-core/src/settlement.rs b/crates/warp-core/src/settlement.rs index a7860aea4..6fd7cc759 100644 --- a/crates/warp-core/src/settlement.rs +++ b/crates/warp-core/src/settlement.rs @@ -566,6 +566,12 @@ fn canonical_slot_bytes(slot: &SlotId) -> Vec { bytes.extend_from_slice(warp_id.as_bytes()); bytes.extend_from_slice(&port_key.to_le_bytes()); } + SlotId::OutgoingEdges(range) => { + bytes.push(5); + bytes.extend_from_slice(range.source.warp_id.as_bytes()); + bytes.extend_from_slice(range.source.local_id.as_bytes()); + bytes.extend_from_slice(range.edge_type.as_bytes()); + } } bytes } @@ -1171,6 +1177,7 @@ enum RevalidationSlotValue { Edge(Option), Attachment(Option), Port(WarpScopedPortKey), + OutgoingEdges(Option>), } fn overlap_slots_are_clean( @@ -1213,6 +1220,9 @@ fn revalidation_slot_value(state: &WorldlineState, slot: &SlotId) -> Revalidatio .cloned(), }), SlotId::Port(port) => RevalidationSlotValue::Port(port), + SlotId::OutgoingEdges(range) => RevalidationSlotValue::OutgoingEdges( + crate::slot_reading::materialized_outgoing_edges(state.warp_state(), range), + ), } } diff --git a/crates/warp-core/src/slot_reading.rs b/crates/warp-core/src/slot_reading.rs index 71884bf19..a07ed2f61 100644 --- a/crates/warp-core/src/slot_reading.rs +++ b/crates/warp-core/src/slot_reading.rs @@ -11,6 +11,11 @@ //! //! The index is derived from patches and can always be rebuilt from them. It //! is a reading aid, never an authority. +//! +//! Outgoing-edge ranges are aggregates rather than absolute single-op writes. +//! Their historical host reading clones the initial state and replays the +//! selected prefix. It is not an indexed constant-cost or metered interpreter +//! read; point-slot readings retain the indexed path. use std::collections::BTreeMap; @@ -22,6 +27,10 @@ use crate::record::{EdgeRecord, NodeRecord}; use crate::tick_patch::{PortalInit, SlotId, WarpOp, WarpTickPatchV1}; use crate::warp_state::WarpState; +#[cfg(test)] +#[path = "slot_reading/range_tests.rs"] +mod range_tests; + /// Position of one operation in a lane's patch sequence: tick index, then /// operation index within that tick's canonical operation order. type OpPosition = (usize, usize); @@ -35,6 +44,8 @@ pub enum SlotReading { Edge(Option), /// An attachment value, or `None` when the slot holds no attachment. Attachment(Option), + /// Canonically ordered matching edges; `None` means the source is absent. + OutgoingEdges(Option>), } /// Reasons a slot reading cannot be produced. @@ -57,6 +68,9 @@ pub enum SlotReadingError { /// The supplied patches are not the patches the index was built from. #[error("the supplied patches do not match the index")] PatchMismatch, + /// A range's selected historical prefix could not be replayed. + #[error("range history could not be replayed")] + InvalidRangeHistory, } #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -83,8 +97,8 @@ pub struct SlotWriteIndex { /// A [`SlotWriteIndex`] bound to the patches it was built from and the /// boundary those patches apply to. /// -/// Binding checks every patch digest once. Reads then cost no more than the -/// index lookup. +/// Binding checks every patch digest once. Point reads use the index; range +/// reads replay the selected prefix over an owned copy of the initial state. #[derive(Debug, Clone, Copy)] pub struct SlotReader<'a> { index: &'a SlotWriteIndex, @@ -212,7 +226,7 @@ impl SlotReader<'_> { /// /// Returns [`SlotReadingError`] when the cursor lies beyond the indexed /// ticks, the slot is a port slot, or an attachment key pairs an owner - /// with a plane that owner does not have. + /// with a plane that owner does not have, or a range history prefix cannot replay. pub fn read(&self, slot: SlotId, cursor: usize) -> Result { let Self { index, @@ -223,6 +237,18 @@ impl SlotReader<'_> { if cursor > available { return Err(SlotReadingError::CursorBeyondHistory { cursor, available }); } + if matches!(slot, SlotId::OutgoingEdges(_)) { + let mut state = initial.clone(); + for patch in patches + .get(..cursor) + .ok_or(SlotReadingError::PatchMismatch)? + { + patch + .apply_to_state(&mut state) + .map_err(|_| SlotReadingError::InvalidRangeHistory)?; + } + return materialized_slot(&state, slot).ok_or(SlotReadingError::UnsupportedSlot); + } if let SlotId::Attachment(key) = slot { if !key.is_plane_valid() { return Err(SlotReadingError::InvalidAttachmentKey); @@ -300,6 +326,9 @@ pub(crate) fn materialized_slot(state: &WarpState, slot: SlotId) -> Option return None, + SlotId::OutgoingEdges(range) => { + SlotReading::OutgoingEdges(materialized_outgoing_edges(state, range)) + } }) } @@ -308,6 +337,7 @@ const fn absent(slot: SlotId) -> SlotReading { SlotId::Node(_) => SlotReading::Node(None), SlotId::Edge(_) => SlotReading::Edge(None), SlotId::Attachment(_) | SlotId::Port(_) => SlotReading::Attachment(None), + SlotId::OutgoingEdges(_) => SlotReading::OutgoingEdges(None), } } @@ -320,9 +350,25 @@ const fn slot_warp(slot: SlotId) -> Option { AttachmentOwner::Edge(edge) => edge.warp_id, }), SlotId::Port(_) => None, + SlotId::OutgoingEdges(range) => Some(range.source.warp_id), } } +pub(crate) fn materialized_outgoing_edges( + state: &WarpState, + range: crate::footprint::OutgoingEdgeRange, +) -> Option> { + let store = state.store(&range.source.warp_id)?; + store.node(&range.source.local_id)?; + let mut rows: Vec<_> = store + .edges_from(&range.source.local_id) + .filter(|edge| edge.ty == range.edge_type) + .cloned() + .collect(); + rows.sort_by_key(|edge| edge.id); + Some(rows) +} + /// Slots whose value `op` sets, including values it clears as a consequence. fn written_slots(op: &WarpOp) -> [Option; 2] { match op { diff --git a/crates/warp-core/src/slot_reading/range_tests.rs b/crates/warp-core/src/slot_reading/range_tests.rs new file mode 100644 index 000000000..0e41571d8 --- /dev/null +++ b/crates/warp-core/src/slot_reading/range_tests.rs @@ -0,0 +1,164 @@ +// SPDX-License-Identifier: Apache-2.0 +// © James Ross Ω FLYING•ROBOTS +#![allow(clippy::expect_used, clippy::panic)] // Test assertions may abort the witness. +use super::*; +use crate::{GraphStore, NodeId, OutgoingEdgeRange, TickCommitStatus, TypeId, WarpInstance}; + +fn edge(id: u8, source: u8, ty: u8) -> WarpOp { + WarpOp::UpsertEdge { + warp_id: WarpId([1; 32]), + record: EdgeRecord { + id: crate::EdgeId([id; 32]), + from: NodeId([source; 32]), + to: NodeId([3; 32]), + ty: TypeId([ty; 32]), + }, + } +} + +#[test] +fn historical_range_reads_aggregate_insertions_relabels_moves_and_deletions() { + let warp_id = WarpId([1; 32]); + let mut store = GraphStore::new(warp_id); + for node in [2, 3, 4] { + store.insert_node( + NodeId([node; 32]), + NodeRecord { + ty: TypeId([8; 32]), + }, + ); + } + let mut initial = WarpState::new(); + initial.upsert_instance( + WarpInstance { + warp_id, + root_node: NodeId([2; 32]), + parent: None, + }, + store, + ); + let slot = SlotId::OutgoingEdges(OutgoingEdgeRange { + source: NodeKey { + warp_id, + local_id: NodeId([2; 32]), + }, + edge_type: TypeId([9; 32]), + }); + let ops = [ + vec![edge(22, 2, 9), edge(20, 2, 9), edge(21, 2, 10)], + vec![edge(20, 2, 10)], + vec![edge(22, 4, 9)], + vec![edge(24, 2, 9)], + vec![ + WarpOp::DeleteEdge { + warp_id, + from: NodeId([2; 32]), + edge_id: crate::EdgeId([20; 32]), + }, + WarpOp::DeleteEdge { + warp_id, + from: NodeId([2; 32]), + edge_id: crate::EdgeId([21; 32]), + }, + WarpOp::DeleteEdge { + warp_id, + from: NodeId([4; 32]), + edge_id: crate::EdgeId([22; 32]), + }, + WarpOp::DeleteEdge { + warp_id, + from: NodeId([2; 32]), + edge_id: crate::EdgeId([24; 32]), + }, + WarpOp::DeleteNode { + node: NodeKey { + warp_id, + local_id: NodeId([3; 32]), + }, + }, + ], + vec![WarpOp::DeleteNode { + node: NodeKey { + warp_id, + local_id: NodeId([2; 32]), + }, + }], + ]; + let patches: Vec<_> = ops + .into_iter() + .map(|ops| { + WarpTickPatchV1::new(1, [0; 32], TickCommitStatus::Committed, vec![], vec![], ops) + }) + .collect(); + let mut index = SlotWriteIndex::new(); + for patch in &patches { + index.push_patch(patch); + } + let reader = index.reader(&patches, &initial).expect("bound history"); + let expected = [ + Some(vec![]), + Some(vec![20, 22]), + Some(vec![22]), + Some(vec![]), + Some(vec![24]), + Some(vec![]), + None, + ]; + for (cursor, ids) in expected.into_iter().enumerate() { + let SlotReading::OutgoingEdges(rows) = reader.read(slot, cursor).expect("range reading") + else { + panic!("range value") + }; + assert_eq!( + rows.map(|rows| rows + .into_iter() + .map(|row| { + assert_eq!(row.from, NodeId([2; 32])); + assert_eq!(row.to, NodeId([3; 32])); + assert_eq!(row.ty, TypeId([9; 32])); + row.id.0[0] + }) + .collect::>()), + ids, + "cursor {cursor}" + ); + } + assert_eq!( + materialized_slot(&initial, slot), + Some(SlotReading::OutgoingEdges(Some(vec![]))) + ); +} + +#[test] +fn range_reading_refuses_a_prefix_that_cannot_replay() { + let warp_id = WarpId([1; 32]); + let initial = WarpState::new(); + let slot = SlotId::OutgoingEdges(OutgoingEdgeRange { + source: NodeKey { + warp_id, + local_id: NodeId([2; 32]), + }, + edge_type: TypeId([9; 32]), + }); + let patch = WarpTickPatchV1::new( + 1, + [0; 32], + TickCommitStatus::Committed, + vec![], + vec![], + vec![edge(20, 2, 9)], + ); + let patches = [patch]; + let mut index = SlotWriteIndex::new(); + index.push_patch(&patches[0]); + let reader = index.reader(&patches, &initial).expect("bound history"); + assert_eq!(reader.read(slot, 0), Ok(SlotReading::OutgoingEdges(None))); + assert_eq!( + reader.read(slot, 1), + Err(SlotReadingError::InvalidRangeHistory) + ); + assert_eq!( + materialized_slot(&initial, slot), + Some(SlotReading::OutgoingEdges(None)) + ); +} diff --git a/crates/warp-core/src/tick_patch.rs b/crates/warp-core/src/tick_patch.rs index 571efd773..34addb42c 100644 --- a/crates/warp-core/src/tick_patch.rs +++ b/crates/warp-core/src/tick_patch.rs @@ -56,6 +56,8 @@ pub enum SlotId { Attachment(AttachmentKey), /// Boundary port value (warp-scoped: `(WarpId, PortKey)`). Port(WarpScopedPortKey), + /// All outgoing edges of one type from one source, including an empty range. + OutgoingEdges(crate::footprint::OutgoingEdgeRange), } impl SlotId { @@ -65,6 +67,7 @@ impl SlotId { Self::Edge(_) => 2, Self::Attachment(_) => 3, Self::Port(_) => 4, + Self::OutgoingEdges(_) => 5, } } } @@ -86,6 +89,7 @@ impl Ord for SlotId { (Self::Edge(a), Self::Edge(b)) => a.cmp(&b), (Self::Attachment(a), Self::Attachment(b)) => a.cmp(&b), (Self::Port(a), Self::Port(b)) => a.cmp(&b), + (Self::OutgoingEdges(a), Self::OutgoingEdges(b)) => a.cmp(&b), // SAFETY: tag comparison above guarantees matching variants. _ => unreachable!("tag mismatch in SlotId::cmp"), } @@ -1027,6 +1031,12 @@ fn encode_slots(h: &mut Hasher, slots: &[SlotId]) { h.update(&warp_id.0); h.update(&port_key.to_le_bytes()); } + SlotId::OutgoingEdges(range) => { + h.update(&[5]); + h.update(range.source.warp_id.as_bytes()); + h.update(range.source.local_id.as_bytes()); + h.update(range.edge_type.as_bytes()); + } } } } diff --git a/crates/warp-core/src/witnessed_suffix.rs b/crates/warp-core/src/witnessed_suffix.rs index f2e43e657..4366d180b 100644 --- a/crates/warp-core/src/witnessed_suffix.rs +++ b/crates/warp-core/src/witnessed_suffix.rs @@ -1149,6 +1149,12 @@ fn hash_settlement_slot(hasher: &mut Hasher, slot: &SlotId) { hasher.update(warp_id.as_bytes()); hasher.update(&port_key.to_le_bytes()); } + SlotId::OutgoingEdges(range) => { + hasher.update(&[5]); + hasher.update(range.source.warp_id.as_bytes()); + hasher.update(range.source.local_id.as_bytes()); + hasher.update(range.edge_type.as_bytes()); + } } } diff --git a/crates/warp-core/tests/common/mod.rs b/crates/warp-core/tests/common/mod.rs index 9cd145366..4d49ca533 100644 --- a/crates/warp-core/tests/common/mod.rs +++ b/crates/warp-core/tests/common/mod.rs @@ -296,6 +296,8 @@ fn make_parallel_touch_rule() -> RewriteRule { })); } Footprint { + r_read: warp_core::OutgoingEdgeRangeSet::default(), + r_write: warp_core::OutgoingEdgeRangeSet::default(), n_read: warp_core::NodeSet::default(), n_write: warp_core::NodeSet::default(), e_read: warp_core::EdgeSet::default(), @@ -1020,6 +1022,8 @@ macro_rules! make_touch_rule { })); } warp_core::Footprint { + r_read: warp_core::OutgoingEdgeRangeSet::default(), + r_write: warp_core::OutgoingEdgeRangeSet::default(), n_read: warp_core::NodeSet::default(), n_write: warp_core::NodeSet::default(), e_read: warp_core::EdgeSet::default(), diff --git a/crates/warp-core/tests/parallel_footprints.rs b/crates/warp-core/tests/parallel_footprints.rs index 5babb9210..60218eb8a 100644 --- a/crates/warp-core/tests/parallel_footprints.rs +++ b/crates/warp-core/tests/parallel_footprints.rs @@ -130,6 +130,8 @@ mod enforcement { let mut n_read = NodeSet::default(); n_read.insert_with_warp(view.warp_id(), *scope); Footprint { + r_read: warp_core::OutgoingEdgeRangeSet::default(), + r_write: warp_core::OutgoingEdgeRangeSet::default(), n_read, n_write: NodeSet::default(), e_read: EdgeSet::default(), @@ -366,6 +368,8 @@ mod enforcement { let mut n_read = NodeSet::default(); n_read.insert_with_warp(view.warp_id(), *scope); Footprint { + r_read: warp_core::OutgoingEdgeRangeSet::default(), + r_write: warp_core::OutgoingEdgeRangeSet::default(), n_read, n_write: NodeSet::default(), e_read: EdgeSet::default(), @@ -425,6 +429,8 @@ mod enforcement { n_read.insert_with_warp(warp_id, *scope); n_write.insert_with_warp(warp_id, *scope); Footprint { + r_read: warp_core::OutgoingEdgeRangeSet::default(), + r_write: warp_core::OutgoingEdgeRangeSet::default(), n_read, n_write, e_read: EdgeSet::default(), @@ -471,6 +477,8 @@ mod enforcement { let mut n_read = NodeSet::default(); n_read.insert_with_warp(view.warp_id(), *scope); Footprint { + r_read: warp_core::OutgoingEdgeRangeSet::default(), + r_write: warp_core::OutgoingEdgeRangeSet::default(), n_read, n_write: NodeSet::default(), e_read: EdgeSet::default(), @@ -523,6 +531,8 @@ mod enforcement { let mut n_read = NodeSet::default(); n_read.insert_with_warp(view.warp_id(), *scope); Footprint { + r_read: warp_core::OutgoingEdgeRangeSet::default(), + r_write: warp_core::OutgoingEdgeRangeSet::default(), n_read, n_write: NodeSet::default(), e_read: EdgeSet::default(), @@ -589,6 +599,8 @@ mod enforcement { a_read.insert(key); a_write.insert(key); Footprint { + r_read: warp_core::OutgoingEdgeRangeSet::default(), + r_write: warp_core::OutgoingEdgeRangeSet::default(), n_read, n_write: NodeSet::default(), e_read: EdgeSet::default(), @@ -630,6 +642,8 @@ mod enforcement { let mut n_read = NodeSet::default(); n_read.insert_with_warp(view.warp_id(), *scope); Footprint { + r_read: warp_core::OutgoingEdgeRangeSet::default(), + r_write: warp_core::OutgoingEdgeRangeSet::default(), n_read, n_write: NodeSet::default(), e_read: EdgeSet::default(), @@ -678,6 +692,8 @@ mod enforcement { n_read.insert_with_warp(warp_id, *scope); e_write.insert_with_warp(warp_id, make_edge_id("t3-11-edge")); Footprint { + r_read: warp_core::OutgoingEdgeRangeSet::default(), + r_write: warp_core::OutgoingEdgeRangeSet::default(), n_read, n_write: NodeSet::default(), // Missing scope! e_read: EdgeSet::default(), @@ -735,6 +751,8 @@ mod enforcement { let mut n_read = NodeSet::default(); n_read.insert_with_warp(view.warp_id(), *scope); Footprint { + r_read: warp_core::OutgoingEdgeRangeSet::default(), + r_write: warp_core::OutgoingEdgeRangeSet::default(), n_read, n_write: NodeSet::default(), e_read: EdgeSet::default(), @@ -790,6 +808,8 @@ mod enforcement { let mut n_read = NodeSet::default(); n_read.insert_with_warp(view.warp_id(), *scope); Footprint { + r_read: warp_core::OutgoingEdgeRangeSet::default(), + r_write: warp_core::OutgoingEdgeRangeSet::default(), n_read, n_write: NodeSet::default(), e_read: EdgeSet::default(), @@ -853,6 +873,8 @@ mod enforcement { local_id: missing, })); Footprint { + r_read: warp_core::OutgoingEdgeRangeSet::default(), + r_write: warp_core::OutgoingEdgeRangeSet::default(), n_read, n_write, e_read: EdgeSet::default(), @@ -937,6 +959,8 @@ mod enforcement { // Intentionally OMIT a_write(node_alpha) — this should trigger enforcement Footprint { + r_read: warp_core::OutgoingEdgeRangeSet::default(), + r_write: warp_core::OutgoingEdgeRangeSet::default(), n_read, n_write, e_read: EdgeSet::default(), diff --git a/crates/warp-core/tests/provenance_retention_codec_tests.rs b/crates/warp-core/tests/provenance_retention_codec_tests.rs index 95d8c5f48..9c8ffe256 100644 --- a/crates/warp-core/tests/provenance_retention_codec_tests.rs +++ b/crates/warp-core/tests/provenance_retention_codec_tests.rs @@ -39,6 +39,10 @@ fn encoded_field_offset(bytes: &[u8], field: &[u8]) -> usize { } fn fixture_entry() -> ProvenanceEntry { + fixture_entry_with_ranges(false) +} + +fn fixture_entry_with_ranges(include_ranges: bool) -> ProvenanceEntry { let worldline_id = WorldlineId::from_bytes(digest("worldline")); let root_warp = make_warp_id("root"); let child_warp = make_warp_id("child"); @@ -122,22 +126,34 @@ fn fixture_entry() -> ProvenanceEntry { value: None, }, ]; + let mut in_slots = vec![ + SlotId::Node(root_node), + SlotId::Edge(edge_key), + SlotId::Attachment(root_attachment), + SlotId::Port((root_warp, 41)), + ]; + let mut out_slots = vec![ + SlotId::Node(sibling_node), + SlotId::Edge(edge_key), + SlotId::Attachment(edge_attachment), + SlotId::Port((root_warp, 42)), + ]; + if include_ranges { + in_slots.push(SlotId::OutgoingEdges(warp_core::OutgoingEdgeRange { + source: root_node, + edge_type, + })); + out_slots.push(SlotId::OutgoingEdges(warp_core::OutgoingEdgeRange { + source: sibling_node, + edge_type, + })); + } let patch = WarpTickPatchV1::new( 17, digest("rule-pack"), TickCommitStatus::Committed, - vec![ - SlotId::Node(root_node), - SlotId::Edge(edge_key), - SlotId::Attachment(root_attachment), - SlotId::Port((root_warp, 41)), - ], - vec![ - SlotId::Node(sibling_node), - SlotId::Edge(edge_key), - SlotId::Attachment(edge_attachment), - SlotId::Port((root_warp, 42)), - ], + in_slots, + out_slots, ops, ); let parent = ProvenanceRef { @@ -147,12 +163,6 @@ fn fixture_entry() -> ProvenanceEntry { }; let state_root = digest("state-root"); let patch_digest = patch.digest(); - let commit_hash = compute_commit_hash_v2( - &state_root, - &[parent.commit_hash], - &patch_digest, - patch.policy_id(), - ); let receipt = TickReceipt::try_from_retained_parts( TxId::from_raw(8), vec![TickReceiptEntry { @@ -164,6 +174,23 @@ fn fixture_entry() -> ProvenanceEntry { vec![Vec::new()], ) .expect("fixture receipt parts should be parallel"); + let receipt = if include_ranges { + receipt + .with_bound_attribution(vec![warp_core::TickSubmissionAttribution { + submission_id: digest("range-submission"), + decision: warp_core::TickSubmissionDecision::ReceiptEntries(vec![0]), + }]) + .expect("bound range receipt") + } else { + receipt + }; + let commit_hash = warp_core::compute_commit_hash_for_receipt( + &state_root, + &[parent.commit_hash], + &patch_digest, + patch.policy_id(), + Some(&receipt), + ); let decision_digest = receipt.digest(); ProvenanceEntry::local_commit( worldline_id, @@ -252,35 +279,52 @@ fn fixture_contract() -> ContractEvidenceIdentity { #[test] fn replayable_state_delta_round_trips_every_operation_and_slot_variant() { - let entry = fixture_entry(); - let receipt_digest = fixture_receipt_digest(&entry); - let contract = fixture_contract(); - let record = WalRuntimeStateDeltaRecord::from_provenance_entry( - receipt_digest, - Some(warp_core::InstalledInvocationEvidence::LegacyContract( - contract.clone(), - )), - entry.clone(), - ) - .expect("canonical local commit should be retainable"); - let bytes = record - .to_payload_bytes() - .expect("validated record should encode"); - let decoded = WalRuntimeStateDeltaRecord::from_payload_bytes(&bytes) - .expect("retained state delta should decode"); + for entry in [fixture_entry(), fixture_entry_with_ranges(true)] { + let receipt_digest = fixture_receipt_digest(&entry); + let contract = fixture_contract(); + let record = WalRuntimeStateDeltaRecord::from_provenance_entry( + receipt_digest, + Some(warp_core::InstalledInvocationEvidence::LegacyContract( + contract.clone(), + )), + entry.clone(), + ) + .expect("canonical local commit should be retainable"); + let bytes = record + .to_payload_bytes() + .expect("validated record should encode"); + let magic = if entry.tick_receipt.as_ref().expect("receipt").binds_commit() { + b"EPRV0003" + } else { + b"EPRV0001" + }; + assert_eq!(&bytes[..8], b"ERSD0001"); + let provenance_offset = encoded_field_offset(&bytes, magic); + let length_offset = provenance_offset + .checked_sub(8) + .expect("entry length prefix"); + let entry_length = u64::from_le_bytes( + bytes[length_offset..provenance_offset] + .try_into() + .expect("eight-byte entry length"), + ); + assert_eq!(entry_length, (bytes.len() - provenance_offset) as u64); + let decoded = WalRuntimeStateDeltaRecord::from_payload_bytes(&bytes) + .expect("retained state delta should decode"); - assert_eq!(decoded.receipt_digest(), receipt_digest); - assert_eq!( - decoded.contract(), - Some(&warp_core::InstalledInvocationEvidence::LegacyContract( - contract - )) - ); - assert_eq!(decoded.provenance_entry(), &entry); - assert_eq!( - decoded.to_payload_bytes().expect("decoded record encodes"), - bytes - ); + assert_eq!(decoded.receipt_digest(), receipt_digest); + assert_eq!( + decoded.contract(), + Some(&warp_core::InstalledInvocationEvidence::LegacyContract( + contract + )) + ); + assert_eq!(decoded.provenance_entry(), &entry); + assert_eq!( + decoded.to_payload_bytes().expect("decoded record encodes"), + bytes + ); + } } #[test] @@ -457,3 +501,37 @@ fn replayable_state_delta_rejects_corrupt_commitment_and_non_local_event() { Err(RetainedProvenanceError::UnsupportedEventKind) ); } + +#[test] +fn retained_range_substitution_refuses_the_original_patch_commitment() { + let entry = fixture_entry_with_ranges(true); + let record = WalRuntimeStateDeltaRecord::from_provenance_entry( + fixture_receipt_digest(&entry), + None, + entry.clone(), + ) + .expect("retained range entry"); + let mut bytes = record.to_payload_bytes().expect("range encoding"); + let patch = entry.patch.as_ref().expect("patch"); + let range = patch + .in_slots + .iter() + .find_map(|slot| match slot { + SlotId::OutgoingEdges(key) => Some(key), + _ => None, + }) + .expect("range slot"); + let key_bytes = [ + vec![5], + range.source.warp_id.as_bytes().to_vec(), + range.source.local_id.as_bytes().to_vec(), + range.edge_type.as_bytes().to_vec(), + ] + .concat(); + let offset = encoded_field_offset(&bytes, &key_bytes); + bytes[offset + 96] ^= 1; + assert_eq!( + WalRuntimeStateDeltaRecord::from_payload_bytes(&bytes), + Err(RetainedProvenanceError::NonCanonicalPatch) + ); +} diff --git a/crates/warp-core/tests/slice_theorem_proof.rs b/crates/warp-core/tests/slice_theorem_proof.rs index b2ed6b39a..cc5d24ae3 100644 --- a/crates/warp-core/tests/slice_theorem_proof.rs +++ b/crates/warp-core/tests/slice_theorem_proof.rs @@ -110,6 +110,8 @@ fn r1_footprint(view: GraphView<'_>, _scope: &NodeId) -> Footprint { local_id: node_id(1), })); Footprint { + r_read: warp_core::OutgoingEdgeRangeSet::default(), + r_write: warp_core::OutgoingEdgeRangeSet::default(), n_read, n_write: NodeSet::default(), e_read: EdgeSet::default(), @@ -166,6 +168,8 @@ fn r2_footprint(view: GraphView<'_>, _scope: &NodeId) -> Footprint { local_id: node_id(3), })); Footprint { + r_read: warp_core::OutgoingEdgeRangeSet::default(), + r_write: warp_core::OutgoingEdgeRangeSet::default(), n_read, n_write: NodeSet::default(), e_read: EdgeSet::default(), @@ -222,6 +226,8 @@ fn r3_footprint(view: GraphView<'_>, _scope: &NodeId) -> Footprint { local_id: node_id(5), })); Footprint { + r_read: warp_core::OutgoingEdgeRangeSet::default(), + r_write: warp_core::OutgoingEdgeRangeSet::default(), n_read, n_write: NodeSet::default(), e_read: EdgeSet::default(), @@ -291,6 +297,8 @@ fn r4_footprint(view: GraphView<'_>, _scope: &NodeId) -> Footprint { local_id: node_id(6), })); Footprint { + r_read: warp_core::OutgoingEdgeRangeSet::default(), + r_write: warp_core::OutgoingEdgeRangeSet::default(), n_read, n_write: NodeSet::default(), e_read: EdgeSet::default(), @@ -347,6 +355,8 @@ fn r5_footprint(view: GraphView<'_>, _scope: &NodeId) -> Footprint { local_id: node_id(8), })); Footprint { + r_read: warp_core::OutgoingEdgeRangeSet::default(), + r_write: warp_core::OutgoingEdgeRangeSet::default(), n_read, n_write: NodeSet::default(), e_read: EdgeSet::default(), @@ -398,6 +408,8 @@ fn r6_footprint(view: GraphView<'_>, scope: &NodeId) -> Footprint { let mut n_read = NodeSet::default(); n_read.insert_with_warp(warp_id, *scope); Footprint { + r_read: warp_core::OutgoingEdgeRangeSet::default(), + r_write: warp_core::OutgoingEdgeRangeSet::default(), n_read, n_write: NodeSet::default(), e_read: EdgeSet::default(), diff --git a/docs/spec/scheduler-warp-core.md b/docs/spec/scheduler-warp-core.md index d9ec7daf6..1d028b458 100644 --- a/docs/spec/scheduler-warp-core.md +++ b/docs/spec/scheduler-warp-core.md @@ -26,7 +26,7 @@ should admit a candidate. ## Decision 1: Pending rewrites carry explicit footprints -A pending rewrite carries scope hash, rule id, scope, footprint, and phase. The footprint is a conservative resource contract across nodes, edges, attachments, boundary ports, and factor mask. +A pending rewrite carries scope hash, rule id, scope, footprint, and phase. The footprint is a conservative resource contract across nodes, edges, outgoing-edge ranges, attachments, boundary ports, and factor mask. ## Decision 2: `reserve()` is check-then-mark @@ -36,6 +36,31 @@ Reservation checks all candidate resources for conflict and marks resources only A candidate conflicts when it writes a resource another admitted candidate reads or writes, or when its boundary port claim overlaps another admitted boundary port claim. Reads may overlap reads. +Outgoing-edge range claims use `(source NodeKey, edge TypeId)`, so they retain +the WARP, source node and edge type even when the observed range is empty. +`Footprint::independent` and the reservation index both check range reads +against range writes, and range writes against both reads and writes. Range +reads can share a reservation. Footprint union preserves these claims, and a +rejected reservation leaves no range marks behind. The admission-side affected +region retains the range write wake as well as point-resource writes. +`GraphStore::edges_from` exposes an exact-size iterator over its stored source +bucket. Its remaining length is available without advancing or inspecting any +edge, allowing a bounded reader to charge the whole bucket before type filtering. +This storage capability does not itself meter or authorize a read. + +This is explicit claim tracking. It does not infer range observations from +existing edge-ID sets or automatically instrument edge writers. Range-bearing +operation footprints use digest domain +`echo:operation-footprint:v2\0`, followed by the existing eight counted sets +and factor mask, then `rr` and `rw` counted sets. Each range row contains the +32-byte source WARP, source node and edge type in that order. Counts use +little-endian `u64`, and set rows follow key order. Footprints without ranges +retain the exact V1 digest stream. Range claims also project to retained input +and output slots under the [tick-patch contract](warp-tick-patch.md#outgoing-edge-range-slots). +A native Tick witness checks conflict rejection, blockers, range metadata and +replay. This is not an admitted authored-effect profile. No Edict program +currently receives graph mutation authority through this addition. + The broad WARP outcome algebra is `Derived | Plural | Conflict | Obstruction`. Echo's local `TickReceipt` entries realize the narrower tick-scale shape `Applied / Rejected(FootprintConflict | diff --git a/docs/spec/warp-tick-patch.md b/docs/spec/warp-tick-patch.md index 6d6c08f7c..ebb9a799a 100644 --- a/docs/spec/warp-tick-patch.md +++ b/docs/spec/warp-tick-patch.md @@ -31,7 +31,7 @@ Replay applies ops; it does not invoke matchers, executors, or scheduler decisio ## Decision 2: Slots are unversioned in the patch -Patch slots name locations: `Node(NodeKey)`, `Edge(EdgeKey)`, `Attachment(AttachmentKey)`, and `Port(PortKey)`. Version identity is interpreted from patch position in the worldline. +Patch slots name locations: `Node(NodeKey)`, `Edge(EdgeKey)`, `Attachment(AttachmentKey)`, `Port(PortKey)`, and `OutgoingEdges(OutgoingEdgeRange)`. Version identity is interpreted from patch position in the worldline. ## Decision 3: Ops are explicit carrier edits @@ -39,7 +39,7 @@ Patch v2 supports `OpenPortal`, instance upsert/delete, node upsert/delete, edge ## Decision 4: Canonical ordering is part of the format -Slot tag order is node, edge, attachment, port. Op replay order is open portals, upsert instances, delete instances, delete edges, delete nodes, upsert nodes, upsert edges, set attachments. +Slot tag order is node, edge, attachment, port, outgoing-edge range. Op replay order is open portals, upsert instances, delete instances, delete edges, delete nodes, upsert nodes, upsert edges, set attachments. ## Decision 5: `patch_digest` commits to the patch core @@ -48,3 +48,33 @@ Slot tag order is node, edge, attachment, port. Op replay order is open portals, ## Slicing contract For a target slot, find the latest producer patch, add that patch, enqueue its `in_slots`, and repeat until the producer frontier closes or reaches the boundary state. Over-approximating `in_slots` is acceptable; under-approximating them is not. + +## Outgoing-edge range slots + +An outgoing-edge range key contains the source `NodeKey` and edge `TypeId`. +Slot tag 5 is followed by the 32-byte source WARP, 32-byte source node and +32-byte edge type, in that order, both in the patch digest stream and retained +provenance slot encoding. Existing point-slot bytes and digest inputs are +unchanged. A reader without range-tag support refuses the new tag rather than +interpreting it as a point slot. + +A declared range read contributes an input slot. A declared range write +contributes both input and output slots, so tracing a range's next producer +also traces the prior aggregate. The engine's blocker reconstruction compares +these same range claims when explaining a scheduler rejection. A rejected +candidate contributes no range outputs. + +The host slot reader returns matching full edge records sorted by edge ID. +An existing source with no matching edges is `Some([])`; an absent source is +`None`. A producer that depends on source presence must also record the source +node slot; the range key does not replace that point dependency. A range is an aggregate, so the point index's latest absolute write +cannot reconstruct it. Historical range reading currently clones the initial +state and replays the selected prefix. This includes relabeling, moving and +removing edges, and source deletion. Invalid replay history returns +`InvalidRangeHistory`. The borrowed initial state is never mutated. + +That host reading has full-prefix replay and state-copy cost. It is not the +bounded, metered read aperture of an interpreted Edict operation. Native Tick +and retained-codec witnesses establish declared claim retention and replay; +they do not establish the authored operation effect bridge or automatically +record range reads and writes in existing graph accessors. diff --git a/docs/topics/GeneratedRules.md b/docs/topics/GeneratedRules.md index 8b6c4ba7a..c9cbc446d 100644 --- a/docs/topics/GeneratedRules.md +++ b/docs/topics/GeneratedRules.md @@ -336,6 +336,13 @@ but it still needs a deliberately false-footprint negative oracle before it may claim release qualification. Wesley still needs both a package emitter and its own positive installed generated-pack witness. +The scheduler can reserve explicit outgoing-edge range claims, including empty +ranges, as described in the [scheduler contract](../spec/scheduler-warp-core.md). +This claim API does not instrument existing graph reads or edge writers. Range +keys survive patch identity, retained provenance encoding and native Tick +metadata. This does not admit a new authored-effect profile or qualify a +generated package for release. + Conflicts remain explicit receipt rejections. A footprint failure must not trigger hidden retries or widen access.