diff --git a/CHANGELOG.md b/CHANGELOG.md index af1e4e9b..998b054c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -33,6 +33,8 @@ ### Fixed +- A worldline frontier retains the state root computed by its last executable-operation Action commit and returns it without hashing until the graph is next accessed mutably. An Action tick after the first now hashes the whole reachable state once instead of twice. Debug builds hash again to check each retained root, so they do not get the saving. The graph field of `WorldlineState` is private to its module so that every mutation goes through the accessor that drops the retained root. + - A filesystem WAL writer that takes over an existing log syncs its segment files before reading them as recovered history. A record that an earlier writer wrote and never synced was readable after reopen and was reported as a durable acceptance; a later power loss could still lose it. - A scheduler pass that commits executable-operation Actions hashes the worldline's whole reachable state twice instead of three times. The root before the commit is the root the pass already computed for the frontier basis; the commit no longer computes it again. diff --git a/crates/warp-core/src/coordinator.rs b/crates/warp-core/src/coordinator.rs index 62dca774..08484db8 100644 --- a/crates/warp-core/src/coordinator.rs +++ b/crates/warp-core/src/coordinator.rs @@ -7751,7 +7751,10 @@ mod tests { { let frontier = runtime.worldlines.frontier_mut(&worldline_b).unwrap(); let broken_root = frontier.state.root.warp_id; - assert!(frontier.state.warp_state.delete_instance(&broken_root)); + assert!(frontier + .state + .warp_state_mut() + .delete_instance(&broken_root)); } SchedulerCoordinator::super_tick(&mut runtime, &mut provenance, &mut engine).unwrap_err(); @@ -7842,7 +7845,10 @@ mod tests { { let frontier = runtime.worldlines.frontier_mut(&worldline_b).unwrap(); let broken_root = frontier.state.root.warp_id; - assert!(frontier.state.warp_state.delete_instance(&broken_root)); + assert!(frontier + .state + .warp_state_mut() + .delete_instance(&broken_root)); } let mut provenance = mirrored_provenance(&runtime); @@ -7999,7 +8005,10 @@ mod tests { { let frontier = runtime.worldlines.frontier_mut(&worldline_b).unwrap(); let broken_root = frontier.state.root.warp_id; - assert!(frontier.state.warp_state.delete_instance(&broken_root)); + assert!(frontier + .state + .warp_state_mut() + .delete_instance(&broken_root)); } let mut provenance = mirrored_provenance(&runtime); @@ -8088,7 +8097,10 @@ mod tests { { let frontier = runtime.worldlines.frontier_mut(&worldline_b).unwrap(); let broken_root = frontier.state.root.warp_id; - assert!(frontier.state.warp_state.delete_instance(&broken_root)); + assert!(frontier + .state + .warp_state_mut() + .delete_instance(&broken_root)); } let mut provenance = mirrored_provenance(&runtime); @@ -8167,7 +8179,10 @@ mod tests { { let frontier = runtime.worldlines.frontier_mut(&worldline_b).unwrap(); let broken_root = frontier.state.root.warp_id; - assert!(frontier.state.warp_state.delete_instance(&broken_root)); + assert!(frontier + .state + .warp_state_mut() + .delete_instance(&broken_root)); } let mut provenance = mirrored_provenance(&runtime); diff --git a/crates/warp-core/src/echo_operation.rs b/crates/warp-core/src/echo_operation.rs index c8fcd5ad..cdd705c4 100644 --- a/crates/warp-core/src/echo_operation.rs +++ b/crates/warp-core/src/echo_operation.rs @@ -6299,8 +6299,11 @@ pub(crate) fn commit_scheduler_action_batch_to_state_v1( .map(u32::try_from) .collect::, _>>() .map_err(|_| EchoOperationCommitErrorV1::TooManyCandidates)?; - patch.apply_journaled(&mut state.warp_state, journal)?; + state.apply_patch_journaled(&patch, journal)?; let state_root_after = state.state_root(); + // The next reading of this frontier's root, for example the basis of the + // next tick, does not hash the state again. + state.retain_state_root(state_root_after); let parents = state .last_snapshot .as_ref() @@ -6618,12 +6621,12 @@ pub(crate) fn commit_prepared_to_state( vec![Vec::new()], ); // Nothing after this application can fail, so the journal is not kept. - prepared.patch.apply_journaled( - &mut state.warp_state, + state.apply_patch_journaled( + &prepared.patch, &mut crate::tick_patch::StateJournal::default(), )?; - let state_root_after = - crate::snapshot::compute_state_root_for_warp_state(&state.warp_state, state.root()); + let state_root_after = state.state_root(); + state.retain_state_root(state_root_after); let parents = state .last_snapshot .as_ref() @@ -8618,7 +8621,7 @@ mod tests { ); state - .warp_state + .warp_state_mut() .store_mut(&root_warp) .expect("the fixture retains its root store") .set_node_attachment( diff --git a/crates/warp-core/src/echo_operation/observed.rs b/crates/warp-core/src/echo_operation/observed.rs index 79ad2204..b4ccd671 100644 --- a/crates/warp-core/src/echo_operation/observed.rs +++ b/crates/warp-core/src/echo_operation/observed.rs @@ -285,7 +285,10 @@ mod tests { local_id: crate::make_node_id("unrelated"), } }; - let store = state.warp_state.store_mut(&changed.warp_id).expect("store"); + let store = state + .warp_state_mut() + .store_mut(&changed.warp_id) + .expect("store"); store.insert_node( changed.local_id, NodeRecord { @@ -336,7 +339,10 @@ mod tests { local_id: crate::make_node_id("second-observation"), }; for length in [1_900, 2_000] { - let store = state.warp_state.store_mut(&first.warp_id).expect("store"); + let store = state + .warp_state_mut() + .store_mut(&first.warp_id) + .expect("store"); store.insert_node( second.local_id, NodeRecord { @@ -391,7 +397,7 @@ mod tests { let node = *state.root(); for length in [0, 23, 24, 255, 256, 512] { state - .warp_state + .warp_state_mut() .store_mut(&node.warp_id) .expect("store") .set_node_attachment( @@ -413,7 +419,7 @@ mod tests { let (_, mut state, _, _, _, _) = super::super::tests::projected_create_fixture(1_024); let node = *state.root(); state - .warp_state + .warp_state_mut() .store_mut(&node.warp_id) .expect("store") .set_node_attachment( diff --git a/crates/warp-core/src/engine_impl.rs b/crates/warp-core/src/engine_impl.rs index 7e80d909..d7004ac5 100644 --- a/crates/warp-core/src/engine_impl.rs +++ b/crates/warp-core/src/engine_impl.rs @@ -555,7 +555,7 @@ struct RuntimeCommitStateGuard<'a> { impl<'a> RuntimeCommitStateGuard<'a> { fn enter(engine: &'a mut Engine, state: &'a mut WorldlineState) -> Self { let saved_engine_state = - std::mem::replace(&mut engine.state, std::mem::take(&mut state.warp_state)); + std::mem::replace(&mut engine.state, std::mem::take(state.warp_state_mut())); let state_before_runtime_commit = engine.state.clone(); let saved_state_root = state.root; let saved_root = engine.current_root; @@ -614,7 +614,7 @@ impl<'a> RuntimeCommitStateGuard<'a> { } fn finish_success(&mut self) { - *self.state = WorldlineState { + *self.state = WorldlineState::from_parts(crate::worldline_state::WorldlineStateParts { warp_state: std::mem::replace( &mut self.engine.state, self.saved_engine_state @@ -650,9 +650,7 @@ impl<'a> RuntimeCommitStateGuard<'a> { committed_ingress: self .committed_ingress .take("runtime commit guard missing committed ingress"), - #[cfg(any(test, feature = "host_test"))] - state_root_computations: crate::worldline_state::TestCallCount::default(), - }; + }); self.engine.current_root = self.saved_root; self.engine.tx_counter = self.saved_tx_counter; self.engine.scheduler = self @@ -670,7 +668,7 @@ impl<'a> RuntimeCommitStateGuard<'a> { return; } - *self.state = WorldlineState { + *self.state = WorldlineState::from_parts(crate::worldline_state::WorldlineStateParts { warp_state: self .state_before_runtime_commit .take("runtime commit guard missing pre-commit state"), @@ -704,9 +702,7 @@ impl<'a> RuntimeCommitStateGuard<'a> { committed_ingress: self .committed_ingress .take("runtime commit guard missing committed ingress"), - #[cfg(any(test, feature = "host_test"))] - state_root_computations: crate::worldline_state::TestCallCount::default(), - }; + }); self.engine.state = self .saved_engine_state .take("runtime commit guard missing saved engine state"); @@ -2261,7 +2257,7 @@ impl Engine { /// Returns a snapshot view over an external worldline frontier state. #[must_use] pub fn snapshot_for_state(&self, state: &WorldlineState) -> Snapshot { - let state_root = compute_state_root(&state.warp_state, &state.root); + let state_root = compute_state_root(state.warp_state(), &state.root); let parents: Vec = state .last_snapshot .as_ref() diff --git a/crates/warp-core/src/trusted_runtime_host.rs b/crates/warp-core/src/trusted_runtime_host.rs index 7991ff70..c85a69af 100644 --- a/crates/warp-core/src/trusted_runtime_host.rs +++ b/crates/warp-core/src/trusted_runtime_host.rs @@ -7137,7 +7137,7 @@ mod tests { let mut node_occupied_parent = parent_state.clone(); node_occupied_parent - .warp_state + .warp_state_mut() .store_mut(&node.warp_id) .expect("the fixture retains its descendant store") .insert_node( @@ -7158,7 +7158,7 @@ mod tests { let mut attachment_occupied_parent = parent_state.clone(); attachment_occupied_parent - .warp_state + .warp_state_mut() .store_mut(&node.warp_id) .expect("the fixture retains its descendant store") .set_node_attachment( @@ -7180,7 +7180,7 @@ mod tests { let mut fully_occupied_parent = node_occupied_parent; fully_occupied_parent - .warp_state + .warp_state_mut() .store_mut(&node.warp_id) .expect("the fixture retains its descendant store") .set_node_attachment( diff --git a/crates/warp-core/src/worldline.rs b/crates/warp-core/src/worldline.rs index 1d4b7c2a..ba265a68 100644 --- a/crates/warp-core/src/worldline.rs +++ b/crates/warp-core/src/worldline.rs @@ -127,7 +127,7 @@ impl WorldlineTickPatchV1 { }); } - apply_ops_to_state(&mut state.warp_state, &self.ops)?; + apply_ops_to_state(state.warp_state_mut(), &self.ops)?; Ok(()) } } @@ -408,7 +408,7 @@ mod tests { }; state - .warp_state + .warp_state_mut() .store_mut(&warp_id) .expect("root store missing") .insert_node( @@ -466,7 +466,7 @@ mod tests { let ty = make_type_id("EdgeType"); let store = state - .warp_state + .warp_state_mut() .store_mut(&warp_id) .expect("root store missing"); store.insert_node(from_id, NodeRecord { ty }); @@ -589,7 +589,7 @@ mod tests { let ty = make_type_id("TestType"); state - .warp_state + .warp_state_mut() .store_mut(&warp_id) .expect("root store missing") .insert_node(node_id, NodeRecord { ty }); diff --git a/crates/warp-core/src/worldline_state.rs b/crates/warp-core/src/worldline_state.rs index 9117ab72..3884a2a1 100644 --- a/crates/warp-core/src/worldline_state.rs +++ b/crates/warp-core/src/worldline_state.rs @@ -81,7 +81,9 @@ pub enum WorldlineStateError { #[derive(Debug, Clone)] pub struct WorldlineState { /// The underlying multi-instance warp state. - pub(crate) warp_state: WarpState, + /// Private to this module: every mutation goes through + /// [`Self::warp_state_mut`], which drops the retained state root. + warp_state: WarpState, /// Root key for snapshot hashing and commit execution. pub(crate) root: NodeKey, /// Initial worldline state preserved for replay. @@ -107,9 +109,27 @@ pub struct WorldlineState { /// restarts; Phase 3 intentionally keeps lifetime idempotence rather than a /// bounded replay horizon. pub(crate) committed_ingress: BTreeSet<(WriterHeadKey, Hash)>, - /// Counts whole-state root computations for tests. + /// The state root of `warp_state` under the given root key, when it is + /// known without hashing. A commit retains the root it computed; any + /// mutable access to the graph drops it. + retained_state_root: Option<(NodeKey, Hash)>, + /// Counts whole-state root computations for tests, excluding the hash by + /// which a debug build checks a retained root. #[cfg(any(test, feature = "host_test"))] - pub(crate) state_root_computations: TestCallCount, + state_root_computations: TestCallCount, +} + +/// The fields from which a worldline state is assembled outside this module. +pub(crate) struct WorldlineStateParts { + pub(crate) warp_state: WarpState, + pub(crate) root: NodeKey, + pub(crate) initial_state: Arc, + pub(crate) last_snapshot: Option, + pub(crate) tick_history: Vec<(Snapshot, TickReceipt, WarpTickPatchV1)>, + pub(crate) last_materialization: Vec, + pub(crate) last_materialization_errors: Vec, + pub(crate) tx_counter: u64, + pub(crate) committed_ingress: BTreeSet<(WriterHeadKey, Hash)>, } /// Test instrumentation that counts calls. @@ -179,6 +199,7 @@ impl WorldlineState { last_materialization_errors: Vec::new(), tx_counter: 0, committed_ingress: BTreeSet::new(), + retained_state_root: None, #[cfg(any(test, feature = "host_test"))] state_root_computations: TestCallCount::default(), } @@ -299,14 +320,76 @@ impl WorldlineState { /// Equal state roots therefore do not prove equal stores. A committed /// detached write remains bound by the tick patch digest and commit identity. /// - /// Every call hashes the whole reachable state. + /// A call hashes the whole reachable state, unless the state holds a root + /// retained by its last commit: then the retained root is returned + /// without hashing. Debug builds hash in that case as well, to check the + /// retained root. #[must_use] pub fn state_root(&self) -> Hash { + if let Some((root, state_root)) = self.retained_state_root { + if root == self.root { + debug_assert_eq!( + state_root, + crate::snapshot::compute_state_root_for_warp_state( + &self.warp_state, + &self.root + ), + "the retained state root must be the root of the current graph" + ); + return state_root; + } + } #[cfg(any(test, feature = "host_test"))] self.state_root_computations.increment(); crate::snapshot::compute_state_root_for_warp_state(&self.warp_state, &self.root) } + /// Returns the graph for mutation and drops the retained state root. + pub(crate) fn warp_state_mut(&mut self) -> &mut WarpState { + self.retained_state_root = None; + &mut self.warp_state + } + + /// Retains `state_root` as the root of the current graph. + /// + /// The caller has just computed it from this state. Later calls to + /// [`Self::state_root`] return it without hashing until the graph is + /// accessed mutably or the root key changes. + pub(crate) fn retain_state_root(&mut self, state_root: Hash) { + self.retained_state_root = Some((self.root, state_root)); + } + + /// Applies `patch` to the graph, recording in `journal` what is needed to + /// take it back. On error the state, including its retained root, is + /// unchanged. + pub(crate) fn apply_patch_journaled( + &mut self, + patch: &WarpTickPatchV1, + journal: &mut StateJournal, + ) -> Result<(), crate::tick_patch::TickPatchError> { + patch.apply_journaled(&mut self.warp_state, journal)?; + self.retained_state_root = None; + Ok(()) + } + + /// Assembles a state from parts built outside this module. + pub(crate) fn from_parts(parts: WorldlineStateParts) -> Self { + Self { + warp_state: parts.warp_state, + root: parts.root, + initial_state: parts.initial_state, + last_snapshot: parts.last_snapshot, + tick_history: parts.tick_history, + last_materialization: parts.last_materialization, + last_materialization_errors: parts.last_materialization_errors, + tx_counter: parts.tx_counter, + committed_ingress: parts.committed_ingress, + retained_state_root: None, + #[cfg(any(test, feature = "host_test"))] + state_root_computations: TestCallCount::default(), + } + } + /// Resets test instrumentation for whole-state root computations. #[cfg(any(test, feature = "host_test"))] pub fn reset_state_root_computation_count_for_test(&self) { @@ -315,6 +398,10 @@ impl WorldlineState { /// Returns the number of whole-state root computations on this state /// since the last instrumentation reset. + /// + /// It counts the computations that every build performs. It does not + /// count the extra hash by which a debug build checks a retained root, so + /// the count is the same in debug and release builds. #[cfg(any(test, feature = "host_test"))] #[must_use] pub fn state_root_computation_count_for_test(&self) -> usize { @@ -407,6 +494,7 @@ impl WorldlineState { last_materialization_errors: self.last_materialization_errors.clone(), tx_counter: self.tx_counter, committed_ingress: BTreeSet::new(), + retained_state_root: self.retained_state_root, #[cfg(any(test, feature = "host_test"))] state_root_computations: TestCallCount::default(), } @@ -426,6 +514,7 @@ impl WorldlineState { last_materialization_errors: self.last_materialization_errors.clone(), tx_counter: self.tx_counter, committed_ingress: BTreeSet::new(), + retained_state_root: self.retained_state_root, #[cfg(any(test, feature = "host_test"))] state_root_computations: TestCallCount::default(), } @@ -447,6 +536,7 @@ impl WorldlineState { last_materialization_errors: Vec::new(), tx_counter: 0, committed_ingress: BTreeSet::new(), + retained_state_root: None, #[cfg(any(test, feature = "host_test"))] state_root_computations: TestCallCount::default(), } @@ -746,6 +836,57 @@ mod tests { assert_eq!(frontier.frontier_tick(), WorldlineTick::from_raw(42)); } + #[test] + fn a_retained_state_root_is_dropped_by_any_mutable_graph_access() { + let mut state = WorldlineState::empty(); + let root = state.state_root(); + assert_eq!(state.state_root_computation_count_for_test(), 1); + + state.retain_state_root(root); + assert_eq!(state.state_root(), root); + assert_eq!( + state.state_root_computation_count_for_test(), + 1, + "a retained root is returned without hashing" + ); + assert_eq!(state.clone().state_root_computation_count_for_test(), 1); + + let node = *state.root(); + if let Some(store) = state.warp_state_mut().store_mut(&node.warp_id) { + store.insert_node( + crate::ident::make_node_id("retained-root-extra"), + crate::record::NodeRecord { + ty: crate::ident::make_type_id("retained-root-extra"), + }, + ); + } + let _ = state.state_root(); + assert_eq!( + state.state_root_computation_count_for_test(), + 2, + "mutable access to the graph drops the retained root" + ); + } + + #[test] + fn a_retained_state_root_is_bound_to_the_root_key() { + let mut state = WorldlineState::empty(); + let root = state.state_root(); + state.retain_state_root(root); + state.reset_state_root_computation_count_for_test(); + + state.root = NodeKey { + warp_id: state.root.warp_id, + local_id: crate::ident::make_node_id("retained-root-other-key"), + }; + let _ = state.state_root(); + assert_eq!( + state.state_root_computation_count_for_test(), + 1, + "a root retained under another root key is not reused" + ); + } + #[test] fn try_from_warp_state() { let result = WorldlineState::try_from(WorldlineState::empty().warp_state().clone()); diff --git a/crates/warp-core/tests/edict_projected_cas_tests.rs b/crates/warp-core/tests/edict_projected_cas_tests.rs index 9b3f7d6c..c16bccdb 100644 --- a/crates/warp-core/tests/edict_projected_cas_tests.rs +++ b/crates/warp-core/tests/edict_projected_cas_tests.rs @@ -363,6 +363,66 @@ fn compiler_cas_tick_wal_failure_restores_exact_runtime_and_provenance() { } } +#[cfg(feature = "host_test")] +#[test] +fn compiler_cas_tick_hashes_the_state_once_after_the_first_commit() { + let dir = WalDir::new(); + let mut fixture = Fixture::new(true); + fixture.enable_wal(&dir); + fixture.install(); + fixture + .host + .install_echo_operation_action_admission_policy_v1(policy()); + let submit = |fixture: &mut Fixture, current: &[u8], next: &str| { + let digest = echo_operation_atom_value_digest_v1(attachment_type(), current); + let envelope = echo_operation_action_envelope_v1( + IngressTarget::ExactHead { key: fixture.head }, + fixture.invocation(current, digest, digest, next), + ) + .expect("Action envelope"); + fixture + .host + .app() + .submit_intent_with_runtime_wal_ack(envelope) + .expect("durable Action acknowledgement"); + }; + let root_computations = |fixture: &Fixture| { + fixture + .host + .runtime() + .worldlines() + .get(&fixture.head.worldline_id) + .expect("the worldline remains live") + .state() + .state_root_computation_count_for_test() + }; + submit(&mut fixture, INITIAL, "first"); + fixture.host.tick_once().expect("first Action commits"); + fixture + .host + .runtime() + .worldlines() + .get(&fixture.head.worldline_id) + .expect("the worldline remains live") + .state() + .reset_state_root_computation_count_for_test(); + + submit(&mut fixture, b"first", "second"); + fixture.host.tick_once().expect("second Action commits"); + assert_eq!(fixture.value(), b"second"); + assert_eq!( + root_computations(&fixture), + 1, + "the frontier basis reuses the root the previous commit computed; \ + only the root after this commit is hashed (the count excludes the \ + hash by which a debug build checks the retained root)" + ); + + submit(&mut fixture, b"second", "third"); + fixture.host.tick_once().expect("third Action commits"); + assert_eq!(root_computations(&fixture), 2); +} + fn initial_digest() -> [u8; 32] { echo_operation_atom_value_digest_v1(attachment_type(), INITIAL) } diff --git a/docs/architecture/sliced-readings-and-copy-on-write-strands.md b/docs/architecture/sliced-readings-and-copy-on-write-strands.md index 81325ad4..1990cc3c 100644 --- a/docs/architecture/sliced-readings-and-copy-on-write-strands.md +++ b/docs/architecture/sliced-readings-and-copy-on-write-strands.md @@ -158,12 +158,21 @@ The runtime does not implement this design. replay and registers the child with an independent copy. - `ReplayCheckpoint` stores a complete materialized graph for one tick. - Canonical state hashing and state comparison visit the graph. Each admitted - tick computes a state root over the whole reachable state. + tick computes a state root over the whole reachable state. A tick of + executable-operation Actions does so once, after its commit: the frontier + retains that root until its graph is next mutated, and the next tick reads + it for its basis without hashing. The first tick on a frontier, and a tick + that runs native rules through the engine, hash more than once. So does + every tick of a debug build, which hashes again to check each retained + root. - A native rule's matcher and footprint function read a view over a complete instance. - Tick patches record their input and output slots, and `slice_worldline_indices` computes the tick indices in a slot's dependency cone. Only tests call it. +- `SlotWriteIndex` and `SlotReader` resolve the value of one slot at a tick + cursor from the patch history, without replay. The runtime does not + maintain or use them yet. - Wormhole remains a reserved term for tick-range compression. Implementation state is owned by