From d4d19eaa210bf6c921dff3aa012667ac7ec9969d Mon Sep 17 00:00:00 2001 From: Dashiel Lopez Mendez Date: Fri, 28 Aug 2026 23:41:19 -0400 Subject: [PATCH 01/14] feat(journal): opt-in retention via RetainableJournal::truncate_before A log that only grows cannot satisfy a retention policy or a right-to-erasure obligation. `truncate_before(stream, at)` drops the oldest records in a stream, and the snapshot machinery that makes it safe already exists. Retention is a capability, not a universal contract -- an append-only store may have no way to drop records, and some domains forbid it -- so this is a supertrait rather than part of `Journal`. Two things it must get right, both now enforced: - **Truncation never renumbers.** A `Seq` is an identity, and subscriptions hold high-water marks that refer to it, so retained records keep their sequence numbers. `MemoryJournal`'s streams carry a `truncated` base for this; a `Seq` below the horizon reports `UnknownSeq` (the record existed, but this journal can no longer answer for it) rather than silently returning a neighbour. - **Truncation is refused when it would strand the stream.** Replay resumes from the newest snapshot and reads every record after it, so truncating past that point would leave the stream unresumable. The new `JournalError::NoSnapshotForTruncation` names the snapshot it found and the one it needed. Contract property 11 -- truncation preserves resume -- runs via `run_contract_retainable`, reachable from `journal_contract_test!` with a `retainable` marker: after truncating at a snapshot boundary the stream must resume to a bit-identical aggregate at the same entropy position, and report the new horizon. Also records the decision to defer the hash-chain proposal and let the adopter who needs it build it downstream first, with the design constraints that evaluation surfaced -- notably that a `Journal` decorator is unreachable from an async adapter, so any upstreamed version must be pure functions over bytes the adapter already persists. Co-Authored-By: Claude Opus 5 --- AGENTS.md | 17 ++++ app/crates/ironstate-journal/src/contract.rs | 83 ++++++++++++++- app/crates/ironstate-journal/src/journal.rs | 48 ++++++++- app/crates/ironstate-journal/src/lib.rs | 5 +- app/crates/ironstate-journal/src/macros.rs | 21 +++- app/crates/ironstate-journal/src/memory.rs | 102 ++++++++++++++++--- 6 files changed, 256 insertions(+), 20 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 923bf66..304cea3 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -181,6 +181,23 @@ says why it isn't there and what would change that. - **Kani for aggregates** — state-space explosion over struct state needs its own design. +- **Hash-chained append / per-entry tamper evidence.** `replay_hash` proves the + *outcome* (a terminal `AuditDigest` over final state); it cannot answer "was + entry 4,102 altered?". A per-entry chain can. It is deferred, and deliberately + being built **downstream first** by the adopter who needs it, for three + reasons. The claim is only worth its verification, and a chain is evidence + only when its root is *externally anchored* — anchoring is application + infrastructure, so ironstate would ship the half that does not deliver the + property alone. Of the two known consumers only one has the requirement; the + other has no audit machinery and could not consume it anyway. And the obvious + shape is wrong: a `Journal` decorator is unreachable from an async adapter, so + if this is ever upstreamed it must be **pure functions** — + `link_hash(prev, seq, encoded_events, entropy_pos)` plus a `verify_chain` over + an iterator — hashing bytes the adapter already persists, per *batch* (one + record), and carrying a per-link algorithm tag so the chain survives an + algorithm migration. *Activates when a downstream chain has proven the design + against a real anchor.* + **Out of scope — downstream, or unneeded.** - **Storage adapters** (Postgres, SQLite, …) — still downstream, written against diff --git a/app/crates/ironstate-journal/src/contract.rs b/app/crates/ironstate-journal/src/contract.rs index 2488109..c0657a8 100644 --- a/app/crates/ironstate-journal/src/contract.rs +++ b/app/crates/ironstate-journal/src/contract.rs @@ -11,7 +11,8 @@ //! [`run_contract_forkable`]. use crate::journal::{ - ExecuteError, ForkableJournal, Journal, JournalError, Seq, Snapshot, StreamId, VersionedEvent, + ExecuteError, ForkableJournal, Journal, JournalError, RetainableJournal, Seq, Snapshot, + StreamId, VersionedEvent, }; use crate::memory::MemoryJournal; use crate::replay::{execute, replay, resume}; @@ -156,6 +157,29 @@ where } } +/// Run the retention property against a journal that can truncate. +/// +/// # Panics +/// +/// Panics with a property-numbered message if truncating at a snapshot boundary +/// changes what the stream resumes to. +pub fn run_contract_retainable(cases: u32, max_steps: usize, seed_base: u64) +where + J: ContractJournal + RetainableJournal + for<'a> Journal = ()>, + A: AggregateArbitrary + StableHash, + A::Ctx: CtxEntropy, +{ + let stream = test_stream(); + let mut runner = seeded_runner(seed_base); + for case in 0..cases { + let genesis = sample(A::initial_state_strategy(), &mut runner); + let seed = run_seed(seed_base, case); + let (mut journal, _live, _steps) = + drive::(genesis.clone(), &seed, &mut runner, max_steps); + property_11_truncation_preserves_resume(&mut journal, &stream, &seed, case); + } +} + /// Run every property, including the two that need [`ForkableJournal`]. /// /// # Panics @@ -397,6 +421,63 @@ where } } +/// Property 11 — **truncation preserves resume**. After truncating at a +/// snapshot boundary, the stream resumes to a bit-identical aggregate and the +/// same entropy position. +/// +/// This is what makes retention safe to offer at all: dropping history must be +/// invisible to everything downstream of the snapshot it was taken against. +fn property_11_truncation_preserves_resume( + journal: &mut J, + stream: &StreamId, + seed: &Seed, + case: u32, +) where + J: RetainableJournal + for<'a> Journal = ()>, + A: AggregateRules + Clone + StableHash, +{ + let Some(head) = journal.head(stream) else { + return; + }; + let Ok((before, entropy_before)) = resume::(journal, stream, seed) else { + return; + }; + + // Snapshot the resumed state at the head, then drop everything before the + // midpoint — a truncation the snapshot fully covers. + let snapshot = Snapshot { + state: before.state().clone(), + schema_version: 0, + at: head, + entropy_pos: entropy_before.draws(), + }; + journal + .snapshot_in(&mut (), stream, snapshot) + .expect("snapshot"); + + let at = Seq(head.0.div_ceil(2).max(1)); + journal + .truncate_before(stream, at) + .expect("truncating below a snapshot taken at the head must be allowed"); + + let (after, entropy_after) = resume::(journal, stream, seed).expect("resume after"); + assert_eq!( + digest128(after.state()), + digest128(before.state()), + "property 11: truncation changed what the stream resumes to, case {case}", + ); + assert_eq!( + entropy_after.draws(), + entropy_before.draws(), + "property 11: truncation moved the resume entropy position, case {case}", + ); + assert_eq!( + journal.retained_from(stream), + at, + "property 11: retained_from must report the new horizon, case {case}", + ); +} + /// Property 8 — **stream independence**. Appends to one stream never move /// another stream's head, entropy position, or snapshot. /// diff --git a/app/crates/ironstate-journal/src/journal.rs b/app/crates/ironstate-journal/src/journal.rs index 56d7832..cd1c708 100644 --- a/app/crates/ironstate-journal/src/journal.rs +++ b/app/crates/ironstate-journal/src/journal.rs @@ -85,6 +85,14 @@ pub enum JournalError { /// The sequence number that was not found. at: Seq, }, + /// Truncation was refused: it would have discarded records that are still + /// needed to replay from the newest snapshot. + NoSnapshotForTruncation { + /// The sequence truncation was requested before. + at: Seq, + /// The newest snapshot in the stream, if it has one at all. + latest_snapshot: Option, + }, } impl core::fmt::Display for JournalError { @@ -97,6 +105,17 @@ impl core::fmt::Display for JournalError { The sequence is past the head, or below the earliest retained record.\n\ Check `head()` and the snapshot horizon before addressing a Seq.", ), + Self::NoSnapshotForTruncation { + at, + latest_snapshot, + } => write!( + f, + "refusing to truncate before {at:?}: the newest snapshot is {latest_snapshot:?}.\n\ + Truncating here would discard records still needed to replay from that \ + snapshot, leaving the stream unresumable.\n\ + Take a snapshot at or after {:?} first, then truncate.", + Seq(at.0.saturating_sub(1)), + ), } } } @@ -105,7 +124,7 @@ impl std::error::Error for JournalError { fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { match self { Self::Storage(source) => Some(source.as_ref()), - Self::UnknownSeq { .. } => None, + Self::UnknownSeq { .. } | Self::NoSnapshotForTruncation { .. } => None, } } } @@ -251,3 +270,30 @@ pub trait ForkableJournal: Journal { where Self: Sized; } + +/// A journal that can discard the oldest part of a stream. +/// +/// Retention is a capability, not a universal contract: an append-only store +/// may have no way to drop records, and some domains forbid it outright. It is +/// therefore opt-in — but where it exists, anyone with a retention policy, a +/// right-to-erasure obligation, or simply a large log needs it, and the +/// snapshot machinery that makes it safe already exists. +pub trait RetainableJournal: Journal { + /// Discard every record in `stream` before `at`, keeping `at` onward. + /// + /// Truncation must be refused unless a snapshot at or after `at - 1` + /// exists, since replay resumes from the newest snapshot and would + /// otherwise need records that are gone. Sequence numbers of the *retained* + /// records never change: a `Seq` is an identity, and subscriptions hold + /// high-water marks that refer to it. + /// + /// # Errors + /// + /// Returns [`JournalError::NoSnapshotForTruncation`] if no snapshot covers + /// the retained prefix, or [`JournalError::Storage`] if the store failed. + fn truncate_before(&mut self, stream: &StreamId, at: Seq) -> Result<(), JournalError>; + + /// The earliest sequence still retained in `stream` — everything below it + /// has been truncated away. + fn retained_from(&self, stream: &StreamId) -> Seq; +} diff --git a/app/crates/ironstate-journal/src/lib.rs b/app/crates/ironstate-journal/src/lib.rs index 0ed502a..2998016 100644 --- a/app/crates/ironstate-journal/src/lib.rs +++ b/app/crates/ironstate-journal/src/lib.rs @@ -97,7 +97,8 @@ mod sim; mod subscription; pub use journal::{ - ExecuteError, ForkableJournal, Journal, JournalError, Seq, Snapshot, StreamId, VersionedEvent, + ExecuteError, ForkableJournal, Journal, JournalError, RetainableJournal, Seq, Snapshot, + StreamId, VersionedEvent, }; pub use replay::{ Pending, Prepared, ResumeError, execute, execute_in, prepare, replay, replay_hash, resume, @@ -121,6 +122,6 @@ pub mod testkit { #[cfg(feature = "sim")] #[doc(hidden)] pub mod testkit_support { - pub use crate::contract::{run_contract, run_contract_forkable}; + pub use crate::contract::{run_contract, run_contract_forkable, run_contract_retainable}; pub use crate::sim::run_scenario; } diff --git a/app/crates/ironstate-journal/src/macros.rs b/app/crates/ironstate-journal/src/macros.rs index 4c20486..e038eb3 100644 --- a/app/crates/ironstate-journal/src/macros.rs +++ b/app/crates/ironstate-journal/src/macros.rs @@ -8,7 +8,9 @@ /// adapter implements [`ForkableJournal`](crate::ForkableJournal), which brings /// in the two extra properties that need branching: round-trip **at each /// recorded step**, and fork-position equality. Round-trip of the whole log is -/// part of the base contract and runs for every adapter. +/// part of the base contract and runs for every adapter. Add `retainable` when +/// it implements [`RetainableJournal`](crate::RetainableJournal), which adds +/// the truncation-preserves-resume property. /// /// # Your journal must have `Tx<'_> = ()` /// @@ -23,11 +25,12 @@ /// ironstate_journal::journal_contract_test!(MatchState); // the memory journal /// ironstate_journal::journal_contract_test!(MyPostgresJournal, MatchState); /// ironstate_journal::journal_contract_test!(MyForkingJournal, MatchState, forkable); +/// ironstate_journal::journal_contract_test!(MyStore, MatchState, forkable, retainable); /// ``` #[macro_export] macro_rules! journal_contract_test { ($agg:ty) => { - $crate::journal_contract_test!($crate::MemoryJournal<$agg>, $agg, forkable); + $crate::journal_contract_test!($crate::MemoryJournal<$agg>, $agg, forkable, retainable); }; ($journal:ty, $agg:ty) => { #[test] @@ -41,6 +44,20 @@ macro_rules! journal_contract_test { $crate::testkit_support::run_contract_forkable::<$journal, $agg>(64, 24, 0xC047); } }; + ($journal:ty, $agg:ty, retainable) => { + #[test] + fn journal_contract() { + $crate::testkit_support::run_contract::<$journal, $agg>(64, 24, 0xC047); + $crate::testkit_support::run_contract_retainable::<$journal, $agg>(32, 24, 0xC047); + } + }; + ($journal:ty, $agg:ty, forkable, retainable) => { + #[test] + fn journal_contract() { + $crate::testkit_support::run_contract_forkable::<$journal, $agg>(64, 24, 0xC047); + $crate::testkit_support::run_contract_retainable::<$journal, $agg>(32, 24, 0xC047); + } + }; } /// Generate a `#[test]` for the seeded whole-tier simulation: a fault-injected diff --git a/app/crates/ironstate-journal/src/memory.rs b/app/crates/ironstate-journal/src/memory.rs index 3ed1d99..bdfb09f 100644 --- a/app/crates/ironstate-journal/src/memory.rs +++ b/app/crates/ironstate-journal/src/memory.rs @@ -2,7 +2,8 @@ //! judged against by `journal_contract_test!`. use crate::journal::{ - ForkableJournal, Journal, JournalError, Seq, Snapshot, StreamId, VersionedEvent, + ForkableJournal, Journal, JournalError, RetainableJournal, Seq, Snapshot, StreamId, + VersionedEvent, }; use ironstate_aggregate::{AggregateRules, DrawPos}; use std::borrow::Cow; @@ -14,21 +15,51 @@ struct Record { } /// One stream's history: its records and the snapshots taken over them. +/// +/// `truncated` counts the records dropped off the front, so a retained record's +/// `Seq` never changes when older ones are discarded — `records[i]` is always +/// `Seq(truncated + i + 1)`. struct Stream { records: Vec>, snapshots: Vec>, + truncated: u64, } impl Stream { + fn new(genesis: Snapshot) -> Self { + Self { + records: Vec::new(), + snapshots: vec![genesis], + truncated: 0, + } + } + + /// The latest sequence in this stream, or `None` if it holds no records. + fn head(&self) -> Option { + (!self.records.is_empty()).then_some(Seq(self.truncated + self.records.len() as u64)) + } + + /// The earliest sequence still retained. + fn retained_from(&self) -> Seq { + Seq(self.truncated + 1) + } + fn record_at(&self, at: Seq) -> Result<&Record, JournalError> { // Seq is public, so a caller can pass an out-of-range value. Compare in // u64 and only cast once it is within bounds — otherwise on a 32-bit // target an out-of-range Seq could truncate to a valid index instead of - // returning UnknownSeq. - if at.0 == 0 || at.0 > self.records.len() as u64 { + // returning UnknownSeq. A Seq below the retained horizon is equally + // unknown: the record existed once, but this journal can no longer + // answer for it. + if at.0 <= self.truncated || at.0 > self.truncated + self.records.len() as u64 { return Err(JournalError::UnknownSeq { at }); } - Ok(&self.records[(at.0 - 1) as usize]) + Ok(&self.records[(at.0 - self.truncated - 1) as usize]) + } + + /// The newest snapshot's sequence, if the stream has one. + fn latest_snapshot_at(&self) -> Option { + self.snapshots.iter().map(|s| s.at).max() } } @@ -68,10 +99,7 @@ impl MemoryJournal { /// first time it has been touched. fn stream_mut(&mut self, id: &StreamId) -> &mut Stream { if !self.streams.contains_key(id) { - let seeded = Stream { - records: Vec::new(), - snapshots: vec![self.genesis_snapshot()], - }; + let seeded = Stream::new(self.genesis_snapshot()); self.streams.insert(id.clone(), seeded); } self.streams @@ -106,7 +134,7 @@ impl Journal for MemoryJournal { events: events.to_vec(), entropy_pos, }); - Ok(Seq(stream.records.len() as u64)) + Ok(stream.head().expect("a record was just pushed")) } fn snapshot_in( @@ -140,8 +168,7 @@ impl Journal for MemoryJournal { } fn head(&self, stream: &StreamId) -> Option { - let stream = self.streams.get(stream)?; - (!stream.records.is_empty()).then_some(Seq(stream.records.len() as u64)) + self.streams.get(stream)?.head() } fn events_since( @@ -154,7 +181,10 @@ impl Journal for MemoryJournal { }; // Saturate rather than truncate: an out-of-range `after` (possible only on // a 32-bit target, since Seq is public) means "past the end", so skip all. - let start = after.map_or(0, |s| usize::try_from(s.0).unwrap_or(usize::MAX)); + // `after` is an absolute Seq, so discount whatever was truncated away. + let start = after.map_or(0, |s| { + usize::try_from(s.0.saturating_sub(stream.truncated)).unwrap_or(usize::MAX) + }); let type_name = Cow::Borrowed(core::any::type_name::()); Ok(stream .records @@ -193,13 +223,14 @@ impl ForkableJournal for MemoryJournal { Err(JournalError::UnknownSeq { at }) }; }; - if at.0 > source.records.len() as u64 { + if at.0 < source.truncated || at.0 > source.truncated + source.records.len() as u64 { return Err(JournalError::UnknownSeq { at }); } - let cutoff = at.0 as usize; + let cutoff = (at.0 - source.truncated) as usize; forked.streams.insert( stream.clone(), Stream { + truncated: source.truncated, records: source .records .iter() @@ -220,3 +251,46 @@ impl ForkableJournal for MemoryJournal { Ok(forked) } } + +impl RetainableJournal for MemoryJournal { + fn truncate_before(&mut self, stream: &StreamId, at: Seq) -> Result<(), JournalError> { + let stream = self.stream_mut(stream); + let latest_snapshot = stream.latest_snapshot_at(); + + // Replay resumes from the newest snapshot and reads every record after + // it, so the retained prefix must start at or before that point. + let needed = at.0.saturating_sub(1); + if latest_snapshot.is_none_or(|s| s.0 < needed) { + return Err(JournalError::NoSnapshotForTruncation { + at, + latest_snapshot, + }); + } + + let drop_count = at.0.saturating_sub(1).saturating_sub(stream.truncated); + let drop_count = usize::try_from(drop_count) + .unwrap_or(usize::MAX) + .min(stream.records.len()); + stream.records.drain(..drop_count); + stream.truncated += drop_count as u64; + + // Snapshots below the new horizon are no longer reachable bases, but the + // newest one at or before it must survive — it is what replay starts from. + if let Some(keep) = stream + .snapshots + .iter() + .filter(|s| s.at.0 <= stream.truncated) + .map(|s| s.at) + .max() + { + stream.snapshots.retain(|s| s.at >= keep); + } + Ok(()) + } + + fn retained_from(&self, stream: &StreamId) -> Seq { + self.streams + .get(stream) + .map_or(Seq(1), Stream::retained_from) + } +} From 8b8ae4d3f4748d81b026a89593814520ff463ed7 Mon Sep 17 00:00:00 2001 From: Dashiel Lopez Mendez Date: Fri, 28 Aug 2026 23:50:33 -0400 Subject: [PATCH 02/14] fix(journal): stop truncation from stranding a fork on a stale snapshot Two defects in the retention work, both found by testing truncation against forking -- an interaction neither feature's own tests reached, since the fork properties never truncate and the truncation property never forks. **A stale snapshot could produce a silently wrong aggregate.** Truncation kept every snapshot at or above the newest one *below* the horizon, so a genesis snapshot survived a truncation that discarded the records replay from it needs. Forking between that stale base and the next real snapshot then resumed from genesis over only the retained records and returned the result as `Ok` -- a wrong state, not an error. Truncation now drops every snapshot below the horizon; the refusal check already guarantees at least one valid base survives. **A fork at the horizon claimed a head it did not have.** The bounds check admitted `at == truncated`, producing a branch with no records whose `head` was `None` rather than the fork point, violating the fork-position property. Fork points at or below the horizon are now refused as `UnknownSeq`: those records are gone, so no branch can reproduce them. Adds `tests/retention.rs` covering both regressions plus the properties the generated suite cannot express for a specific adapter -- sequence identity across truncation, `UnknownSeq` below the horizon, refusal when no snapshot covers the retained prefix, and resume after truncation. Co-Authored-By: Claude Opus 5 --- app/crates/ironstate-journal/src/memory.rs | 25 +- .../ironstate-journal/tests/retention.rs | 272 ++++++++++++++++++ 2 files changed, 285 insertions(+), 12 deletions(-) create mode 100644 app/crates/ironstate-journal/tests/retention.rs diff --git a/app/crates/ironstate-journal/src/memory.rs b/app/crates/ironstate-journal/src/memory.rs index bdfb09f..9a14d3d 100644 --- a/app/crates/ironstate-journal/src/memory.rs +++ b/app/crates/ironstate-journal/src/memory.rs @@ -223,7 +223,13 @@ impl ForkableJournal for MemoryJournal { Err(JournalError::UnknownSeq { at }) }; }; - if at.0 < source.truncated || at.0 > source.truncated + source.records.len() as u64 { + // A fork must reproduce the source's records through `at`. Anything at + // or below the truncation horizon is gone, so it cannot be forked — + // and forking *at* the horizon would yield a branch whose head is not + // the fork point. + let head = source.truncated + source.records.len() as u64; + let below_horizon = source.truncated > 0 && at.0 <= source.truncated; + if at.0 > head || below_horizon { return Err(JournalError::UnknownSeq { at }); } let cutoff = (at.0 - source.truncated) as usize; @@ -274,17 +280,12 @@ impl RetainableJournal for MemoryJournal { stream.records.drain(..drop_count); stream.truncated += drop_count as u64; - // Snapshots below the new horizon are no longer reachable bases, but the - // newest one at or before it must survive — it is what replay starts from. - if let Some(keep) = stream - .snapshots - .iter() - .filter(|s| s.at.0 <= stream.truncated) - .map(|s| s.at) - .max() - { - stream.snapshots.retain(|s| s.at >= keep); - } + // A snapshot below the new horizon is no longer a valid base: replaying + // from it would need records that are now gone, and would silently + // produce a *wrong* aggregate rather than an error. The refusal check + // above guarantees at least one snapshot survives this. + let horizon = stream.truncated; + stream.snapshots.retain(|s| s.at.0 >= horizon); Ok(()) } diff --git a/app/crates/ironstate-journal/tests/retention.rs b/app/crates/ironstate-journal/tests/retention.rs new file mode 100644 index 0000000..3926ead --- /dev/null +++ b/app/crates/ironstate-journal/tests/retention.rs @@ -0,0 +1,272 @@ +//! Retention behaviour that the generated contract properties do not reach: +//! how truncation interacts with sequence identity, with addressing below the +//! horizon, and with forking. + +use ironstate::prelude::*; +use ironstate_aggregate::{ + Aggregate, AggregateRules, DrawPos, LogicalTime, OwnedDeterministicCtx, Seed, SeededEntropy, + StableHash, +}; +use ironstate_journal::{ + ForkableJournal, Journal, JournalError, MemoryJournal, RetainableJournal, Seq, Snapshot, + StreamId, execute, resume, +}; + +#[derive(StateMachine, StableHash, Clone, Debug, PartialEq)] +#[state_machine(initial = Open, terminal = [Closed])] +enum Phase { + Open, + Closed, +} +#[derive(Event, Clone, Debug, PartialEq)] +enum Step { + Close, +} +impl TransitionRules for Phase { + type Event = Step; + fn transition(&self, _: &Step) -> Option { + matches!(self, Phase::Open).then_some(Phase::Closed) + } +} + +#[derive(Event, Clone, Debug, PartialEq)] +enum Command { + Tick, +} +#[derive(Clone, Debug, PartialEq)] +enum Ev { + Rolled(u8), +} +#[derive(Debug, thiserror::Error)] +#[error("closed")] +struct ClosedErr; + +#[derive(StableHash, Clone, Debug, PartialEq)] +struct Counter { + phase: Phase, + total: u32, +} + +impl AggregateRules for Counter { + type Phase = Phase; + type Command = Command; + type Event = Ev; + type Error = ClosedErr; + type Ctx = OwnedDeterministicCtx; + + fn phase(&self) -> Phase { + self.phase.clone() + } + fn decide(&self, _cmd: &Command, ctx: &mut Self::Ctx) -> Result, ClosedErr> { + Ok(vec![Ev::Rolled(ctx.entropy.draw_range(1..7) as u8)]) + } + fn evolve(&mut self, event: &Ev) { + let Ev::Rolled(n) = event; + self.total += u32::from(*n); + } +} + +fn genesis() -> Counter { + Counter { + phase: Phase::Open, + total: 0, + } +} + +fn stream() -> StreamId { + StreamId::new("retention") +} + +fn ctx(seed: &Seed, pos: DrawPos) -> OwnedDeterministicCtx { + OwnedDeterministicCtx { + entropy: Box::new(SeededEntropy::at(seed, pos)), + actor: 0, + now: LogicalTime(0), + } +} + +/// Drive `steps` appends, then snapshot at the head so truncation is legal. +fn driven(steps: u64) -> (MemoryJournal, Aggregate, Seed) { + let seed = Seed([3; 32]); + let mut journal = MemoryJournal::new(genesis()); + let mut agg = Aggregate::new(genesis()).unwrap(); + for _ in 0..steps { + let pos = journal + .head(&stream()) + .map_or(DrawPos(0), |h| journal.entropy_pos(&stream(), h).unwrap()); + let mut c = ctx(&seed, pos); + execute(&mut journal, &stream(), &mut agg, &Command::Tick, &mut c).unwrap(); + } + let head = journal.head(&stream()).unwrap(); + let pos = journal.entropy_pos(&stream(), head).unwrap(); + journal + .snapshot_in( + &mut (), + &stream(), + Snapshot { + state: agg.state().clone(), + schema_version: 0, + at: head, + entropy_pos: pos, + }, + ) + .unwrap(); + (journal, agg, seed) +} + +#[test] +fn retained_records_keep_their_sequence_numbers() { + let (mut journal, _agg, _seed) = driven(6); + let before = journal.entropy_pos(&stream(), Seq(5)).unwrap(); + + journal.truncate_before(&stream(), Seq(4)).unwrap(); + + assert_eq!(journal.retained_from(&stream()), Seq(4)); + assert_eq!( + journal.head(&stream()), + Some(Seq(6)), + "truncation must not renumber the head", + ); + assert_eq!( + journal.entropy_pos(&stream(), Seq(5)).unwrap(), + before, + "a retained record must keep its Seq and its recorded position", + ); +} + +#[test] +fn addressing_below_the_horizon_is_unknown_not_a_neighbour() { + let (mut journal, _agg, _seed) = driven(6); + journal.truncate_before(&stream(), Seq(4)).unwrap(); + + for gone in [Seq(1), Seq(2), Seq(3)] { + match journal.entropy_pos(&stream(), gone) { + Err(JournalError::UnknownSeq { at }) => assert_eq!(at, gone), + other => panic!("expected UnknownSeq for truncated {gone:?}, got {other:?}"), + } + } +} + +#[test] +fn truncation_is_refused_when_no_snapshot_covers_the_retained_prefix() { + let seed = Seed([3; 32]); + let mut journal = MemoryJournal::new(genesis()); + let mut agg = Aggregate::new(genesis()).unwrap(); + for _ in 0..4 { + let pos = journal + .head(&stream()) + .map_or(DrawPos(0), |h| journal.entropy_pos(&stream(), h).unwrap()); + let mut c = ctx(&seed, pos); + execute(&mut journal, &stream(), &mut agg, &Command::Tick, &mut c).unwrap(); + } + // The only snapshot is the genesis at Seq(0); truncating before Seq(3) + // would strand records 1..2 that replay from it still needs. + match journal.truncate_before(&stream(), Seq(3)) { + Err(JournalError::NoSnapshotForTruncation { + at, + latest_snapshot, + }) => { + assert_eq!(at, Seq(3)); + assert_eq!(latest_snapshot, Some(Seq(0))); + } + other => panic!("expected refusal, got {other:?}"), + } + assert_eq!( + journal.head(&stream()), + Some(Seq(4)), + "a refused truncation must change nothing", + ); +} + +#[test] +fn a_truncated_stream_still_resumes() { + let (mut journal, agg, seed) = driven(6); + let before = agg.state().clone(); + journal.truncate_before(&stream(), Seq(4)).unwrap(); + let (resumed, _) = resume::(&journal, &stream(), &seed).unwrap(); + assert_eq!(resumed.state(), &before); +} + +#[test] +fn forking_below_the_retained_horizon_is_refused() { + let (mut journal, _agg, _seed) = driven(6); + journal.truncate_before(&stream(), Seq(4)).unwrap(); + + // Records 1..3 are gone, so no fork can reproduce history through them. + for gone in [Seq(1), Seq(2), Seq(3)] { + assert!( + matches!( + journal.fork(&stream(), gone), + Err(JournalError::UnknownSeq { .. }) + ), + "forking at {gone:?}, below the horizon, must be refused", + ); + } +} + +#[test] +fn a_fork_of_a_truncated_stream_keeps_the_fork_point_as_its_head() { + let (mut journal, _agg, _seed) = driven(6); + journal.truncate_before(&stream(), Seq(4)).unwrap(); + + let branch = journal.fork(&stream(), Seq(5)).expect("fork above horizon"); + assert_eq!( + branch.head(&stream()), + Some(Seq(5)), + "a fork's head must sit at the fork point, truncated or not", + ); + assert_eq!( + branch.entropy_pos(&stream(), Seq(5)).unwrap(), + journal.entropy_pos(&stream(), Seq(5)).unwrap(), + "a fork must agree with its source at the fork point", + ); + assert_eq!(branch.retained_from(&stream()), Seq(4)); +} + +#[test] +fn a_fork_of_a_truncated_stream_still_has_a_replay_base() { + let (mut journal, _agg, seed) = driven(6); + journal.truncate_before(&stream(), Seq(4)).unwrap(); + + let branch = journal.fork(&stream(), Seq(6)).expect("fork at head"); + resume::(&branch, &stream(), &seed) + .expect("a fork must carry a snapshot to replay from"); +} + +/// A snapshot taken *below* the truncation horizon is no longer a valid replay +/// base: replaying from it needs records that are gone. Truncation must drop +/// such snapshots, so a fork that lands between them fails honestly rather than +/// silently replaying to a wrong state. +/// +/// Regression: before this was fixed, the fork below resumed to a state built +/// from the genesis snapshot plus only the *retained* records — a wrong +/// aggregate returned as `Ok`. +#[test] +fn a_stale_base_below_the_horizon_never_produces_a_wrong_state() { + // The true state at the fork point, captured before anything is truncated. + let (untruncated, _agg, seed) = driven(6); + let branch = untruncated + .fork(&stream(), Seq(5)) + .expect("fork of an untruncated stream"); + let truth = resume::(&branch, &stream(), &seed) + .expect("resume") + .0 + .state() + .clone(); + + let (mut journal, _agg, _seed) = driven(6); + journal.truncate_before(&stream(), Seq(4)).unwrap(); + + match journal.fork(&stream(), Seq(5)) { + Err(_) => {} + Ok(branch) => match resume::(&branch, &stream(), &seed) { + // Honest: no surviving snapshot covers this fork point. + Err(_) => {} + Ok((resumed, _)) => assert_eq!( + resumed.state(), + &truth, + "a fork must never replay from a base whose records were truncated away", + ), + }, + } +} From 4dda451b5572234ca8eef7a8f1e5ba917cac373a Mon Sep 17 00:00:00 2001 From: Dashiel Lopez Mendez Date: Sat, 29 Aug 2026 00:22:02 -0400 Subject: [PATCH 03/14] fix(journal): close the truncation, conformance and keying gaps from review Addresses a review of the whole stack. Several findings were confirmed by probe tests before being fixed. **Truncation and entropy bookkeeping.** - A stream truncated to its head reported `head() == None`, so `execute` took its rewind anchor as `DrawPos(0)` and reissued draws already consumed -- positions running backwards while sequences ran forwards, breaking the determinism contract. The records are gone; the history is not. - `entropy_pos` now consults a snapshot at the requested `Seq` before the records, since a snapshot records the position at its own `Seq` and outlives the record there. - `entropy_pos(_, Seq(0))` short-circuited before the horizon check and fabricated `DrawPos(0)` for a genesis that truncation had discarded. - `events_since` silently returned a *gapped* list when read from below the horizon. A subscription resuming from a stale mark was handed non-contiguous records and dropped one with no error anywhere. It now refuses; `retained_from` says where a valid read starts. - `truncate_before` created the stream through `stream_mut`, so a refused truncation left a phantom entry behind, and it never bounded `at` against the head -- a snapshot recorded past the head authorised discarding everything, after which `retained_from` disagreed with the `at` requested. - `fork` could return a branch with no snapshot at or below the fork point, unresumable but returned as `Ok`. New `JournalError::NoBaseForFork` refuses it where it happens instead of failing later with a vaguer error. **Conformance.** - Property 1 (round trip) had moved behind `run_contract_forkable`, so the two-argument `journal_contract_test!` silently stopped checking the single most important journal property. It does not need `fork` -- it is now expressed over `events_since` and runs for every adapter, with the per-step variant staying behind forking. - `run_contract_forkable` drove every history twice from a differently advanced runner, doubling suite time and making a fork failure irreproducible against the reported case. One pass now. - Property 11 additionally checks positions stay total and monotonic over the retained range and that everything below the horizon is `UnknownSeq` -- the failure mode retention introduces, which properties 2 and 10 cannot see on an untruncated journal. **Keying.** `VersionedEvent` now carries the `Seq` of the record it came from. `events_since` flattens events across records, so the index into its result is not a sequence: as soon as one `decide` emitted two events, the `hidden-info` subscription recorded a mark past the stream head and dropped every later event as a duplicate. The example uses the real `Seq`. **Surface.** `StreamId::MAIN`/`main()` are removed (unused), and `streams()` moves from `Journal` to `RetainableJournal`, where sweeping for expired history is an actual use -- no ordinary adapter should owe a full-table scan for a method nothing calls. **Documented rather than fixed**, each where an implementor will meet it: `execute_in` is one-call-per-transaction while `head`/`entropy_pos` take no `tx`; `Subscription::deliver` is idempotent but not atomic and cannot use the `Tx` seam; and the conformance suite cannot yet drive an adapter whose `Tx` is a real transaction, so such adapters are held by a `Tx = ()` twin as `async-store` already is. A rewritten rollback test in tests/transactional.rs replaces one whose assertion was vacuously true. Co-Authored-By: Claude Opus 5 --- app/crates/examples/async-store/src/main.rs | 10 +- app/crates/examples/hidden-info/src/main.rs | 12 +- app/crates/ironstate-journal/src/contract.rs | 54 +++++- app/crates/ironstate-journal/src/journal.rs | 39 ++++- app/crates/ironstate-journal/src/memory.rs | 109 +++++++++--- app/crates/ironstate-journal/src/replay.rs | 14 ++ .../ironstate-journal/src/subscription.rs | 11 ++ .../ironstate-journal/tests/retention.rs | 157 +++++++++++++++++- .../ironstate-journal/tests/transactional.rs | 9 +- 9 files changed, 371 insertions(+), 44 deletions(-) diff --git a/app/crates/examples/async-store/src/main.rs b/app/crates/examples/async-store/src/main.rs index 3fc6c31..0b0b982 100644 --- a/app/crates/examples/async-store/src/main.rs +++ b/app/crates/examples/async-store/src/main.rs @@ -261,10 +261,16 @@ impl Log { Ok(self .records(stream) .iter() + .enumerate() .skip(start) - .flat_map(|record| record.events.iter()) - .map(|event| VersionedEvent { + .flat_map(|(i, record)| { + // The record's Seq, shared by every event it holds. + let seq = Seq(i as u64 + 1); + record.events.iter().map(move |event| (seq, event)) + }) + .map(|(seq, event)| VersionedEvent { event: event.clone(), + seq, type_name: type_name.clone(), version: 1, }) diff --git a/app/crates/examples/hidden-info/src/main.rs b/app/crates/examples/hidden-info/src/main.rs index eea689b..6d9643b 100644 --- a/app/crates/examples/hidden-info/src/main.rs +++ b/app/crates/examples/hidden-info/src/main.rs @@ -503,17 +503,15 @@ fn run_demo() -> Result<()> { })?; let mut subscription: Subscription = Subscription::new(); let mut profile_ctx = ctx(&seed, DrawPos(0), 0); - for (i, event) in journal - .events_since(&match_stream(), None) - .unwrap() - .iter() - .enumerate() - { + for event in journal.events_since(&match_stream(), None).unwrap().iter() { subscription .deliver( SourceEvent { stream: &match_stream(), - at: Seq(i as u64 + 1), + // The record's own Seq, not its index in the flattened list: + // one record can hold several events, so an index would run + // ahead of the head and mark real events as duplicates. + at: event.seq, event: &event.event, }, &StreamId::new("profile"), diff --git a/app/crates/ironstate-journal/src/contract.rs b/app/crates/ironstate-journal/src/contract.rs index c0657a8..1029e4b 100644 --- a/app/crates/ironstate-journal/src/contract.rs +++ b/app/crates/ironstate-journal/src/contract.rs @@ -37,7 +37,9 @@ use proptest::test_runner::{Config, RngAlgorithm, TestRng, TestRunner}; /// Until it does, hold such an adapter to the contract with a twin over the /// same storage whose `Tx` is `()` — the pattern the `async-store` example /// already uses for a store that cannot implement the synchronous trait at -/// all. This is a limitation of the harness, not of the adapter. +/// all. This is a limitation of the harness, not of the adapter. The rollback +/// behaviour only a real `Tx` can exhibit is covered separately, in +/// `tests/transactional.rs`. pub trait ContractJournal: Journal { /// A fresh, empty journal seeded with the aggregate's genesis state. fn fresh(genesis: A) -> Self; @@ -192,15 +194,32 @@ where A: AggregateArbitrary + StableHash, A::Ctx: CtxEntropy, { - run_contract::(cases, max_steps, seed_base); - let stream = test_stream(); let mut runner = seeded_runner(seed_base); for case in 0..cases { let genesis = sample(A::initial_state_strategy(), &mut runner); let seed = run_seed(seed_base, case); - let (journal, _live, steps) = drive::(genesis.clone(), &seed, &mut runner, max_steps); + let (journal, live, steps) = drive::(genesis.clone(), &seed, &mut runner, max_steps); + property_2_positions_total_and_monotonic(&journal, &stream, case); + property_1_round_trip(&journal, &stream, &genesis, &steps, case); + property_7_version_tagging(&journal, &stream, case); + property_3_resume_identity(&journal, &stream, live, &seed, &mut runner, case); + property_5_snapshot_vs_head(&journal, &stream, &seed, case); + property_9_out_of_range_seq(&journal, &stream, case); + + property_6_failed_append_atomicity::(&seed, &mut runner, case); + property_8_stream_independence::( + genesis.clone(), + &seed, + &mut runner, + max_steps, + case, + ); + + // The branching properties, in the same pass over the same histories — + // driving a second, separately-seeded set would double the cost and make + // a fork failure irreproducible against the case just reported. property_1_round_trip_at_each_step(&journal, &stream, &genesis, &steps, case); property_4_fork_position_equality(&journal, &stream, case); } @@ -460,6 +479,33 @@ fn property_11_truncation_preserves_resume( .truncate_before(stream, at) .expect("truncating below a snapshot taken at the head must be allowed"); + // Positions stay total and monotonic over what remains, and everything below + // the horizon is now out of range — the new failure mode retention adds, + // which property 2 and property 10 cannot see on an untruncated journal. + let horizon = journal.retained_from(stream); + let mut previous = DrawPos(0); + for seq in horizon.0..=head.0 { + let pos = journal.entropy_pos(stream, Seq(seq)).unwrap_or_else(|_| { + panic!( + "[proven] property 11: entropy_pos undefined at retained Seq({seq}), case {case}" + ) + }); + assert!( + pos >= previous, + "[proven] property 11: entropy_pos decreased at Seq({seq}), case {case}", + ); + previous = pos; + } + for gone in 1..horizon.0 { + assert!( + matches!( + journal.entropy_pos(stream, Seq(gone)), + Err(JournalError::UnknownSeq { .. }) + ), + "[proven] property 11: truncated Seq({gone}) must be UnknownSeq, case {case}", + ); + } + let (after, entropy_after) = resume::(journal, stream, seed).expect("resume after"); assert_eq!( digest128(after.state()), diff --git a/app/crates/ironstate-journal/src/journal.rs b/app/crates/ironstate-journal/src/journal.rs index cd1c708..41ad2ad 100644 --- a/app/crates/ironstate-journal/src/journal.rs +++ b/app/crates/ironstate-journal/src/journal.rs @@ -68,6 +68,16 @@ pub struct Snapshot { pub struct VersionedEvent { /// The event payload. pub event: A::Event, + /// The sequence of the **record** this event was appended in. + /// + /// One record can hold several events, so this is not a per-event ordinal + /// and several returned events may share it. It is the identity a + /// [`Subscription`](crate::Subscription) keys its high-water mark by, and + /// the only correct thing to pass as + /// [`SourceEvent::at`](crate::SourceEvent::at) — a position derived from the + /// index within the returned list is wrong as soon as any `decide` emits + /// more than one event. + pub seq: Seq, /// The event type's name when stored. pub type_name: Cow<'static, str>, /// The event enum's version when stored. @@ -85,6 +95,12 @@ pub enum JournalError { /// The sequence number that was not found. at: Seq, }, + /// A fork was refused: no snapshot at or below the fork point survives, so + /// the branch would have no base to replay from. + NoBaseForFork { + /// The requested fork point. + at: Seq, + }, /// Truncation was refused: it would have discarded records that are still /// needed to replay from the newest snapshot. NoSnapshotForTruncation { @@ -105,6 +121,13 @@ impl core::fmt::Display for JournalError { The sequence is past the head, or below the earliest retained record.\n\ Check `head()` and the snapshot horizon before addressing a Seq.", ), + Self::NoBaseForFork { at } => write!( + f, + "cannot fork at {at:?}: no snapshot at or below it survives.\n\ + Truncation discarded the bases that covered this point, so the branch \ + would be unresumable.\n\ + Fork at or above the newest snapshot instead, or take one first.", + ), Self::NoSnapshotForTruncation { at, latest_snapshot, @@ -124,7 +147,9 @@ impl std::error::Error for JournalError { fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { match self { Self::Storage(source) => Some(source.as_ref()), - Self::UnknownSeq { .. } | Self::NoSnapshotForTruncation { .. } => None, + Self::UnknownSeq { .. } + | Self::NoBaseForFork { .. } + | Self::NoSnapshotForTruncation { .. } => None, } } } @@ -296,4 +321,16 @@ pub trait RetainableJournal: Journal { /// The earliest sequence still retained in `stream` — everything below it /// has been truncated away. fn retained_from(&self, stream: &StreamId) -> Seq; + + /// Every stream this journal holds, so a retention sweep can enumerate what + /// it might expire. + /// + /// This lives here rather than on [`Journal`] because sweeping is the only + /// thing that needs it, and on a relational store it is a full-table scan + /// no ordinary adapter should be made to implement. + /// + /// # Errors + /// + /// Returns [`JournalError::Storage`] if the underlying store failed. + fn streams(&self) -> Result, JournalError>; } diff --git a/app/crates/ironstate-journal/src/memory.rs b/app/crates/ironstate-journal/src/memory.rs index 9a14d3d..3f2939d 100644 --- a/app/crates/ironstate-journal/src/memory.rs +++ b/app/crates/ironstate-journal/src/memory.rs @@ -34,9 +34,17 @@ impl Stream { } } - /// The latest sequence in this stream, or `None` if it holds no records. + /// The latest sequence in this stream, or `None` if nothing was ever + /// appended to it. + /// + /// A stream truncated all the way to its head still *has* a head — the + /// records are gone, the history is not. Reporting `None` there would tell + /// `execute` the stream is empty and rewind the live entropy stream to + /// `DrawPos(0)`, so positions would run backwards while sequences ran + /// forwards. fn head(&self) -> Option { - (!self.records.is_empty()).then_some(Seq(self.truncated + self.records.len() as u64)) + let head = self.truncated + self.records.len() as u64; + (head > 0).then_some(Seq(head)) } /// The earliest sequence still retained. @@ -156,13 +164,21 @@ impl Journal for MemoryJournal { Err(JournalError::UnknownSeq { at }) }; }; + // A snapshot records the position at its own `Seq` and outlives the + // record there, so after a truncation it may be the only thing that + // still knows it — including at the head of a fully truncated stream. + if let Some(snapshot) = stream.snapshots.iter().find(|s| s.at == at) { + return Ok(snapshot.entropy_pos); + } if at.0 == 0 { - // Genesis position. (A snapshot may also sit at Seq(0).) - return Ok(stream - .snapshots - .iter() - .find(|s| s.at == Seq(0)) - .map_or(DrawPos(0), |s| s.entropy_pos)); + // Genesis is below the horizon once anything has been truncated, so + // it is as unknown as any other discarded sequence — answering + // `DrawPos(0)` here would fabricate a position for history that is + // gone. + if stream.truncated > 0 { + return Err(JournalError::UnknownSeq { at }); + } + return Ok(DrawPos(0)); } Ok(stream.record_at(at)?.entropy_pos) } @@ -179,20 +195,36 @@ impl Journal for MemoryJournal { let Some(stream) = self.streams.get(stream) else { return Ok(Vec::new()); }; + // Reading from below the horizon would silently return a *gapped* list: + // the caller asked to continue from a point whose successors are partly + // discarded. A subscription handed that list would drop records without + // ever seeing an error, so refuse instead — `retained_from` says where a + // valid read starts. `None` means "from this stream's start", which is + // the horizon, so it is only valid on an untruncated stream. + let from = after.map_or(0, |s| s.0); + if from < stream.truncated { + return Err(JournalError::UnknownSeq { + at: after.unwrap_or(Seq(0)), + }); + } // Saturate rather than truncate: an out-of-range `after` (possible only on // a 32-bit target, since Seq is public) means "past the end", so skip all. - // `after` is an absolute Seq, so discount whatever was truncated away. - let start = after.map_or(0, |s| { - usize::try_from(s.0.saturating_sub(stream.truncated)).unwrap_or(usize::MAX) - }); + let start = usize::try_from(from - stream.truncated).unwrap_or(usize::MAX); let type_name = Cow::Borrowed(core::any::type_name::()); Ok(stream .records .iter() + .enumerate() .skip(start) - .flat_map(|record| record.events.iter()) - .map(|event| VersionedEvent { + .flat_map(|(i, record)| { + // Every event in a record carries that record's Seq, not its own + // index in the flattened list. + let seq = Seq(stream.truncated + i as u64 + 1); + record.events.iter().map(move |event| (seq, event)) + }) + .map(|(seq, event)| VersionedEvent { event: event.clone(), + seq, type_name: type_name.clone(), version: 1, }) @@ -232,6 +264,20 @@ impl ForkableJournal for MemoryJournal { if at.0 > head || below_horizon { return Err(JournalError::UnknownSeq { at }); } + // A branch with no snapshot at or below the fork point cannot be + // replayed at all. Truncation can discard the bases that covered this + // point, so refuse here rather than hand back a journal whose `resume` + // fails later with a less specific error. + let base: Vec> = source + .snapshots + .iter() + .filter(|s| s.at <= at) + .map(clone_snapshot) + .collect(); + if base.is_empty() { + return Err(JournalError::NoBaseForFork { at }); + } + let cutoff = (at.0 - source.truncated) as usize; forked.streams.insert( stream.clone(), @@ -246,12 +292,7 @@ impl ForkableJournal for MemoryJournal { entropy_pos: r.entropy_pos, }) .collect(), - snapshots: source - .snapshots - .iter() - .filter(|s| s.at <= at) - .map(clone_snapshot) - .collect(), + snapshots: base, }, ); Ok(forked) @@ -259,8 +300,23 @@ impl ForkableJournal for MemoryJournal { } impl RetainableJournal for MemoryJournal { - fn truncate_before(&mut self, stream: &StreamId, at: Seq) -> Result<(), JournalError> { - let stream = self.stream_mut(stream); + fn truncate_before(&mut self, id: &StreamId, at: Seq) -> Result<(), JournalError> { + // Look up read-only first: a refused truncation must leave the journal + // exactly as it was, and `stream_mut` would create the stream — so a + // typo'd id would leave a phantom entry behind that `streams()` reports. + let Some(stream) = self.streams.get(id) else { + return Err(JournalError::UnknownSeq { at }); + }; + + // `at` may sit one past the head (discard everything) but no further. + // Without this a snapshot recorded beyond the head would authorise an + // arbitrary truncation, and `retained_from` would then disagree with the + // `at` that was asked for. + let head = stream.truncated + stream.records.len() as u64; + if at.0 > head + 1 { + return Err(JournalError::UnknownSeq { at }); + } + let latest_snapshot = stream.latest_snapshot_at(); // Replay resumes from the newest snapshot and reads every record after @@ -273,6 +329,11 @@ impl RetainableJournal for MemoryJournal { }); } + let stream = self + .streams + .get_mut(id) + .expect("the stream was found immediately above"); + let drop_count = at.0.saturating_sub(1).saturating_sub(stream.truncated); let drop_count = usize::try_from(drop_count) .unwrap_or(usize::MAX) @@ -294,4 +355,8 @@ impl RetainableJournal for MemoryJournal { .get(stream) .map_or(Seq(1), Stream::retained_from) } + + fn streams(&self) -> Result, JournalError> { + Ok(self.streams.keys().cloned().collect()) + } } diff --git a/app/crates/ironstate-journal/src/replay.rs b/app/crates/ironstate-journal/src/replay.rs index 0891f80..041c89b 100644 --- a/app/crates/ironstate-journal/src/replay.rs +++ b/app/crates/ironstate-journal/src/replay.rs @@ -166,6 +166,20 @@ where /// } /// ``` /// +/// # One call per transaction +/// +/// This reads the stream head through [`Journal::head`] / [`Journal::entropy_pos`], +/// which see **committed** state only — they take no `tx`. A second +/// `execute_in` against the same open transaction would therefore compute its +/// rewind anchor from the pre-transaction head, and aborting it would rewind the +/// entropy stream past the first append's draws, leaving a committed record +/// whose position is ahead of the live stream. +/// +/// So: one `execute_in` per unit of work, unless your adapter's reads observe +/// its own uncommitted writes. Batching several commands atomically needs a +/// journal whose `head`/`entropy_pos` are transaction-aware, which the trait +/// does not yet express. +/// /// # Errors /// /// Returns [`ExecuteError::Rejected`] if the command never produced events, or diff --git a/app/crates/ironstate-journal/src/subscription.rs b/app/crates/ironstate-journal/src/subscription.rs index e32aebd..bead07a 100644 --- a/app/crates/ironstate-journal/src/subscription.rs +++ b/app/crates/ironstate-journal/src/subscription.rs @@ -78,6 +78,17 @@ impl> Subscription { /// The source stream and `target_stream` are independent: the first names the /// history being read, the second the history being written. /// + /// # Delivery is idempotent, not atomic + /// + /// Each command is `execute`d separately. If `react` returns several and a + /// later one fails, the earlier ones are already durable while the mark + /// stays put — so redelivery re-runs `react` and re-applies them. Keep + /// `react` emitting commands that are safe to re-apply, or have it emit one. + /// + /// Making the whole batch atomic needs [`execute_in`](crate::execute_in) and + /// a journal with a real [`Journal::Tx`], which this method cannot use: it + /// is bound to `Tx<'a> = ()`. + /// /// # Errors /// /// Returns whatever [`execute`] returned for the first command that failed; diff --git a/app/crates/ironstate-journal/tests/retention.rs b/app/crates/ironstate-journal/tests/retention.rs index 3926ead..43a5e9d 100644 --- a/app/crates/ironstate-journal/tests/retention.rs +++ b/app/crates/ironstate-journal/tests/retention.rs @@ -4,8 +4,8 @@ use ironstate::prelude::*; use ironstate_aggregate::{ - Aggregate, AggregateRules, DrawPos, LogicalTime, OwnedDeterministicCtx, Seed, SeededEntropy, - StableHash, + Aggregate, AggregateRules, DrawPos, EntropySource, LogicalTime, OwnedDeterministicCtx, Seed, + SeededEntropy, StableHash, }; use ironstate_journal::{ ForkableJournal, Journal, JournalError, MemoryJournal, RetainableJournal, Seq, Snapshot, @@ -209,20 +209,165 @@ fn a_fork_of_a_truncated_stream_keeps_the_fork_point_as_its_head() { let (mut journal, _agg, _seed) = driven(6); journal.truncate_before(&stream(), Seq(4)).unwrap(); - let branch = journal.fork(&stream(), Seq(5)).expect("fork above horizon"); + // Seq(6) carries the surviving snapshot, so it is a forkable point. + let branch = journal.fork(&stream(), Seq(6)).expect("fork above horizon"); assert_eq!( branch.head(&stream()), - Some(Seq(5)), + Some(Seq(6)), "a fork's head must sit at the fork point, truncated or not", ); assert_eq!( - branch.entropy_pos(&stream(), Seq(5)).unwrap(), - journal.entropy_pos(&stream(), Seq(5)).unwrap(), + branch.entropy_pos(&stream(), Seq(6)).unwrap(), + journal.entropy_pos(&stream(), Seq(6)).unwrap(), "a fork must agree with its source at the fork point", ); assert_eq!(branch.retained_from(&stream()), Seq(4)); } +/// A fork point with no surviving snapshot at or below it is refused outright, +/// rather than handed back as a branch whose `resume` fails later. +#[test] +fn forking_where_no_base_survives_is_refused() { + let (mut journal, _agg, _seed) = driven(6); + journal.truncate_before(&stream(), Seq(4)).unwrap(); + + // Truncation dropped the genesis snapshot; the only one left is at Seq(6). + match journal.fork(&stream(), Seq(5)) { + Err(JournalError::NoBaseForFork { at }) => assert_eq!(at, Seq(5)), + Err(other) => panic!("expected NoBaseForFork, got {other:?}"), + Ok(_) => panic!("expected NoBaseForFork, got a branch with no replay base"), + } +} + +/// A stream truncated all the way to its head still has a head. Reporting +/// `None` would tell `execute` the stream is empty, rewinding the live entropy +/// stream to `DrawPos(0)` — so positions would run backwards while sequences +/// ran forwards, breaking the determinism contract. +#[test] +fn a_fully_truncated_stream_still_reports_its_head() { + let (mut journal, _agg, seed) = driven(6); + let head_pos = journal.entropy_pos(&stream(), Seq(6)).unwrap(); + + journal.truncate_before(&stream(), Seq(7)).unwrap(); + + assert_eq!(journal.retained_from(&stream()), Seq(7)); + assert_eq!( + journal.head(&stream()), + Some(Seq(6)), + "the records are gone, the history is not", + ); + + // The next append must continue the entropy stream, not restart it. + let (mut resumed, entropy) = resume::(&journal, &stream(), &seed).unwrap(); + assert_eq!(entropy.draws(), head_pos, "resume must not rewind entropy"); + let mut c = ctx(&seed, entropy.draws()); + execute( + &mut journal, + &stream(), + &mut resumed, + &Command::Tick, + &mut c, + ) + .unwrap(); + assert!( + journal.entropy_pos(&stream(), Seq(7)).unwrap() > head_pos, + "positions must keep moving forward across a full truncation", + ); +} + +/// Reading from below the horizon must refuse rather than return a list with a +/// silent hole in it — a subscription resuming from a stale high-water mark +/// would otherwise drop records and corrupt its projection with no error. +#[test] +fn reading_from_below_the_horizon_is_refused_not_gapped() { + let (mut journal, _agg, _seed) = driven(6); + let before = journal.events_since(&stream(), Some(Seq(2))).unwrap().len(); + assert_eq!(before, 4); + + journal.truncate_before(&stream(), Seq(4)).unwrap(); + + match journal.events_since(&stream(), Some(Seq(2))) { + Err(JournalError::UnknownSeq { at }) => assert_eq!(at, Seq(2)), + other => panic!( + "expected UnknownSeq for a stale mark, got {:?}", + other.map(|e| e.len()) + ), + } + // `None` means "from this stream's start", which is now the horizon. + match journal.events_since(&stream(), None) { + Err(JournalError::UnknownSeq { at }) => assert_eq!(at, Seq(0)), + other => panic!( + "expected UnknownSeq from genesis, got {:?}", + other.map(|e| e.len()) + ), + } + // Reading from the horizon itself is the valid way to say "everything you have". + assert_eq!( + journal.events_since(&stream(), Some(Seq(3))).unwrap().len(), + 3, + ); +} + +/// Genesis is below the horizon once anything is truncated, so it must be as +/// unknown as any other discarded sequence. +#[test] +fn genesis_is_unknown_once_it_is_below_the_horizon() { + let (mut journal, _agg, _seed) = driven(6); + journal.truncate_before(&stream(), Seq(4)).unwrap(); + + match journal.entropy_pos(&stream(), Seq(0)) { + Err(JournalError::UnknownSeq { at }) => assert_eq!(at, Seq(0)), + other => panic!("expected UnknownSeq for genesis below the horizon, got {other:?}"), + } +} + +/// A refused truncation must leave the journal byte-for-byte as it was — +/// including not bringing a stream into existence. +#[test] +fn a_refused_truncation_does_not_materialise_the_stream() { + let (mut journal, _agg, _seed) = driven(6); + let before = journal.streams().unwrap(); + + let typo = StreamId::new("typo"); + assert!(journal.truncate_before(&typo, Seq(5)).is_err()); + + assert_eq!( + journal.streams().unwrap(), + before, + "a refused truncation must not add a phantom stream", + ); +} + +/// A snapshot recorded beyond the head must not authorise an arbitrary +/// truncation — `retained_from` would then disagree with the `at` requested. +#[test] +fn truncation_past_the_head_is_refused() { + let (mut journal, agg, _seed) = driven(6); + journal + .snapshot_in( + &mut (), + &stream(), + Snapshot { + state: agg.state().clone(), + schema_version: 0, + at: Seq(1000), + entropy_pos: DrawPos(0), + }, + ) + .unwrap(); + + assert!( + journal.truncate_before(&stream(), Seq(500)).is_err(), + "a snapshot past the head must not authorise truncating past the head", + ); + assert_eq!(journal.head(&stream()), Some(Seq(6))); + assert_eq!(journal.retained_from(&stream()), Seq(1)); + + // One past the head is the legitimate "discard everything" call. + journal.truncate_before(&stream(), Seq(7)).unwrap(); + assert_eq!(journal.retained_from(&stream()), Seq(7)); +} + #[test] fn a_fork_of_a_truncated_stream_still_has_a_replay_base() { let (mut journal, _agg, seed) = driven(6); diff --git a/app/crates/ironstate-journal/tests/transactional.rs b/app/crates/ironstate-journal/tests/transactional.rs index 4ce365f..8036373 100644 --- a/app/crates/ironstate-journal/tests/transactional.rs +++ b/app/crates/ironstate-journal/tests/transactional.rs @@ -190,10 +190,15 @@ impl Journal for StagedJournal { Ok(self .records(stream) .iter() + .enumerate() .skip(start) - .flat_map(|r| r.events.iter()) - .map(|event| VersionedEvent { + .flat_map(|(i, r)| { + let seq = Seq(i as u64 + 1); + r.events.iter().map(move |event| (seq, event)) + }) + .map(|(seq, event)| VersionedEvent { event: event.clone(), + seq, type_name: type_name.clone(), version: 1, }) From 293e1ad8f754fb9f5a2107e4ff03b7545dd2cf50 Mon Sep 17 00:00:00 2001 From: Dashiel Lopez Mendez Date: Sat, 29 Aug 2026 11:07:29 -0400 Subject: [PATCH 04/14] docs(journal): renumber the retention property to close the gap Follows the renumbering in the parent commit: out-of-range addressing took the free number 9, so truncation-preserves-resume becomes 10 rather than leaving a hole an adapter author would have to ask about. Co-Authored-By: Claude Opus 5 --- app/crates/ironstate-journal/src/contract.rs | 18 +++++++++--------- 1 file changed, 9 insertions(+), 9 deletions(-) diff --git a/app/crates/ironstate-journal/src/contract.rs b/app/crates/ironstate-journal/src/contract.rs index 1029e4b..4746ebb 100644 --- a/app/crates/ironstate-journal/src/contract.rs +++ b/app/crates/ironstate-journal/src/contract.rs @@ -178,7 +178,7 @@ where let seed = run_seed(seed_base, case); let (mut journal, _live, _steps) = drive::(genesis.clone(), &seed, &mut runner, max_steps); - property_11_truncation_preserves_resume(&mut journal, &stream, &seed, case); + property_10_truncation_preserves_resume(&mut journal, &stream, &seed, case); } } @@ -440,13 +440,13 @@ where } } -/// Property 11 — **truncation preserves resume**. After truncating at a +/// Property 10 — **truncation preserves resume**. After truncating at a /// snapshot boundary, the stream resumes to a bit-identical aggregate and the /// same entropy position. /// /// This is what makes retention safe to offer at all: dropping history must be /// invisible to everything downstream of the snapshot it was taken against. -fn property_11_truncation_preserves_resume( +fn property_10_truncation_preserves_resume( journal: &mut J, stream: &StreamId, seed: &Seed, @@ -487,12 +487,12 @@ fn property_11_truncation_preserves_resume( for seq in horizon.0..=head.0 { let pos = journal.entropy_pos(stream, Seq(seq)).unwrap_or_else(|_| { panic!( - "[proven] property 11: entropy_pos undefined at retained Seq({seq}), case {case}" + "[proven] property 10: entropy_pos undefined at retained Seq({seq}), case {case}" ) }); assert!( pos >= previous, - "[proven] property 11: entropy_pos decreased at Seq({seq}), case {case}", + "[proven] property 10: entropy_pos decreased at Seq({seq}), case {case}", ); previous = pos; } @@ -502,7 +502,7 @@ fn property_11_truncation_preserves_resume( journal.entropy_pos(stream, Seq(gone)), Err(JournalError::UnknownSeq { .. }) ), - "[proven] property 11: truncated Seq({gone}) must be UnknownSeq, case {case}", + "[proven] property 10: truncated Seq({gone}) must be UnknownSeq, case {case}", ); } @@ -510,17 +510,17 @@ fn property_11_truncation_preserves_resume( assert_eq!( digest128(after.state()), digest128(before.state()), - "property 11: truncation changed what the stream resumes to, case {case}", + "property 10: truncation changed what the stream resumes to, case {case}", ); assert_eq!( entropy_after.draws(), entropy_before.draws(), - "property 11: truncation moved the resume entropy position, case {case}", + "property 10: truncation moved the resume entropy position, case {case}", ); assert_eq!( journal.retained_from(stream), at, - "property 11: retained_from must report the new horizon, case {case}", + "property 10: retained_from must report the new horizon, case {case}", ); } From 4ec492f6aee44b34daf50ca525499d433071cd0f Mon Sep 17 00:00:00 2001 From: Dashiel Lopez Mendez Date: Sat, 29 Aug 2026 11:17:38 -0400 Subject: [PATCH 05/14] fix(journal): remove the panic paths from MemoryJournal Review flagged an `expect` inside `append_in`, a public trait method. The crate's standing invariant is that a fallible operation returns a typed error rather than panicking, and "the invariant holds one line above" is a weaker guarantee than not needing one. - `append_in` computes the new head from `truncated + records.len()`, the same arithmetic `Stream::head()` does, instead of unwrapping its `Option`. - `stream_mut` uses the entry API rather than insert-then-`expect`. - `truncate_before`, also a public trait method, re-fetches mutably with a `let ... else` returning the same `UnknownSeq` its read-only checks would have, instead of asserting the lookup cannot fail. `MemoryJournal` now contains no `expect` or `unwrap` at all. Co-Authored-By: Claude Opus 5 --- app/crates/ironstate-journal/src/memory.rs | 23 +++++++++++----------- 1 file changed, 12 insertions(+), 11 deletions(-) diff --git a/app/crates/ironstate-journal/src/memory.rs b/app/crates/ironstate-journal/src/memory.rs index 3f2939d..36f583f 100644 --- a/app/crates/ironstate-journal/src/memory.rs +++ b/app/crates/ironstate-journal/src/memory.rs @@ -106,13 +106,10 @@ impl MemoryJournal { /// The stream's history, seeded with its genesis snapshot if this is the /// first time it has been touched. fn stream_mut(&mut self, id: &StreamId) -> &mut Stream { - if !self.streams.contains_key(id) { - let seeded = Stream::new(self.genesis_snapshot()); - self.streams.insert(id.clone(), seeded); - } + let genesis = self.genesis_snapshot(); self.streams - .get_mut(id) - .expect("the stream was just inserted") + .entry(id.clone()) + .or_insert_with(|| Stream::new(genesis)) } } @@ -142,7 +139,9 @@ impl Journal for MemoryJournal { events: events.to_vec(), entropy_pos, }); - Ok(stream.head().expect("a record was just pushed")) + // Computed directly rather than via `head()`, which returns an Option + // and would need unwrapping inside a public trait method. + Ok(Seq(stream.truncated + stream.records.len() as u64)) } fn snapshot_in( @@ -329,10 +328,12 @@ impl RetainableJournal for MemoryJournal { }); } - let stream = self - .streams - .get_mut(id) - .expect("the stream was found immediately above"); + // Re-fetch mutably now the read-only checks have passed. The lookup + // cannot fail here, but saying so with the same error the read phase + // would have returned keeps this method panic-free. + let Some(stream) = self.streams.get_mut(id) else { + return Err(JournalError::UnknownSeq { at }); + }; let drop_count = at.0.saturating_sub(1).saturating_sub(stream.truncated); let drop_count = usize::try_from(drop_count) From b59c06fc3bb2ae59c7e6da1e3bd1cfb31be7aeb4 Mon Sep 17 00:00:00 2001 From: Dashiel Lopez Mendez Date: Sat, 29 Aug 2026 11:26:06 -0400 Subject: [PATCH 06/14] fix(journal): make VersionedEvent forward-extensible, and point UnknownSeq at a real API MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two review notes on public surface. **`VersionedEvent` gained a field this release**, which is breaking for any adapter constructing or exhaustively destructuring it — and adapters must construct it, in `events_since`. Rather than spend that break silently, the struct is now `#[non_exhaustive]` with a `VersionedEvent::new` constructor. The next field is then additive for every adapter instead of breaking. This is the release to do it in: 0.2.0 already breaks the trait, and the alternative is paying the same cost again later. Making it `#[non_exhaustive]` immediately broke the three out-of-crate construction sites, which is the mechanism proving the constructor was needed; all three now go through it. **`JournalError::UnknownSeq` pointed at a "snapshot horizon"**, which is a phrase rather than an API. It now names `head(stream)` for the upper bound and `RetainableJournal::retained_from(stream)` for the lower — the two calls that actually answer the question, the second being the one retention introduces. Co-Authored-By: Claude Opus 5 --- app/crates/examples/async-store/src/main.rs | 7 +---- app/crates/ironstate-journal/src/journal.rs | 28 ++++++++++++++++++- app/crates/ironstate-journal/src/memory.rs | 7 +---- .../ironstate-journal/tests/transactional.rs | 7 +---- 4 files changed, 30 insertions(+), 19 deletions(-) diff --git a/app/crates/examples/async-store/src/main.rs b/app/crates/examples/async-store/src/main.rs index 0b0b982..0e9f61f 100644 --- a/app/crates/examples/async-store/src/main.rs +++ b/app/crates/examples/async-store/src/main.rs @@ -268,12 +268,7 @@ impl Log { let seq = Seq(i as u64 + 1); record.events.iter().map(move |event| (seq, event)) }) - .map(|(seq, event)| VersionedEvent { - event: event.clone(), - seq, - type_name: type_name.clone(), - version: 1, - }) + .map(|(seq, event)| VersionedEvent::new(event.clone(), seq, type_name.clone(), 1)) .collect()) } diff --git a/app/crates/ironstate-journal/src/journal.rs b/app/crates/ironstate-journal/src/journal.rs index 41ad2ad..accc7f8 100644 --- a/app/crates/ironstate-journal/src/journal.rs +++ b/app/crates/ironstate-journal/src/journal.rs @@ -65,6 +65,7 @@ pub struct Snapshot { /// A stored event tagged with the type and version it was written as, so a /// mixed-version stream can be upcast per event at load. +#[non_exhaustive] pub struct VersionedEvent { /// The event payload. pub event: A::Event, @@ -84,6 +85,29 @@ pub struct VersionedEvent { pub version: u32, } +impl VersionedEvent { + /// A stored event, tagged with the record it came from and the schema it + /// was written under. + /// + /// Adapters build these in `events_since`. The struct is + /// `#[non_exhaustive]`, so it is constructed through here rather than by + /// literal: a future field is then an additive change for every adapter + /// instead of a breaking one. + pub fn new( + event: A::Event, + seq: Seq, + type_name: impl Into>, + version: u32, + ) -> Self { + Self { + event, + seq, + type_name: type_name.into(), + version, + } + } +} + /// A failure from the storage layer. #[non_exhaustive] #[derive(Debug)] @@ -119,7 +143,9 @@ impl core::fmt::Display for JournalError { f, "no record at sequence {at:?}.\n\ The sequence is past the head, or below the earliest retained record.\n\ - Check `head()` and the snapshot horizon before addressing a Seq.", + Check `head(stream)` for the upper bound, and — on a journal that \ + truncates — `RetainableJournal::retained_from(stream)` for the lower \ + one, before addressing a Seq.", ), Self::NoBaseForFork { at } => write!( f, diff --git a/app/crates/ironstate-journal/src/memory.rs b/app/crates/ironstate-journal/src/memory.rs index 36f583f..ad56612 100644 --- a/app/crates/ironstate-journal/src/memory.rs +++ b/app/crates/ironstate-journal/src/memory.rs @@ -221,12 +221,7 @@ impl Journal for MemoryJournal { let seq = Seq(stream.truncated + i as u64 + 1); record.events.iter().map(move |event| (seq, event)) }) - .map(|(seq, event)| VersionedEvent { - event: event.clone(), - seq, - type_name: type_name.clone(), - version: 1, - }) + .map(|(seq, event)| VersionedEvent::new(event.clone(), seq, type_name.clone(), 1)) .collect()) } diff --git a/app/crates/ironstate-journal/tests/transactional.rs b/app/crates/ironstate-journal/tests/transactional.rs index 8036373..9670d0f 100644 --- a/app/crates/ironstate-journal/tests/transactional.rs +++ b/app/crates/ironstate-journal/tests/transactional.rs @@ -196,12 +196,7 @@ impl Journal for StagedJournal { let seq = Seq(i as u64 + 1); r.events.iter().map(move |event| (seq, event)) }) - .map(|(seq, event)| VersionedEvent { - event: event.clone(), - seq, - type_name: type_name.clone(), - version: 1, - }) + .map(|(seq, event)| VersionedEvent::new(event.clone(), seq, type_name.clone(), 1)) .collect()) } From 734db4aa890a426acf4b9c6c175215d77dbdb79e Mon Sep 17 00:00:00 2001 From: Dashiel Lopez Mendez Date: Sat, 29 Aug 2026 11:28:06 -0400 Subject: [PATCH 07/14] fix(journal): tighten truncation's preconditions and the macro's arg order MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three more review findings. **`truncate_before` accepted `Seq(0)`.** `at` names the first record to keep, and genesis is not a record — so the call reported success while `retained_from` still said `Seq(1)`, having kept nothing "from Seq(0) onward" because there is no such thing. Now refused. **A snapshot recorded past the head could authorise a truncation.** The safety check took the newest snapshot by `Seq` over *all* stored snapshots, so a bogus one beyond the head satisfied it — vouching for a truncation no real base covered. `latest_snapshot_at` now ignores snapshots above the head, since replay could never start from one anyway. Both cases get a regression test. **`journal_contract_test!` only matched `forkable, retainable`.** The markers name capabilities, which are a set rather than a sequence, so the reversed order failing with a macro error was a pure footgun. A second arm forwards it. Also corrects the `events_since` comment: it claimed `None` means "from this stream's start", which reads as the horizon. `None` is `after = Seq(0)` — genesis — so on a truncated stream it is itself below the horizon and refused. The comment now says that, and names the call that does read everything retained. Co-Authored-By: Claude Opus 5 --- app/crates/ironstate-journal/src/macros.rs | 6 ++ app/crates/ironstate-journal/src/memory.rs | 30 +++++++--- .../ironstate-journal/tests/retention.rs | 56 +++++++++++++++++++ 3 files changed, 83 insertions(+), 9 deletions(-) diff --git a/app/crates/ironstate-journal/src/macros.rs b/app/crates/ironstate-journal/src/macros.rs index e038eb3..c009860 100644 --- a/app/crates/ironstate-journal/src/macros.rs +++ b/app/crates/ironstate-journal/src/macros.rs @@ -26,6 +26,7 @@ /// ironstate_journal::journal_contract_test!(MyPostgresJournal, MatchState); /// ironstate_journal::journal_contract_test!(MyForkingJournal, MatchState, forkable); /// ironstate_journal::journal_contract_test!(MyStore, MatchState, forkable, retainable); +/// // capability markers may be given in either order /// ``` #[macro_export] macro_rules! journal_contract_test { @@ -58,6 +59,11 @@ macro_rules! journal_contract_test { $crate::testkit_support::run_contract_retainable::<$journal, $agg>(32, 24, 0xC047); } }; + // The capabilities are a set, not a sequence — accept either order rather + // than failing to match with a macro error. + ($journal:ty, $agg:ty, retainable, forkable) => { + $crate::journal_contract_test!($journal, $agg, forkable, retainable); + }; } /// Generate a `#[test]` for the seeded whole-tier simulation: a fault-injected diff --git a/app/crates/ironstate-journal/src/memory.rs b/app/crates/ironstate-journal/src/memory.rs index ad56612..87bb739 100644 --- a/app/crates/ironstate-journal/src/memory.rs +++ b/app/crates/ironstate-journal/src/memory.rs @@ -66,8 +66,17 @@ impl Stream { } /// The newest snapshot's sequence, if the stream has one. + /// The newest snapshot's sequence, ignoring any recorded past the head. + /// + /// A snapshot beyond the head describes state this stream does not have, so + /// it cannot vouch for a truncation — replay could never start from it. fn latest_snapshot_at(&self) -> Option { - self.snapshots.iter().map(|s| s.at).max() + let head = self.truncated + self.records.len() as u64; + self.snapshots + .iter() + .map(|s| s.at) + .filter(|at| at.0 <= head) + .max() } } @@ -197,9 +206,11 @@ impl Journal for MemoryJournal { // Reading from below the horizon would silently return a *gapped* list: // the caller asked to continue from a point whose successors are partly // discarded. A subscription handed that list would drop records without - // ever seeing an error, so refuse instead — `retained_from` says where a - // valid read starts. `None` means "from this stream's start", which is - // the horizon, so it is only valid on an untruncated stream. + // ever seeing an error, so refuse instead. + // + // `None` means "from genesis" — it is `after = Seq(0)` — so on a + // truncated stream it is itself below the horizon and refused. To read + // everything still retained, pass `Some(Seq(retained_from - 1))`. let from = after.map_or(0, |s| s.0); if from < stream.truncated { return Err(JournalError::UnknownSeq { @@ -302,12 +313,13 @@ impl RetainableJournal for MemoryJournal { return Err(JournalError::UnknownSeq { at }); }; - // `at` may sit one past the head (discard everything) but no further. - // Without this a snapshot recorded beyond the head would authorise an - // arbitrary truncation, and `retained_from` would then disagree with the - // `at` that was asked for. + // `at` names the first record to keep, so it must be a record sequence: + // `Seq(0)` is genesis and keeping "from genesis onward" is not something + // this can express — it would report success while `retained_from` still + // said `Seq(1)`. The upper bound is one past the head, which means + // discard everything. let head = stream.truncated + stream.records.len() as u64; - if at.0 > head + 1 { + if at.0 == 0 || at.0 > head + 1 { return Err(JournalError::UnknownSeq { at }); } diff --git a/app/crates/ironstate-journal/tests/retention.rs b/app/crates/ironstate-journal/tests/retention.rs index 43a5e9d..e35b56e 100644 --- a/app/crates/ironstate-journal/tests/retention.rs +++ b/app/crates/ironstate-journal/tests/retention.rs @@ -415,3 +415,59 @@ fn a_stale_base_below_the_horizon_never_produces_a_wrong_state() { }, } } + +/// `at` names the first record to keep, so `Seq(0)` — genesis, not a record — +/// is not a truncation point. Accepting it would report success while +/// `retained_from` still said `Seq(1)`. +#[test] +fn truncating_before_genesis_is_refused() { + let (mut journal, _agg, _seed) = driven(6); + match journal.truncate_before(&stream(), Seq(0)) { + Err(JournalError::UnknownSeq { at }) => assert_eq!(at, Seq(0)), + other => panic!("expected UnknownSeq for Seq(0), got {other:?}"), + } + assert_eq!(journal.retained_from(&stream()), Seq(1)); +} + +/// A snapshot recorded past the head describes state the stream does not have, +/// so it must not vouch for a truncation that would otherwise be illegal. +#[test] +fn a_snapshot_past_the_head_cannot_authorise_truncation() { + let seed = Seed([3; 32]); + let mut journal = MemoryJournal::new(genesis()); + let mut agg = Aggregate::new(genesis()).unwrap(); + for _ in 0..4 { + let pos = journal + .head(&stream()) + .map_or(DrawPos(0), |h| journal.entropy_pos(&stream(), h).unwrap()); + let mut c = ctx(&seed, pos); + execute(&mut journal, &stream(), &mut agg, &Command::Tick, &mut c).unwrap(); + } + // The only real snapshot is genesis at Seq(0). Add a bogus one past the head. + journal + .snapshot_in( + &mut (), + &stream(), + Snapshot { + state: agg.state().clone(), + schema_version: 0, + at: Seq(99), + entropy_pos: DrawPos(0), + }, + ) + .unwrap(); + + // Truncating before Seq(3) needs a base at Seq(2) or later; only the bogus + // snapshot qualifies, and it must not count. + match journal.truncate_before(&stream(), Seq(3)) { + Err(JournalError::NoSnapshotForTruncation { + latest_snapshot, .. + }) => assert_eq!( + latest_snapshot, + Some(Seq(0)), + "the out-of-range snapshot must be ignored when reporting the newest base", + ), + other => panic!("expected refusal, got {other:?}"), + } + assert_eq!(journal.retained_from(&stream()), Seq(1)); +} From a04c92fee20e9452615503b97a7f228c208dd776 Mon Sep 17 00:00:00 2001 From: Dashiel Lopez Mendez Date: Sat, 29 Aug 2026 11:32:24 -0400 Subject: [PATCH 08/14] docs(journal): say what `streams()` promises, and fix a contradictory comment MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three review notes, all on clarity rather than behaviour. A test comment said `None` means "from this stream's start, which is now the horizon" while the assertion directly below expects `UnknownSeq { at: Seq(0) }`. The assertion was right: `None` is `after = Seq(0)` — genesis — so after truncation it is itself below the horizon. The comment now says that. `RetainableJournal::streams()` promised nothing about order, and a test was comparing the returned `Vec` directly. `MemoryJournal` happens to be backed by a `BTreeMap`, so its order is in fact stable — but that is an implementation detail no adapter owes, and the test's intent is that the *set* of streams is unchanged. The rustdoc now says the order is unspecified, and the test sorts before comparing. The semver consequences of `VersionedEvent` gaining a required field and `#[non_exhaustive]` are now stated in the PR description rather than left under an "additive" summary that stopped being true. Co-Authored-By: Claude Opus 5 --- app/crates/ironstate-journal/src/journal.rs | 4 ++++ app/crates/ironstate-journal/tests/retention.rs | 11 +++++++---- 2 files changed, 11 insertions(+), 4 deletions(-) diff --git a/app/crates/ironstate-journal/src/journal.rs b/app/crates/ironstate-journal/src/journal.rs index accc7f8..e553b82 100644 --- a/app/crates/ironstate-journal/src/journal.rs +++ b/app/crates/ironstate-journal/src/journal.rs @@ -351,6 +351,10 @@ pub trait RetainableJournal: Journal { /// Every stream this journal holds, so a retention sweep can enumerate what /// it might expire. /// + /// The order is unspecified — treat the result as a set. An adapter is free + /// to return rows in whatever order its store yields them, and callers that + /// need a stable order should sort. + /// /// This lives here rather than on [`Journal`] because sweeping is the only /// thing that needs it, and on a relational store it is a full-table scan /// no ordinary adapter should be made to implement. diff --git a/app/crates/ironstate-journal/tests/retention.rs b/app/crates/ironstate-journal/tests/retention.rs index e35b56e..7e70fc7 100644 --- a/app/crates/ironstate-journal/tests/retention.rs +++ b/app/crates/ironstate-journal/tests/retention.rs @@ -293,7 +293,8 @@ fn reading_from_below_the_horizon_is_refused_not_gapped() { other.map(|e| e.len()) ), } - // `None` means "from this stream's start", which is now the horizon. + // `None` is `after = Seq(0)` — genesis — not "whatever this stream still + // has", so after truncation it is itself below the horizon and refused. match journal.events_since(&stream(), None) { Err(JournalError::UnknownSeq { at }) => assert_eq!(at, Seq(0)), other => panic!( @@ -326,14 +327,16 @@ fn genesis_is_unknown_once_it_is_below_the_horizon() { #[test] fn a_refused_truncation_does_not_materialise_the_stream() { let (mut journal, _agg, _seed) = driven(6); - let before = journal.streams().unwrap(); + let mut before = journal.streams().unwrap(); + before.sort(); let typo = StreamId::new("typo"); assert!(journal.truncate_before(&typo, Seq(5)).is_err()); + let mut after = journal.streams().unwrap(); + after.sort(); assert_eq!( - journal.streams().unwrap(), - before, + after, before, "a refused truncation must not add a phantom stream", ); } From d80e26fa5632e801af2d777220cec65782536774 Mon Sep 17 00:00:00 2001 From: Dashiel Lopez Mendez Date: Sat, 29 Aug 2026 11:36:05 -0400 Subject: [PATCH 09/14] docs(journal): document the error paths this stack added MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `truncate_before`'s rustdoc listed only `NoSnapshotForTruncation` and `Storage`, but it also returns `UnknownSeq` for a `Seq(0)` truncation point, for an `at` beyond one past the head, and for an unknown stream. A caller cannot handle deliberately what the contract does not mention. Checking the rest of the surface for the same omission found two more, both introduced by this stack: - `fork` gained `NoBaseForFork`, and its refusal below the retained horizon, neither of which were listed. - `events_since` gained a refusal when read from below the horizon — including `None`, which means genesis — and still claimed only `Storage`. Co-Authored-By: Claude Opus 5 --- app/crates/ironstate-journal/src/journal.rs | 18 ++++++++++++++---- 1 file changed, 14 insertions(+), 4 deletions(-) diff --git a/app/crates/ironstate-journal/src/journal.rs b/app/crates/ironstate-journal/src/journal.rs index e553b82..071417d 100644 --- a/app/crates/ironstate-journal/src/journal.rs +++ b/app/crates/ironstate-journal/src/journal.rs @@ -287,7 +287,10 @@ pub trait Journal { /// /// # Errors /// - /// Returns [`JournalError::Storage`] if the underlying store failed. + /// Returns [`JournalError::UnknownSeq`] if `after` is below the stream's + /// retained horizon — including `None`, which means genesis — since the + /// result would otherwise have a silent gap in it. Returns + /// [`JournalError::Storage`] if the underlying store failed. fn events_since( &self, stream: &StreamId, @@ -316,7 +319,10 @@ pub trait ForkableJournal: Journal { /// /// # Errors /// - /// Returns [`JournalError::UnknownSeq`] if `at` is past the stream's head. + /// Returns [`JournalError::UnknownSeq`] if `at` is past the stream's head + /// or at or below its retained horizon, and + /// [`JournalError::NoBaseForFork`] if no snapshot at or below `at` + /// survives — the branch would have nothing to replay from. fn fork(&self, stream: &StreamId, at: Seq) -> Result where Self: Sized; @@ -340,8 +346,12 @@ pub trait RetainableJournal: Journal { /// /// # Errors /// - /// Returns [`JournalError::NoSnapshotForTruncation`] if no snapshot covers - /// the retained prefix, or [`JournalError::Storage`] if the store failed. + /// Returns [`JournalError::UnknownSeq`] if `at` is not a truncation point + /// this stream can express — `Seq(0)` is genesis rather than a record, `at` + /// beyond one past the head would discard more than exists, and an unknown + /// stream has nothing to truncate. Returns + /// [`JournalError::NoSnapshotForTruncation`] if no snapshot covers the + /// retained prefix, or [`JournalError::Storage`] if the store failed. fn truncate_before(&mut self, stream: &StreamId, at: Seq) -> Result<(), JournalError>; /// The earliest sequence still retained in `stream` — everything below it From 5d7f21c447dc415508498604aa7b6ea8e3d6b9da Mon Sep 17 00:00:00 2001 From: Dashiel Lopez Mendez Date: Sat, 29 Aug 2026 11:42:24 -0400 Subject: [PATCH 10/14] docs(journal): make the "read everything retained" guidance compile MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The comment told callers to pass `Some(Seq(retained_from - 1))`, which is not valid Rust twice over: `retained_from` is a method, and `Seq` has no `Sub`. It now shows the real expression — `events_since(stream, Some(Seq(journal.retained_from(stream).0 - 1)))` — and notes that it is `Seq(0)` on an untruncated stream, so it degenerates to the same read as `None`. That last claim is now a test rather than an assertion in prose: the two agree before truncation, and afterwards `None` is refused while the horizon expression still returns exactly the retained records. Co-Authored-By: Claude Opus 5 --- app/crates/ironstate-journal/src/memory.rs | 5 ++- .../ironstate-journal/tests/retention.rs | 35 +++++++++++++++++++ 2 files changed, 39 insertions(+), 1 deletion(-) diff --git a/app/crates/ironstate-journal/src/memory.rs b/app/crates/ironstate-journal/src/memory.rs index 87bb739..871a930 100644 --- a/app/crates/ironstate-journal/src/memory.rs +++ b/app/crates/ironstate-journal/src/memory.rs @@ -210,7 +210,10 @@ impl Journal for MemoryJournal { // // `None` means "from genesis" — it is `after = Seq(0)` — so on a // truncated stream it is itself below the horizon and refused. To read - // everything still retained, pass `Some(Seq(retained_from - 1))`. + // everything still retained, ask for the record before the horizon: + // `events_since(stream, Some(Seq(journal.retained_from(stream).0 - 1)))`, + // which is `Seq(0)` on an untruncated stream and so equivalent to `None` + // there. let from = after.map_or(0, |s| s.0); if from < stream.truncated { return Err(JournalError::UnknownSeq { diff --git a/app/crates/ironstate-journal/tests/retention.rs b/app/crates/ironstate-journal/tests/retention.rs index 7e70fc7..3ab19da 100644 --- a/app/crates/ironstate-journal/tests/retention.rs +++ b/app/crates/ironstate-journal/tests/retention.rs @@ -474,3 +474,38 @@ fn a_snapshot_past_the_head_cannot_authorise_truncation() { } assert_eq!(journal.retained_from(&stream()), Seq(1)); } + +/// The documented way to read "everything still retained" — ask for the record +/// before the horizon — works on a truncated stream and degenerates to `None` +/// on an untruncated one. +#[test] +fn reading_from_the_horizon_is_the_way_to_read_everything_retained() { + let (mut journal, _agg, _seed) = driven(6); + + // Untruncated: the horizon expression is Seq(0), equivalent to `None`. + let from_horizon = Seq(journal.retained_from(&stream()).0 - 1); + assert_eq!(from_horizon, Seq(0)); + assert_eq!( + journal + .events_since(&stream(), Some(from_horizon)) + .unwrap() + .len(), + journal.events_since(&stream(), None).unwrap().len(), + "on an untruncated stream the two must agree", + ); + + journal.truncate_before(&stream(), Seq(4)).unwrap(); + + // Truncated: `None` is refused, the horizon expression still works. + assert!(journal.events_since(&stream(), None).is_err()); + let from_horizon = Seq(journal.retained_from(&stream()).0 - 1); + assert_eq!(from_horizon, Seq(3)); + assert_eq!( + journal + .events_since(&stream(), Some(from_horizon)) + .unwrap() + .len(), + 3, + "records 4..6 are what remains", + ); +} From 0c4d9cb029c9a7dcf5daf6e402ed3ce2d1a5c6a9 Mon Sep 17 00:00:00 2001 From: Dashiel Lopez Mendez Date: Sat, 29 Aug 2026 11:49:03 -0400 Subject: [PATCH 11/14] chore(journal): release 0.3.0, and correct the README the reshape left behind MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three things this PR needed and did not have. **The version was already published.** 0.2.0 went to crates.io when the reshape merged, so this PR's breaking changes — `VersionedEvent` gaining a required field and `#[non_exhaustive]`, `streams` moving to `RetainableJournal` — would have shipped nothing: `release.yml` publishes only what is ahead of crates.io, and 0.2.0 is not ahead of 0.2.0. Bumped to 0.3.0, with the workspace requirement following. **0.2.0 had no changelog entry at all.** It was published by version-ahead detection rather than a release-plz PR, so nothing generated one. Recorded now, after the fact and marked as such, because a published version with no record of what changed is worse than a late entry. **The README documented the pre-reshape API.** It is the front page on crates.io and docs.rs for a release that has been live since the reshape merged: `execute`/`resume` without a `StreamId`, `fork` listed as a core operation, "seven-property conformance suite", and no mention of either headline feature. It now shows the real signatures, introduces streams, has a section on appending inside the caller's transaction — including why `execute_in` does not evolve the aggregate — and names the two opt-in capabilities. Also refuses `truncate_before` at or below an existing horizon, per review: those records are gone, so the call could not keep anything "from `at` onward" and returned `Ok(())` while `retained_from` stayed put. Truncating to exactly the current horizon remains an honest no-op. Regression test added. Co-Authored-By: Claude Opus 5 --- app/Cargo.lock | 2 +- app/Cargo.toml | 2 +- app/crates/ironstate-journal/CHANGELOG.md | 62 +++++++++++++++++++ app/crates/ironstate-journal/Cargo.toml | 2 +- app/crates/ironstate-journal/README.md | 49 ++++++++++++--- app/crates/ironstate-journal/src/memory.rs | 16 +++-- .../ironstate-journal/tests/retention.rs | 29 +++++++++ 7 files changed, 145 insertions(+), 17 deletions(-) diff --git a/app/Cargo.lock b/app/Cargo.lock index cd13af6..8e89202 100644 --- a/app/Cargo.lock +++ b/app/Cargo.lock @@ -216,7 +216,7 @@ dependencies = [ [[package]] name = "ironstate-journal" -version = "0.2.0" +version = "0.3.0" dependencies = [ "ironstate", "ironstate-aggregate", diff --git a/app/Cargo.toml b/app/Cargo.toml index fa9d53f..69995be 100644 --- a/app/Cargo.toml +++ b/app/Cargo.toml @@ -25,7 +25,7 @@ ironstate = { version = "0.1", path = "crates/ironstate" } ironstate-derive = { version = "0.1", path = "crates/ironstate-derive" } ironstate-aggregate = { version = "0.1", path = "crates/ironstate-aggregate" } ironstate-aggregate-derive = { version = "0.1", path = "crates/ironstate-aggregate-derive" } -ironstate-journal = { version = "0.2", path = "crates/ironstate-journal" } +ironstate-journal = { version = "0.3", path = "crates/ironstate-journal" } # external — pin minor, let patch float rustc-stable-hash = "0.1" # rust-lang-maintained engine behind Digest128 diff --git a/app/crates/ironstate-journal/CHANGELOG.md b/app/crates/ironstate-journal/CHANGELOG.md index 929551d..b575a9c 100644 --- a/app/crates/ironstate-journal/CHANGELOG.md +++ b/app/crates/ironstate-journal/CHANGELOG.md @@ -5,6 +5,68 @@ on [Keep a Changelog](https://keepachangelog.com/); from the next release on, th entries are maintained by release-plz. This crate is the journal tier of the [ironstate](https://github.com/kassian-dev/ironstate) family. +## [0.3.0](https://github.com/kassian-dev/ironstate/compare/ironstate-journal-v0.2.0...ironstate-journal-v0.3.0) - 2026-08-29 + +### Added + +- `RetainableJournal`: `truncate_before` discards the oldest records in a + stream, refusing anything that would leave it unresumable; `retained_from` + reports the horizon and `streams` enumerates what a retention sweep could + expire. Conformance property 10 proves a truncated stream resumes to a + bit-identical aggregate. +- `VersionedEvent::new`, the constructor adapters build results with. + +### Changed + +- **Breaking:** `VersionedEvent` carries the `Seq` of the record an event came + from, and is now `#[non_exhaustive]`. `events_since` flattens events across + records, so an index into its result is not a sequence — a subscription keyed + on one silently dropped events as duplicates once any `decide` emitted more + than one. Construct through `VersionedEvent::new`; the next field is then + additive rather than breaking. +- `JournalError` gains `NoBaseForFork` and `NoSnapshotForTruncation`, and + `UnknownSeq` now also covers reads, forks and truncations below a retained + horizon. + +### Fixed + +- `MemoryJournal` no longer panics anywhere: `append_in` and `truncate_before` + return typed errors rather than asserting internal invariants. +- Truncation no longer strands a fork on a snapshot whose records it discarded, + which could return a wrong aggregate as `Ok`. +- A fully truncated stream reports its head instead of `None`, so `execute` no + longer rewinds the entropy stream to `DrawPos(0)`. +- Reading from below the retained horizon is refused rather than returning a + silently gapped list. + +## [0.2.0](https://github.com/kassian-dev/ironstate/compare/ironstate-journal-v0.1.4...ironstate-journal-v0.2.0) - 2026-08-29 + +Recorded after the fact: 0.2.0 was published by the release workflow on a +version bump rather than through a release-plz PR, so no entry was generated +at the time. + +### Changed + +- **Breaking:** `Journal` takes a `&StreamId` on every operation, and `Seq` is + per-stream. One journal value now holds many aggregate instances, which makes + the write side agree with `Subscription`, whose idempotency key was already + `(StreamId, Seq)`. +- **Breaking:** `Journal` gains an associated `Tx<'a>`, the caller's unit of + work, with `append_in`/`snapshot_in`. A journal owning its durability sets it + to `()`, and `execute`/`resume` keep their previous shape through a + `for<'a> Journal = ()>` bound. +- **Breaking:** `fork` moves to `ForkableJournal`. With per-stream sequences a + whole-journal fork point is ill-defined, and a relational adapter should not + owe row-copying for a method it never calls. + +### Added + +- `execute_in` and `Pending`, for a journal enlisting in the caller's + transaction. `execute_in` deliberately does not evolve the aggregate: a + rollback after the append would otherwise leave it ahead of the durable log. +- Conformance properties for stream independence and out-of-range addressing; + `run_contract_forkable` for adapters that branch. + ## [0.1.4](https://github.com/kassian-dev/ironstate/compare/ironstate-journal-v0.1.3...ironstate-journal-v0.1.4) - 2026-06-23 ### Other diff --git a/app/crates/ironstate-journal/Cargo.toml b/app/crates/ironstate-journal/Cargo.toml index 8d46d9c..17e6610 100644 --- a/app/crates/ironstate-journal/Cargo.toml +++ b/app/crates/ironstate-journal/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "ironstate-journal" -version = "0.2.0" +version = "0.3.0" edition.workspace = true rust-version.workspace = true license.workspace = true diff --git a/app/crates/ironstate-journal/README.md b/app/crates/ironstate-journal/README.md index e1750c6..90a058c 100644 --- a/app/crates/ironstate-journal/README.md +++ b/app/crates/ironstate-journal/README.md @@ -1,32 +1,65 @@ # →(Fe) ironstate-journal The event journal for [ironstate](https://docs.rs/ironstate) aggregates: -append, snapshot, replay, fork, and subscribe — with the entropy position -recorded atomically beside every event, so an aggregate can be replayed and -resumed bit-identically. +append, snapshot, replay and subscribe — with the entropy position recorded +atomically beside every event, so an aggregate can be replayed and resumed +bit-identically. Because replay consumes no entropy (only `decide` draws), the draw position cannot be recomputed from the events — it must be stored. Every append persists the post-`decide` position together with its events, in one atomic unit. +One journal holds **many streams**, each an aggregate instance with its own +`Seq` line, so the write side addresses history the same way `Subscription` +always has. + ```rust,ignore // The canonical persistent loop: structural checks → decide → append → evolve. -let seq = execute(&mut journal, &mut aggregate, &command, &mut ctx)?; +let stream = StreamId::new("account-42"); +let seq = execute(&mut journal, &stream, &mut aggregate, &command, &mut ctx)?; // Rebuild an aggregate (and its entropy stream) from the journal. -let (aggregate, entropy) = resume(&journal, &seed)?; +let (aggregate, entropy) = resume(&journal, &stream, &seed)?; // A published, collision-resistant digest anyone can recompute to audit a match. let digest = replay_hash(snapshot, &events)?; ``` +## Appending inside your own transaction + +A journal that owns its durability sets `type Tx<'a> = ()` and the calls above +apply unchanged. A journal living in the same database as your read models sets +`Tx` to that database's transaction, so the append and your other writes commit +together or not at all: + +```rust,ignore +let mut tx = pool.begin()?; +let pending = execute_in(&mut journal, &mut tx, &stream, &aggregate, &cmd, &mut ctx)?; +write_read_model(&mut tx, ...)?; // your other writes, +enqueue_outbound_job(&mut tx, ...)?; // in the same transaction +match tx.commit() { + Ok(()) => { let seq = pending.commit(&mut aggregate); } + Err(e) => { pending.abort(&mut ctx); return Err(e.into()); } +} +``` + +`execute_in` deliberately does *not* evolve the aggregate — a rollback after the +append would otherwise leave it silently ahead of the durable log. The returned +`Pending` is driven once your transaction resolves. + +Two capabilities are opt-in, because not every store can or should offer them: +`ForkableJournal` branches a stream's history (essential for simulation, +meaningless for a statutory record), and `RetainableJournal` truncates the +oldest records once a snapshot covers what remains. + A `Subscription` delivers one aggregate's events to another exactly once (`(StreamId, Seq)` idempotency keys), and `Versioned` events upcast through a `MigrateFrom` chain on load. -The reference in-memory journal (feature `memory`, default) passes a -seven-property conformance suite (`journal_contract_test!`) that every storage -adapter is judged against; storage adapters themselves live downstream. Under +The reference in-memory journal (feature `memory`, default) passes the +conformance suite (`journal_contract_test!`) that every storage adapter is +judged against — eight properties for any journal, plus two for forkable ones +and one for retainable ones; storage adapters themselves live downstream. Under feature `sim`, `scenario_test!` drives a seeded whole-tier fault simulation (append failures, crash/resume, fork/continue) and checks faults are invisible to outcomes, with the `FaultInjector`/`ReferenceRun` testkit exposed for diff --git a/app/crates/ironstate-journal/src/memory.rs b/app/crates/ironstate-journal/src/memory.rs index 871a930..ebb6a86 100644 --- a/app/crates/ironstate-journal/src/memory.rs +++ b/app/crates/ironstate-journal/src/memory.rs @@ -316,13 +316,17 @@ impl RetainableJournal for MemoryJournal { return Err(JournalError::UnknownSeq { at }); }; - // `at` names the first record to keep, so it must be a record sequence: - // `Seq(0)` is genesis and keeping "from genesis onward" is not something - // this can express — it would report success while `retained_from` still - // said `Seq(1)`. The upper bound is one past the head, which means - // discard everything. + // `at` names the first record to keep, so it must name a record this + // stream still has. Anything at or below the horizon cannot be kept + // "from `at` onward" — the records are gone — so reporting success + // while `retained_from` stayed put would be a silent no-op. That also + // covers `Seq(0)`, which is genesis rather than a record. The upper + // bound is one past the head, meaning discard everything. + // + // `at == retained_from` is the one no-op that is honest: the stream is + // already truncated to exactly there, so truncation is idempotent. let head = stream.truncated + stream.records.len() as u64; - if at.0 == 0 || at.0 > head + 1 { + if at.0 <= stream.truncated || at.0 > head + 1 { return Err(JournalError::UnknownSeq { at }); } diff --git a/app/crates/ironstate-journal/tests/retention.rs b/app/crates/ironstate-journal/tests/retention.rs index 3ab19da..ba25357 100644 --- a/app/crates/ironstate-journal/tests/retention.rs +++ b/app/crates/ironstate-journal/tests/retention.rs @@ -509,3 +509,32 @@ fn reading_from_the_horizon_is_the_way_to_read_everything_retained() { "records 4..6 are what remains", ); } + +/// Truncating to a point already below the horizon cannot keep records "from +/// `at` onward" — they are gone — so it must refuse rather than report success +/// while `retained_from` stays put. This matches `record_at` and `fork`, which +/// both treat sequences at or below the horizon as unknown. +#[test] +fn truncating_below_an_existing_horizon_is_refused() { + let (mut journal, _agg, _seed) = driven(6); + journal.truncate_before(&stream(), Seq(4)).unwrap(); + assert_eq!(journal.retained_from(&stream()), Seq(4)); + + for stale in [Seq(1), Seq(2), Seq(3)] { + match journal.truncate_before(&stream(), stale) { + Err(JournalError::UnknownSeq { at }) => assert_eq!(at, stale), + other => panic!("expected UnknownSeq truncating to {stale:?}, got {other:?}"), + } + assert_eq!( + journal.retained_from(&stream()), + Seq(4), + "a refused truncation must not move the horizon", + ); + } + + // Truncating to exactly the current horizon is the one honest no-op: + // the stream is already there, so truncation is idempotent. + journal.truncate_before(&stream(), Seq(4)).unwrap(); + assert_eq!(journal.retained_from(&stream()), Seq(4)); + assert_eq!(journal.head(&stream()), Some(Seq(6))); +} From 3f9830c4e263dc2da7a4d37c91e88317388b001a Mon Sep 17 00:00:00 2001 From: Dashiel Lopez Mendez Date: Sat, 29 Aug 2026 11:57:09 -0400 Subject: [PATCH 12/14] docs(journal): label every contract assertion, not most of them MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit AGENTS.md is unambiguous — "every analysis/test-macro claim is labeled `[proven]` or `[sampled]`. No unlabeled claims" — and the conformance suite was following it about half the time. Properties 2, 7, 8 and 9 carried `[proven]`; 1, 3, 4, 5 and 6 carried nothing, and property 10 was split down the middle, three labelled and three not. Review caught the property 10 half. The rest is the same defect with an older date, so all 26 assertions now carry the label. Failure output is classifiable by grep again, which is the point of having a vocabulary. One thing this does not settle, deliberately: whether a suite driven by generated command streams should say `[proven]` at all, rather than `[sampled]`. The checks that loop over every `Seq` in a journal are exhaustive *for that journal*, but the journal itself was sampled — so the honest label arguably differs per assertion, and for some of them it is not `[proven]`. That question is about the crate's honesty vocabulary and deserves deciding on its own rather than inside a retention PR; consistency first, correctness of the choice second. Co-Authored-By: Claude Opus 5 --- app/crates/ironstate-journal/src/contract.rs | 24 ++++++++++---------- 1 file changed, 12 insertions(+), 12 deletions(-) diff --git a/app/crates/ironstate-journal/src/contract.rs b/app/crates/ironstate-journal/src/contract.rs index 4746ebb..d363c57 100644 --- a/app/crates/ironstate-journal/src/contract.rs +++ b/app/crates/ironstate-journal/src/contract.rs @@ -269,7 +269,7 @@ fn property_1_round_trip( assert_eq!( digest128(rebuilt.state()), *live_digest, - "property 1: replay of the whole log did not reproduce the live digest, case {case}", + "[proven] property 1: replay of the whole log did not reproduce the live digest, case {case}", ); } @@ -293,7 +293,7 @@ fn property_1_round_trip_at_each_step( assert_eq!( digest128(rebuilt.state()), *live_digest, - "property 1b: replay did not reproduce the live digest at {seq:?}, case {case}", + "[proven] property 1b: replay did not reproduce the live digest at {seq:?}, case {case}", ); } } @@ -348,7 +348,7 @@ fn property_3_resume_identity( assert_eq!( digest128(resumed.state()), digest128(live.state()), - "property 3: resume-to-head then handle diverged from the live handle, case {case}", + "[proven] property 3: resume-to-head then handle diverged from the live handle, case {case}", ); } @@ -363,12 +363,12 @@ where assert_eq!( branch.entropy_pos(stream, at).expect("branch position"), journal.entropy_pos(stream, at).expect("main position"), - "property 4: entropy_pos disagreed at the fork point, case {case}", + "[proven] property 4: entropy_pos disagreed at the fork point, case {case}", ); assert_eq!( branch.head(stream), Some(at), - "property 4: a fork's head should sit at the fork point, case {case}", + "[proven] property 4: a fork's head should sit at the fork point, case {case}", ); } } @@ -386,7 +386,7 @@ where assert_eq!( entropy.draws(), pos, - "property 5: resume must position entropy at the head, not an earlier snapshot, case {case}", + "[proven] property 5: resume must position entropy at the head, not an earlier snapshot, case {case}", ); } @@ -417,18 +417,18 @@ where assert_eq!( journal.head(&stream), None, - "property 6: a failed append journaled something, case {case}" + "[proven] property 6: a failed append journaled something, case {case}" ); assert_eq!( digest128(aggregate.state()), before, - "property 6: a failed append mutated the state, case {case}", + "[proven] property 6: a failed append mutated the state, case {case}", ); let pos = ctx.entropy_mut().map_or(DrawPos(0), |e| e.draws()); assert_eq!( pos, DrawPos(0), - "property 6: a failed append left the entropy advanced, case {case}" + "[proven] property 6: a failed append left the entropy advanced, case {case}" ); return; } @@ -510,17 +510,17 @@ fn property_10_truncation_preserves_resume( assert_eq!( digest128(after.state()), digest128(before.state()), - "property 10: truncation changed what the stream resumes to, case {case}", + "[proven] property 10: truncation changed what the stream resumes to, case {case}", ); assert_eq!( entropy_after.draws(), entropy_before.draws(), - "property 10: truncation moved the resume entropy position, case {case}", + "[proven] property 10: truncation moved the resume entropy position, case {case}", ); assert_eq!( journal.retained_from(stream), at, - "property 10: retained_from must report the new horizon, case {case}", + "[proven] property 10: retained_from must report the new horizon, case {case}", ); } From ca7068148f3f8b2e84764c616d849487d454eac6 Mon Sep 17 00:00:00 2001 From: Dashiel Lopez Mendez Date: Sat, 29 Aug 2026 11:59:46 -0400 Subject: [PATCH 13/14] perf(journal): stop cloning the genesis state on every append MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two review findings. `stream_mut` hoisted `genesis_snapshot()` above the entry lookup, so every append cloned the whole aggregate genesis state — only to discard it when the stream already existed. That came from the earlier refactor away from insert-then-`expect`; the entry API was right, the eager clone was not. Destructuring `self` splits the borrow so the clone can happen inside the vacant arm, where it belongs: once per stream rather than once per append. The `events_since` rustdoc overstated the refusal. It read as though `None` is refused whenever it is below the horizon, but a stream with no history is not below anything — reads from it succeed with an empty list, and the conformance suite depends on exactly that: property 8 calls `events_since(untouched, None)` on a stream nothing has touched. The doc now separates the two cases: a stale mark on a truncated stream is refused; an empty stream is not. Co-Authored-By: Claude Opus 5 --- app/crates/ironstate-journal/src/journal.rs | 12 +++++++++--- app/crates/ironstate-journal/src/memory.rs | 17 +++++++++++++---- 2 files changed, 22 insertions(+), 7 deletions(-) diff --git a/app/crates/ironstate-journal/src/journal.rs b/app/crates/ironstate-journal/src/journal.rs index 071417d..c8a0480 100644 --- a/app/crates/ironstate-journal/src/journal.rs +++ b/app/crates/ironstate-journal/src/journal.rs @@ -288,9 +288,15 @@ pub trait Journal { /// # Errors /// /// Returns [`JournalError::UnknownSeq`] if `after` is below the stream's - /// retained horizon — including `None`, which means genesis — since the - /// result would otherwise have a silent gap in it. Returns - /// [`JournalError::Storage`] if the underlying store failed. + /// retained horizon, since the result would otherwise have a silent gap in + /// it. That is a stale mark on a stream some of whose history has been + /// truncated away — a stream that simply has no history yet is not below + /// anything, and reads from it (including `None`) succeed with an empty + /// list. + /// + /// `None` means genesis, so on a *truncated* stream it is itself below the + /// horizon and refused; ask for the record before the horizon instead. + /// Returns [`JournalError::Storage`] if the underlying store failed. fn events_since( &self, stream: &StreamId, diff --git a/app/crates/ironstate-journal/src/memory.rs b/app/crates/ironstate-journal/src/memory.rs index ebb6a86..38fec17 100644 --- a/app/crates/ironstate-journal/src/memory.rs +++ b/app/crates/ironstate-journal/src/memory.rs @@ -115,10 +115,19 @@ impl MemoryJournal { /// The stream's history, seeded with its genesis snapshot if this is the /// first time it has been touched. fn stream_mut(&mut self, id: &StreamId) -> &mut Stream { - let genesis = self.genesis_snapshot(); - self.streams - .entry(id.clone()) - .or_insert_with(|| Stream::new(genesis)) + // Destructured so the genesis clone can happen *inside* the vacant arm: + // `genesis_snapshot()` borrows all of `self`, which `entry` has already + // borrowed mutably, and hoisting it would clone the whole aggregate + // state on every append rather than once per stream. + let Self { genesis, streams } = self; + streams.entry(id.clone()).or_insert_with(|| { + Stream::new(Snapshot { + state: genesis.clone(), + schema_version: 0, + at: Seq(0), + entropy_pos: DrawPos(0), + }) + }) } } From bfdb6bac78a7ae3807552cff2026e598cf8c2873 Mon Sep 17 00:00:00 2001 From: Dashiel Lopez Mendez Date: Sat, 29 Aug 2026 12:05:18 -0400 Subject: [PATCH 14/14] fix(journal): never resume from a snapshot recorded past the head MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `latest_snapshot` took the highest-`Seq` snapshot without regard for whether the stream actually reaches that far, and `resume` uses it as its replay base — so a snapshot recorded beyond the head was handed back verbatim as the resumed aggregate. Wrong state, returned as `Ok`, which is the failure mode this crate exists to make impossible. Verified before and after: with a bogus snapshot at `Seq(1000)` on a six-record stream, `resume` returned `total: 9999` instead of the true `28`. The regression test fails without the fix. This is the same defect I fixed one call site earlier: `latest_snapshot_at` already ignores out-of-range snapshots so one cannot authorise a truncation. Filtering there and not here meant a snapshot too bogus to permit truncation was still trusted to reconstruct state. Co-Authored-By: Claude Opus 5 --- app/crates/ironstate-journal/src/memory.rs | 8 ++++- .../ironstate-journal/tests/retention.rs | 34 +++++++++++++++++++ 2 files changed, 41 insertions(+), 1 deletion(-) diff --git a/app/crates/ironstate-journal/src/memory.rs b/app/crates/ironstate-journal/src/memory.rs index 38fec17..450dd46 100644 --- a/app/crates/ironstate-journal/src/memory.rs +++ b/app/crates/ironstate-journal/src/memory.rs @@ -253,10 +253,16 @@ impl Journal for MemoryJournal { let Some(stream) = self.streams.get(stream) else { return Ok(Some(self.genesis_snapshot())); }; - // The highest-`at` snapshot — the most useful base for replay. + // The highest-`at` snapshot at or below the head — the most useful base + // for replay. One recorded beyond the head describes state this stream + // does not have, and `resume` would hand it back verbatim, so it is + // ignored here for the same reason `latest_snapshot_at` ignores it when + // authorising a truncation. + let head = stream.truncated + stream.records.len() as u64; Ok(stream .snapshots .iter() + .filter(|s| s.at.0 <= head) .max_by_key(|s| s.at) .map(clone_snapshot)) } diff --git a/app/crates/ironstate-journal/tests/retention.rs b/app/crates/ironstate-journal/tests/retention.rs index ba25357..241cdd4 100644 --- a/app/crates/ironstate-journal/tests/retention.rs +++ b/app/crates/ironstate-journal/tests/retention.rs @@ -538,3 +538,37 @@ fn truncating_below_an_existing_horizon_is_refused() { assert_eq!(journal.retained_from(&stream()), Seq(4)); assert_eq!(journal.head(&stream()), Some(Seq(6))); } + +/// A snapshot recorded past the head is not a base `resume` may use: it +/// describes state the stream does not have, and returning it verbatim would +/// be a wrong aggregate handed back as `Ok`. +#[test] +fn resume_ignores_a_snapshot_recorded_past_the_head() { + let (mut journal, agg, seed) = driven(6); + let truth = agg.state().clone(); + + // A bogus snapshot, past the head and with state that is deliberately not + // the truth, so using it would be visible. + journal + .snapshot_in( + &mut (), + &stream(), + Snapshot { + state: Counter { + phase: Phase::Open, + total: 9999, + }, + schema_version: 0, + at: Seq(1000), + entropy_pos: DrawPos(0), + }, + ) + .unwrap(); + + let (resumed, _) = resume::(&journal, &stream(), &seed).unwrap(); + assert_eq!( + resumed.state(), + &truth, + "resume must replay from a real base, not one recorded past the head", + ); +}