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/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/examples/async-store/src/main.rs b/app/crates/examples/async-store/src/main.rs index 3fc6c31..0e9f61f 100644 --- a/app/crates/examples/async-store/src/main.rs +++ b/app/crates/examples/async-store/src/main.rs @@ -261,13 +261,14 @@ impl Log { Ok(self .records(stream) .iter() + .enumerate() .skip(start) - .flat_map(|record| record.events.iter()) - .map(|event| VersionedEvent { - event: event.clone(), - type_name: type_name.clone(), - version: 1, + .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::new(event.clone(), seq, type_name.clone(), 1)) .collect()) } 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/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/contract.rs b/app/crates/ironstate-journal/src/contract.rs index 2488109..d363c57 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}; @@ -36,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; @@ -156,6 +159,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_10_truncation_preserves_resume(&mut journal, &stream, &seed, case); + } +} + /// Run every property, including the two that need [`ForkableJournal`]. /// /// # Panics @@ -168,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); } @@ -226,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}", ); } @@ -250,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}", ); } } @@ -305,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}", ); } @@ -320,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}", ); } } @@ -343,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}", ); } @@ -374,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; } @@ -397,6 +440,90 @@ where } } +/// 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_10_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"); + + // 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 10: entropy_pos undefined at retained Seq({seq}), case {case}" + ) + }); + assert!( + pos >= previous, + "[proven] property 10: 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 10: truncated Seq({gone}) must be UnknownSeq, case {case}", + ); + } + + let (after, entropy_after) = resume::(journal, stream, seed).expect("resume after"); + assert_eq!( + digest128(after.state()), + digest128(before.state()), + "[proven] property 10: truncation changed what the stream resumes to, case {case}", + ); + assert_eq!( + entropy_after.draws(), + entropy_before.draws(), + "[proven] property 10: truncation moved the resume entropy position, case {case}", + ); + assert_eq!( + journal.retained_from(stream), + at, + "[proven] property 10: 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..c8a0480 100644 --- a/app/crates/ironstate-journal/src/journal.rs +++ b/app/crates/ironstate-journal/src/journal.rs @@ -65,15 +65,49 @@ 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, + /// 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. 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)] @@ -85,6 +119,20 @@ 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 { + /// 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 { @@ -95,7 +143,27 @@ 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, + "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, + } => 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 +173,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 { .. } => None, + Self::UnknownSeq { .. } + | Self::NoBaseForFork { .. } + | Self::NoSnapshotForTruncation { .. } => None, } } } @@ -217,6 +287,15 @@ pub trait Journal { /// /// # Errors /// + /// Returns [`JournalError::UnknownSeq`] if `after` is below the stream's + /// 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, @@ -246,8 +325,58 @@ 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; } + +/// 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::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 + /// 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. + /// + /// 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. + /// + /// # Errors + /// + /// Returns [`JournalError::Storage`] if the underlying store failed. + fn streams(&self) -> Result, JournalError>; +} 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..c009860 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,13 @@ /// 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); +/// // capability markers may be given in either order /// ``` #[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 +45,25 @@ 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); + } + }; + // 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 3ed1d99..450dd46 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,68 @@ 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 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 { + let head = self.truncated + self.records.len() as u64; + (head > 0).then_some(Seq(head)) + } + + /// 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. + /// 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 { + let head = self.truncated + self.records.len() as u64; + self.snapshots + .iter() + .map(|s| s.at) + .filter(|at| at.0 <= head) + .max() } } @@ -67,16 +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 { - if !self.streams.contains_key(id) { - let seeded = Stream { - records: Vec::new(), - snapshots: vec![self.genesis_snapshot()], - }; - self.streams.insert(id.clone(), seeded); - } - self.streams - .get_mut(id) - .expect("the stream was just inserted") + // 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), + }) + }) } } @@ -106,7 +157,9 @@ impl Journal for MemoryJournal { events: events.to_vec(), entropy_pos, }); - Ok(Seq(stream.records.len() as u64)) + // 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( @@ -128,20 +181,27 @@ 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) } 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( @@ -152,20 +212,39 @@ 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. + // + // `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, 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 { + 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. - let start = after.map_or(0, |s| usize::try_from(s.0).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 { - event: event.clone(), - type_name: type_name.clone(), - version: 1, + .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::new(event.clone(), seq, type_name.clone(), 1)) .collect()) } @@ -174,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)) } @@ -193,13 +278,34 @@ impl ForkableJournal for MemoryJournal { Err(JournalError::UnknownSeq { at }) }; }; - if at.0 > 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 as usize; + // 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(), Stream { + truncated: source.truncated, records: source .records .iter() @@ -209,14 +315,78 @@ 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) } } + +impl RetainableJournal for MemoryJournal { + 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` 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 <= stream.truncated || 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 + // 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, + }); + } + + // 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) + .unwrap_or(usize::MAX) + .min(stream.records.len()); + stream.records.drain(..drop_count); + stream.truncated += drop_count as u64; + + // 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(()) + } + + fn retained_from(&self, stream: &StreamId) -> Seq { + self.streams + .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 new file mode 100644 index 0000000..241cdd4 --- /dev/null +++ b/app/crates/ironstate-journal/tests/retention.rs @@ -0,0 +1,574 @@ +//! 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, EntropySource, 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(); + + // 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(6)), + "a fork's head must sit at the fork point, truncated or not", + ); + assert_eq!( + 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` 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!( + "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 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!( + after, 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); + 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", + ), + }, + } +} + +/// `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)); +} + +/// 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", + ); +} + +/// 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))); +} + +/// 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", + ); +} diff --git a/app/crates/ironstate-journal/tests/transactional.rs b/app/crates/ironstate-journal/tests/transactional.rs index 4ce365f..9670d0f 100644 --- a/app/crates/ironstate-journal/tests/transactional.rs +++ b/app/crates/ironstate-journal/tests/transactional.rs @@ -190,13 +190,13 @@ impl Journal for StagedJournal { Ok(self .records(stream) .iter() + .enumerate() .skip(start) - .flat_map(|r| r.events.iter()) - .map(|event| VersionedEvent { - event: event.clone(), - type_name: type_name.clone(), - version: 1, + .flat_map(|(i, r)| { + let seq = Seq(i as u64 + 1); + r.events.iter().map(move |event| (seq, event)) }) + .map(|(seq, event)| VersionedEvent::new(event.clone(), seq, type_name.clone(), 1)) .collect()) }