From 5462952691600c988c5eab310e979b889dd1d945 Mon Sep 17 00:00:00 2001 From: Levi Neuwirth Date: Thu, 30 Jul 2026 19:10:18 -0400 Subject: [PATCH] Advance the committed prefix with a checkpoint MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `StoreEngine::checkpoint` was the frozen D0 signature that returned NotImplemented. It now checkpoints every shard on its own writer thread and returns a lease that really pins. The ordering inside it is the correctness. The checkpoint format carries catalog, refs and receipts and no object index, and advancing committed_shard_sequence stops recovery replaying the frames it covers — so entries that live only in the root's delta layers describe objects that are on disk, named by a segment, and unreachable at the next open. Every layer through the committed sequence is sealed into a run the checkpoint's own manifest names and discarded from the root before the sequence moves; coverage short of that is refused rather than partially applied. CheckpointLease holds the exact Arc each shard published, which pins segment and checkpoint descriptors and index-run mappings. Counting retained generations cannot stand in for that: two later checkpoints prune the generation the lease names. The group boundary is structural rather than a lock. The checkpoint travels the same channel as submissions and the writer loop publishes any open group before running one, so it cannot advance past sequenced-but-unfenced frames. Four edge cases, each tested: an empty shard is skipped rather than checkpointed at sequence 0; a repeat with no new work reuses instead of renaming onto its own name; a finalized but unreferenced checkpoint left by a crash is validated through recovery's reader and adopted rather than wedging the shard on a permanent EEXIST; and a journal holding no frame seals nothing. store-bench found that last one by calling the entry point that had always refused. It now earns index_maintenance from a measurement as well — index_maintenance().unsealed_delta_layers where it recorded None for "no such reading exists", which is the interface request its own guard had written down. Acceptance: write below seal pressure, checkpoint, reopen, resolve every object with the replay delta excluded as the answer. Both named mutations fail it on reachability rather than bookkeeping. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01XKzM69CHmBuDcA3qN1jFdh --- crates/levcs-store/src/bin/store-bench.rs | 19 +- crates/levcs-store/src/engine.rs | 887 +++++++++++++++++++++- doc/instance-throughput-rewrite-plan.md | 47 ++ doc/phase1-storage-spine-scope.md | 6 + 4 files changed, 940 insertions(+), 19 deletions(-) diff --git a/crates/levcs-store/src/bin/store-bench.rs b/crates/levcs-store/src/bin/store-bench.rs index 7387a73..54d9471 100644 --- a/crates/levcs-store/src/bin/store-bench.rs +++ b/crates/levcs-store/src/bin/store-bench.rs @@ -3819,6 +3819,10 @@ fn run_engine( // asserted about it. Asked after the reconciliation so a checkpoint the day // this stops refusing cannot move what the reconciliation read. let checkpoint = probe_checkpointing(&reopened)?; + // Read from the live engine, after the checkpoint probe, so it describes + // the state the bundle is about — and before the handle is dropped, because + // the reading only exists while the root is open. + let unsealed_delta_backlog = reopened.index_maintenance().unsealed_delta_layers; drop(reopened); // Per-repository sequence integrity, through the one shared checker. @@ -3859,11 +3863,16 @@ fn run_engine( validated_runs: index_scan.validated_runs, unvalidatable: index_scan.unvalidatable, groups: fences, - // Nothing seals, and `StoreEngine` exposes no reading of how - // many published deltas are outstanding. `None` is that - // absence, and it is what makes `runs_sealed` unearnable until - // the reading exists rather than inferable from a file count. - unsealed_delta_backlog: None, + // The reading the earlier `None` recorded the absence of. + // `IndexMaintenanceSnapshot` is taken from one captured + // committed root, so the backlog and the runs it is compared + // against describe the same instant — which is what makes + // `runs_sealed` a measurement rather than a file count. + // + // Taken from `reopened`, after the checkpoint probe, so it + // reports the state the bundle describes rather than the one + // before it. + unsealed_delta_backlog: Some(unsealed_delta_backlog), }, configured_max_index_runs, default_max_index_runs, diff --git a/crates/levcs-store/src/engine.rs b/crates/levcs-store/src/engine.rs index f5e32b3..2cd0185 100644 --- a/crates/levcs-store/src/engine.rs +++ b/crates/levcs-store/src/engine.rs @@ -52,22 +52,25 @@ use crate::completion::SharedCompletion; use crate::failpoints::{self, Failpoint, FailpointAction}; use crate::format::{ frame_total_len, object_type_code, Frame, FrameHeader, FrameObjectV1, FrameObjectsV1, - FrameReceiptFieldsV1, RepositoryCreateV1, TransactionFramePayloadV1, + FrameReceiptFieldsV1, Manifest, RepositoryCreateV1, TailRange, TransactionFramePayloadV1, }; use crate::index::{ delta_pressure, DeltaPressure, IndexDelta, IndexKey, IndexLocation, IndexRun, IndexRunBuilder, - NamespaceLifecycle, NamespaceRecord, NamespaceStorageMode, + NamespaceCatalog, NamespaceLifecycle, NamespaceRecord, NamespaceStorageMode, }; +use std::path::{Path, PathBuf}; + use crate::journal::{GroupBuilder, Journal}; use crate::options::StoreOptions; use crate::recovery::{RecoveredShard, RecoveryConfig, RecoverySession}; use crate::roots::{ CommittedRoot, GenerationId, GenerationMap, LayeredObjectIndex, OperationKey, - OperationStatusMetricSnapshot, OperationStatusMetrics, OperationStatusRoot, RepoMap, RepoState, - RetainedGeneration, RetainedIndexRun, RetainedReceipt, ShardSequenceMap, ShardSubtree, + OperationStatusMetricSnapshot, OperationStatusMetrics, OperationStatusRoot, PinnedFile, + RepoMap, RepoState, RetainedGeneration, RetainedIndexRun, RetainedReceipt, + RetainedSegment as RootRetainedSegment, RetainedTail, ShardSequenceMap, ShardSubtree, StatusEntry, StatusReservation, TerminalStatusEntry, TerminalStatusMap, TypedRefMap, }; -use crate::segment::{self, RootLayout}; +use crate::segment::{self, RootLayout, SegmentReader}; use crate::snapshot::RepoSnapshot; use crate::staging::ProjectionStaging; use crate::transaction::ValidatedTransaction; @@ -88,8 +91,19 @@ pub struct StoreEngine { /// A held checkpoint export lease. Pins the referenced generation's segments, /// indexes, and checkpoints against reclamation until dropped. +/// +/// The pin is the exact `RetainedGeneration` each shard published, so it holds +/// open file descriptors for segments and checkpoints and the mappings behind +/// index runs — the same objects the committed root pins, kept alive +/// independently of it. Count-based retention cannot stand in for this: two +/// further checkpoints prune the generation this lease names, and a lease that +/// stopped being true after two unrelated operations would not be a lease. +/// +/// Nothing reads the field, and that is what it is for. Holding it is the whole +/// contract; dropping it releases the pin. pub struct CheckpointLease { - _private: (), + #[allow(dead_code)] + generations: Vec>, } impl std::fmt::Debug for StoreEngine { @@ -164,7 +178,7 @@ struct EngineShared { struct ShardHandle { /// `Option` so `Drop` can close the channel before joining. A shard thread /// exits on disconnect, so dropping the sender is the shutdown signal. - submissions: Option>, + submissions: Option>, thread: Option>, /// **Not reachable from outside the crate yet.** The counters are the only /// witness for "one fence per group", and B4's harness will need them @@ -180,6 +194,19 @@ struct Submission { completion: SharedCompletion, } +/// What a shard writer accepts, on the one channel that reaches it. +/// +/// A checkpoint is a message rather than a lock taken from the caller's thread +/// because the writer thread is the only thing that may touch the journal, the +/// open group, or the shard's publication path. Sending it here is also what +/// makes "at a group boundary" structural: the loop publishes any open group +/// before it runs one, so a checkpoint cannot advance the committed sequence +/// past frames that are still unfenced. +enum Command { + Submit(Submission), + Checkpoint(Sender>, StoreError>>), +} + impl StoreEngine { /// Open or initialize a store root. /// @@ -348,10 +375,10 @@ impl StoreEngine { let completion = SharedCompletion::new(); let waiter = completion.subscribe(); - let submission = Submission { + let submission = Command::Submit(Submission { transaction: txn, completion, - }; + }); // The queue bound is backpressure, not a wait. Blocking the caller's // task inside a crate that starts no runtime would hand the store a // scheduling decision that belongs to its embedder (plan §7). @@ -434,10 +461,36 @@ impl StoreEngine { } /// Force a checkpoint and hold a lease on the resulting generation. + /// + /// Every shard, in order, each on its own writer thread at a group + /// boundary. Shards with nothing committed are skipped rather than + /// checkpointed at sequence 0 — "no committed sequence" and "committed + /// through the first frame" are different facts about a shard, and a + /// checkpoint file is a claim about frames. + /// + /// Blocking, because the frozen signature is synchronous and because a + /// checkpoint is an operator action rather than a request path. A shard + /// that fails stops the operation: the shards that already checkpointed + /// keep their durable state, which is sound but partial, and reporting the + /// error is the only honest thing to do with it. pub fn checkpoint(&self) -> Result { - Err(StoreError::NotImplemented( - "StoreEngine::checkpoint — B1 NamespaceTxn, scope 6-B1", - )) + let mut generations = Vec::with_capacity(self.shards.len()); + for handle in &self.shards { + let sender = handle.submissions.as_ref().ok_or(StoreError::NotReady)?; + let (reply, answer) = crossbeam_channel::bounded(1); + sender + .send(Command::Checkpoint(reply)) + .map_err(|_| StoreError::NotReady)?; + match answer.recv() { + Ok(Ok(Some(generation))) => generations.push(generation), + Ok(Ok(None)) => {} + Ok(Err(error)) => return Err(error), + // The writer thread died holding this request. Whatever it did + // before that is recorded on disk; this process cannot say what. + Err(_) => return Err(StoreError::NotReady), + } + } + Ok(CheckpointLease { generations }) } /// Durability counters for one shard, so "one fence per group" and "no @@ -1552,7 +1605,7 @@ struct Prepared { transaction: Arc, } -fn run_shard_writer(mut writer: ShardWriter, submissions: Receiver) { +fn run_shard_writer(mut writer: ShardWriter, submissions: Receiver) { let unwound = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { loop { // The idle bound is a receive deadline, not a timer thread: an @@ -1565,7 +1618,18 @@ fn run_shard_writer(mut writer: ShardWriter, submissions: Receiver) .map_err(|_| RecvTimeoutError::Disconnected), }; match message { - Ok(submission) => writer.accept(submission), + Ok(Command::Submit(submission)) => writer.accept(submission), + // The group boundary. Publishing first is not tidiness: the + // checkpoint is about to declare a committed prefix durable, + // and an open group holds sequenced frames that are not fenced + // yet. A reply the caller can no longer receive is dropped, and + // the checkpoint still stands — it is durable state, not a + // response. + Ok(Command::Checkpoint(reply)) => { + writer.publish_open_group(); + let outcome = writer.checkpoint(); + let _ = reply.send(outcome); + } Err(RecvTimeoutError::Timeout) => writer.publish_open_group(), Err(RecvTimeoutError::Disconnected) => { // A transaction already accepted is already sequenced. @@ -2837,6 +2901,437 @@ impl ShardWriter { /// no longer describe the device. A false poison costs an operator a /// recovery; a missed one lets the writer continue on an ambiguous root, /// which is the failure this boundary exists to prevent. + /// Scope 3.7: make a committed prefix durable outside the active journal + /// and move the replay horizon past it. + /// + /// Runs on the writer thread with no open group, which is the only place + /// the committed sequence means what a checkpoint needs it to mean. + /// + /// # Why an index run comes first + /// + /// The checkpoint format holds catalog, refs and receipts — **no object + /// index**. Once `committed_shard_sequence` advances past a frame, recovery + /// stops replaying it, so nothing rebuilds the `(namespace, object) -> + /// location` entries that frame carried. If those entries live only in the + /// root's delta layers, the objects vanish at the next open: still on disk, + /// still named by a segment, and unreachable. + /// + /// So every layer through the committed sequence is sealed into a run the + /// checkpoint's own manifest names, and removed from the root, before the + /// sequence moves. That ordering is the correctness of this function. + /// + /// `Ok(None)` is a shard with nothing committed — not sequence 0, which is + /// a real frame. + fn checkpoint(&mut self) -> Result>, StoreError> { + if let Some(poison) = self.poison.clone() { + return Err(poison); + } + let root = self.shared.committed.load(); + let Some(committed) = root.shard_committed_sequence(self.shard_index) else { + return Ok(None); + }; + + 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)?; + + // Already checkpointed at this sequence and nothing has been committed + // since. Reusing is not an optimization: `.checkpoint` is the + // installed name, so writing again would collide on a rename that + // refuses to replace, and a second manifest generation would claim + // progress that did not happen. + if manifest.committed_shard_sequence == committed + && manifest + .checkpoints + .iter() + .any(|(sequence, _)| *sequence == committed) + { + return Ok(root + .retained_generations() + .get(&GenerationId::new(self.shard_index, current_generation)) + .map(Arc::clone)); + } + + // --- index first, and completely ------------------------------------ + let backlog = self.unsealed_backlog(&root); + if backlog.layers != 0 { + match self.coverable_through(&root) { + Some(through) if through >= committed => self.seal_index(&root, committed)?, + other => { + return Err(StoreError::Corruption(format!( + "shard {} cannot checkpoint through {committed}: index coverage reaches \ + {other:?}, so advancing the committed sequence would drop the objects \ + those frames carry", + self.shard_index + ))) + } + } + } + + // Re-read: sealing published a generation and discarded the layers it + // covered, and everything below must build on that root, not the one + // this function started with. + let root = self.shared.committed.load(); + 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; + + let checkpoint = self.checkpoint_body(&root, committed)?; + + // --- durable from here: every failure below poisons ------------------ + let installed = match self.install_or_adopt_checkpoint(&paths, &checkpoint, &counters) { + Ok(installed) => installed, + Err(error) => return Err(self.poison_now(format!("installing a checkpoint: {error}"))), + }; + + // The manifest may not commit through a sequence no retained segment + // names, so the prefix in `active/` is sealed in the same publication. + // The segment inherits the tail's logical generation: the frames keep + // the identity every `IndexLocation` above already names them by + // (contract review 2026-07-30-A). + // + // Unless there is no prefix. A journal holding no frame is the state a + // reopen leaves when everything committed is already in segments — the + // shard is checkpointable and has nothing to seal, and A1 correctly + // refuses to seal an empty journal. The checkpoint is then a checkpoint + // row and nothing else, and the manifest must already commit exactly + // where the root does or the two disagree about frames neither holds. + let sealed = match self.journal.last_appended_shard_sequence() { + None if manifest.committed_shard_sequence != committed => { + return Err(StoreError::Corruption(format!( + "shard {} commits through {committed} with an empty journal, but its manifest commits through {}; nothing names the difference", + self.shard_index, manifest.committed_shard_sequence + ))) + } + None => None, + Some(_) => { + 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 the checkpointed prefix: {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())), + }; + Some(SealedPrefix { + generation: sealed_generation, + first, + filename, + segment_path, + active_path, + }) + } + }; + + let mut retained_tail_ranges = manifest.retained_tail_ranges.clone(); + if let Some(sealed) = &sealed { + retained_tail_ranges.push(TailRange { + generation: sealed.generation, + first_shard_sequence: sealed.first, + last_shard_sequence: committed, + filename: sealed.filename.clone(), + }); + } + let mut checkpoints = manifest.checkpoints.clone(); + checkpoints.retain(|(sequence, _)| *sequence != committed); + checkpoints.push((committed, checkpoint.file_name())); + let retain = self.shared.options.checkpoint_retain.max(2) as usize; + if checkpoints.len() > retain { + checkpoints.drain(..checkpoints.len() - retain); + } + let published = Manifest { + root_uuid, + generation: next_generation, + base_generation: manifest.base_generation, + retained_tail_ranges, + index_runs: manifest.index_runs.clone(), + checkpoints, + committed_shard_sequence: committed, + }; + if let Err(error) = segment::install_manifest( + &paths, + &published, + self.shared.options.manifest_retain, + &counters, + ) { + return Err(self.poison_now(format!("publishing a checkpoint manifest: {error}"))); + } + + // The sealed name is now authoritative, so the active one is dead. Only + // a shard that sealed replaces its journal; one that had nothing to + // seal keeps writing into the journal it already has. + if let Some(sealed) = &sealed { + if let Err(error) = + segment::unlink_sealed_journal(&sealed.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, committed), + committed.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-checkpoint journal: {error}")) + ) + } + }; + self.journal = journal; + self.tail_generation = sealed.generation + 1; + } + + let successor = match self.retain_checkpointed_generation( + &root, + current_generation, + next_generation, + sealed.as_ref(), + committed, + &installed, + ) { + Ok(successor) => successor, + Err(error) => return Err(self.poison_now(format!("pinning a checkpoint: {error}"))), + }; + + let mut generations = GenerationMap::new(); + generations.insert(successor.id, Arc::clone(&successor)); + let mut subtree = ShardSubtree::new( + self.shard_index, + committed, + Arc::new(IndexDelta::from_options(&self.shared.options)), + Some(committed), + 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 checkpoint: {error}"))); + } + + // Only after both the checkpoint and its authoritative manifest are + // fenced, so every retained manifest keeps a retained referent. + if let Err(error) = crate::checkpoint::prune( + &paths.checkpoints(), + self.shared.options.checkpoint_retain, + &counters, + ) { + return Err(self.poison_now(format!("pruning checkpoints: {error}"))); + } + Ok(Some(successor)) + } + + fn current_manifest_generation(&self, root: &CommittedRoot) -> Result { + root.retained_generations() + .keys() + .filter(|id| id.shard_index == self.shard_index) + .map(|id| id.manifest_generation) + .max() + .ok_or_else(|| { + StoreError::Corruption(format!( + "shard {} has no retained manifest generation to checkpoint against", + self.shard_index + )) + }) + } + + /// Derived shard state through `committed`, read from the published root. + fn checkpoint_body( + &self, + root: &CommittedRoot, + committed: u64, + ) -> Result { + let shard_count = self.shared.shard_count; + let mut catalog = NamespaceCatalog::new(); + let mut refs = Vec::new(); + for (namespace, state) in root.repositories().iter() { + if StoreOptions::shard_of(namespace, shard_count) != self.shard_index { + continue; + } + catalog.bind(NamespaceRecord { + namespace: *namespace, + genesis_authority: state.genesis_authority, + current_authority: state.current_authority, + lifecycle: state.lifecycle, + storage_mode: state.storage_mode, + repo_sequence: state.repo_sequence, + previous_event_digest: state.previous_event_digest, + })?; + for (target, object) in state.refs.iter() { + refs.push(RefRecord::from_target(*namespace, target, *object)); + } + } + + let mut receipts = Vec::new(); + for (key, entry) in root.terminal_statuses().iter() { + if StoreOptions::shard_of(&key.namespace, shard_count) != self.shard_index { + continue; + } + let TerminalStatusEntry::Committed(retained) = entry else { + continue; + }; + receipts.push(ReceiptRecord { + namespace: key.namespace, + operation_id: key.operation_id, + operation_digest: retained.operation_digest, + repo_sequence: retained.receipt.repo_sequence, + shard_sequence: retained.shard_sequence, + current_authority: retained.receipt.current_authority, + refs: retained.receipt.refs.clone(), + objects_new: retained.receipt.objects_new, + retry_until_micros: retained.retry_until_micros, + first_receipt_visibility_micros: Some(retained.first_visible_at_micros), + receipt_visible_until_micros: retained.receipt_visible_until_micros, + }); + } + + Ok(crate::checkpoint::Checkpoint { + root_uuid: self.shared.root_uuid, + shard_index: self.shard_index, + shard_committed_sequence: committed, + // The resume point. `journal_id` accompanies the offset because an + // offset alone is meaningless once the journal has rotated, and + // recovery checks the identity before it trusts the offset. This + // checkpoint seals the current journal away, so the pair names the + // journal being replaced and recovery correctly declines to resume + // into the fresh one. + active_journal_id: self.journal.journal_id(), + active_journal_offset: self.journal.cursor(), + created_at_micros: now_micros(), + catalog, + refs, + receipts, + }) + } + + /// Install the checkpoint, or adopt a finalized one already at its name. + /// + /// A crash between `install`'s rename and the manifest that publishes it + /// leaves a complete, fenced checkpoint no manifest references. Its name is + /// derived from the sequence, so the next attempt at the same sequence + /// renames onto it — and `rename_noreplace` refuses, permanently, for a + /// file that is not wrong but merely unreferenced. Validating and adopting + /// it is what turns a wedged shard back into a checkpointed one. + fn install_or_adopt_checkpoint( + &self, + paths: &crate::segment::ShardPaths, + checkpoint: &crate::checkpoint::Checkpoint, + counters: &Arc, + ) -> Result { + let directory = paths.checkpoints(); + let destination = directory.join(checkpoint.file_name()); + if !destination.exists() { + return crate::checkpoint::install(&directory, checkpoint, counters); + } + + // Adopted only through the reader recovery itself uses. A file at the + // right name that does not validate is not a checkpoint, and silently + // replacing it would destroy the evidence of whatever wrote it. + match crate::checkpoint::load_newest_valid( + &directory, + &self.shared.root_uuid, + self.shard_index, + self.shared.options.manifest_retain.max(2) as usize, + )? { + crate::checkpoint::CheckpointLoad::Loaded { + checkpoint: existing, + path, + .. + } if existing.shard_committed_sequence == checkpoint.shard_committed_sequence + && path == destination => + { + Ok(destination) + } + _ => Err(StoreError::Corruption(format!( + "{} already exists and is not a valid checkpoint for shard {} through {}", + destination.display(), + self.shard_index, + checkpoint.shard_committed_sequence + ))), + } + } + + /// The successor pin: what the predecessor held, plus the sealed segment + /// and the checkpoint, with the active tail replaced by the fresh journal. + fn retain_checkpointed_generation( + &self, + root: &CommittedRoot, + current_generation: u64, + next_generation: u64, + sealed: Option<&SealedPrefix>, + last: u64, + checkpoint_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 checkpointing" + )) + })?; + + let mut segments = previous.segments.to_vec(); + if let Some(sealed) = sealed { + segments.push(RootRetainedSegment::new( + sealed.generation, + self.journal_id_of_sealed_prefix(&sealed.segment_path)?, + sealed.first, + last, + PinnedFile::open(&sealed.segment_path)?, + )); + } + let mut checkpoints = previous.checkpoints.to_vec(); + checkpoints.push(PinnedFile::open(checkpoint_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), + checkpoints.into(), + active_tails.into(), + Arc::clone(&previous.projection_artifacts), + ))) + } + + /// Read back through the same reader recovery uses, rather than trusting + /// the id the writer happened to be holding. + fn journal_id_of_sealed_prefix(&self, segment_path: &Path) -> Result<[u8; 16], StoreError> { + let reader = SegmentReader::open(segment_path, &self.shared.root_uuid)?; + Ok(reader.footer().journal_id) + } + fn seal_index(&mut self, root: &CommittedRoot, covered_through: u64) -> Result<(), StoreError> { // Both ceilings, before anything is durable, and against the *same* // numbers recovery enforces when it reopens the manifest (scope 3.8). A @@ -6828,6 +7323,307 @@ mod index_maintenance_tests { } } + /// The checkpoint acceptance point: nothing acknowledged may become + /// unreachable by making it durable. + /// + /// Written **below** seal pressure on purpose, so no index run exists when + /// the checkpoint begins. The checkpoint format carries no object index, and + /// advancing `committed_shard_sequence` stops recovery replaying the frames + /// it covers — so unless the checkpoint seals the delta layers into a run + /// its own manifest names, every object committed here is on disk, named by + /// a segment, and unreachable after the reopen. + /// + /// The post-reopen lookups are guarded against being answered by a replay + /// delta, because a store that replayed the frames anyway would pass a bare + /// lookup while proving nothing about the run. + #[test] + fn a_checkpoint_keeps_every_object_resolvable_through_a_reopen() { + let serial = writer_serial(); + let temporary = tempfile::tempdir().expect("tempdir"); + let namespace = NamespaceId([0x51; 32]); + let configure = || { + let mut options = sealing_options(&serial, temporary.path(), 4_000_000); + options.max_index_runs = 8; + options.max_open_index_runs = 8; + options + }; + let mut objects = vec![genesis_object().id]; + { + let engine = StoreEngine::open(configure()).expect("open a fresh root"); + block_on(engine.submit(create_transaction(namespace, 2))).expect("create"); + objects.extend(push_groups(&engine, namespace, 0x60, 3)); + + let before = engine.index_maintenance(); + assert_eq!( + before.sealed_runs, 0, + "this must checkpoint from below seal pressure, or the run it depends \ + on was already there and the test proves nothing" + ); + assert!( + before.unsealed_delta_layers > 0, + "there must be unsealed layers for the checkpoint to have to persist" + ); + + let lease = engine.checkpoint().expect("checkpoint"); + let after = engine.index_maintenance(); + assert_eq!( + after.sealed_runs, 1, + "the checkpoint must seal what it is about to stop replaying" + ); + assert_eq!( + after.unsealed_delta_layers, 0, + "and discard it from the root, or the layers outlive the frames" + ); + drop(lease); + } + + let engine = StoreEngine::open(configure()).expect("reopen through production recovery"); + let root = engine.committed_root(); + for object in &objects { + let key = IndexKey::new(namespace, *object); + assert!( + root.index() + .delta_layers() + .iter() + .all(|layer| layer.delta.get(&key).is_none()), + "{} is answered by a replay delta, so this asserts nothing about the \ + checkpointed run", + object.to_hex() + ); + let location = root.index().get(&key).unwrap_or_else(|| { + panic!( + "{} was acknowledged before the checkpoint and is now unreachable", + object.to_hex() + ) + }); + assert!( + root.object_source(0, location.segment_generation) + .expect("resolve") + .is_some(), + "{} resolves to a generation the reopened root does not pin", + object.to_hex() + ); + } + } + + fn checkpoint_files(root: &Path, shard: u16) -> Vec { + let mut names: Vec = std::fs::read_dir(shard_paths(root, shard).checkpoints()) + .map(|entries| { + entries + .filter_map(Result::ok) + .filter_map(|entry| entry.file_name().into_string().ok()) + .collect() + }) + .unwrap_or_default(); + names.sort(); + names + } + + /// A shard with nothing committed has nothing to make durable. + /// + /// "No committed sequence" is not "committed through sequence 0" — the + /// second is a real frame. Writing `0.checkpoint` for an empty shard would + /// claim a frame that does not exist, and the next checkpoint at the real + /// sequence 0 would then collide with it. + #[test] + fn an_empty_shard_is_skipped_rather_than_checkpointed_at_sequence_zero() { + let serial = writer_serial(); + let temporary = tempfile::tempdir().expect("tempdir"); + let engine = StoreEngine::open(sealing_options(&serial, temporary.path(), 4_000_000)) + .expect("open a fresh root"); + + drop( + engine + .checkpoint() + .expect("checkpointing an empty root is not an error"), + ); + assert!( + checkpoint_files(temporary.path(), 0).is_empty(), + "an empty shard must install no checkpoint at all" + ); + } + + /// Checkpointing twice with nothing in between must reuse, not collide. + /// + /// The installed name is derived from the committed sequence, so a second + /// attempt at the same sequence renames onto a name that already exists — + /// and `rename_noreplace` refuses. Reuse is also the honest answer: no new + /// frame means no new durable state to record. + #[test] + fn checkpointing_again_with_no_new_work_reuses_rather_than_collides() { + let serial = writer_serial(); + let temporary = tempfile::tempdir().expect("tempdir"); + let namespace = NamespaceId([0x52; 32]); + let engine = StoreEngine::open(sealing_options(&serial, temporary.path(), 4_000_000)) + .expect("open a fresh root"); + block_on(engine.submit(create_transaction(namespace, 2))).expect("create"); + push_groups(&engine, namespace, 0x60, 2); + + drop(engine.checkpoint().expect("first checkpoint")); + let after_first = checkpoint_files(temporary.path(), 0); + let runs_after_first = engine.index_maintenance().sealed_runs; + + drop( + engine + .checkpoint() + .expect("a repeat checkpoint must not collide"), + ); + assert_eq!( + checkpoint_files(temporary.path(), 0), + after_first, + "a repeat with no new work must install nothing new" + ); + assert_eq!( + engine.index_maintenance().sealed_runs, + runs_after_first, + "and must not seal a run over an empty backlog either" + ); + } + + /// A crash between the checkpoint's rename and the manifest that publishes + /// it leaves a complete file no manifest references. + /// + /// Its name is the committed sequence, so the next attempt renames onto it + /// and is refused — permanently, for a file that is valid and merely + /// unreferenced. The shard would never checkpoint again. Validating and + /// adopting it is what makes the crash recoverable. + #[test] + fn a_finalized_but_unreferenced_checkpoint_is_adopted_rather_than_wedging() { + let serial = writer_serial(); + let temporary = tempfile::tempdir().expect("tempdir"); + let namespace = NamespaceId([0x53; 32]); + let engine = StoreEngine::open(sealing_options(&serial, temporary.path(), 4_000_000)) + .expect("open a fresh root"); + block_on(engine.submit(create_transaction(namespace, 2))).expect("create"); + let objects = push_groups(&engine, namespace, 0x60, 2); + let committed = engine + .committed_root() + .shard_committed_sequence(0) + .expect("the shard has committed frames"); + + // Exactly what the crash leaves: a valid checkpoint at the name this + // checkpoint is about to use, referenced by nothing. + let counters = Arc::new(DurabilityCounters::default()); + let stranded = crate::checkpoint::Checkpoint::empty(root_uuid_of(temporary.path()), 0); + let stranded = crate::checkpoint::Checkpoint { + shard_committed_sequence: committed, + ..stranded + }; + let path = crate::checkpoint::install( + &shard_paths(temporary.path(), 0).checkpoints(), + &stranded, + &counters, + ) + .expect("strand a finalized checkpoint"); + assert!(path.is_file()); + + drop( + engine + .checkpoint() + .expect("a finalized but unreferenced checkpoint must be adopted"), + ); + let root = engine.committed_root(); + for object in &objects { + assert!( + root.index() + .get(&IndexKey::new(namespace, *object)) + .is_some(), + "{} must survive an adopted checkpoint", + object.to_hex() + ); + } + } + + /// A checkpoint may not advance past an open, unfenced group. + /// + /// The frames of an open group are sequenced but not yet fenced or + /// published. A checkpoint that ran beside one would either commit through + /// a sequence no segment holds, or seal a prefix while the writer was still + /// appending to it. The boundary is structural rather than a lock: the + /// checkpoint arrives on the same channel as submissions, and the writer + /// loop publishes any open group before it runs one. + /// + /// Deterministic by construction. The submission is polled exactly once, so + /// it is inside the writer's open group and nothing has closed it, and the + /// checkpoint is issued from this thread while it sits there. + #[test] + fn a_checkpoint_publishes_the_open_group_it_finds_before_advancing() { + struct ThreadWaker(std::thread::Thread); + impl Wake for ThreadWaker { + fn wake(self: Arc) { + self.0.unpark(); + } + fn wake_by_ref(self: &Arc) { + self.0.unpark(); + } + } + + let serial = writer_serial(); + let temporary = tempfile::tempdir().expect("tempdir"); + let namespace = NamespaceId([0x54; 32]); + let configure = || { + let mut options = sealing_options(&serial, temporary.path(), 4_000_000); + // Room for the group to stay open. The idle bound is what closes a + // one-member group, so it is long relative to the microseconds + // between polling the submission and issuing the checkpoint, and + // short enough that the setup transaction still publishes. + options.max_group_transactions = 8; + options.max_group_idle = Duration::from_millis(250); + options + }; + let engine = StoreEngine::open(configure()).expect("open a fresh root"); + block_on(engine.submit(create_transaction(namespace, 2))).expect("create"); + + let waker = Waker::from(Arc::new(ThreadWaker(std::thread::current()))); + let mut context = Context::from_waker(&waker); + let pending = engine.submit(push_transaction(namespace, 0x61, 0x61, None)); + let mut pending = std::pin::pin!(pending); + assert!( + pending.as_mut().poll(&mut context).is_pending(), + "the submission must still be in the writer's open group" + ); + + drop( + engine + .checkpoint() + .expect("checkpoint beside an open group"), + ); + + let start = std::time::Instant::now(); + let receipt = loop { + if let Poll::Ready(output) = pending.as_mut().poll(&mut context) { + break output.expect("the open group publishes rather than being abandoned"); + } + assert!( + start.elapsed() < Duration::from_secs(30), + "the submission the checkpoint published never completed" + ); + std::thread::park_timeout(Duration::from_millis(5)); + }; + assert_eq!(receipt.repo_sequence, 1); + + // The claim is that the group was published *into* the checkpoint, not + // merely allowed to finish afterwards: the manifest commits through + // exactly what the root does. + let manifest = read_manifest( + &shard_paths(temporary.path(), 0), + read_current( + &shard_paths(temporary.path(), 0), + &root_uuid_of(temporary.path()), + ) + .expect("CURRENT") + .expect("a checkpointed root has a CURRENT") + .generation, + &root_uuid_of(temporary.path()), + ) + .expect("read the checkpoint's manifest"); + assert_eq!( + Some(manifest.committed_shard_sequence), + engine.committed_root().shard_committed_sequence(0), + "the checkpoint committed through the sequence the open group produced" + ); + } + /// The discard is scoped to the shard that sealed. #[test] fn only_the_sealing_shards_layers_are_discarded() { @@ -7215,3 +8011,66 @@ mod index_maintenance_tests { ); } } + +/// What a checkpoint sealed out of `active/`, when it sealed anything. +struct SealedPrefix { + generation: u64, + first: u64, + filename: String, + segment_path: PathBuf, + active_path: PathBuf, +} + +/// The manifest a generation names, or the empty one a shard that has never +/// published a manifest implies. +/// +/// A root opened fresh has a retained generation and no manifest file: nothing +/// has been sealed and nothing published, so there is no generation to read. +/// The synthesized manifest says exactly that — no retained tail, no runs, a +/// committed prefix of zero — and only **absence** synthesizes. A manifest that +/// exists and does not decode is corrupt, and replacing it would erase the +/// predecessor recovery falls back to. +fn read_manifest_or_empty( + paths: &crate::segment::ShardPaths, + generation: u64, + root_uuid: &[u8; 16], +) -> Result { + match segment::read_manifest(paths, generation, root_uuid) { + Ok(manifest) => Ok(manifest), + Err(StoreError::Io(error)) if error.kind() == std::io::ErrorKind::NotFound => { + Ok(Manifest { + root_uuid: *root_uuid, + generation, + base_generation: 0, + retained_tail_ranges: Vec::new(), + index_runs: Vec::new(), + checkpoints: Vec::new(), + committed_shard_sequence: 0, + }) + } + Err(error) => Err(error), + } +} + +/// A journal identity for the journal a checkpoint opens. +/// +/// Distinct domain from recovery's so the two cannot collide, and it carries +/// the sequence the journal begins at so a name in a crash image says which +/// checkpoint opened it. +fn fresh_checkpoint_journal_id(root_uuid: [u8; 16], shard: u16, committed: u64) -> [u8; 16] { + use std::sync::atomic::{AtomicU64, Ordering}; + static NEXT: AtomicU64 = AtomicU64::new(0); + + let mut hasher = blake3::Hasher::new(); + hasher.update(b"levcs-checkpoint-journal-id/v1\0"); + hasher.update(&root_uuid); + hasher.update(&shard.to_le_bytes()); + hasher.update(&committed.to_le_bytes()); + hasher.update(&now_micros().to_le_bytes()); + hasher.update(&std::process::id().to_le_bytes()); + hasher.update(&NEXT.fetch_add(1, Ordering::Relaxed).to_le_bytes()); + let digest = hasher.finalize(); + let mut id = [0u8; 16]; + id.copy_from_slice(&digest.as_bytes()[..16]); + id +} diff --git a/doc/instance-throughput-rewrite-plan.md b/doc/instance-throughput-rewrite-plan.md index b60005a..e30a3b8 100644 --- a/doc/instance-throughput-rewrite-plan.md +++ b/doc/instance-throughput-rewrite-plan.md @@ -1629,6 +1629,53 @@ exposed it. turns this refusal into successful reclamation. Recorded in scope §6.5 with the checkpointing work that will exercise it. +##### Contract review 2026-07-30-C + +**`StoreEngine::checkpoint` is implemented, and the ordering inside it is the correctness.** The +frozen D0 signature returned `NotImplemented`; it now checkpoints every shard on its own writer +thread and returns a lease. + +**The index run comes first, and completely.** The checkpoint format carries catalog, refs and +receipts and **no object index**. Once `committed_shard_sequence` advances past a frame, recovery +stops replaying it, so nothing rebuilds the entries that frame carried — if they live only in the +root's delta layers, the objects are on disk, named by a segment, and unreachable. Every layer +through the committed sequence is therefore sealed into a run the checkpoint's own manifest names, +and discarded from the root, before the sequence moves. Coverage short of the committed sequence is +refused rather than partially applied. + +**Real pinning.** `CheckpointLease` holds the exact `Arc` each shard published, +so it pins segment and checkpoint descriptors and index-run mappings directly. Count-based retention +cannot stand in: two further checkpoints prune the generation, and a lease that stopped being true +after two unrelated operations would not be a lease. + +**The group boundary is structural.** The checkpoint travels the same channel as submissions, and +the writer loop publishes any open group before running one. No lock, and no way to advance past +sequenced-but-unfenced frames. + +**Four edge cases, each with a test.** A shard with no committed sequence is skipped rather than +checkpointed at sequence 0 — the second is a real frame, and writing `0.checkpoint` for an empty +shard would collide with the real one later. A repeat with no new work reuses rather than renaming +onto its own name. A finalized but unreferenced checkpoint left by a crash between the rename and +the manifest is validated through recovery's own reader and adopted, instead of wedging the shard on +a permanent `EEXIST` for a file that is correct and merely unpublished. A journal holding no frame +seals nothing: that is the state a reopen leaves when everything committed is already in segments, +and A1 rightly refuses to seal an empty journal — the checkpoint is then a checkpoint row and +nothing else. + +**Evidence.** `a_checkpoint_keeps_every_object_resolvable_through_a_reopen` writes below seal +pressure, checkpoints, reopens, and resolves every object with the replay delta explicitly excluded +as the answer. Both mutations the review named fail it on reachability, not on bookkeeping: dropping +the run publication loses the genesis object, and sealing one sequence short loses the last push. + +**`store-bench` found the empty-journal case before any test did**, by calling the entry point that +had always refused. It now also earns `index_maintenance` from a measurement: the emitter reads +`index_maintenance().unsealed_delta_layers` where it previously recorded `None` for "no such reading +exists". That was the interface request the guard itself had written down, and it is the one line of +B4 surface this commit touches — the emitter's remaining claims are unchanged. + +**Next, and in this dispatch:** recovery discarding an index run whose covered identity was not +preserved, which turns 2026-07-30-B's refusal into reclamation. + ##### Contract review 2026-07-28-C B4 re-pointed `store-bench` at a real `StoreEngine::submit` and found that four verification diff --git a/doc/phase1-storage-spine-scope.md b/doc/phase1-storage-spine-scope.md index 7902762..c502558 100644 --- a/doc/phase1-storage-spine-scope.md +++ b/doc/phase1-storage-spine-scope.md @@ -1680,6 +1680,12 @@ This is a scope consequence and not a defect: nothing here is unsound, and both asserted by tests. It is written down so that "the shard seals under pressure" is not read as "the shard can run indefinitely under pressure", which is what checkpointing will make true. +**Closed by contract review 2026-07-30-C.** `StoreEngine::checkpoint` advances the committed prefix, +so the replay set is what lies above the newest checkpoint rather than everything the shard ever +wrote. The ordering that makes it safe is recorded there: the checkpoint format holds no object +index, so every delta layer through the committed sequence is sealed into a run its manifest names, +and discarded, before the sequence moves. + #### Closed: a run may cover frames the active journal still holds Recorded as a carry-forward with the index-maintenance slice and **closed** by contract review