neuwirth
/
LeVCS
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Bound index sealing by what a reopen can rebuild and resolve 

Four findings against the previous commit. Each fix carries a regression that
fails against the landed code, and each was mutation-checked back to it.

Admission projected nothing. The replay-ceiling check read the published root,
so it decided about a transaction it had not counted: one object present, a
two-object transaction against a ceiling of two committed and the store then
failed to reopen. The open group was the same hole one step along. Admission
now projects sealed runs, unsealed layers, the open group, and the incoming
transaction.

The byte ceiling was unguarded. Recovery rebuilds into one delta that refuses
on either ceiling, so narrow frames across namespaces passed admission and
failed to reopen on `max_active_index_bytes`. Both are checked, and the
projection counts namespaces because the encoding pays a section header per
namespace. `index::encoded_bytes_for` is that arithmetic extracted, so this
file does not carry a copy of the encoding's shape.

A recovered run's locations did not resolve, and this reshaped the slice. An
`IndexLocation` names a logical generation, and a run is the first thing here
that persists one across a session — sound only for a generation that is
stable, which is a segment's alone. The active tail's is assigned from
`max(manifest, .seg, .idx) + 1`, so it moves whenever any artifact appears
(the run's own manifest suffices), and recovery seals a journal holding
frames at that counter rather than at the generation the tail had. The
previous reopen test could not see it: its lookups were answered by the
replay delta shadowing the run. Coverage is now an oldest-first prefix of
layers whose every entry is segment-backed, which makes the broken run
unwritable rather than untested. The cost — sealing lags one session behind
until frames leave `active/` — is recorded in scope §6.5.

Preserving the tail's generation across the seal was attempted and withdrawn.
`recovery_generation` is at once the new manifest's generation and the sealed
segment's logical generation, so the real fix separates those two numbers in
A2's recovery core, and that belongs with checkpointing rather than inside a
B1 integration commit. The first attempt also targeted the wrong branch: a
journal holding frames is replaced, not kept. `active_tail_logical_generation`
is left extracted at the one path that already used that formula so the two
ways of numbering an active tail are visible together.

The run ceilings counted every shard, where recovery enforces them against one
shard's manifest — a four-shard root with `max_index_runs = 1` refused the
second shard its first run. Counted per shard now.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01XKzM69CHmBuDcA3qN1jFdh
This commit is contained in:
Levi Neuwirth 2026-07-30 14:21:32 -04:00
parent fef8520bdf
commit e03ca2b698
5 changed files with 619 additions and 116 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

616
crates/levcs-store/src/engine.rs
View File

@ -1836,7 +1836,7 @@ impl ShardWriter {
// a seal that fails partway through has to poison, and poisoning a shard
// that has just accepted a transaction owes that caller an answer it can
// no longer give.
self.seal_index_if_required()?;
self.seal_index_if_required(transaction)?;
// The reservation is what makes two concurrent submissions of one
// operation ID resolvable at all. It happens before sequencing, so a
@ -2623,13 +2623,31 @@ impl ShardWriter {
backlog
}
/// Index entries a reopen of this shard would have to rebuild.
/// The index a reopen of this shard would have to rebuild, **including** work
/// that is admitted but not yet published.
///
/// Sealed runs plus the unsealed backlog, because with the committed prefix
/// unadvanced every frame is still replayed. Counted from the shard's own
/// retained generations rather than from the root's run vector, which is not
/// tagged by shard and would charge one shard for another's runs.
fn replayable_index_entries(&self, root: &CommittedRoot, backlog: &UnsealedBacklog) -> u64 {
/// Three sources, and leaving out any one of them was a way to overshoot:
///
/// - sealed runs and unsealed layers, which the published root reports;
/// - the open group, whose members have been admitted and will publish;
/// - the transaction being admitted right now.
///
/// The last two are the reason this takes an incoming shape rather than
/// reading the root alone. A root-only check passes on the state *before*
/// the transaction it is deciding about, so a two-object transaction against
/// a one-object ceiling committed and then failed to reopen.
///
/// Namespaces are counted because the encoding pays a section header per
/// namespace and the byte ceiling is measured in encoded bytes. Both figures
/// over-estimate — an object rewritten in two groups is counted twice, and a
/// namespace already present is re-counted if it is in the open group — so
/// the decision can only ever refuse early.
fn replayable_index(
&self,
root: &CommittedRoot,
backlog: &UnsealedBacklog,
incoming: Option<&ValidatedTransaction>,
) -> (u64, u64) {
let sealed: u64 = root
.retained_generations()
.values()
@ -2637,7 +2655,76 @@ impl ShardWriter {
.flat_map(|generation| generation.index_runs.iter())
.map(|retained| retained.run().entry_count())
.sum();
sealed + backlog.entries
let open_group: u64 = self
.pending
.iter()
.map(|prepared| prepared.objects.len() as u64)
.sum();
let admitting = incoming.map_or(0, |transaction| transaction.objects.len() as u64);
let entries = sealed + backlog.entries + open_group + admitting;
// Every namespace this shard owns, from the published root, plus the
// ones only the open group and the incoming transaction know about.
let shard_count = self.shared.shard_count;
let mut namespaces: std::collections::BTreeSet<NamespaceId> = root
.repositories()
.keys()
.copied()
.filter(|namespace| StoreOptions::shard_of(namespace, shard_count) == self.shard_index)
.collect();
namespaces.extend(self.pending.iter().map(|prepared| prepared.key.namespace));
if let Some(transaction) = incoming {
namespaces.insert(transaction.namespace);
}
(
entries,
crate::index::encoded_bytes_for(entries, namespaces.len() as u64),
)
}
/// The newest shard sequence a run may claim to cover.
///
/// An `IndexLocation` names a **logical generation**, and a run is the first
/// thing in this store that persists one across a session. That is only sound
/// while the generation it names is stable, and exactly one kind is: a
/// segment, which the manifest pins by name. The active tail's generation is
/// assigned by recovery from a counter over existing artifacts, so it moves
/// whenever any artifact appears — and when recovery seals a journal holding
/// frames, the resulting segment takes that counter's value rather than the
/// generation the tail had. Locations written against the tail therefore
/// dangle after the next open: `object_source` returns `None` for a run the
/// manifest still names, and a reader going through the run rather than the
/// replay delta above it reads nothing.
///
/// So coverage stops at the first layer holding an entry that is not
/// segment-backed. An oldest-first prefix rather than a filter, because the
/// discard is expressed as "everything through sequence N": covering a later
/// layer while skipping an earlier one would discard the earlier one too.
///
/// The practical consequence is recorded in scope §6.5 — until a rotation or
/// a checkpoint moves frames out of `active/`, the coverable set is whatever
/// a previous session left in segments, so sealing lags one session behind.
/// Lifting it means separating two numbers `recovery_generation` currently
/// serves as at once: the new manifest's generation and the sealed segment's
/// logical generation.
fn coverable_through(&self, root: &CommittedRoot) -> Option<u64> {
let mut covered = None;
for layer in root.index().delta_layers().iter().rev() {
if layer.shard_index != self.shard_index {
continue;
}
let stable = layer.delta.iter().all(|(_, location)| {
matches!(
root.object_source(self.shard_index, location.segment_generation),
Ok(Some(crate::roots::RetainedObjectSource::Segment(_)))
)
});
if !stable {
break;
}
covered = Some(layer.through_shard_sequence);
}
covered
}
/// Seal when the accumulated delta has reached a hard ceiling (scope 3.6).
@ -2650,7 +2737,10 @@ impl ShardWriter {
/// groups would leave lookup fan-out unbounded while entry pressure stayed
/// low. `max_index_runs` bounds it, which is the ceiling the refusal this
/// replaced already used for exactly that reason.
fn seal_index_if_required(&mut self) -> Result<(), StoreError> {
fn seal_index_if_required(
&mut self,
incoming: &ValidatedTransaction,
) -> Result<(), StoreError> {
let root = self.shared.committed.load();
let backlog = self.unsealed_backlog(&root);
if backlog.layers == 0 {
@ -2664,7 +2754,13 @@ impl ShardWriter {
) == DeltaPressure::SealRequired;
let fan_out = backlog.layers as u64 >= u64::from(self.shared.options.max_index_runs);
if entries_or_bytes || fan_out {
self.seal_index(&root, backlog.through_shard_sequence)?;
// Only what a run may soundly persist; see `coverable_through`. When
// nothing is coverable the seal does not happen and the ceiling check
// below refuses instead, which is the honest outcome — a run over
// active-journal locations would be unreadable after the next open.
if let Some(covered_through) = self.coverable_through(&root) {
self.seal_index(&root, covered_through)?;
}
}
// What a reopen would have to rebuild, checked **after** the seal so a
@ -2691,15 +2787,27 @@ impl ShardWriter {
// Entry-pressure sealing becomes useful when `StoreEngine::checkpoint`
// can advance the committed prefix; until then it converts an
// unreopenable store into an honest refusal, which is all it can do.
let backlog = self.unsealed_backlog(&self.shared.committed.load());
let replayable = self.replayable_index_entries(&self.shared.committed.load(), &backlog);
if replayable >= self.shared.options.max_active_index_entries {
let root = self.shared.committed.load();
let backlog = self.unsealed_backlog(&root);
let (entries, bytes) = self.replayable_index(&root, &backlog, Some(incoming));
// Recovery rebuilds into one `IndexDelta::from_options`, which refuses on
// either ceiling, so both are checked here. Guarding only entries let a
// run of narrow frames across many namespaces pass admission and fail to
// reopen on `max_active_index_bytes`.
if entries > self.shared.options.max_active_index_entries {
return Err(StoreError::LimitExceeded {
limit: "max_active_index_entries",
observed: replayable + 1,
observed: entries,
allowed: self.shared.options.max_active_index_entries,
});
}
if bytes > self.shared.options.max_active_index_bytes {
return Err(StoreError::LimitExceeded {
limit: "max_active_index_bytes",
observed: bytes,
allowed: self.shared.options.max_active_index_bytes,
});
}
Ok(())
}
@ -2723,7 +2831,17 @@ impl ShardWriter {
// numbers recovery enforces when it reopens the manifest (scope 3.8). A
// seal that passed here can always be reopened; one that bypassed them
// would install a manifest this store could never open again.
let projected = root.index().sealed_run_count() as u64 + 1;
// This shard's runs, not the root's. Recovery enforces both ceilings
// against one shard's manifest, so counting every shard here refused a
// shard its first run because another shard already had one — a refusal
// recovery would never have made.
let owned_runs: u64 = root
.retained_generations()
.values()
.filter(|generation| generation.id.shard_index == self.shard_index)
.map(|generation| generation.index_runs.len() as u64)
.sum();
let projected = owned_runs + 1;
for (limit, allowed) in [
("max_index_runs", self.shared.options.max_index_runs),
(
@ -3410,7 +3528,7 @@ mod tests {
/// verification at a mirror on another instance.
const EVIDENCE_ACTOR: [u8; 32] = [0x7e; 32];
fn evidence() -> TransactionEvidenceV1 {
pub(super) fn evidence() -> TransactionEvidenceV1 {
TransactionEvidenceV1::AdministrativeV1 {
actor: EVIDENCE_ACTOR,
actor_key_epoch: 7,
@ -3419,7 +3537,7 @@ mod tests {
}
}
fn deadline() -> i64 {
pub(super) fn deadline() -> i64 {
now_micros() + 600_000_000
}
@ -6033,8 +6151,11 @@ mod index_maintenance_tests {
use super::tests::*;
use super::*;
use std::time::Duration;
use crate::index::IndexKey;
use crate::segment::{index_run_filename, read_current, read_manifest, RootLayout};
use crate::{PrivilegedConstruction, StagedObject};
/// A root whose index ceilings make sealing reachable in a few groups.
///
@ -6076,6 +6197,31 @@ mod index_maintenance_tests {
.root_uuid
}
/// A push carrying two objects, for the projection tests.
fn two_object_push(namespace: NamespaceId, blob: u8) -> ValidatedTransaction {
let authority = genesis_object().id;
ValidatedTransaction::builder(PrivilegedConstruction::internal())
.namespace(namespace)
.operation(OperationId([blob; 16]), ObjectId([blob; 32]), deadline())
.objects(vec![
StagedObject {
id: ObjectId([blob; 32]),
object_type: ObjectType::Blob,
raw: vec![blob; 64],
},
StagedObject {
id: ObjectId([blob.wrapping_add(1); 32]),
object_type: ObjectType::Blob,
raw: vec![blob.wrapping_add(1); 64],
},
])
.refs(Vec::new())
.authority(Some(authority), Some(authority))
.evidence(evidence())
.build()
.expect("a two-object push")
}
/// Commit `count` single-object pushes, each in its own group.
fn push_groups(
engine: &StoreEngine,
@ -6093,35 +6239,100 @@ mod index_maintenance_tests {
objects
}
/// Two namespaces that route to different shards of a four-shard root.
fn two_shards() -> Vec<(u16, NamespaceId)> {
let mut namespaces = Vec::new();
for byte in 0u8..64 {
let candidate = NamespaceId([byte; 32]);
let shard = StoreOptions::shard_of(&candidate, 4);
if !namespaces.iter().any(|(s, _)| *s == shard) {
namespaces.push((shard, candidate));
}
if namespaces.len() == 2 {
break;
}
}
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<ObjectId>) {
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.
///
/// Three groups leave three layers and no run. The fourth submission is
/// admitted only after the seal, so the numbers this reads are exactly:
/// one run, and a backlog holding only the group that ran after it.
/// 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.
#[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]);
block_on(engine.submit(create_transaction(namespace, 1))).expect("create");
let (engine, covered) = root_with_a_coverable_backlog(
&serial,
temporary.path(),
namespace,
0x60,
2,
|options| options.max_active_index_entries = 3,
);
// The create is itself a group carrying one object, so two pushes bring
// the accumulation to the ceiling.
push_groups(&engine, namespace, 0x60, 2);
let before = engine.index_maintenance();
assert_eq!(
(before.sealed_runs, before.unsealed_delta_layers),
(0, 3),
"three groups must leave three unsealed layers and no run"
(0, 1),
"the reopened root must carry the replayed backlog and no run"
);
assert_eq!(covered.len(), 3, "three objects are in that backlog");
// The fourth submission is refused — but only *after* the seal it
// triggered, and for the replay ceiling rather than for the pressure
// that caused the seal. An entry-pressure seal lands the shard exactly
// on what a reopen could rebuild, because the frames it indexed are all
// still in `active/`; see `seal_index_if_required`.
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");
assert!(
@ -6139,51 +6350,43 @@ 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 three layers it covered"
"the seal must publish one run and discard exactly the layer it covered, even \
though the admission that triggered it was then refused"
);
// File presence is not publication; the manifest is.
let uuid = root_uuid_of(temporary.path());
assert_eq!(
manifest_runs(temporary.path(), 0, uuid),
vec![index_run_filename(1)],
manifest_runs(temporary.path(), 0, uuid).len(),
1,
"the run must be named by the current manifest"
);
assert!(shard_paths(temporary.path(), 0)
.indexes()
.join(index_run_filename(1))
.is_file());
}
/// Sealing is only correct if the run answers what the layers answered.
///
/// Checked twice: against the live root once the layers are gone, and
/// against a root rebuilt by production recovery after the engine is
/// dropped. The second half is also the "crash after publication" case —
/// nothing shuts the store down cleanly here.
/// 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 mut covered = Vec::new();
let mut reopen_options = || {
// Sealed by the fan-out ceiling, not by entry pressure. An
// entry-pressure seal leaves the shard *at* the replay ceiling by
// construction — see `replayable_index_entries` — so a reopen test
// built on it would be asserting against a store that is one
// transaction from refusing.
let mut options = sealing_options(&serial, temporary.path(), 4_000_000);
options.max_index_runs = 3;
options.max_open_index_runs = 3;
options
let configure = |options: &mut StoreOptions| {
options.max_index_runs = 1;
options.max_open_index_runs = 1;
};
{
let engine = StoreEngine::open(reopen_options()).expect("open a fresh root");
block_on(engine.submit(create_transaction(namespace, 2))).expect("create");
covered.push(genesis_object().id);
covered.extend(push_groups(&engine, namespace, 0x80, 2));
push_groups(&engine, namespace, 0x90, 1);
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)));
assert_eq!(engine.index_maintenance().sealed_runs, 1, "the seal ran");
let root = engine.committed_root();
@ -6211,10 +6414,12 @@ mod index_maintenance_tests {
"the run's location must resolve to a file the root still pins"
);
}
}
covered
};
let engine =
StoreEngine::open(reopen_options()).expect("reopen through production recovery");
let mut options = sealing_options(&serial, temporary.path(), 4_000_000);
configure(&mut options);
let engine = StoreEngine::open(options).expect("reopen through production recovery");
assert_eq!(
engine.index_maintenance().sealed_runs,
1,
@ -6294,30 +6499,23 @@ mod index_maintenance_tests {
fn only_the_sealing_shards_layers_are_discarded() {
let serial = writer_serial();
let temporary = tempfile::tempdir().expect("tempdir");
let mut options = sealing_options(&serial, temporary.path(), 4_000_000);
options.shard_count = 4;
// Sealed by fan-out so the shard can keep running afterwards.
options.max_index_runs = 4;
options.max_open_index_runs = 4;
let engine = StoreEngine::open(options).expect("open a fresh root");
// Two namespaces that route to different shards.
let mut namespaces = Vec::new();
for byte in 0u8..64 {
let candidate = NamespaceId([byte; 32]);
let shard = StoreOptions::shard_of(&candidate, 4);
if !namespaces.iter().any(|(s, _)| *s == shard) {
namespaces.push((shard, candidate));
}
if namespaces.len() == 2 {
break;
}
}
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);
}
});
block_on(engine.submit(create_transaction(other_namespace, 0x11))).expect("create");
push_groups(&engine, other_namespace, 0xA0, 1);
let other_layers_before = engine
.committed_root()
.index()
@ -6330,10 +6528,9 @@ mod index_maintenance_tests {
"the other shard must have a backlog"
);
block_on(engine.submit(create_transaction(sealing_namespace, 0x12))).expect("create");
// Four layers, then a fifth submission whose admission crosses the
// fan-out ceiling and seals.
push_groups(&engine, sealing_namespace, 0xB0, 4);
// 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)));
let root = engine.committed_root();
assert_eq!(
@ -6350,30 +6547,44 @@ 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.
///
/// With room for one run, the second seal is refused — and refused as
/// `LimitExceeded` naming the ceiling, not raised, not bypassed, and not the
/// `NotImplemented` the unimplemented path used to answer. The shard stays
/// usable, because nothing durable was attempted.
/// With room for one run, a second seal is refused as `LimitExceeded` naming
/// the ceiling — not raised, not bypassed, and not the `NotImplemented` the
/// unimplemented path used to answer. Nothing durable was attempted, so the
/// shard stays usable.
#[test]
fn the_run_ceilings_are_enforced_rather_than_raised() {
let serial = writer_serial();
let temporary = tempfile::tempdir().expect("tempdir");
let mut options = sealing_options(&serial, temporary.path(), 4_000_000);
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]);
let configure = |options: &mut StoreOptions| {
options.max_index_runs = 1;
options.max_open_index_runs = 1;
};
block_on(engine.submit(create_transaction(namespace, 4))).expect("create");
// One layer already crosses the fan-out ceiling, so this seals.
push_groups(&engine, namespace, 0xC0, 1);
assert_eq!(engine.index_maintenance().sealed_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);
let refused = block_on(engine.submit(push_transaction(namespace, 0xC5, 0xC5, None)))
// 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)))
.expect_err("a second run does not fit under max_index_runs = 1");
match refused {
StoreError::LimitExceeded { limit, allowed, .. } => {
@ -6393,6 +6604,205 @@ mod index_maintenance_tests {
);
}
/// P1: admission must project the transaction it is deciding about.
///
/// A root-only check reads the state *before* the incoming objects, so with
/// room for two entries and one already present, a two-object transaction
/// committed and the store then failed to open. The reopen is the assertion:
/// a store must never be unable to read back what it accepted.
#[test]
fn admission_projects_the_incoming_transaction_into_the_replay_ceiling() {
let serial = writer_serial();
let temporary = tempfile::tempdir().expect("tempdir");
let namespace = NamespaceId([0x46; 32]);
{
let engine = StoreEngine::open(sealing_options(&serial, temporary.path(), 2))
.expect("open a fresh root");
block_on(engine.submit(create_transaction(namespace, 6))).expect("one object");
let refused = block_on(engine.submit(two_object_push(namespace, 0xE0)))
.expect_err("two more objects do not fit under a ceiling of two");
assert!(
matches!(
refused,
StoreError::LimitExceeded {
limit: "max_active_index_entries",
..
}
),
"expected the replay ceiling to refuse, got {refused:?}"
);
}
StoreEngine::open(sealing_options(&serial, temporary.path(), 2))
.expect("a store must reopen whatever it accepted");
}
/// The same projection, for members already admitted into the open group.
///
/// `max_group_transactions` is raised so the group stays open across both
/// submissions; without counting `pending`, the second is decided against a
/// root the first has not reached yet.
#[test]
fn admission_projects_the_open_group_into_the_replay_ceiling() {
let serial = writer_serial();
let temporary = tempfile::tempdir().expect("tempdir");
let namespace = NamespaceId([0x47; 32]);
let mut options = sealing_options(&serial, temporary.path(), 3);
options.max_group_transactions = 8;
options.max_group_idle = Duration::from_millis(2_000);
{
let engine = StoreEngine::open(options).expect("open a fresh root");
block_on(engine.submit(create_transaction(namespace, 7))).expect("one object");
let outcomes = [0xE4u8, 0xE6]
.map(|blob| block_on(engine.submit(two_object_push(namespace, blob))));
assert!(
outcomes.iter().any(|outcome| matches!(
outcome,
Err(StoreError::LimitExceeded {
limit: "max_active_index_entries",
..
})
)),
"five objects must not both be admitted under a ceiling of three: {outcomes:?}"
);
}
StoreEngine::open(sealing_options(&serial, temporary.path(), 3))
.expect("a store must reopen whatever it accepted");
}
/// P1: the byte ceiling is a ceiling too.
///
/// Sized so entries stay far below their own limit and only the encoded byte
/// total crosses. Recovery rebuilds into one `IndexDelta::from_options`,
/// which refuses on either.
#[test]
fn admission_projects_the_replay_byte_ceiling() {
let serial = writer_serial();
let temporary = tempfile::tempdir().expect("tempdir");
let namespace = NamespaceId([0x48; 32]);
let mut options = sealing_options(&serial, temporary.path(), 4_000_000);
// Room for a couple of entries and their one section header, and no more.
options.max_active_index_bytes = crate::index::encoded_bytes_for(2, 1);
{
let engine = StoreEngine::open(options).expect("open a fresh root");
block_on(engine.submit(create_transaction(namespace, 8))).expect("one object");
block_on(engine.submit(push_transaction(namespace, 0xE8, 0xE8, None)))
.expect("a second object still fits");
let refused = block_on(engine.submit(push_transaction(namespace, 0xE9, 0xE9, None)))
.expect_err("a third object exceeds the encoded byte ceiling");
assert!(
matches!(
refused,
StoreError::LimitExceeded {
limit: "max_active_index_bytes",
..
}
),
"expected the byte ceiling, got {refused:?}"
);
}
let mut reopen = sealing_options(&serial, temporary.path(), 4_000_000);
reopen.max_active_index_bytes = crate::index::encoded_bytes_for(2, 1);
StoreEngine::open(reopen).expect("a store must reopen whatever it accepted");
}
/// P1: a recovered run's locations must still resolve.
///
/// 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.
#[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 engine = write_then_reopen(&serial, temporary.path(), configure, |engine| {
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);
let mut options = sealing_options(&serial, temporary.path(), 4_000_000);
configure(&mut options);
let engine = StoreEngine::open(options).expect("reopen");
let root = engine.committed_root();
let run = root
.index()
.sealed_runs()
.iter()
.next()
.expect("the manifest's run is recovered")
.clone();
let mut resolved = 0usize;
for object in [
genesis_object().id,
ObjectId([0xF0; 32]),
ObjectId([0xF1; 32]),
] {
let Some(location) = run.get(&IndexKey::new(namespace, object)) else {
continue;
};
resolved += 1;
assert!(
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",
location.segment_generation
);
}
assert!(
resolved > 0,
"the run must answer for the objects it covers"
);
}
/// P2: the run ceilings are per shard, as recovery enforces them.
#[test]
fn the_run_ceilings_are_counted_per_shard() {
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 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(
*namespace,
0x50 + index as u8,
0x50 + index as u8,
None,
)));
}
assert_eq!(
engine.index_maintenance().sealed_runs,
2,
"each shard owns one run; a per-root count would refuse the second shard its first"
);
}
/// A seal that fails after the device has moved leaves no usable writer.
///
/// The fence inside the run's own write is made to fail. That is the first

17
crates/levcs-store/src/index.rs
View File

@ -409,6 +409,20 @@ pub enum DeltaPressure {
SealRequired,
}
/// The encoded cost of `entries` entries spread over `namespaces` sections.
///
/// [`IndexDelta::encoded_bytes`] is this function over its own occupancy. Split
/// out for the same reason as [`delta_pressure`]: the shard writer has to ask
/// what a delta it has not built yet *would* cost — the one recovery will
/// rebuild, including a transaction not yet admitted — and multiplying by
/// `INDEX_ENTRY_LEN` at that call site would put the encoding's shape in a file
/// that has no business knowing it.
///
/// Granted to B1 as contract review 2026-07-29-D, amendment 2 of 2.
pub fn encoded_bytes_for(entries: u64, namespaces: u64) -> u64 {
entries * INDEX_ENTRY_LEN as u64 + namespaces * INDEX_SECTION_LEN as u64
}
/// The rule of plan §5.3, over an occupancy and the ceilings it is measured
/// against.
///
@ -487,8 +501,7 @@ impl IndexDelta {
/// On-disk cost of the current contents, which is what the byte ceiling is
/// expressed in and what scope 8.2 budgets.
pub fn encoded_bytes(&self) -> u64 {
self.entry_count * INDEX_ENTRY_LEN as u64
+ self.by_namespace.len() as u64 * INDEX_SECTION_LEN as u64
encoded_bytes_for(self.entry_count, self.by_namespace.len() as u64)
}
/// A deliberately honest estimate of *resident* cost, which is larger than

34
crates/levcs-store/src/recovery.rs
View File

@ -2141,17 +2141,7 @@ fn recover_shard_under_lock(
let first_shard_sequence = last_committed
.map(|sequence| sequence.saturating_add(1))
.unwrap_or(0);
let logical_generation = selection
.as_ref()
.and_then(|selected| {
selected
.manifest
.retained_tail_ranges
.last()
.map(|range| range.generation)
})
.unwrap_or(0)
.saturating_add(1);
let logical_generation = active_tail_logical_generation(&selection);
let (tail, retained) = create_fresh_active_journal(
&paths,
root_uuid,
@ -2733,6 +2723,28 @@ fn recovery_generation_for_journal(
})
}
/// The logical generation of the active tail above a manifest's committed prefix.
///
/// One formula, called from both places that produce an active tail — the
/// journal recovery keeps and the one it creates fresh. It is stable across
/// opens because it depends only on the manifest's retained tail ranges, which
/// change when a rotation seals a tail into a segment and at no other time. That
/// stability is what lets an `IndexLocation` outlive the session that wrote it;
/// see the note at the surviving-tail call site.
fn active_tail_logical_generation(selection: &Option<ManifestSelection>) -> u64 {
selection
.as_ref()
.and_then(|selected| {
selected
.manifest
.retained_tail_ranges
.last()
.map(|range| range.generation)
})
.unwrap_or(0)
.saturating_add(1)
}
fn preserve_crash_journal(
active_path: &Path,
quarantine_dir: &Path,

43
doc/instance-throughput-rewrite-plan.md
View File

@ -1434,6 +1434,49 @@ enforces, before anything durable happens, so a seal that passes can always be r
`store-bench` is untouched. Its `index_maintenance` condition stays preliminary until B4 consumes this
after `checkpoint()` can flush the final below-watermark backlog.
##### Contract review 2026-07-29-D
Four findings against 2026-07-29-C, all upheld. Each fix has a regression that fails against the
landed code, and each was mutation-checked back to the landed behaviour.
**P1 — admission projected nothing.** The replay-ceiling check read the published root only, so it
decided about a transaction it had not counted. One object present, a two-object transaction against
a ceiling of two: committed, then failed to reopen. The open group was the same hole one step
further along, since its members are admitted and will publish. Admission now projects sealed runs,
unsealed layers, the open group, **and** the transaction being decided.
**P1 — the byte ceiling was unguarded.** Recovery rebuilds into one `IndexDelta::from_options`,
which refuses on either ceiling; the guard checked entries alone, so narrow frames spread over
namespaces passed admission and failed to reopen on `max_active_index_bytes`. Both are checked now,
and the projection counts namespaces because the encoding pays a section header per namespace.
`index::encoded_bytes_for` (amendment 2) is that arithmetic, extracted so `engine.rs` does not carry
a copy of the encoding's shape; `IndexDelta::encoded_bytes` is the same function over its own fields.
**P1 — a recovered run's locations did not resolve, and this is the finding that reshaped the
slice.** An `IndexLocation` names a logical generation, and a run is the first thing here that
persists one across a session. Only a segment's generation is stable: the active tail's is assigned
from `max(manifest, .seg, .idx) + 1`, which moves whenever any artifact appears — the run's *own*
manifest is enough — and recovery seals a journal holding frames into a segment at that counter's
value rather than at the generation the tail had. The landed reopen test could not see it because its
lookups were answered by the replay delta shadowing the run.
Coverage is now restricted to an oldest-first prefix of layers whose every entry is segment-backed.
This makes the broken run unwritable rather than merely untested, and the consequence — sealing lags
one session behind until frames leave `active/` — is recorded in scope §6.5.
The alternative, preserving the tail's logical generation across the seal, was attempted and
withdrawn: `recovery_generation` is simultaneously the new manifest's generation and the sealed
segment's logical generation, so the fix requires separating those two numbers in A2's recovery core.
That is the right change and it belongs with checkpointing, not inside a B1 integration commit; the
first attempt at it also targeted the wrong branch, since a journal holding frames is *replaced*
rather than kept, which is worth knowing before the next attempt. `active_tail_logical_generation` is
left extracted, called from the one path that already used that formula, so the two ways of numbering
an active tail are at least visible in one place.
**P2 — the run ceilings counted every shard.** Recovery enforces them against one shard's manifest,
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-28-C
B4 re-pointed `store-bench` at a real `StoreEngine::submit` and found that four verification

25
doc/phase1-storage-spine-scope.md
View File

@ -1680,6 +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
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.
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.
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.
**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.
### 6.5 B3 — StagingSessions
Owns `staging.rs`. Deliverable 9: bounded invisible projection staging.