diff --git a/crates/levcs-store/src/engine.rs b/crates/levcs-store/src/engine.rs index ca416b0..c391ce7 100644 --- a/crates/levcs-store/src/engine.rs +++ b/crates/levcs-store/src/engine.rs @@ -2716,7 +2716,10 @@ impl ShardWriter { let stable = layer.delta.iter().all(|(_, location)| { matches!( root.object_source(self.shard_index, location.segment_generation), - Ok(Some(crate::roots::RetainedObjectSource::Segment(_))) + Ok(Some( + crate::roots::RetainedObjectSource::Segment(_) + | crate::roots::RetainedObjectSource::ActiveTail(_) + )) ) }); if !stable { @@ -6314,83 +6317,28 @@ mod index_maintenance_tests { namespaces } - /// Run `write` against a fresh engine, close it, and reopen. - /// - /// The reopen is what makes the previous session's frames coverable: recovery - /// seals a journal holding frames into a segment, and only segment-backed - /// locations may be persisted into a run. - fn write_then_reopen( - serial: &WriterSerial, - root: &Path, - configure: impl Fn(&mut StoreOptions), - write: impl FnOnce(&StoreEngine), - ) -> StoreEngine { - { - let mut options = sealing_options(serial, root, 4_000_000); - configure(&mut options); - write(&StoreEngine::open(options).expect("open")); - } - let mut options = sealing_options(serial, root, 4_000_000); - configure(&mut options); - StoreEngine::open(options).expect("reopen through production recovery") - } - - /// A root whose backlog is **coverable**: frames written, then reopened, so - /// recovery has sealed them into a segment and the replayed layer's - /// locations name a generation the manifest pins. - /// - /// Sealing cannot cover frames still in `active/` — see - /// `ShardWriter::coverable_through` — so every test that needs a seal to - /// happen goes through here rather than writing and sealing in one session. - fn root_with_a_coverable_backlog( - serial: &WriterSerial, - root: &Path, - namespace: NamespaceId, - tag: u8, - pushes: u8, - configure: impl Fn(&mut StoreOptions), - ) -> (StoreEngine, Vec) { - let mut covered = vec![genesis_object().id]; - { - let mut options = sealing_options(serial, root, 4_000_000); - configure(&mut options); - let engine = StoreEngine::open(options).expect("open a fresh root"); - block_on(engine.submit(create_transaction(namespace, tag))).expect("create"); - covered.extend(push_groups(&engine, namespace, tag.wrapping_add(1), pushes)); - } - let mut options = sealing_options(serial, root, 4_000_000); - configure(&mut options); - let engine = StoreEngine::open(options).expect("reopen through production recovery"); - (engine, covered) - } - /// The acceptance point, and the discard it pays for. /// - /// The backlog is one replayed layer whose locations name a sealed segment. - /// Its entries are at the ceiling, so the next admission seals before it is - /// allowed to proceed — and is then refused by the replay ceiling, which the - /// seal cannot relieve while the committed prefix is unchanged. + /// Three groups leave three layers at a ceiling of three. The fourth + /// submission is admitted only after the seal — and is then refused by the + /// replay ceiling, which a seal cannot relieve while every frame is still in + /// `active/`. #[test] fn crossing_seal_required_seals_before_admitting_more_work() { let serial = writer_serial(); let temporary = tempfile::tempdir().expect("tempdir"); + let engine = StoreEngine::open(sealing_options(&serial, temporary.path(), 3)) + .expect("open a fresh root"); let namespace = NamespaceId([0x41; 32]); - let (engine, covered) = root_with_a_coverable_backlog( - &serial, - temporary.path(), - namespace, - 0x60, - 2, - |options| options.max_active_index_entries = 3, - ); + block_on(engine.submit(create_transaction(namespace, 1))).expect("create"); + push_groups(&engine, namespace, 0x60, 2); let before = engine.index_maintenance(); assert_eq!( (before.sealed_runs, before.unsealed_delta_layers), - (0, 1), - "the reopened root must carry the replayed backlog and no run" + (0, 3), + "three groups must leave three unsealed layers and no run" ); - assert_eq!(covered.len(), 3, "three objects are in that backlog"); let refused = block_on(engine.submit(push_transaction(namespace, 0x70, 0x70, None))) .expect_err("the shard is at the ceiling a reopen would have to rebuild"); @@ -6409,43 +6357,35 @@ mod index_maintenance_tests { assert_eq!( (after.sealed_runs, after.unsealed_delta_layers), (1, 0), - "the seal must publish one run and discard exactly the layer it covered, even \ - though the admission that triggered it was then refused" + "the seal must publish one run and discard exactly the three layers it covered" ); - - let uuid = root_uuid_of(temporary.path()); assert_eq!( - manifest_runs(temporary.path(), 0, uuid).len(), - 1, + manifest_runs(temporary.path(), 0, root_uuid_of(temporary.path())), + vec![index_run_filename(1)], "the run must be named by the current manifest" ); } /// Sealing is only correct if the run answers what the layers answered. - /// - /// Checked against the live root once the layer is gone, and then against a - /// root rebuilt by production recovery — which is also the "crash after - /// publication" case, since nothing shuts this store down cleanly. #[test] fn a_sealed_run_answers_every_covered_object_before_and_after_reopen() { let serial = writer_serial(); let temporary = tempfile::tempdir().expect("tempdir"); let namespace = NamespaceId([0x42; 32]); - let configure = |options: &mut StoreOptions| { - options.max_index_runs = 1; - options.max_open_index_runs = 1; + let configure = || { + let mut options = sealing_options(&serial, temporary.path(), 4_000_000); + options.max_index_runs = 3; + options.max_open_index_runs = 3; + options }; - let covered = { - let (engine, covered) = root_with_a_coverable_backlog( - &serial, - temporary.path(), - namespace, - 0x80, - 2, - configure, - ); - // One layer already crosses a fan-out ceiling of one, so this seals. - let _ = block_on(engine.submit(push_transaction(namespace, 0x88, 0x88, None))); + let mut covered = vec![genesis_object().id]; + { + let engine = StoreEngine::open(configure()).expect("open a fresh root"); + block_on(engine.submit(create_transaction(namespace, 2))).expect("create"); + covered.extend(push_groups(&engine, namespace, 0x80, 2)); + // Three layers cross the fan-out ceiling, so this submission seals. + block_on(engine.submit(push_transaction(namespace, 0x88, 0x88, None))) + .expect("the group after the seal commits"); assert_eq!(engine.index_maintenance().sealed_runs, 1, "the seal ran"); let root = engine.committed_root(); @@ -6473,12 +6413,9 @@ mod index_maintenance_tests { "the run's location must resolve to a file the root still pins" ); } - covered - }; + } - let mut options = sealing_options(&serial, temporary.path(), 4_000_000); - configure(&mut options); - let engine = StoreEngine::open(options).expect("reopen through production recovery"); + let engine = StoreEngine::open(configure()).expect("reopen through production recovery"); assert_eq!( engine.index_maintenance().sealed_runs, 1, @@ -6561,20 +6498,15 @@ mod index_maintenance_tests { let namespaces = two_shards(); let (sealing_shard, sealing_namespace) = namespaces[0]; let (other_shard, other_namespace) = namespaces[1]; - let configure = |options: &mut StoreOptions| { - options.shard_count = 4; - options.max_index_runs = 1; - options.max_open_index_runs = 1; - }; - - let engine = write_then_reopen(&serial, temporary.path(), configure, |engine| { - for (index, (_, namespace)) in namespaces.iter().enumerate() { - let tag = 0x11 + index as u8; - block_on(engine.submit(create_transaction(*namespace, tag))).expect("create"); - push_groups(engine, *namespace, 0xA0 + index as u8 * 8, 1); - } - }); + let mut options = sealing_options(&serial, temporary.path(), 4_000_000); + options.shard_count = 4; + options.max_index_runs = 1; + options.max_open_index_runs = 1; + let engine = StoreEngine::open(options).expect("open a fresh root"); + // One layer on the other shard and no further submission to it, so it + // never reaches an admission that would seal. + block_on(engine.submit(create_transaction(other_namespace, 0x11))).expect("create"); let other_layers_before = engine .committed_root() .index() @@ -6587,9 +6519,10 @@ mod index_maintenance_tests { "the other shard must have a backlog" ); - // One layer on the sealing shard already crosses a fan-out ceiling of - // one, so this submission seals before it is admitted. - let _ = block_on(engine.submit(push_transaction(sealing_namespace, 0xB0, 0xB0, None))); + block_on(engine.submit(create_transaction(sealing_namespace, 0x12))).expect("create"); + // Its one layer crosses a fan-out ceiling of one, so this seals. + block_on(engine.submit(push_transaction(sealing_namespace, 0xB0, 0xB0, None))) + .expect("the group after the seal commits"); let root = engine.committed_root(); assert_eq!( @@ -6606,7 +6539,6 @@ mod index_maintenance_tests { 1, "exactly the sealing shard published a run" ); - let _ = other_namespace; } /// The ceilings recovery enforces are the ceilings the writer enforces. @@ -6619,31 +6551,21 @@ mod index_maintenance_tests { fn the_run_ceilings_are_enforced_rather_than_raised() { let serial = writer_serial(); let temporary = tempfile::tempdir().expect("tempdir"); - let namespace = NamespaceId([0x44; 32]); - let configure = |options: &mut StoreOptions| { - options.max_index_runs = 1; - options.max_open_index_runs = 1; - }; - - // First session's frames, sealed into a run in the second session. - let engine = write_then_reopen(&serial, temporary.path(), configure, |engine| { - block_on(engine.submit(create_transaction(namespace, 4))).expect("create"); - push_groups(engine, namespace, 0xC0, 1); - }); - let _ = block_on(engine.submit(push_transaction(namespace, 0xC4, 0xC4, None))); - assert_eq!( - engine.index_maintenance().sealed_runs, - 1, - "the first seal ran" - ); - drop(engine); - - // A third session: the second session's frames are now segment-backed - // too, so a second seal is due — and does not fit. let mut options = sealing_options(&serial, temporary.path(), 4_000_000); - configure(&mut options); - let engine = StoreEngine::open(options).expect("reopen"); - let refused = block_on(engine.submit(push_transaction(namespace, 0xC8, 0xC8, None))) + options.max_index_runs = 1; + options.max_open_index_runs = 1; + let engine = StoreEngine::open(options).expect("open a fresh root"); + let namespace = NamespaceId([0x44; 32]); + + block_on(engine.submit(create_transaction(namespace, 4))).expect("create"); + // One layer crosses the fan-out ceiling of one, so this seals. + block_on(engine.submit(push_transaction(namespace, 0xC0, 0xC0, None))) + .expect("the group after the seal commits"); + assert_eq!(engine.index_maintenance().sealed_runs, 1); + + // That group left a layer of its own, so the next admission is due a + // second seal — which does not fit. + let refused = block_on(engine.submit(push_transaction(namespace, 0xC5, 0xC5, None))) .expect_err("a second run does not fit under max_index_runs = 1"); match refused { StoreError::LimitExceeded { limit, allowed, .. } => { @@ -6749,16 +6671,18 @@ mod index_maintenance_tests { let serial = writer_serial(); let temporary = tempfile::tempdir().expect("tempdir"); let namespace = NamespaceId([0x4A; 32]); - let configure = |options: &mut StoreOptions| { - options.max_active_index_entries = 3; - options.max_index_runs = 2; - options.max_open_index_runs = 2; + // Fan-out room to spare, so the seal is driven purely by entry pressure + // at the third entry and not by the layer count part-way through. + let configure = || { + let mut options = sealing_options(&serial, temporary.path(), 3); + options.max_index_runs = 8; + options.max_open_index_runs = 8; + options }; { - let engine = write_then_reopen(&serial, temporary.path(), configure, |engine| { - block_on(engine.submit(create_transaction(namespace, 0x0A))).expect("create"); - push_groups(engine, namespace, 0x2A, 2); - }); + let engine = StoreEngine::open(configure()).expect("open a fresh root"); + block_on(engine.submit(create_transaction(namespace, 0x0A))).expect("create"); + push_groups(&engine, namespace, 0x2A, 2); // Seals the three-entry backlog, then refuses this submission. let _ = block_on(engine.submit(push_transaction(namespace, 0x3A, 0x3A, None))); @@ -6766,9 +6690,10 @@ mod index_maintenance_tests { assert_eq!( (maintenance.sealed_runs, maintenance.unsealed_delta_layers), (1, 0), - "the fixture must leave one run and an empty backlog" + "the seal must leave one run and an empty backlog" ); + // The admission the early return used to skip entirely. let refused = block_on(engine.submit(push_transaction(namespace, 0x3B, 0x3B, None))) .expect_err("the run already holds everything the ceiling allows"); assert!( @@ -6782,10 +6707,7 @@ mod index_maintenance_tests { "expected the replay ceiling, got {refused:?}" ); } - - let mut reopen = sealing_options(&serial, temporary.path(), 4_000_000); - configure(&mut reopen); - StoreEngine::open(reopen).expect("a store must reopen whatever it accepted"); + StoreEngine::open(configure()).expect("a store must reopen whatever it accepted"); } /// P1: the byte ceiling is a ceiling too. @@ -6824,36 +6746,38 @@ mod index_maintenance_tests { StoreEngine::open(reopen).expect("a store must reopen whatever it accepted"); } - /// P1: a recovered run's locations must still resolve. + /// The property the generation split exists for: a run written against the + /// **active tail** still resolves after recovery seals that tail away. /// - /// The reopen test above cannot see this: its lookups are answered by the - /// replay delta, which shadows the run. This one queries the **run itself** - /// out of the recovered root and resolves the location it returns, which is - /// what a checkpoint — reading through the run rather than around it — would - /// do. It is the assertion that makes `coverable_through` load-bearing. + /// This is the case that was unwritable before. Every location in the run + /// names the tail's logical generation; recovery then moves those frames into + /// a segment, and the segment inherits the identity rather than taking a + /// fresh one. The run is queried directly out of the recovered root — not + /// through the layered index, where the replay delta would shadow it — which + /// is what a checkpoint reading through the run would do. #[test] fn a_recovered_runs_locations_still_resolve_to_a_pinned_source() { let serial = writer_serial(); let temporary = tempfile::tempdir().expect("tempdir"); let namespace = NamespaceId([0x49; 32]); - // One layer is enough to cross the fan-out ceiling, which is what the - // reopened root carries: recovery replays the whole journal into one. - let configure = |options: &mut StoreOptions| { - options.max_index_runs = 1; - options.max_open_index_runs = 1; + let configure = || { + let mut options = sealing_options(&serial, temporary.path(), 4_000_000); + options.max_index_runs = 2; + options.max_open_index_runs = 2; + options }; - - let engine = write_then_reopen(&serial, temporary.path(), configure, |engine| { + { + let engine = StoreEngine::open(configure()).expect("open a fresh root"); block_on(engine.submit(create_transaction(namespace, 9))).expect("create"); - push_groups(engine, namespace, 0xF0, 2); - }); - let _ = block_on(engine.submit(push_transaction(namespace, 0xF8, 0xF8, None))); - assert_eq!(engine.index_maintenance().sealed_runs, 1, "the seal ran"); - drop(engine); + push_groups(&engine, namespace, 0xF0, 1); + // Two layers cross the fan-out ceiling, so this seals — over + // locations that name the active tail. + block_on(engine.submit(push_transaction(namespace, 0xF8, 0xF8, None))) + .expect("the group after the seal commits"); + assert_eq!(engine.index_maintenance().sealed_runs, 1, "the seal ran"); + } - let mut options = sealing_options(&serial, temporary.path(), 4_000_000); - configure(&mut options); - let engine = StoreEngine::open(options).expect("reopen"); + let engine = StoreEngine::open(configure()).expect("reopen"); let root = engine.committed_root(); let run = root .index() @@ -6863,11 +6787,7 @@ mod index_maintenance_tests { .expect("the manifest's run is recovered") .clone(); let mut resolved = 0usize; - for object in [ - genesis_object().id, - ObjectId([0xF0; 32]), - ObjectId([0xF1; 32]), - ] { + for object in [genesis_object().id, ObjectId([0xF0; 32])] { let Some(location) = run.get(&IndexKey::new(namespace, object)) else { continue; }; @@ -6876,8 +6796,8 @@ mod index_maintenance_tests { root.object_source(0, location.segment_generation) .expect("resolve") .is_some(), - "the recovered run points at logical generation {} which nothing pins: a \ - reader going through the run rather than the replay delta reads nothing", + "the recovered run points at logical generation {} which nothing pins: the \ + sealed segment did not inherit the tail's identity", location.segment_generation ); } @@ -6893,26 +6813,22 @@ mod index_maintenance_tests { let serial = writer_serial(); let temporary = tempfile::tempdir().expect("tempdir"); let namespaces = two_shards(); - let configure = |options: &mut StoreOptions| { - options.shard_count = 4; - options.max_index_runs = 1; - options.max_open_index_runs = 1; - }; + let mut options = sealing_options(&serial, temporary.path(), 4_000_000); + options.shard_count = 4; + options.max_index_runs = 1; + options.max_open_index_runs = 1; + let engine = StoreEngine::open(options).expect("open a fresh root"); - let engine = write_then_reopen(&serial, temporary.path(), configure, |engine| { - for (index, (_, namespace)) in namespaces.iter().enumerate() { - let tag = 0x20 + index as u8; - block_on(engine.submit(create_transaction(*namespace, tag))).expect("create"); - push_groups(engine, *namespace, 0x30 + index as u8 * 8, 1); - } - }); for (index, (_, namespace)) in namespaces.iter().enumerate() { - let _ = block_on(engine.submit(push_transaction( + let tag = 0x20 + index as u8; + block_on(engine.submit(create_transaction(*namespace, tag))).expect("create"); + block_on(engine.submit(push_transaction( *namespace, - 0x50 + index as u8, - 0x50 + index as u8, + 0x30 + index as u8, + 0x30 + index as u8, None, - ))); + ))) + .expect("the group after the seal commits"); } assert_eq!( engine.index_maintenance().sealed_runs, diff --git a/crates/levcs-store/src/recovery.rs b/crates/levcs-store/src/recovery.rs index 9e77532..da1cd9f 100644 --- a/crates/levcs-store/src/recovery.rs +++ b/crates/levcs-store/src/recovery.rs @@ -1936,10 +1936,14 @@ fn recover_shard_under_lock( must_create_fresh = true; } ActiveJournalDisposition::Replay => { - let recovery_generation = recovery_generation_for_journal( + let RecoveryGenerations { + logical: logical_generation, + manifest: manifest_generation, + } = recovery_generations_for_journal( &paths, &root_uuid, &header.journal_id, + &selection, config, )?; let from = if checkpointed && checkpoint.active_journal_id == header.journal_id { @@ -1998,7 +2002,7 @@ fn recover_shard_under_lock( replayed.push(ReplayedFrame { facts, payload, - generation: recovery_generation, + generation: logical_generation, offset: scanned.offset, len: scanned.len, }); @@ -2029,7 +2033,7 @@ fn recover_shard_under_lock( &header, &full_scan, &paths, - recovery_generation, + logical_generation, &counters, )?; let filename = segment_path @@ -2055,14 +2059,14 @@ fn recover_shard_under_lock( }) .unwrap_or((Vec::new(), Vec::new(), Vec::new())); retained_tail_ranges.push(TailRange { - generation: recovery_generation, + generation: logical_generation, first_shard_sequence: first, last_shard_sequence: last, filename, }); let manifest = Manifest { root_uuid, - generation: recovery_generation, + generation: manifest_generation, base_generation: 0, retained_tail_ranges, index_runs, @@ -2080,29 +2084,29 @@ fn recover_shard_under_lock( let reader = Arc::new(SegmentReader::open(&segment_path, &root_uuid)?); retained_segments.push(RootRetainedSegment::new( - recovery_generation, + logical_generation, header.journal_id, first, last, PinnedFile::open(segment_path.clone())?, )); segments.push(RecoveredSegment { - generation: recovery_generation, + generation: logical_generation, first_shard_sequence: first, last_shard_sequence: last, path: segment_path, reader, }); - let manifest_path = paths.manifest(recovery_generation); + let manifest_path = paths.manifest(manifest_generation); selection = Some(ManifestSelection { manifest, - generation: recovery_generation, + generation: manifest_generation, path: manifest_path, source: ManifestSource::Current, rejected: Vec::new(), }); - report.manifest_generation = Some(recovery_generation); + report.manifest_generation = Some(manifest_generation); } drop(file); @@ -2113,13 +2117,13 @@ fn recover_shard_under_lock( let retained_tail = PinnedFile::from_shared(path.clone(), Arc::clone(&shared_file)); recovered_tail = Some(RecoveredTail { - logical_generation: recovery_generation, + logical_generation, journal_id: header.journal_id, path, validated_through: JOURNAL_HEADER_LEN as u64, file: shared_file, }); - retained_tails.push(RetainedTail::new(recovery_generation, retained_tail)); + retained_tails.push(RetainedTail::new(logical_generation, retained_tail)); } } } @@ -2574,12 +2578,50 @@ fn apply_recovered_refs( Ok(()) } -fn recovery_generation_for_journal( +/// The two generations a recovery needs, which are not the same number. +/// +/// Contract review 2026-07-30-A. They were one, and the conflation was invisible +/// for as long as nothing outlived a session holding an `IndexLocation`. +pub(crate) struct RecoveryGenerations { + /// The **logical** generation of the frames: the identity an `IndexLocation` + /// names, inherited from the active tail so that sealing it into a segment + /// moves the bytes without changing what they are called. + logical: u64, + /// The generation of the manifest this recovery installs. A counter over + /// manifests, which an index run's manifest also advances. + manifest: u64, +} + +/// Choose both. +/// +/// # Why they had to be separated +/// +/// A single `max(manifest, .seg, .idx) + 1` served as the sealed segment's +/// logical generation *and* as the new manifest's generation. Every index run +/// publishes a manifest, so every seal moved the number — and the segment +/// recovery then wrote took the moved value while the frames it holds were +/// already named by the old one. Persisted index locations dangled: +/// `object_source` returned `None` for a run the manifest still named. +/// +/// Separated, each number answers its own question. The logical generation is +/// [`active_tail_logical_generation`] — the identity the frames already have, +/// derived from the manifest's committed prefix and stable across opens. The +/// manifest generation is the next free one, so an index run's manifest cannot +/// collide with a recovery's. +/// +/// # Resumption +/// +/// An interrupted recovery leaves a `.seg` whose footer names this journal, or a +/// `.recovery--.prefix`. Either fixes the logical generation: +/// finishing an interrupted seal must reuse the identity it already chose, not +/// pick a fresh one and orphan the artifact. +fn recovery_generations_for_journal( paths: &ShardPaths, root_uuid: &[u8; 16], active_journal_id: &[u8; 16], + selection: &Option, config: &RecoveryConfig<'_>, -) -> Result { +) -> Result { let per_manifest = usize::try_from(config.max_index_runs) .unwrap_or(usize::MAX) .saturating_add(config.checkpoint_retain as usize) @@ -2590,7 +2632,13 @@ fn recovery_generation_for_journal( .saturating_mul(per_manifest) .max(64); let mut inspected = 0usize; - let mut maximum = 0u64; + // Two questions, two answers. A single maximum over all three namespaces — + // manifest generations, segment logical generations, index-run identities — + // is what conflated them: an index run's manifest moved a number the sealed + // segment's identity was read from. Only a segment can collide with a + // segment, and only a manifest with a manifest. + let maximum_manifest; + let mut occupied_segments: BTreeSet = BTreeSet::new(); let mut resumable = BTreeSet::new(); let manifests = segment::list_manifest_generations(paths)?; @@ -2602,7 +2650,7 @@ fn recovery_generation_for_journal( allowed: budget as u64, }); } - maximum = maximum.max(manifests.into_iter().max().unwrap_or(0)); + maximum_manifest = manifests.into_iter().max().unwrap_or(0); for (dir, extension) in [(paths.segments(), "seg"), (paths.indexes(), "idx")] { let entries = match std::fs::read_dir(&dir) { @@ -2643,6 +2691,7 @@ fn recovery_generation_for_journal( if &reader.footer().journal_id == active_journal_id { resumable.insert(reader.footer().generation); } + occupied_segments.insert(reader.footer().generation); reader.footer().generation }) } else { @@ -2665,7 +2714,13 @@ fn recovery_generation_for_journal( path.display() )) })?; - maximum = maximum.max(generation); + // A `.seg` occupies a logical generation whether or not it opens: + // an interrupted seal can leave a name holding something that is not + // a segment at all, and that name is still taken. Recorded from the + // parsed name precisely because the reader above could not read it. + if extension == "seg" { + occupied_segments.insert(generation); + } } } @@ -2696,7 +2751,6 @@ fn recovery_generation_for_journal( path.display() )) })?; - maximum = maximum.max(generation); resumable.insert(generation); } @@ -2707,20 +2761,47 @@ fn recovery_generation_for_journal( resumable ))); } - if let Some(generation) = resumable.into_iter().next() { - if generation != maximum { - return Err(StoreError::Corruption(format!( - "recovery artifact generation {generation} for journal {} is below occupied \ - immutable generation {maximum}", - hex::encode(active_journal_id) - ))); - } - return Ok(generation); + let manifest = maximum_manifest.checked_add(1).ok_or_else(|| { + StoreError::Corruption("no generation remains for recovery manifests".into()) + })?; + + // Resuming: the identity was already chosen and the artifact carries it. + if let Some(logical) = resumable.into_iter().next() { + return Ok(RecoveryGenerations { logical, manifest }); } - maximum.checked_add(1).ok_or_else(|| { - StoreError::Corruption("no generation remains for recovery artifacts".into()) - }) + // Otherwise the frames keep the name they already have. + let preferred = active_tail_logical_generation(selection); + if !occupied_segments.contains(&preferred) { + return Ok(RecoveryGenerations { + logical: preferred, + manifest, + }); + } + + // The identity is taken — by an orphan from an interrupted seal, or by a + // segment no manifest references. Recovery must still succeed: an orphan is + // a documented state in which the active journal remains the authority, and + // refusing here would turn a recoverable root into an outage. + // + // So the seal falls back to a free generation, and the frames are renamed. + // That is the pre-2026-07-30-A behaviour and it carries its hazard with it: + // an index run holding locations against `preferred` no longer resolves. The + // closure is to drop such runs during recovery rather than let a reader find + // nothing through them — recorded in scope §6.5 and not attempted here, + // because it is a change to what recovery *discards* and belongs with the + // checkpointing work that will exercise it. + let logical = occupied_segments + .iter() + .copied() + .max() + .unwrap_or(preferred) + .max(preferred) + .checked_add(1) + .ok_or_else(|| { + StoreError::Corruption("no logical generation remains for a recovery segment".into()) + })?; + Ok(RecoveryGenerations { logical, manifest }) } /// The logical generation of the active tail above a manifest's committed prefix. diff --git a/doc/instance-throughput-rewrite-plan.md b/doc/instance-throughput-rewrite-plan.md index f0aed75..3fbb830 100644 --- a/doc/instance-throughput-rewrite-plan.md +++ b/doc/instance-throughput-rewrite-plan.md @@ -1477,6 +1477,54 @@ an active tail are at least visible in one place. so a four-shard root with `max_index_runs = 1` refused the second shard its first run. Counted per shard now, from the shard's own retained generations. +##### Contract review 2026-07-30-A + +**`recovery_generation` was two numbers wearing one name, and they are now separated.** Granted and +landed in `recovery.rs` (A2) as the prerequisite the checkpointing dispatch was to open with. + +One `max(manifest, .seg, .idx) + 1` served as the logical generation of the segment recovery seals +*and* as the generation of the manifest recovery installs. Every index run publishes a manifest, so +every seal moved the number; the segment recovery later wrote took the moved value while the frames +inside it were already named by the old one. Persisted index locations dangled — `object_source` +returning `None` for a run the manifest still named. + +Separated: + +- **logical** — the identity of the frames, taken from `active_tail_logical_generation`, which is + derived from the manifest's committed prefix and moves only when a rotation seals a tail. Sealing a + journal into a segment now changes where the bytes are, not what they are called. An interrupted + recovery still resumes: a `.seg` footer naming this journal, or a `.recovery--.prefix`, + fixes the identity that was already chosen. +- **manifest** — the next free manifest generation, so an index run's manifest and a recovery's + cannot collide. + +Collision detection moved with them. The old check compared a resumable generation against a maximum +that mixed all three namespaces, which an index run could raise on its own; only a segment can +collide with a segment, so the check is now against segment generations alone, and claiming one that +another journal holds is `Corruption`. + +**What this closes.** The scope §6.5 carry-forward "a run may only cover frames a segment already +holds" is gone, and with it the restriction in `ShardWriter::coverable_through`: a run may now cover +locations naming the active tail, because the tail's identity survives being sealed away. Sealing +covers current frames in the session that wrote them rather than lagging one behind. Mutation-checked +by putting the segment's identity back on the manifest counter, which reproduces the original defect +exactly — a recovered run pointing at a logical generation nothing pins. + +**One case is not closed, and it was found by a test rather than predicted.** An orphan `.seg` from +an interrupted seal occupies a logical generation whether or not it is a readable segment — and the +identity the tail wants may be exactly the one it holds. `frame_golden`'s +`an_orphan_segment_leaves_the_active_journal_the_authority` failed on the first attempt for that +reason, which is also the test that documents why recovery must not refuse in this state: the active +journal is still the authority and an outage here would be the wrong answer. The seal therefore falls +back to a free generation and renames the frames, exactly as before this review, and an index run +against the old identity dangles. Confined to roots carrying an orphan segment, where it was +previously universal; the closure — recovery discarding runs whose identity was not preserved — is +recorded in scope §6.5 for the checkpointing work. + +The other §6.5 carry-forward stands: sealing still moves no frame out of `active/`, so the replay +ceiling still bounds admission and an entry-pressure seal still lands on it. That is what +`StoreEngine::checkpoint` is for, and it is now unblocked. + ##### 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 b7c7644..7876ef6 100644 --- a/doc/phase1-storage-spine-scope.md +++ b/doc/phase1-storage-spine-scope.md @@ -1680,30 +1680,31 @@ 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. -#### Carry-forward: a run may only cover frames a segment already holds +#### Closed: a run may cover frames the active journal still holds -The second half of the same fact, found in review (contract review 2026-07-29-D) and stated -separately because it constrains what a seal may *contain* rather than when one happens. +Recorded as a carry-forward with the index-maintenance slice and **closed** by contract review +2026-07-30-A, which is why the restriction it describes is no longer in the code. -An `IndexLocation` names a **logical generation**, and a sealed run is the first thing in this store -that persists one beyond the session that wrote it. That is sound only for a generation that is -stable across opens, and exactly one kind is: a segment, pinned by name in the manifest. The active -tail's generation is assigned by recovery from `max(manifest, .seg, .idx) + 1`, so it moves whenever -any artifact appears — including the index run's own manifest. When recovery seals a journal holding -frames, the resulting segment takes that counter's value and not the generation the tail had, so a -run written against tail locations dangles at the next open: `object_source` returns `None` for a run -the manifest still names, and a reader going *through* the run rather than around it reads nothing. +The constraint was that an `IndexLocation` names a logical generation, and only a segment's was +stable: the active tail's came from `max(manifest, .seg, .idx) + 1`, so it moved whenever any +artifact appeared — an index run's own manifest was enough — and recovery then sealed the tail into +a segment under the moved value rather than the identity the frames already had. -Sealing therefore covers an oldest-first prefix of layers whose every entry is segment-backed, and -stops at the first that is not. Until a rotation or checkpoint moves frames out of `active/`, that -means a seal covers what a previous session left in segments — sealing lags one session behind. +`recovery_generation` was two numbers wearing one name. Split, each answers its own question: the +sealed segment inherits the **tail's** logical generation, so moving frames from `active/` to +`segments/` changes where they are and not what they are called; the recovery manifest takes the +next free **manifest** generation, so an index run's manifest cannot collide with a recovery's. +Sealing therefore covers current frames, in the session that wrote them. -**The blocker to lift it**, and it should be lifted with checkpointing rather than after: -`recovery_generation` currently serves as two different numbers at once — the generation of the new -manifest recovery installs, and the logical generation of the segment it seals. Preserving a location -across the move from `active/` to `segments/` requires the segment to inherit the tail's logical -generation, which requires those two to be separated first. That is an A2 recovery-core change and -was deliberately not attempted inside a B1 integration commit. +**Residual, one case wide.** If a `.seg` already occupies the tail's identity — an orphan from an +interrupted seal, or a segment no manifest references — the seal cannot use that name and falls back +to a free generation, renaming the frames. An index run holding locations against the old identity +then resolves to nothing. Recovery must not refuse here: an orphan segment is a documented state in +which the active journal stays the authority, and refusing would turn a recoverable root into an +outage. The closure is for recovery to **discard index runs whose covered identity was not +preserved**, which is a change to what recovery reclaims and belongs with the checkpointing work that +will exercise it. Until then the hazard is confined to roots carrying an orphan segment, where it was +previously universal. ### 6.5 B3 — StagingSessions