From 9b2117a406d50fa95c1abb245c843af779a40ba6 Mon Sep 17 00:00:00 2001 From: Levi Neuwirth Date: Sun, 9 Aug 2026 16:13:21 +0200 Subject: [PATCH] Seal a full journal and continue in a fresh one A journal is preallocated, and a shard that filled it stopped accepting work. Nothing about the refused transactions was wrong and no amount of retrying made room, so the shard was simply done -- the one failure mode a store cannot have. Rotation cannot happen from inside preparation. The open group's frames are built but not appended and each carries `journal_id` in its header, so sealing mid-formation would leave them naming a journal that no longer takes writes. `accept` therefore publishes the open group, rotates, and re-prepares, which is the shape the group-full path already used. The reservation is released first; speculative state is not advanced until past the fit test, so that release is the whole rollback. Two triggers, and the second is not implied by the first: a group that no longer fits must rotate or be refused, and a cursor that has reached `segment_max_bytes` must rotate so segments stay near their configured size. With `segment_max_bytes` below `journal_preallocate_bytes` that boundary arrives first and every time, and checking only the fit let segments grow to the whole preallocation whatever the ceiling said. The threshold is deliberately not a per-group cap: it is read before a group is added rather than inside one, so a segment may overshoot by at most a group and no committed group is ever split. A frame no *empty* journal could hold is a ceiling, not a rotation -- sealing would produce a fresh journal that refuses it again, forever. It is measured against the preallocation less the journal header, because a new journal's cursor starts past that header; comparing against the whole preallocation called frames in that gap rotatable and retried them into the same refusal. The manifest advances `committed_shard_sequence` to the sealed segment's last. `validate_manifest_sequence_coverage` requires the final retained tail range to end exactly there, and sealing does make that prefix durable under a second, manifest-referenced name. Installing an index run is the case that differs and correctly carries the field forward: a run makes no journal frame more durable than it already was. The successor generation pins the new segment and the fresh tail. The predecessor's active tail is not carried forward -- that file is the segment now, and retaining both would leave one logical generation naming two sources, which `object_source` refuses. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01JGdH5V43XWnj1PdHqiPktQ --- crates/levcs-store/src/engine.rs | 300 +++++++++++++++++- crates/levcs-store/tests/journal_rotation.rs | 301 +++++++++++++++++++ 2 files changed, 588 insertions(+), 13 deletions(-) create mode 100644 crates/levcs-store/tests/journal_rotation.rs diff --git a/crates/levcs-store/src/engine.rs b/crates/levcs-store/src/engine.rs index f6e6510..00e5828 100644 --- a/crates/levcs-store/src/engine.rs +++ b/crates/levcs-store/src/engine.rs @@ -53,6 +53,7 @@ use crate::failpoints::{self, Failpoint, FailpointAction}; use crate::format::{ frame_total_len, object_type_code, Frame, FrameHeader, FrameObjectV1, FrameObjectsV1, FrameReceiptFieldsV1, Manifest, RepositoryCreateV1, TailRange, TransactionFramePayloadV1, + JOURNAL_HEADER_LEN, }; use crate::index::{ delta_pressure, DeltaPressure, IndexDelta, IndexKey, IndexLocation, IndexRun, IndexRunBuilder, @@ -1541,6 +1542,7 @@ fn spawn_shard_writer( in_flight_reservation: None, repositories, tail_generation: tail.logical_generation, + rotation_wanted: false, run_entry_baseline, checkpoint_namespaces, journal, @@ -1622,6 +1624,11 @@ struct ShardWriter { /// Logical index generation of the active journal, so an index location /// built here resolves to the same pinned file after a reopen. tail_generation: u64, + /// Set by `sequence_into_frame` when the forming group no longer fits the + /// active journal, and consumed by `accept`. A flag rather than an error + /// variant because the condition is not a refusal — nothing is wrong with + /// the transaction, and the answer is to seal and continue. + rotation_wanted: bool, /// Sealed-run entries that some other term of the replay ceiling already /// accounts for — a **duplicate-representation baseline**, not a horizon. /// @@ -1863,7 +1870,29 @@ impl ShardWriter { let (mut prepared, frame) = match self.prepare_caught(&transaction) { Ok(pair) => pair, - Err(error) => refuse!(error), + Err(error) => { + if !std::mem::take(&mut self.rotation_wanted) { + refuse!(error); + } + // `prepare` reserved before sequencing and the fit was tested + // after, so the reservation is held and the speculative state + // is not yet advanced -- that insertion happens past the test. + // Releasing the reservation is therefore the whole rollback. + if let Some(key) = self.in_flight_reservation.take() { + self.release_reservation(&key); + } + self.publish_open_group(); + if let Some(poison) = self.poison.clone() { + refuse!(poison); + } + if let Err(error) = self.rotate_journal() { + refuse!(error); + } + match self.prepare_caught(&transaction) { + Ok(pair) => pair, + Err(error) => refuse!(error), + } + } }; if let Err(frame) = self.builder.push(frame) { @@ -2158,19 +2187,51 @@ impl ShardWriter { let payload_len = payload.len() as u64; let total_len = frame_total_len(payload_len); - // Rotation and sealing are not in this slice. Refusing a group that no - // longer fits the preallocated journal here — with the exact length in - // hand, before anything is marked `Resolving` — keeps the missing - // deliverable a clean pre-append refusal rather than a `LimitExceeded` - // discovered between the `Resolving` mark and the fence. - if !self + // Two independent rotation triggers, and the second is not implied by + // the first. A group that no longer fits the preallocated file must + // rotate or be refused; a cursor that has reached `segment_max_bytes` + // must rotate so segments stay near their configured size, and with + // `segment_max_bytes < journal_preallocate_bytes` that boundary arrives + // first and every time. Checking only the fit let a shard run to the + // end of its preallocation whatever the segment ceiling said. + // + // The exact frame length is known here and nowhere earlier, so this is + // where both are discovered — before anything is marked `Resolving`, + // which keeps it a clean pre-append condition. + let group_end = self.builder.bytes().saturating_add(total_len); + let over_segment_ceiling = self .journal - .fits(self.builder.bytes().saturating_add(total_len)) - { - return Err(StoreError::NotImplemented( - "active journal rotation and sealing — B1 NamespaceTxn, scope 6-B1 \ - deliverable 1 and scope 3.4", - )); + .should_rotate(self.shared.options.segment_max_bytes); + if !self.journal.fits(group_end) || over_segment_ceiling { + // A frame no *empty* journal could hold is a ceiling and not a + // rotation: sealing would produce a fresh journal that refuses it + // again, so the shard would seal once per submit forever and still + // never accept it. + // + // Measured against the space a fresh journal actually offers, which + // is the preallocation less its header — a new journal's cursor + // starts past that header, so comparing the frame with the whole + // preallocation calls a frame rotatable when no rotation can ever + // seat it. + let preallocated = self.shared.options.journal_preallocate_bytes; + let usable = preallocated.saturating_sub(JOURNAL_HEADER_LEN as u64); + if total_len > usable { + return Err(StoreError::LimitExceeded { + limit: "journal_preallocate_bytes", + observed: total_len, + allowed: usable, + }); + } + // Rotation cannot happen from in here. The open group's frames are + // built but not appended, and each carries `journal_id` in its + // header; sealing now would leave them naming a journal that no + // longer takes writes. `accept` publishes the open group first and + // then rotates, which is the same shape the group-full path uses. + self.rotation_wanted = true; + return Err(StoreError::Overloaded { + limit: "active_journal", + retry_after_micros: 0, + }); } let frame = Frame { @@ -3698,6 +3759,219 @@ impl ShardWriter { } /// Derived shard state through `committed`, read from the published root. + /// Seal the active journal into a segment and continue in a fresh one. + /// + /// Scope 3.4, driven by the journal filling rather than by a checkpoint. + /// The two share a sealing sequence and differ in what they *additionally* + /// claim: a checkpoint installs a checkpoint row and moves the replay + /// horizon, while this installs only the segment. Both advance + /// `committed_shard_sequence`, because both make a prefix durable under a + /// manifest-referenced name — and `validate_manifest_sequence_coverage` + /// requires the final retained tail range to end exactly there. The + /// checkpoint list is carried forward untouched; the horizon is the + /// checkpoint's sequence and the active journal's position, not this field. + /// + /// Called from `accept` with the open group already published, so the + /// builder is empty. It cannot be called with one forming: those frames + /// carry `journal_id` in their headers and would name a journal that no + /// longer accepts writes. + /// + /// Every failure past the seal poisons. The segment is linked into + /// `segments/` before the manifest names it and the active name is unlinked + /// only after — §3.4's ordering, which leaves every crash point with either + /// a scannable journal or a manifest that references the segment, and never + /// neither. A failure in the middle is not something this process can + /// resolve by retrying. + fn rotate_journal(&mut self) -> Result<(), StoreError> { + // Nothing appended means nothing to seal. A1 refuses to seal an empty + // journal, and a shard that reached here with one is asking for a + // rotation that would produce a segment naming no frames. + let Some(last) = self.journal.last_appended_shard_sequence() else { + return Ok(()); + }; + + let root = self.shared.committed.load_full(); + let paths = RootLayout::new(&self.shared.options.root).shard(self.shard_index); + let root_uuid = self.shared.root_uuid; + let counters = Arc::clone(self.journal.counters_handle()); + let current_generation = self.current_manifest_generation(&root)?; + let manifest = read_manifest_or_empty(&paths, current_generation, &root_uuid)?; + let next_generation = current_generation + 1; + + // The segment inherits the tail's logical generation, so every + // `IndexLocation` already written against it keeps naming the same + // bytes (contract review 2026-07-30-A). + let sealed_generation = self.tail_generation; + let first = self.journal.header().first_shard_sequence; + let active_path = self.journal.path().to_path_buf(); + let segment_path = + match segment::seal_journal(&mut self.journal, &paths, sealed_generation, &counters) { + Ok(path) => path, + Err(error) => { + return Err(self.poison_now(format!("sealing a full journal: {error}"))) + } + }; + let filename = match segment_path.file_name().and_then(|name| name.to_str()) { + Some(name) => name.to_string(), + None => return Err(self.poison_now("sealed segment name is not UTF-8".into())), + }; + + let mut retained_tail_ranges = manifest.retained_tail_ranges.clone(); + retained_tail_ranges.push(TailRange { + generation: sealed_generation, + first_shard_sequence: first, + last_shard_sequence: last, + filename, + }); + let published = Manifest { + root_uuid, + generation: next_generation, + base_generation: manifest.base_generation, + retained_tail_ranges, + index_runs: manifest.index_runs.clone(), + checkpoints: manifest.checkpoints.clone(), + // Advanced to the sealed segment's last sequence, and this is the + // one field where rotation differs from installing an index run. + // + // `install_index_run` carries this forward untouched because a run + // makes no journal frame more durable than it already was — the + // frames it indexes are still only in `active/`, and advancing + // would tell recovery to stop replaying them. Sealing does the + // opposite: this prefix is now durable under a second name that the + // manifest references, which is exactly what the field records. + // + // `validate_manifest_sequence_coverage` requires the final retained + // tail range to end precisely here, so leaving it behind produced a + // manifest whose coverage stopped short of its own last range and a + // reopen that refused. Advancing skips nothing: with no checkpoint + // installed, recovery replays every manifest-referenced retained + // tail range, and the replay horizon is the checkpoint's sequence + // and the active journal's position rather than this field alone. + committed_shard_sequence: last, + }; + if let Err(error) = segment::install_manifest( + &paths, + &published, + self.shared.options.manifest_retain, + &counters, + ) { + return Err(self.poison_now(format!("publishing a rotation manifest: {error}"))); + } + + if let Err(error) = segment::unlink_sealed_journal(&active_path, &paths, &counters) { + return Err(self.poison_now(format!("unlinking the sealed journal: {error}"))); + } + let journal = match Journal::create( + &paths.active(), + fresh_checkpoint_journal_id(root_uuid, self.shard_index, last), + last.saturating_add(1), + self.shard_index, + root_uuid, + self.shared.options.journal_preallocate_bytes, + now_micros(), + Arc::clone(&counters), + ) { + Ok(journal) => journal, + Err(error) => { + return Err(self.poison_now(format!("opening the post-rotation journal: {error}"))) + } + }; + self.journal = journal; + self.tail_generation = sealed_generation + 1; + + let successor = match self.retain_rotated_generation( + &root, + current_generation, + next_generation, + sealed_generation, + first, + last, + &segment_path, + ) { + Ok(successor) => successor, + Err(error) => return Err(self.poison_now(format!("pinning a rotation: {error}"))), + }; + + let mut generations = GenerationMap::new(); + generations.insert(successor.id, successor); + let mut subtree = ShardSubtree::new( + self.shard_index, + // Unchanged: no frame was appended by this publication. + root.shard_committed_sequence(self.shard_index) + .unwrap_or(last), + Arc::new(IndexDelta::from_options(&self.shared.options)), + // `None`, though a segment was installed. Sealing the *journal* is + // not sealing the *index*: naming a sealed-through sequence here + // would discard delta layers whose entries no run contains. + None, + Vector::new(), + RepoMap::new(), + TerminalStatusMap::new(), + generations, + ); + subtree + .retained_generation_removals + .push_back(GenerationId::new(self.shard_index, current_generation)); + if let Err(error) = self.publish_subtree(&subtree) { + return Err(self.poison_now(format!("publishing a rotation: {error}"))); + } + Ok(()) + } + + /// The successor generation pinning the newly sealed segment and the fresh + /// journal, plus everything the predecessor pinned. + /// + /// The predecessor's active tail is deliberately **not** carried forward: + /// that file is now the segment, reachable at its new name, and retaining + /// the old pin as a tail would leave one logical generation naming two + /// sources -- exactly the collision `object_source` refuses. + #[allow(clippy::too_many_arguments)] + fn retain_rotated_generation( + &self, + root: &CommittedRoot, + current_generation: u64, + next_generation: u64, + sealed_generation: u64, + first: u64, + last: u64, + segment_path: &Path, + ) -> Result, StoreError> { + let previous = root + .retained_generations() + .get(&GenerationId::new(self.shard_index, current_generation)) + .map(Arc::clone) + .ok_or_else(|| { + StoreError::Corruption(format!( + "generation {current_generation} vanished from the root while rotating" + )) + })?; + + let mut segments = previous.segments.to_vec(); + segments.push(RootRetainedSegment::new( + sealed_generation, + self.journal_id_of_sealed_prefix(segment_path)?, + first, + last, + PinnedFile::open(segment_path)?, + )); + let active_tails = vec![RetainedTail::new( + self.tail_generation, + PinnedFile::from_shared( + self.journal.path().to_path_buf(), + Arc::new(self.journal.file().try_clone()?), + ), + )]; + + Ok(Arc::new(RetainedGeneration::new( + GenerationId::new(self.shard_index, next_generation), + segments.into(), + Arc::clone(&previous.index_runs), + Arc::clone(&previous.checkpoints), + active_tails.into(), + Arc::clone(&previous.projection_artifacts), + ))) + } + fn checkpoint_body( &self, root: &CommittedRoot, diff --git a/crates/levcs-store/tests/journal_rotation.rs b/crates/levcs-store/tests/journal_rotation.rs new file mode 100644 index 0000000..fb9a6c2 --- /dev/null +++ b/crates/levcs-store/tests/journal_rotation.rs @@ -0,0 +1,301 @@ +//! Scope 3.4 — sealing the active journal and continuing in a fresh one. +//! +//! A journal is preallocated. When a forming group no longer fits it, the shard +//! has to seal what it has into `segments/`, install the manifest generation +//! that names it, and keep going in a new file. Until that exists the shard +//! simply stops accepting work, which is the one failure mode a store cannot +//! have: nothing about the transactions being refused is wrong, and no amount +//! of retrying makes room. +//! +//! # What this asserts beyond "the submit succeeded" +//! +//! Rotation is easy to do in a way that passes a liveness test and loses data. +//! The frames sealed into the segment are still the only copy of everything +//! committed before the rotation, and every `IndexLocation` naming them was +//! written against the *tail's* logical generation — so a rotation that +//! installs the segment under a different generation, or that fails to retain +//! it, leaves those objects indexed at a generation nothing resolves. The +//! assertions here therefore span the rotation in both directions: objects +//! written before it must still locate after it, and the whole thing must +//! survive a reopen through production recovery. +//! +//! §3.4's link-then-unlink ordering exists for the crash window in the middle +//! of that sequence. This file does not crash the process — that is the crash +//! matrix's job — but it does prove the non-crash path leaves a root recovery +//! reopens without replaying the sealed prefix twice. + +#![cfg(all( + feature = "store-privileged", + feature = "store-internals", + feature = "failpoints" +))] + +use levcs_core::{ObjectId, ObjectType}; +use levcs_store::transaction::StagedObject; +use levcs_store::types::{NamespaceId, OperationId, PrivilegedConstruction}; +use levcs_store::{StoreEngine, StoreOptions, ValidatedTransaction}; + +#[path = "support/engine_matrix.rs"] +mod engine_matrix; + +use engine_matrix::{ + create_transaction, deadline, evidence, genesis_id, namespace_on_shard, submit, + DEFAULT_MAX_INDEX_RUNS, +}; + +const SHARD_COUNT: u16 = 1; +/// Small enough that a handful of transactions fills it, and still at least +/// `max_group_bytes`, which `StoreOptions::validate` requires. +const JOURNAL_BYTES: u64 = 64 * 1024; +const GROUP_BYTES: u64 = 8 * 1024; +/// Four kilobytes per transaction, so the journal fills in about a dozen +/// commits rather than a thousand. +const OBJECT_BYTES: usize = 4 * 1024; + +fn options(root: &std::path::Path) -> StoreOptions { + let mut options = + engine_matrix::options_with_index_runs(root, SHARD_COUNT, DEFAULT_MAX_INDEX_RUNS); + options.journal_preallocate_bytes = JOURNAL_BYTES; + options.segment_max_bytes = JOURNAL_BYTES; + options.max_group_bytes = GROUP_BYTES; + options +} + +fn blob_id(seed: u16) -> ObjectId { + let mut bytes = [0u8; 32]; + bytes[..2].copy_from_slice(&seed.to_le_bytes()); + ObjectId(bytes) +} + +/// A transaction carrying one sizeable object, so the journal fills quickly. +fn filling_transaction(namespace: NamespaceId, seed: u16) -> ValidatedTransaction { + let authority = genesis_id(&namespace); + let mut operation = [0u8; 16]; + operation[..2].copy_from_slice(&seed.to_le_bytes()); + ValidatedTransaction::builder(PrivilegedConstruction::assert_validated()) + .namespace(namespace) + .operation( + OperationId(operation), + ObjectId([seed as u8; 32]), + deadline(), + ) + .objects(vec![StagedObject { + id: blob_id(seed), + object_type: ObjectType::Blob, + raw: vec![seed as u8; OBJECT_BYTES], + }]) + .refs(Vec::new()) + .authority(Some(authority), Some(authority)) + .evidence(evidence()) + .build() + .expect("a complete filling transaction") +} + +fn segment_count(root: &std::path::Path) -> usize { + let segments = root.join("shards").join("00").join("segments"); + std::fs::read_dir(&segments) + .map(|entries| entries.filter_map(Result::ok).count()) + .unwrap_or(0) +} + +/// `segment_max_bytes` is a rotation trigger in its own right. +/// +/// `Journal::should_rotate` names two boundaries — the preallocated length and +/// the configured segment size, whichever comes first — and only the first one +/// is a question about whether the next group *fits*. A shard that rotates +/// solely on fit runs to the end of its preallocation however small the segment +/// ceiling is, and produces segments many times the configured maximum. +/// +/// The two limits are deliberately far apart here. The main test above sets +/// them equal, which is the realistic default and also the one arrangement +/// where this defect is invisible. +#[test] +fn a_segment_ceiling_below_the_preallocation_rotates_first() { + let directory = tempfile::TempDir::new().expect("a temporary root"); + let namespace = namespace_on_shard(0, SHARD_COUNT, 1); + + let mut options = options(directory.path()); + // Room for many groups in the file, but a segment ceiling reached after a + // handful of them. + options.journal_preallocate_bytes = 1024 * 1024; + options.segment_max_bytes = 24 * 1024; + + let engine = StoreEngine::open(options).expect("the root initializes"); + let created = submit(&engine, create_transaction(namespace, 1)); + created + .receipt() + .unwrap_or_else(|| panic!("the repository must be created: {:?}", created.error())); + + const WRITES: u16 = 24; + for seed in 1..=WRITES { + let outcome = submit(&engine, filling_transaction(namespace, seed)); + outcome + .receipt() + .unwrap_or_else(|| panic!("write {seed} must commit: {:?}", outcome.error())); + } + + // Roughly 96 KiB written against a 24 KiB ceiling, so several segments. + // Asserted as "more than one" rather than an exact count: a segment may + // overshoot the ceiling by up to one group, because the boundary is checked + // before a group is added and not in the middle of one. + let sealed = segment_count(directory.path()); + assert!( + sealed > 1, + "with segment_max_bytes far below journal_preallocate_bytes the shard must seal on \ + the segment ceiling, but only {sealed} segment(s) exist after {WRITES} writes" + ); + + for seed in 1..=WRITES { + assert!( + engine + .snapshot(namespace) + .expect("snapshot") + .locate(blob_id(seed)) + .expect("locate") + .is_some(), + "object {seed} must survive a segment-ceiling rotation" + ); + } +} + +/// A frame that no rotation can ever seat is a ceiling, not a rotation. +/// +/// The distinction is off by exactly the journal header. A fresh journal's +/// cursor starts past its own header, so the space a rotation actually offers +/// is the preallocation *less* that header. A frame in between — small enough +/// for the preallocation, too large once the header is there — was classified +/// as rotatable, and the retry against the fresh journal failed the same test +/// again. The caller was promised `LimitExceeded` and got a second rotation +/// instead. +#[test] +fn a_frame_larger_than_a_fresh_journal_is_refused_rather_than_rotated() { + let directory = tempfile::TempDir::new().expect("a temporary root"); + let namespace = namespace_on_shard(0, SHARD_COUNT, 1); + + // The object is sized to land in the gap: its frame fits + // `journal_preallocate_bytes` outright but not once the header is + // accounted for. `max_group_bytes` is raised past it so the group ceiling + // is not what refuses first -- this must be the journal's answer. + const PREALLOCATED: u64 = 32 * 1024; + let mut options = options(directory.path()); + options.journal_preallocate_bytes = PREALLOCATED; + options.segment_max_bytes = PREALLOCATED; + options.max_group_bytes = PREALLOCATED; + + let engine = StoreEngine::open(options).expect("the root initializes"); + let created = submit(&engine, create_transaction(namespace, 1)); + created + .receipt() + .unwrap_or_else(|| panic!("the repository must be created: {:?}", created.error())); + + // Just under the preallocation, so the frame around it is over the usable + // space by roughly the header. + let oversized = ValidatedTransaction::builder(PrivilegedConstruction::assert_validated()) + .namespace(namespace) + .operation(OperationId([0xee; 16]), ObjectId([0xee; 32]), deadline()) + .objects(vec![StagedObject { + id: blob_id(0xeee), + object_type: ObjectType::Blob, + raw: vec![0xee; PREALLOCATED as usize - 768], + }]) + .refs(Vec::new()) + .authority(Some(genesis_id(&namespace)), Some(genesis_id(&namespace))) + .evidence(evidence()) + .build() + .expect("the transaction itself is well formed"); + + let outcome = submit(&engine, oversized); + match outcome.error() { + Some(levcs_store::types::StoreError::LimitExceeded { + limit, + observed, + allowed, + }) => { + assert_eq!( + *limit, "journal_preallocate_bytes", + "the refusal must name the ceiling that stopped it" + ); + // The reported ceiling is what makes this test see the off-by-one + // rather than merely see a refusal. A frame can be far enough over + // to be refused either way; what distinguishes the two is *which* + // ceiling the store believes it has. A fresh journal's cursor + // starts past its header, so the space a rotation can offer is the + // preallocation less that header, and reporting the whole + // preallocation means frames in the gap were called rotatable. + assert_eq!( + *allowed, + PREALLOCATED - levcs_store::format::JOURNAL_HEADER_LEN as u64, + "the ceiling must be the space a fresh journal actually offers, not the raw \ + preallocation" + ); + assert!( + *observed > *allowed, + "the refusal must report a frame that genuinely exceeds the ceiling" + ); + } + other => panic!( + "a frame no fresh journal can hold must be refused as a ceiling, not retried as a \ + rotation, but submit answered {other:?}" + ), + } + + // And the shard is still usable: a ceiling is a refusal of one transaction, + // not a wedged writer. + let after = submit(&engine, filling_transaction(namespace, 1)); + after + .receipt() + .unwrap_or_else(|| panic!("an ordinary write must still commit: {:?}", after.error())); +} + +#[test] +fn a_full_active_journal_seals_and_the_shard_keeps_committing() { + let directory = tempfile::TempDir::new().expect("a temporary root"); + let namespace = namespace_on_shard(0, SHARD_COUNT, 1); + + // Enough transactions to overrun the preallocated journal several times + // over, so the test exercises repeated rotation rather than one boundary. + const WRITES: u16 = 48; + + { + let engine = StoreEngine::open(options(directory.path())).expect("the root initializes"); + let created = submit(&engine, create_transaction(namespace, 1)); + created + .receipt() + .unwrap_or_else(|| panic!("the repository must be created: {:?}", created.error())); + + for seed in 1..=WRITES { + let outcome = submit(&engine, filling_transaction(namespace, seed)); + outcome.receipt().unwrap_or_else(|| { + panic!( + "write {seed} of {WRITES} must commit; the journal filling is a rotation, \ + not a ceiling: {:?}", + outcome.error() + ) + }); + } + + // Before the reopen, because a rotation that lost the segment would + // still answer correctly from a root that has not been rebuilt yet. + let snapshot = engine.snapshot(namespace).expect("snapshot"); + for seed in [1u16, WRITES / 2, WRITES] { + assert!( + snapshot.locate(blob_id(seed)).expect("locate").is_some(), + "object {seed} was committed before a rotation and must still be located \ + after it" + ); + } + } + + // Through production recovery: the sealed segments are now the only copy of + // everything but the last journal's frames, and a reopen is what proves the + // manifest names them. + let reopened = StoreEngine::open(options(directory.path())).expect("the root reopens"); + let snapshot = reopened.snapshot(namespace).expect("snapshot"); + for seed in 1..=WRITES { + assert!( + snapshot.locate(blob_id(seed)).expect("locate").is_some(), + "object {seed} did not survive the reopen; a sealed prefix the manifest does not \ + name is a fenced, acknowledged transaction that recovery cannot find" + ); + } +}