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

Install an adopted projection through submit 

`adopt_projection` reached `build` and stopped there. Wiring it through
`submit` is one change because there is no safe partial: the moment the
builder stops refusing, a submit that ignores the adoption drops a live
pin with no outcome, which is the leak scope 6.5 declares a bug.

The pin cannot live inside the transaction. `Prepared` retains an
`Arc<ValidatedTransaction>` so a group can be re-sequenced when the
pre-mark deadline recheck drops a member, and an `Arc` has no move-out.
It travels beside it in a one-shot `AdoptionSlot` that settles itself
from an append phase rather than from a guess: the phase advances at the
first frame write, not at the fence, so torn and unfenced bytes stay
recovery-owned. That makes the unenumerated routes safe by construction
-- queue rejection, pre-append errors and panics, deadline reforming,
writer unwinding -- instead of correct only where someone remembered.

Two orderings are now statements rather than drop-order accidents. Both
the poison window and every pre-append refusal settle the pin before
resolving the waiter; otherwise a submit could return while staging still
believed the pin was live.

Membership could not travel as an index delta. Adopted generations sit in
a reserved band `1 << 63` away from journal generations, and an index run
packs `segment_generation` as a 16-bit delta from a per-namespace section
base, so one section cannot hold both domains -- the adoption committed
and the next checkpoint poisoned the shard. B3 now materializes
staging-owned runs post-fence, deterministically and idempotently, and
`resolve_committed` only opens, verifies, and pins them. Recovery replays
the same materialization, which is what makes a crash between the fence
and materialization recoverable rather than ambiguous. Runs partition one
per adoption, splitting every 65,536 ordinals, because two adoptions'
sequences differ by more than a section can span.

The run count is therefore knowable before anything is appended, so the
ceiling check is a refusal at revalidation rather than a poison after the
fence.

Payload kind is a matrix axis: all eight Wave B rows drive inline and
staged-projection, and the adoption outcome is read from staging's
counters rather than derived, so the fixture and the run remain two
independent derivations.

Contract review 2026-08-09-A records the interface changes.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01JGdH5V43XWnj1PdHqiPktQ
This commit is contained in:
Levi Neuwirth 2026-08-09 15:39:46 +02:00
parent 9bc298868a
commit fb71bba615
No known key found for this signature in database
11 changed files with 2287 additions and 329 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -44,7 +44,7 @@ use levcs_core::{ObjectId, ObjectType};
use levcs_protocol::oracle::{self, AppendDeadlinePhase, DeadlineDecision};
use levcs_protocol::v2::{
self as protocol, AppliedRefV1, CommittedTransactionV1, DurabilityResultV1, RefMutation,
RefStateV1, SignedCommittedTransactionV1,
RefStateV1, SignedCommittedTransactionV1, StagedProjectionInstallV1,
};
use crate::checkpoint::{ReceiptRecord, RefRecord};
@ -72,8 +72,11 @@ use crate::roots::{
};
use crate::segment::{self, RootLayout, SegmentReader};
use crate::snapshot::RepoSnapshot;
use crate::staging::ProjectionStaging;
use crate::transaction::ValidatedTransaction;
use crate::staging::{
ProjectionAdoptionOutcome, ProjectionRecoveryResolver, ProjectionStaging,
RecoveredProjectionArtifacts,
};
use crate::transaction::{AdoptionSlot, ValidatedTransaction};
use crate::types::{
CommitEvidenceSigner, CommitReceipt, DurabilityCounterSnapshot, DurabilityCounters,
NamespaceId, OperationId, PendingPhase, StoreError, TransactionStatus,
@ -140,8 +143,10 @@ struct EngineShared {
/// — two handles, or two processes, each admitting up to the full global
/// limit. Ownership here is what makes "root-global" true rather than
/// asserted.
_staging: Arc<ProjectionStaging>,
_session: RecoverySession,
staging: Arc<ProjectionStaging>,
/// Shared with [`EngineShared::staging`], which retains its own clone.
/// Whichever is dropped last releases `LOCK`.
_session: Arc<RecoverySession>,
/// What startup state 1 actually cost, or `None` when this open did not
/// initialize.
///
@ -299,9 +304,19 @@ impl StoreEngine {
// without it, and a resolver constructed after recovery would be a
// resolver recovery never had. It then lives in `EngineShared` for
// exactly as long as the session does.
//
// The session is wrapped first because staging *keeps* the lease
// rather than being shown it: a caller may hold staging, a stage
// session, or an adoption pin after this engine is gone, and each can
// still write under the root. Sharing the lease is what makes the lock
// outlive every writer rather than every *shard* writer.
let session = Arc::new(session);
let staging_durability = Arc::new(DurabilityCounters::default());
let staging =
ProjectionStaging::open(&session, options.clone(), Arc::clone(&staging_durability))?;
let staging = ProjectionStaging::open(
Arc::clone(&session),
options.clone(),
Arc::clone(&staging_durability),
)?;
let config = RecoveryConfig::from_store_options(&options)
.with_projection_recovery_resolver(staging.as_ref());
@ -331,7 +346,7 @@ impl StoreEngine {
))),
status_metrics: OperationStatusMetrics::default(),
options,
_staging: staging,
staging,
_session: session,
#[cfg(test)]
initialization,
@ -349,6 +364,32 @@ impl StoreEngine {
Ok(Self { shared, shards })
}
/// The one [`ProjectionStaging`] for this root.
///
/// Staging is reached *through* the engine rather than constructed beside
/// it because its ceilings are root-global and its constructor demands
/// proof of the root lock, which this engine holds. A caller that opened
/// its own would be a second accountant for one root — see the note on
/// [`EngineShared::staging`].
///
/// **Borrowing here guarantees nothing about lifetimes**, and an earlier
/// version of this comment claimed it did. The returned reference cannot
/// escape, but an `Arc::clone` of it can, and so can a
/// `ProjectionStageSession` or an adoption pin — both of which own a clone.
/// Any of them can still write under the root after this engine is gone.
/// What actually holds is ownership: [`ProjectionStaging`] retains a clone
/// of the `RecoverySession` lease, so `LOCK` is released when the last of
/// them is dropped rather than when the engine is.
///
/// This is the production path, not a test hook. Staging a projection and
/// adopting it are two halves of one flow — `begin`/`put_chunk`/`seal`
/// here, then [`ValidatedTransactionBuilder::adopt_projection`] and
/// [`Self::submit`] — and the descriptor and pin that join them are only
/// issued by this object.
pub fn staging(&self) -> &Arc<ProjectionStaging> {
&self.shared.staging
}
/// Acquire exactly one committed root and derive a repository view from
/// it. Readers never observe an index entry newer than their captured
/// root (plan §5.3).
@ -361,7 +402,11 @@ impl StoreEngine {
// One load, so the snapshot is a single point in the publication
// order. Resolving the repository from a second load could pair one
// generation's `RepoState` with another's index.
RepoSnapshot::capture(self.shared.committed.load_full(), repo)
RepoSnapshot::capture(
self.shared.committed.load_full(),
repo,
StoreOptions::shard_of(&repo, self.shared.shard_count),
)
}
/// Submit a validated transaction for sequencing, group append, fence, and
@ -1657,6 +1702,14 @@ struct Prepared {
repo_state: Arc<RepoState>,
objects: Vec<(ObjectId, u8)>,
receipt: CommitReceipt,
/// This member's adoption pin, if it adopts a projection.
///
/// Beside the `Arc` and not inside it: see [`AdoptionSlot`]. It is attached
/// to whichever `Prepared` actually joins `pending` — the group-full path
/// prepares twice, and a pin attached to the discarded first attempt would
/// settle when that attempt dropped, while the transaction it belongs to
/// went on to append.
adoption: Option<AdoptionSlot>,
/// Retained so the whole group can be re-sequenced, re-chained, and
/// re-signed when the pre-mark deadline recheck drops a member.
///
@ -1764,7 +1817,7 @@ fn pre_append_panic_error() -> StoreError {
impl ShardWriter {
fn accept(&mut self, submission: Submission) {
let Submission {
transaction,
mut transaction,
completion,
} = submission;
// Registered before anything that can fail or panic. Until this
@ -1772,18 +1825,45 @@ impl ShardWriter {
// `fail_newest_waiter` is what resolves it.
self.waiters.push(Waiter::new(completion));
if let Some(poison) = self.poison.clone() {
self.fail_newest_waiter(poison);
return;
}
// Out **before** the `Arc`, which is the only moment it can leave: an
// `Arc` has no move-out, and this pin has to be settled by hand. From
// here it is a local, and every failure below settles it explicitly.
let adoption = transaction.take_adoption();
let transaction = Arc::new(transaction);
let (prepared, frame) = match self.prepare_caught(&transaction) {
Ok(pair) => pair,
Err(error) => {
self.fail_newest_waiter(error);
// Every pre-append refusal in this function settles the pin *before*
// resolving the waiter. Letting the local drop at the end of scope
// would settle it too, but only after the caller already had its
// answer — so a submit could return while staging still believed the
// pin was live, and a caller that immediately asked staging would get
// the pre-submit answer. Settling first makes the order a statement
// rather than a consequence of where `adoption` happens to be declared.
macro_rules! refuse {
($error:expr) => {{
drop(adoption);
self.fail_newest_waiter($error);
return;
}};
}
if let Some(poison) = self.poison.clone() {
refuse!(poison);
}
// Pre-append revalidation, and the *only* validation that may use the
// pin's read-only view. `resolve_committed` is the post-fence
// counterpart and its contract requires the frame to already be
// durable, so calling it here would be wrong in the dangerous
// direction — it would treat an unwritten frame as authority.
if let Some(slot) = adoption.as_ref() {
if let Err(error) = self.revalidate_adoption(&transaction, slot) {
refuse!(error);
}
}
let (mut prepared, frame) = match self.prepare_caught(&transaction) {
Ok(pair) => pair,
Err(error) => refuse!(error),
};
if let Err(frame) = self.builder.push(frame) {
@ -1801,23 +1881,25 @@ impl ShardWriter {
self.roll_back_prepared(&prepared);
self.publish_open_group();
if let Some(poison) = self.poison.clone() {
self.fail_newest_waiter(poison);
return;
refuse!(poison);
}
let (reprepared, frame) = match self.prepare_caught(&transaction) {
let (mut reprepared, frame) = match self.prepare_caught(&transaction) {
Ok(pair) => pair,
Err(error) => {
self.fail_newest_waiter(error);
return;
}
Err(error) => refuse!(error),
};
self.builder
.push(frame)
.expect("an empty group admits any frame");
self.attach_newest_reservation(reprepared.key);
// Onto the second preparation, because the first was rolled back
// and dropped. `publish_open_group` above appended the *previous*
// group, never this transaction, so the pin is still `BeforeAppend`
// and this attachment is not late.
reprepared.adoption = adoption;
self.pending.push(reprepared);
} else {
self.attach_newest_reservation(prepared.key);
prepared.adoption = adoption;
self.pending.push(prepared);
}
@ -1826,6 +1908,73 @@ impl ShardWriter {
}
}
/// Prove the pin still protects the projection this frame is about to name.
///
/// The pairing itself is guaranteed by [`StagedProjectionAdoption`], which
/// cannot be assembled from mismatched halves. What that cannot guarantee
/// is *time*: the descriptor was derived when the session finalized, and a
/// submit happens later. This recomputes it from the pin's current
/// read-only resolution and requires the two to be identical.
///
/// Rebuilt through staging's own [`install_from_resolution`] rather than
/// compared field by field. A second derivation of a canonical value is a
/// second opinion about it, and the fields it forgot to compare would be
/// exactly the ones that drifted.
fn revalidate_adoption(
&self,
transaction: &ValidatedTransaction,
slot: &AdoptionSlot,
) -> Result<(), StoreError> {
let Some(descriptor) = transaction.projection() else {
return Err(self.pre_append_conflict(
"a transaction carries an adoption pin but no install descriptor".into(),
));
};
let Some(resolution) = slot.resolution() else {
return Err(self.pre_append_conflict(
"a transaction's adoption pin was already settled before it was sequenced".into(),
));
};
let resolution = resolution?;
// The projection is bound to one repository at `begin`, and adopting it
// into another would install objects under a namespace that never
// agreed to them — the membership boundary §4 makes structural
// everywhere else.
let bound = NamespaceId::from(resolution.session.destination_repo);
if bound != transaction.namespace {
return Err(self.pre_append_conflict(format!(
"staged projection session {} is bound to repository {} and cannot be adopted \
into {}",
hex::encode(resolution.session.session_id),
bound.to_hex(),
transaction.namespace.to_hex()
)));
}
self.adoption_fits_run_ceilings(&resolution)?;
let current = crate::staging::install_from_resolution(&resolution)?;
if current != *descriptor {
return Err(self.pre_append_conflict(format!(
"the install descriptor for staged projection session {} no longer matches what \
its pin resolves to; the session changed between finalize and submit",
hex::encode(descriptor.session_id)
)));
}
Ok(())
}
/// A refusal that happens before anything is sequenced.
///
/// `Conflict` is the taxonomy's definitive-rejection variant, which is what
/// every pre-append refusal in this file returns: nothing was appended and
/// nothing can be, so the caller may retry.
fn pre_append_conflict(&self, message: String) -> StoreError {
StoreError::Conflict(message)
}
/// Scope 6.3 steps 1-3 with their own unwind boundary.
///
/// A panic in here is a pre-append failure by definition, so it must not
@ -2070,6 +2219,9 @@ impl ShardWriter {
repo_state: sequenced.repo_state,
objects: sequenced.objects,
receipt: sequenced.receipt,
// Attached by `accept`, which owns the pin until it knows which
// preparation is the one that lands.
adoption: None,
transaction: Arc::clone(transaction),
},
frame,
@ -2334,17 +2486,24 @@ impl ShardWriter {
receipt,
payload: TransactionFramePayloadV1 {
repository_create: repository_create(transaction)?,
objects: FrameObjectsV1::Inline(
transaction
.objects
.iter()
.map(|object| FrameObjectV1 {
object_type: object.object_type,
object_id: object.id,
raw: object.raw.clone(),
})
.collect(),
),
// One payload or the other, which the builder already refused
// to let a caller ask for both of. An adopting transaction's
// objects are in staging's artifacts; the frame carries only
// the descriptor that names them.
objects: match transaction.projection() {
Some(install) => FrameObjectsV1::StagedProjectionInstall(install.clone()),
None => FrameObjectsV1::Inline(
transaction
.objects
.iter()
.map(|object| FrameObjectV1 {
object_type: object.object_type,
object_id: object.id,
raw: object.raw.clone(),
})
.collect(),
),
},
ref_cas: transaction.refs.clone(),
expected_authority,
new_authority,
@ -2378,7 +2537,7 @@ impl ShardWriter {
// sequencing time. Between the two sit the signer handoff and the whole
// idle delay that closes an unfull group, either of which can be longer
// than what remains of a caller's signed deadline.
let (frames, pending) = self.close_group();
let (frames, mut pending) = self.close_group();
if frames.is_empty() {
debug_assert!(pending.is_empty());
return;
@ -2397,7 +2556,7 @@ impl ShardWriter {
}
let keys: Vec<OperationKey> = pending.iter().map(|prepared| prepared.key).collect();
let outcome = self.publish_window(&frames, &pending, &keys);
let outcome = self.publish_window(&frames, &mut pending, &keys);
match outcome {
Ok(()) => {
// Step 8 removed the status entries, so the reservations are no
@ -2410,6 +2569,19 @@ impl ShardWriter {
self.wake_committed_group(pending.len());
}
Err(error) => {
// Settle every surviving pin *before* any waiter is told.
//
// Dropping `pending` at the end of this function would settle
// them too — that is what the slot's `Drop` is for — but it
// would happen after the caller already had its answer, so a
// submit could return while staging still believed the pin was
// live. Taking them here makes the order a statement: the
// outcome is recorded, from the append phase each slot reached,
// and only then does anyone learn the transaction failed.
for prepared in pending.iter_mut() {
drop(prepared.adoption.take());
}
// Scope 3.7: steps 4 to 8 are the poison window. The shard is
// read-only from here, the group's status entries stay
// `Resolving` because step 8 never ran, and nothing is
@ -2514,15 +2686,23 @@ impl ShardWriter {
// stay aligned with `new_pending`.
let mut cursor = 0usize;
for prepared in pending {
for mut prepared in pending {
let key = prepared.key;
let transaction = Arc::clone(&prepared.transaction);
// Moved out before `prepared` is dropped by this iteration.
// `sequence_into_frame` builds a fresh `Prepared` from the retained
// `Arc`, which cannot carry the pin — an `Arc` has no move-out —
// so a survivor's pin would otherwise settle here as a definitive
// pre-append failure while its transaction went on to append. This
// is the re-sequencing route the slot exists for.
let adoption = prepared.adoption.take();
let next_sequence = base_sequence + new_pending.len() as u64;
let outcome = deadline_permits_append(now, prepared.retry_until_micros)
.map_err(|reason| reason.into_error(key.operation_id))
.and_then(|()| self.sequence_into_frame(&transaction, next_sequence, now));
match outcome {
Ok((reprepared, frame)) => {
Ok((mut reprepared, frame)) => {
reprepared.adoption = adoption;
new_frames.push(frame);
new_pending.push(reprepared);
cursor += 1;
@ -2549,7 +2729,7 @@ impl ShardWriter {
fn publish_window(
&mut self,
frames: &[Frame],
pending: &[Prepared],
pending: &mut [Prepared],
keys: &[OperationKey],
) -> Result<(), StoreError> {
// The journal's sequence counter advances here and nowhere else.
@ -2574,6 +2754,20 @@ impl ShardWriter {
fire_in_poison_window(self.shard_index, Failpoint::AfterMarkedResolving)?;
// --- step 5: exact reserved frames, one fence ----------------------
//
// The phase advances here, before the call and not after the fence
// inside it. From this line on, no failure may claim that nothing was
// written: `append_group_and_fence` can fail with a torn frame or with
// whole frames that were never fenced, and both are states only
// recovery can read the device to resolve. Advancing after the fence
// would leave exactly those two outcomes settling as
// `DefinitivePreAppendFailure` — staging releasing artifacts a durable
// frame may already name.
for prepared in pending.iter_mut() {
if let Some(slot) = prepared.adoption.as_mut() {
slot.entered_append();
}
}
let sequences = self.journal.append_group_and_fence(frames)?;
if sequences.len() != pending.len() {
return Err(self.poison_error(format!(
@ -2609,6 +2803,37 @@ impl ShardWriter {
// of the crate that implements them.
fence(Ordering::Release);
self.remove_status_entries(keys);
// --- step 9: settle every adoption pin ----------------------------
//
// Strictly after publication, because `Adopted` is a claim that a
// committed root references these artifacts and it carries the sequence
// that makes the claim checkable. Staging answers "does any root still
// point into this directory?" against a root a caller supplies, and a
// root captured before this publication references none of them for the
// trivial reason that it predates them — so recording the position is
// what stops absence-measured-too-early from reading as absence.
//
// A settlement that fails poisons: the frame is durable and the root is
// published, so staging believing otherwise is a disagreement about
// committed state, not a lost update. Anything left unsettled by an
// early return above is still correct — the slots are `Appended` by
// now, so dropping them transfers to recovery.
for prepared in pending.iter_mut() {
let Some(slot) = prepared.adoption.take() else {
continue;
};
slot.finish(ProjectionAdoptionOutcome::Adopted {
committed_shard_sequence: prepared.shard_sequence,
})
.map_err(|error| {
self.poison_error(format!(
"recording the adoption of the projection committed at shard sequence {}: \
{error}",
prepared.shard_sequence
))
})?;
}
Ok(())
}
@ -2651,6 +2876,64 @@ impl ShardWriter {
}
}
/// Bind a committed adoption descriptor to the artifacts on disk.
///
/// The post-fence counterpart of [`Self::revalidate_adoption`], and a
/// different question asked of a different authority. That one asked
/// staging's live session state whether the pin still matched, and could
/// refuse. This asks staging to resolve a descriptor whose frame is
/// **already durable**, which is what `resolve_committed` requires and why
/// it cannot be called before the fence. There is no refusing here: the
/// frame naming these artifacts is committed, so a failure is corruption
/// and poisons the shard — every caller of this is inside the poison
/// window.
///
/// The shard sequence is the adoption's identity and is passed rather than
/// looked up: it is what every artifact's logical generation derives from,
/// and staging cannot supply it because a finalizing session has no durable
/// position of its own. Same reason recovery passes it at `recovery.rs`.
fn resolve_adoption(
&self,
prepared: &Prepared,
install: &StagedProjectionInstallV1,
shard_sequence: u64,
) -> Result<RecoveredProjectionArtifacts, StoreError> {
// Materialize before resolving, because resolution only opens what this
// creates. Both take the same sequence, and this is the first moment it
// is real: the frame is fenced, so it can no longer be re-sequenced.
self.shared
.staging
.materialize_committed_index_runs(prepared.key.namespace, install, shard_sequence)
.map_err(|error| {
self.poison_error(format!(
"materializing index runs for committed staged projection session {}: {error}",
hex::encode(install.session_id)
))
})?;
let artifacts = self
.shared
.staging
.resolve_committed(prepared.key.namespace, install, shard_sequence)
.map_err(|error| {
self.poison_error(format!(
"resolving committed staged projection session {}: {error}",
hex::encode(install.session_id)
))
})?;
// The same recheck recovery makes for the same reason: resolution is
// the one step where a descriptor is handed to another package and a
// different one could come back, and every artifact below is trusted on
// the strength of this descriptor.
if artifacts.descriptor() != install {
return Err(self.poison_error(format!(
"staging resolved committed session {} to a different descriptor",
hex::encode(install.session_id)
)));
}
Ok(artifacts)
}
fn build_subtree(&self, pending: &[Prepared]) -> Result<ShardSubtree, StoreError> {
let mut delta = IndexDelta::from_options(&self.shared.options);
// The device offsets the fence just made durable, read back from the
@ -2658,6 +2941,10 @@ impl ShardWriter {
// name bytes that are actually there.
let index = self.journal.frame_index();
let appended = &index[index.len() - pending.len()..];
// `(adoption frame's shard sequence, what staging resolved it to)`.
// Empty for every group that adopts nothing, which is almost all of
// them.
let mut adopted: Vec<(u64, RecoveredProjectionArtifacts)> = Vec::new();
for (prepared, (sequence, offset, len)) in pending.iter().zip(appended) {
if *sequence != prepared.shard_sequence {
return Err(self.poison_error(format!(
@ -2682,6 +2969,18 @@ impl ShardWriter {
},
)?;
}
// An adopted projection's objects are **not** indexed at the loop
// above's frame location. That location names the bytes of the
// frame, and an adopting frame's bytes are a descriptor; the
// objects live in staging's artifacts, at one logical generation
// per chunk. `prepared.objects` is empty for these — the builder
// refuses to pair inline objects with an adoption — so nothing here
// overwrites anything the loop above wrote.
if let Some(install) = prepared.transaction.projection() {
let resolved = self.resolve_adoption(prepared, install, *sequence)?;
adopted.push((*sequence, resolved));
}
}
let mut repositories = RepoMap::new();
@ -2716,20 +3015,149 @@ impl ShardWriter {
.map(|prepared| prepared.shard_sequence)
.ok_or_else(|| self.poison_error("published an empty group".into()))?;
// A group that adopted nothing publishes no generation: the active
// journal these frames landed in is already pinned by the generation
// recovery installed.
let mut generations = GenerationMap::new();
let mut adopted_runs = Vector::new();
if !adopted.is_empty() {
// A resolved projection's membership arrives as a delta, as sealed
// index runs, or as both — B3 decides which by size, and the
// recovery-direction merge handles both for the same reason.
//
// The delta is merged into this group's own rather than layered
// beside it, because a `ShardSubtree` carries exactly one. The keys
// cannot collide: an adopting transaction has no inline objects, so
// the frame loop above wrote nothing for it, and two committed
// frames cannot name one session — staging refuses a second
// finalize.
for (_, artifacts) in &adopted {
for (key, location) in artifacts.index_delta().iter() {
delta.insert(key, location)?;
}
// Into the *root's* run stack, and not only into the generation
// below. Pinning a run keeps its file alive; it is this that
// makes its entries findable, and an adoption whose membership
// came back as a run resolves to nothing without it.
for run in artifacts.index_runs_newest_first() {
adopted_runs.push_back(Arc::clone(run));
}
}
let generation = self.generation_with_adoptions(&adopted)?;
generations.insert(generation.id, generation);
}
Ok(ShardSubtree::new(
self.shard_index,
last,
Arc::new(delta),
// `None`, though runs are installed: these cover the adopted
// projections' own objects, not this shard's delta backlog. Naming
// a sealed-through sequence would discard delta layers whose
// entries no adopted run contains.
None,
Vector::new(),
adopted_runs,
repositories,
terminal_statuses,
// No new generation: the active journal these frames landed in is
// already pinned by the generation recovery installed.
GenerationMap::new(),
generations,
))
}
/// Count an adoption's index runs against the ceilings recovery enforces,
/// **before** anything is appended.
///
/// An adoption installs staging-owned index runs, and a run count this
/// store can publish but not reopen is a root that opens once. `seal_index`
/// makes the same check against the same numbers for its own run.
///
/// Pre-append and not post-fence, though the runs only materialize after
/// the fence: the count is a function of the descriptor's chunk count and
/// the encoder's fixed partition rule, so it is knowable now, and knowing
/// it now is the difference between a clean refusal the caller can act on
/// and a poisoned shard holding a frame it already committed.
fn adoption_fits_run_ceilings(
&self,
resolution: &crate::staging::ProjectionAdoptionResolution,
) -> Result<(), StoreError> {
let planned = crate::staging::planned_index_run_count(resolution.session.chunk_count);
let root = self.shared.committed.load();
let owned: 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.saturating_add(planned);
for (limit, allowed) in [
("max_index_runs", self.shared.options.max_index_runs),
(
"max_open_index_runs",
self.shared.options.max_open_index_runs,
),
] {
if projected > u64::from(allowed) {
return Err(StoreError::LimitExceeded {
limit,
observed: projected,
allowed: u64::from(allowed),
});
}
}
Ok(())
}
/// The current generation, republished with this group's adopted artifacts
/// added to its pin set.
///
/// Under the **same** `GenerationId`, not a successor. A new manifest
/// generation is a durable artifact — `checkpoint` writes one — and this is
/// inside the poison window after the fence, where a write that failed
/// would poison a shard whose frames are already committed. It does not
/// need to be durable: `checkpoint.rs` has no projection-artifact section
/// at all, and an adoption's durability comes from the committed frame plus
/// staging's marker, with recovery rebuilding these exact pins by replaying
/// the frame through `resolve_committed`. What has to survive here is the
/// in-memory pin, so that compaction cannot delete an artifact a live root
/// resolves through.
///
/// Re-publishing the same id is what `CommittedRoot::merge` already does
/// for repositories and statuses, and `object_source`'s rule is satisfied
/// because the pins carried forward are the identical values — same paths,
/// same source kinds — plus new ones for generations no predecessor named.
fn generation_with_adoptions(
&self,
adopted: &[(u64, RecoveredProjectionArtifacts)],
) -> Result<Arc<RetainedGeneration>, StoreError> {
let root = self.shared.committed.load();
// The *manifest* generation, which is not `tail_generation`: that one
// numbers journal segments and is what an index location names, while
// retention is keyed by the manifest the checkpointer publishes. The
// two are independent counters and only coincide by accident.
let id = GenerationId::new(self.shard_index, self.current_manifest_generation(&root)?);
let previous = root.retained_generations().get(&id).ok_or_else(|| {
self.poison_error(format!(
"manifest generation {} vanished from the root while adopting",
id.manifest_generation
))
})?;
let mut index_runs = previous.index_runs.to_vec();
let mut artifacts = previous.projection_artifacts.to_vec();
for (_, resolved) in adopted {
index_runs.extend_from_slice(resolved.retained_index_runs());
artifacts.extend_from_slice(resolved.retained_artifacts());
}
Ok(Arc::new(RetainedGeneration::new(
id,
Arc::clone(&previous.segments),
index_runs.into(),
Arc::clone(&previous.checkpoints),
Arc::clone(&previous.active_tails),
artifacts.into(),
)))
}
/// This shard's unsealed index backlog, as one captured root reports it.
fn unsealed_backlog(&self, root: &CommittedRoot) -> UnsealedBacklog {
let mut backlog = UnsealedBacklog::default();

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

@ -2320,6 +2320,17 @@ fn recover_shard_under_lock(
// The frame's own sequence is the adoption's identity, and the only
// authority that creates one: staging has no durable position for a pin
// that transferred across a process boundary.
// The same materialization the live adopter performs, replayed. A
// process that stopped between the fence and materialization left some
// prefix of this frame's runs on disk, and possibly none; recomputing
// them from the same committed inputs either finds identical bytes or
// writes what is missing. That is what makes that window recoverable
// rather than a state recovery can only report.
resolver.materialize_committed_index_runs(
frame.facts.namespace,
descriptor,
frame.facts.shard_sequence,
)?;
let artifacts = resolver.resolve_committed(
frame.facts.namespace,
descriptor,
@ -3204,6 +3215,18 @@ mod production_session_tests {
Ok(Arc::from([self.descriptor.session_id, self.absent_session]))
}
/// The double's runs are whatever its fixture already holds, so
/// materialization has nothing to create. Recording that it *ran* is
/// what the callers under test care about; producing bytes is not.
fn materialize_committed_index_runs(
&self,
_namespace: NamespaceId,
_descriptor: &StagedProjectionInstallV1,
_adoption_shard_sequence: u64,
) -> Result<(), StoreError> {
Ok(())
}
fn resolve_committed(
&self,
namespace: NamespaceId,

65
crates/levcs-store/src/snapshot.rs
View File

@ -33,7 +33,7 @@ use std::sync::Arc;
use levcs_core::{ObjectId, ObjectType};
use crate::index::{IndexKey, NamespaceLifecycle, NamespaceStorageMode};
use crate::roots::{CommittedRoot, RepoState};
use crate::roots::{CommittedRoot, RepoState, RetainedObjectSource};
use crate::types::{NamespaceId, StoreError};
/// One repository's committed logical state at one generation.
@ -43,6 +43,11 @@ pub struct RepoSnapshot {
/// traversal.
root: Arc<CommittedRoot>,
namespace: NamespaceId,
/// The shard this namespace routes to, which is the space its logical
/// generation numbers are unique within. Carried so
/// [`RepoSnapshot::object_source`] can resolve a location without the
/// caller re-deriving the routing.
shard_index: u16,
/// Resolved once, at capture. `RepoMap` holds each `RepoState` behind its
/// own `Arc`, so this shares that allocation rather than copying the refs
/// map, and it takes the hash lookup off every accessor below.
@ -94,9 +99,16 @@ impl RepoSnapshot {
/// here would make a deleted repository indistinguishable from one that
/// never existed, and the two have different answers to every question
/// below.
///
/// `shard_index` is supplied rather than derived because deriving it needs
/// the shard count, which lives in [`crate::StoreOptions`] and not in the
/// root. It is well defined for a snapshot: a namespace routes to exactly
/// one shard, so every frame and every index entry it owns lives in that
/// shard's generations.
pub(crate) fn capture(
root: Arc<CommittedRoot>,
namespace: NamespaceId,
shard_index: u16,
) -> Result<Self, StoreError> {
let Some(state) = root.repo(&namespace) else {
return Err(StoreError::NoSuchRepository { namespace });
@ -105,6 +117,7 @@ impl RepoSnapshot {
Ok(Self {
root,
namespace,
shard_index,
state,
})
}
@ -187,6 +200,31 @@ impl RepoSnapshot {
shard_sequence: location.shard_sequence,
}))
}
/// Resolve a location's decoder and live pin.
///
/// [`Self::locate`] answers *which generation*, which is not enough to read
/// anything: a logical generation is a segment, an active journal tail, or
/// an adopted projection artifact, and the three are different files in
/// different formats. Before adoption existed every location a reader could
/// hold was a frame in this shard's journal, so the distinction was
/// invisible and the generation number alone was serviceable. An adopted
/// object is indexed at its artifact's generation and its bytes are not in
/// the frame that installed it, so answering "where" without answering
/// "what kind" is now an answer a reader cannot act on.
///
/// `Ok(None)` means no retained generation in this shard claims that
/// number. For a location this snapshot just produced that is corruption in
/// waiting, not an ordinary absence — but it is reported the same way the
/// root reports it, because the root is what decides retention and this is
/// a view onto the root, not a second opinion about it.
pub fn object_source(
&self,
location: &ObjectLocation,
) -> Result<Option<RetainedObjectSource<'_>>, StoreError> {
self.root
.object_source(self.shard_index, location.segment_generation)
}
}
#[cfg(test)]
@ -204,6 +242,11 @@ mod tests {
TerminalStatusMap, TypedRefMap,
};
/// The shard every root below is built in. Nothing here asserts about
/// generation resolution, so the value only has to be the one constant
/// these hand-built roots agree on.
const SHARD: u16 = 0;
// -----------------------------------------------------------------------
// A thread-local allocation meter
// -----------------------------------------------------------------------
@ -331,7 +374,7 @@ mod tests {
let ns = namespace(1);
let root = root_with(&[(ns, repo_state(42, NamespaceLifecycle::Active))], &[]);
let snapshot = RepoSnapshot::capture(root, ns).expect("the namespace is bound");
let snapshot = RepoSnapshot::capture(root, ns, SHARD).expect("the namespace is bound");
assert_eq!(snapshot.namespace(), ns);
assert_eq!(snapshot.repo_sequence(), 42);
@ -347,7 +390,8 @@ mod tests {
let unbound = namespace(2);
let root = root_with(&[(bound, repo_state(1, NamespaceLifecycle::Active))], &[]);
let error = RepoSnapshot::capture(root, unbound).expect_err("nothing is bound for it");
let error =
RepoSnapshot::capture(root, unbound, SHARD).expect_err("nothing is bound for it");
// Asserted by name and by payload: contract review 2026-08-07-A makes
// the identity part of the error, so a caller can recover what it
@ -366,7 +410,8 @@ mod tests {
let ns = namespace(1);
let root = root_with(&[(ns, repo_state(9, NamespaceLifecycle::Deleted))], &[]);
let snapshot = RepoSnapshot::capture(root, ns).expect("a retired repository is bound");
let snapshot =
RepoSnapshot::capture(root, ns, SHARD).expect("a retired repository is bound");
assert_eq!(snapshot.lifecycle(), NamespaceLifecycle::Deleted);
assert_eq!(snapshot.repo_sequence(), 9);
@ -385,7 +430,7 @@ mod tests {
&[(ns, id, location(ObjectType::Commit as u8))],
);
let found = RepoSnapshot::capture(root, ns)
let found = RepoSnapshot::capture(root, ns, SHARD)
.expect("bound")
.locate(id)
.expect("a defined type code decodes")
@ -427,7 +472,7 @@ mod tests {
],
);
let through_a = RepoSnapshot::capture(Arc::clone(&root), a).expect("bound");
let through_a = RepoSnapshot::capture(Arc::clone(&root), a, SHARD).expect("bound");
// The object present in both resolves to A's row, not B's.
let found = through_a
@ -454,7 +499,7 @@ mod tests {
let root = root_with(&[(ns, repo_state(1, NamespaceLifecycle::Active))], &[]);
assert_eq!(
RepoSnapshot::capture(root, ns)
RepoSnapshot::capture(root, ns, SHARD)
.expect("bound")
.locate(object(0xff))
.expect("an absent object is not a failure"),
@ -473,7 +518,7 @@ mod tests {
&[(ns, id, location(6))],
);
let error = RepoSnapshot::capture(root, ns)
let error = RepoSnapshot::capture(root, ns, SHARD)
.expect("bound")
.locate(id)
.expect_err("an undefined type code must not be reported as absence");
@ -555,7 +600,7 @@ mod tests {
drop(copied);
let (snapshot, capture_bytes) =
allocated_bytes(|| RepoSnapshot::capture(Arc::clone(&root), ns));
allocated_bytes(|| RepoSnapshot::capture(Arc::clone(&root), ns, SHARD));
let snapshot = snapshot.expect("bound");
assert_eq!(
@ -590,7 +635,7 @@ mod tests {
let ns = namespace(1);
let added_later = object(0xbb);
let before = root_with(&[(ns, repo_state(1, NamespaceLifecycle::Active))], &[]);
let snapshot = RepoSnapshot::capture(before, ns).expect("bound");
let snapshot = RepoSnapshot::capture(before, ns, SHARD).expect("bound");
// A newer root binds the same namespace and indexes a new object.
let _after = root_with(

802
crates/levcs-store/src/staging.rs
View File

File diff suppressed because it is too large Load Diff

403
crates/levcs-store/src/transaction.rs
View File

@ -19,11 +19,14 @@
//! a builder supplied.
use std::collections::BTreeSet;
use std::sync::Arc;
use levcs_core::{ObjectId, ObjectType};
use levcs_protocol::v2::{RefTarget, TransactionEvidenceV1, TypedRefCas};
use levcs_protocol::v2::{
RefTarget, StagedProjectionInstallV1, TransactionEvidenceV1, TypedRefCas,
};
use crate::staging::{ProjectionAdoptionOutcome, StagedProjectionAdoption};
use crate::staging::{ProjectionAdoption, ProjectionAdoptionOutcome, StagedProjectionAdoption};
use crate::types::{NamespaceId, OperationId, PrivilegedConstruction, StoreError};
/// The immutable output of the instance pipeline's stages 4-8 (plan §7).
@ -49,6 +52,166 @@ pub struct ValidatedTransaction {
pub(crate) expected_authority: Option<ObjectId>,
pub(crate) new_authority: Option<ObjectId>,
pub(crate) evidence: TransactionEvidenceV1,
/// The canonical descriptor of an adopted projection, kept here rather than
/// beside the pin because the frame is rebuilt from this value on **every**
/// sequencing attempt. A pre-mark deadline recheck that drops a group member
/// re-sequences, re-chains, and re-signs everything after it, and a
/// descriptor that had left with the pin would leave the retry with no
/// payload to encode.
pub(crate) projection: Option<StagedProjectionInstallV1>,
/// The live pin, and the one field of this struct that is not reproducible.
///
/// `Option` because the writer **takes** it, exactly once, before wrapping
/// this transaction in the `Arc` that survives re-sequencing — an `Arc` has
/// no move-out, so a pin left in here would be unreachable at the moment it
/// has to be settled. From that point the pin lives in an `AdoptionSlot`
/// beside the `Arc`, and `projection` above is what the retries read.
pub(crate) adoption_pin: Option<AdoptionSlot>,
}
impl ValidatedTransaction {
/// Take the slot out on its way into the writer.
///
/// Called once, before `Arc::new`. Calling it twice yields `None` rather
/// than a second capability: the pin is linear and there is only ever one.
pub(crate) fn take_adoption(&mut self) -> Option<AdoptionSlot> {
self.adoption_pin.take()
}
/// Read by the writer when it encodes the frame payload. See the note on
/// [`AppendPhase`] for why it is unreached today.
#[allow(dead_code)]
pub(crate) fn projection(&self) -> Option<&StagedProjectionInstallV1> {
self.projection.as_ref()
}
}
/// How far this transaction's frame has gone toward the device.
///
/// Only two values, because only two answers matter to a pin that has to be
/// settled without knowing why: either nothing of this frame reached the
/// journal, or something may have.
// `Appended` is constructed only by the writer's append path, and the parts of
// `AdoptionSlot` below that the writer alone calls are likewise unreached until
// `StoreEngine::submit` stops refusing an adoption. Both are exercised by this
// module's tests; the allow covers the non-test build only.
#[allow(dead_code)]
#[derive(Copy, Clone, Debug, PartialEq, Eq)]
pub(crate) enum AppendPhase {
/// No byte of this transaction's frame has been written. An adoption
/// abandoned here is definitively not installed.
BeforeAppend,
/// A write has been attempted. **Not** "a fence succeeded" — the
/// transition is at the first possible frame write, so torn bytes and whole
/// unfenced frames are both on this side of it. Recovery, which can read
/// the device, is the only thing that can say which happened.
Appended,
}
/// The one-shot owner of an admitted adoption pin.
///
/// # Why this exists rather than a field on the transaction
///
/// `Prepared` retains an `Arc<ValidatedTransaction>` so a group can be
/// re-sequenced, re-chained, and re-signed when the pre-mark deadline recheck
/// drops a member. An `Arc` has no move-out, so a linear `ProjectionAdoption`
/// stored inside one could never be taken out to be settled. The pin therefore
/// travels *beside* the `Arc`, in this slot, and the descriptor the retries
/// need stays in the transaction where it can be read any number of times.
///
/// # Why the phase, rather than settling at each site
///
/// Every route out of a submit has to settle the pin, and the dangerous ones
/// are the routes nobody enumerated: a full or disconnected submission queue,
/// a pre-append error, a panic inside the writer, a deadline recheck that
/// reforms the group, an unwind. Naming them one at a time means the next
/// route added is a leak. Instead the slot settles itself on drop and reads the
/// outcome off the phase, so a route that was never considered is still
/// correct — and correct in the safe direction, because the phase advances
/// before the write rather than after the fence.
pub(crate) struct AdoptionSlot {
/// `None` only after an explicit [`Self::finish`]. A slot that still holds
/// its pin at drop time settles it from the phase.
handle: Option<ProjectionAdoption>,
phase: AppendPhase,
}
#[allow(dead_code)] // See the note on `AppendPhase`.
impl AdoptionSlot {
pub(crate) fn new(handle: ProjectionAdoption) -> Self {
Self {
handle: Some(handle),
phase: AppendPhase::BeforeAppend,
}
}
/// Advance to [`AppendPhase::Appended`], at the first possible frame write.
///
/// Deliberately idempotent and one-way: a group that re-forms and appends
/// again must not walk the phase back to `BeforeAppend`, because the
/// earlier attempt's bytes may already be on the device.
pub(crate) fn entered_append(&mut self) {
self.phase = AppendPhase::Appended;
}
pub(crate) fn phase(&self) -> AppendPhase {
self.phase
}
/// Read the pin without consuming it, for the pre-append revalidation.
///
/// `None` cannot happen before [`Self::finish`] and is not an error worth a
/// second failure mode: a caller that gets it has already settled the pin
/// and has nothing left to validate.
pub(crate) fn resolution(
&self,
) -> Option<Result<Arc<crate::staging::ProjectionAdoptionResolution>, StoreError>> {
self.handle.as_ref().map(|handle| handle.resolution())
}
/// Settle with an explicit outcome, and hand back staging's answer.
///
/// The only path that reports a settlement failure. [`Drop`] cannot, which
/// is why the writer calls this at the two points where it knows the
/// outcome — the publication that adopted, and the pre-append refusal that
/// did not.
pub(crate) fn finish(mut self, outcome: ProjectionAdoptionOutcome) -> Result<(), StoreError> {
match self.handle.take() {
Some(handle) => handle.finish(outcome),
None => Ok(()),
}
}
}
impl Drop for AdoptionSlot {
fn drop(&mut self) {
let Some(handle) = self.handle.take() else {
return;
};
let outcome = match self.phase {
AppendPhase::BeforeAppend => ProjectionAdoptionOutcome::DefinitivePreAppendFailure,
AppendPhase::Appended => ProjectionAdoptionOutcome::TransferredToRecovery,
};
// Swallowed because a `Drop` has nowhere to put it, and accounted for
// regardless: a `finish` that fails leaves `ProjectionAdoption` unfinished,
// so its own `Drop` records the pin as dropped without an outcome. The
// failure is visible in staging's counters rather than lost — it is only
// this frame's error text that cannot be carried out of here.
let _ = handle.finish(outcome);
}
}
/// Deliberately no `Debug` derive on the slot: `#[derive]` on a struct holding
/// a capability invites printing it, and what a pin identifies is a session
/// whose id belongs in staging's diagnostics rather than in a transaction dump.
impl std::fmt::Debug for AdoptionSlot {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter
.debug_struct("AdoptionSlot")
.field("settled", &self.handle.is_none())
.field("phase", &self.phase)
.finish()
}
}
/// Deliberately does not print object or evidence bytes. A transaction's
@ -105,7 +268,10 @@ pub struct ValidatedTransactionBuilder {
#[allow(clippy::type_complexity)]
authority: Option<(Option<ObjectId>, Option<ObjectId>)>,
evidence: Option<TransactionEvidenceV1>,
adoption: Option<StagedProjectionAdoption>,
/// The descriptor and its pin, the pin already inside the slot that owns
/// it. Held as a pair because `build` needs the descriptor and the slot to
/// travel to different places, and neither may arrive without the other.
adoption: Option<(StagedProjectionInstallV1, AdoptionSlot)>,
}
fn incomplete(field: &str) -> StoreError {
@ -157,49 +323,53 @@ impl ValidatedTransactionBuilder {
self
}
/// Adopt one sealed staged projection. The opaque handle is the adoption
/// pin and is consumed with the canonical wire descriptor so neither half
/// can be forgotten independently.
/// Adopt one sealed staged projection.
///
/// Takes the whole [`StagedProjectionAdoption`] rather than a descriptor
/// and a pin separately. The frozen signature took two arguments so that
/// neither half could be *forgotten* independently, which this keeps — and
/// it additionally makes them impossible to *mismatch*, which two
/// arguments could not. See the type's own note. Contract review
/// 2026-08-09-A.
pub fn adopt_projection(
mut self,
descriptor: levcs_protocol::v2::StagedProjectionInstallV1,
handle: crate::staging::ProjectionAdoption,
adoption: StagedProjectionAdoption,
) -> Result<Self, StoreError> {
if let Some(previous) = self.adoption.take() {
let StagedProjectionAdoption { descriptor, handle } = adoption;
// Wrapped before anything below can fail. From here the pin is owned by
// a slot, so a builder abandoned mid-chain — `adopt_projection` called
// and `build` never reached — releases it with a pre-append outcome
// instead of dropping a live capability.
let slot = AdoptionSlot::new(handle);
if let Some((_, previous)) = self.adoption.take() {
// Both pins were admitted, so both receive a terminal outcome
// even though the builder rejects the duplicate. Returning early
// after finishing only one would turn the other drop into the
// lifecycle bug this handle exists to expose.
let previous_result = previous
.handle
.finish(ProjectionAdoptionOutcome::DefinitivePreAppendFailure);
// lifecycle bug this handle exists to expose. Settled explicitly
// rather than by drop because only this path can report a
// settlement that itself failed.
let previous_result =
previous.finish(ProjectionAdoptionOutcome::DefinitivePreAppendFailure);
let submitted_result =
handle.finish(ProjectionAdoptionOutcome::DefinitivePreAppendFailure);
slot.finish(ProjectionAdoptionOutcome::DefinitivePreAppendFailure);
previous_result?;
submitted_result?;
return Err(StoreError::Conflict(
"a transaction may adopt exactly one staged projection".into(),
));
}
self.adoption = Some(StagedProjectionAdoption { descriptor, handle });
self.adoption = Some((descriptor, slot));
Ok(self)
}
pub fn build(mut self) -> Result<ValidatedTransaction, StoreError> {
if let Some(adoption) = self.adoption.take() {
// Adoption is the B1/B3 seam of scope 6.2 item 9 and is not part
// of this slice. The pin is released with a definitive pre-append
// outcome before the refusal, because a dropped pin with no
// outcome is the leak §6.5 declares a bug — the refusal must not
// create one.
adoption
.handle
.finish(ProjectionAdoptionOutcome::DefinitivePreAppendFailure)?;
return Err(StoreError::NotImplemented(
"ValidatedTransactionBuilder::adopt_projection — B1 NamespaceTxn, \
scope 6-B1 deliverable 3 and scope 6.2 item 9",
));
}
// Held across every fallible step below rather than unwrapped at the
// end. `AdoptionSlot` settles itself when dropped, so each `?` in this
// function releases the pin with a definitive pre-append outcome
// without naming it — and a refusal that forgot to would create exactly
// the unsettled-drop §6.5 declares a bug. The slot reaches the returned
// transaction only on the success path.
let adoption = self.adoption.take();
let namespace = self.namespace.ok_or_else(|| incomplete("namespace"))?;
let (operation_id, operation_digest, retry_until_micros) =
@ -210,6 +380,23 @@ impl ValidatedTransactionBuilder {
let mut objects = self.objects.ok_or_else(|| incomplete("objects"))?;
let refs = self.refs.ok_or_else(|| incomplete("refs"))?;
// A frame carries inline objects or an install descriptor and never
// both: `FrameObjectsV1` is an enum, so there is no encoding for the
// pair. This is a refusal rather than a silent choice because both
// silent choices are wrong — dropping the inline objects loses bytes
// the caller asked to commit, and dropping the projection commits a
// frame that references artifacts nothing installs.
//
// An adopting transaction is therefore built with an *empty* object
// list, not with the list omitted. The projection is the payload.
if adoption.is_some() && !objects.is_empty() {
return Err(StoreError::Conflict(format!(
"a transaction adopting a staged projection may not also introduce {} inline \
object(s); a frame carries one payload or the other",
objects.len()
)));
}
// Canonical frame order, established once. A duplicate `ObjectId` is
// refused rather than deduplicated: two records for one object make
// `objects_new` and the index disagree about what the transaction
@ -248,6 +435,11 @@ impl ValidatedTransactionBuilder {
}
}
let (projection, adoption_pin) = match adoption {
Some((descriptor, slot)) => (Some(descriptor), Some(slot)),
None => (None, None),
};
Ok(ValidatedTransaction {
namespace,
operation_id,
@ -259,6 +451,8 @@ impl ValidatedTransactionBuilder {
expected_authority,
new_authority,
evidence,
projection,
adoption_pin,
})
}
}
@ -309,6 +503,19 @@ mod tests {
}
}
/// The pairing a caller outside this crate cannot assemble by hand, which
/// is the point of the type. These tests are inside the crate and can, so
/// they build it through one helper rather than at each call site — a
/// second construction site is a second chance to pair a descriptor with a
/// pin from somewhere else, which is exactly what production no longer
/// permits.
fn adoption(session_id: [u8; 16], lifecycle: Arc<Lifecycle>) -> StagedProjectionAdoption {
StagedProjectionAdoption {
descriptor: descriptor(session_id),
handle: ProjectionAdoption::new(lifecycle),
}
}
pub(crate) fn administrative_evidence() -> TransactionEvidenceV1 {
TransactionEvidenceV1::AdministrativeV1 {
actor: [9; 32],
@ -455,17 +662,46 @@ mod tests {
}
#[test]
fn a_projection_adoption_is_refused_and_releases_its_pin() {
fn an_adoption_reaches_the_built_transaction_still_unsettled() {
let lifecycle = Arc::new(Lifecycle::default());
let transaction = complete()
.adopt_projection(adoption([3; 16], lifecycle.clone()))
.expect("one adoption is admitted")
.build()
.expect("an adopting transaction builds");
assert_eq!(
transaction.projection().map(|install| install.session_id),
Some([3; 16]),
"the descriptor the frame is encoded from must survive `build`"
);
// Unsettled on purpose: `build` is not a decision about whether the
// projection installs. Recording an outcome here would either claim an
// append that has not happened or release artifacts a submit is about
// to reference.
assert_eq!(&*lifecycle.outcomes.lock().unwrap(), &[]);
assert_eq!(*lifecycle.dropped.lock().unwrap(), 0);
}
/// The enum has no encoding for both payloads, so the builder refuses
/// rather than choosing which one to lose.
#[test]
fn an_adoption_alongside_inline_objects_is_refused_and_releases_its_pin() {
let lifecycle = Arc::new(Lifecycle::default());
let result = complete()
.adopt_projection(
descriptor([3; 16]),
ProjectionAdoption::new(lifecycle.clone()),
)
.objects(vec![StagedObject {
id: ObjectId([9; 32]),
object_type: ObjectType::Blob,
raw: vec![9; 8],
}])
.adopt_projection(adoption([3; 16], lifecycle.clone()))
.expect("one adoption is admitted")
.build();
assert!(matches!(result, Err(StoreError::NotImplemented(_))));
let Err(StoreError::Conflict(message)) = result else {
panic!("expected a conflict, got {result:?}");
};
assert!(message.contains("one payload or the other"), "{message}");
assert_eq!(
&*lifecycle.outcomes.lock().unwrap(),
&[ProjectionAdoptionOutcome::DefinitivePreAppendFailure]
@ -473,16 +709,107 @@ mod tests {
assert_eq!(*lifecycle.dropped.lock().unwrap(), 0);
}
/// The two abandonment routes that no call site names.
///
/// Neither is reachable by an enumerated release: one drops a builder that
/// was never finished, the other drops a transaction that was never
/// submitted. Both are ordinary things for a caller to do, and both would
/// leak a live pin if settlement lived at the call sites rather than in the
/// slot's `Drop`.
#[test]
fn an_abandoned_builder_or_transaction_releases_its_pin() {
for abandon_before_build in [true, false] {
let lifecycle = Arc::new(Lifecycle::default());
let builder = complete()
.adopt_projection(adoption([4; 16], lifecycle.clone()))
.expect("one adoption is admitted");
if abandon_before_build {
drop(builder);
} else {
drop(builder.build().expect("an adopting transaction builds"));
}
assert_eq!(
&*lifecycle.outcomes.lock().unwrap(),
&[ProjectionAdoptionOutcome::DefinitivePreAppendFailure],
"abandon_before_build = {abandon_before_build}"
);
assert_eq!(
*lifecycle.dropped.lock().unwrap(),
0,
"abandon_before_build = {abandon_before_build}"
);
}
}
/// The phase, and only the phase, decides an unattended settlement.
///
/// Asserted directly on the slot because the writer's routes into
/// `Appended` are failures and panics, and a test that could only reach
/// this rule through one of them would be asserting about that failure
/// rather than about the rule.
#[test]
fn a_dropped_slot_settles_from_its_append_phase() {
let cases = [
(
false,
ProjectionAdoptionOutcome::DefinitivePreAppendFailure,
"no byte was written",
),
(
true,
ProjectionAdoptionOutcome::TransferredToRecovery,
"a write was attempted",
),
];
for (entered_append, expected, why) in cases {
let lifecycle = Arc::new(Lifecycle::default());
let mut slot = AdoptionSlot::new(ProjectionAdoption::new(lifecycle.clone()));
assert_eq!(slot.phase(), AppendPhase::BeforeAppend);
if entered_append {
slot.entered_append();
// One-way: a group that re-forms must not walk the phase back.
slot.entered_append();
assert_eq!(slot.phase(), AppendPhase::Appended);
}
drop(slot);
assert_eq!(&*lifecycle.outcomes.lock().unwrap(), &[expected], "{why}");
assert_eq!(*lifecycle.dropped.lock().unwrap(), 0, "{why}");
}
}
/// An explicit settlement wins over the phase, and happens once.
#[test]
fn an_explicitly_finished_slot_does_not_settle_again_on_drop() {
let lifecycle = Arc::new(Lifecycle::default());
let mut slot = AdoptionSlot::new(ProjectionAdoption::new(lifecycle.clone()));
slot.entered_append();
slot.finish(ProjectionAdoptionOutcome::Adopted {
committed_shard_sequence: 7,
})
.expect("staging accepts the outcome");
assert_eq!(
&*lifecycle.outcomes.lock().unwrap(),
&[ProjectionAdoptionOutcome::Adopted {
committed_shard_sequence: 7
}],
"the drop that follows `finish` must not add a second outcome"
);
assert_eq!(*lifecycle.dropped.lock().unwrap(), 0);
}
#[test]
fn duplicate_projection_adoption_releases_both_pins() {
let first = Arc::new(Lifecycle::default());
let second = Arc::new(Lifecycle::default());
let builder = ValidatedTransaction::builder(PrivilegedConstruction::internal())
.adopt_projection(descriptor([1; 16]), ProjectionAdoption::new(first.clone()))
.adopt_projection(adoption([1; 16], first.clone()))
.expect("first adoption");
let result =
builder.adopt_projection(descriptor([2; 16]), ProjectionAdoption::new(second.clone()));
let result = builder.adopt_projection(adoption([2; 16], second.clone()));
assert!(matches!(result, Err(StoreError::Conflict(_))));
for lifecycle in [first, second] {
assert_eq!(

68
crates/levcs-store/tests/crash_matrix.rs
View File

@ -1550,11 +1550,37 @@ fn wave_b_rows_drive_through_submit_to_their_full_failpoint_expectation() {
continue;
}
let expected_adoption = AdoptionOutcomeExpectation::from_name(&plan.adoption_outcome)
.unwrap_or_else(|| {
panic!(
"row {}: submit.adoption_outcome {:?} names no AdoptionOutcomeExpectation",
row.failpoint, plan.adoption_outcome
)
});
for action in &plan.actions {
let action = engine_matrix::action_from_name(action)
.unwrap_or_else(|| panic!("row {}: unknown action {action:?}", row.failpoint));
let observation = engine_matrix::drive_submit_row(&serial, resolved.point, action);
assert_full_expectation(resolved.point, resolved.class, &observation);
// The payload is a matrix axis, so every row runs once per kind it
// declares. The staged-projection pass is what makes the adoption
// outcome an observation rather than a fixture field: without it the
// column would agree with the class table and describe nothing that
// ran.
for kind in &plan.payload_kinds {
let kind = PayloadKind::from_name(kind).unwrap_or_else(|| {
panic!("row {}: unknown payload kind {kind:?}", row.failpoint)
});
let observation = engine_matrix::drive_submit_row(
&serial,
resolved.point,
action,
kind.carries_adoption(),
);
assert_full_expectation(resolved.point, resolved.class, &observation);
if kind.carries_adoption() {
assert_adoption_outcome(&observation, expected_adoption);
}
}
}
driven.push(row.failpoint.clone());
}
@ -1570,6 +1596,44 @@ fn wave_b_rows_drive_through_submit_to_their_full_failpoint_expectation() {
);
}
/// The pin's terminal outcome, observed rather than derived.
///
/// The expectation comes from the fixture, which `the_adoption_expectation_
/// agrees_with_the_physical_state_class` independently checks against the class
/// table. This compares that expectation with what staging actually counted, so
/// the two derivations the charter requires stay two: a rule about physical
/// state, and a number a run produced.
///
/// Every driven row must settle exactly one pin. Settling none is the leak §6.5
/// declares a bug, and settling two means one adoption reached two terminal
/// states — both are failures here rather than shrugs.
fn assert_adoption_outcome(
observation: &engine_matrix::RowObservation,
expected: AdoptionOutcomeExpectation,
) {
let counters = observation
.adoption
.expect("a staged-projection row records staging's pin counters");
let observed = counters
.sole_outcome()
.unwrap_or_else(|why| panic!("row {}: {why}", observation.row));
let observed = observed.unwrap_or_else(|| {
panic!(
"row {}: the adoption pin was never settled; a pin dropped without an outcome is \
the lifecycle bug the handle exists to expose ({counters:?})",
observation.row
)
});
assert_eq!(
observed,
expected.name(),
"row {}: the adoption pin settled {observed}, but this row's physical state class \
requires {} ({counters:?})",
observation.row,
expected.name()
);
}
/// The row B1 disclosed was never armed.
///
/// `DuringRootCasRetry` sits inside `publish_subtree`'s retry loop, past a

24
crates/levcs-store/tests/fixtures/phase1-failpoints.json vendored
View File

@ -52,7 +52,8 @@
],
"requires_root_cas_contention": false,
"payload_kinds": [
"inline"
"inline",
"staged_projection"
],
"adoption_outcome": "DefinitivePreAppendFailure"
},
@ -95,7 +96,8 @@
],
"requires_root_cas_contention": false,
"payload_kinds": [
"inline"
"inline",
"staged_projection"
],
"adoption_outcome": "DefinitivePreAppendFailure"
},
@ -163,7 +165,8 @@
],
"requires_root_cas_contention": false,
"payload_kinds": [
"inline"
"inline",
"staged_projection"
],
"adoption_outcome": "TransferredToRecovery"
},
@ -182,7 +185,8 @@
],
"requires_root_cas_contention": false,
"payload_kinds": [
"inline"
"inline",
"staged_projection"
],
"adoption_outcome": "TransferredToRecovery"
},
@ -202,7 +206,8 @@
],
"requires_root_cas_contention": false,
"payload_kinds": [
"inline"
"inline",
"staged_projection"
],
"adoption_outcome": "TransferredToRecovery"
},
@ -222,7 +227,8 @@
],
"requires_root_cas_contention": true,
"payload_kinds": [
"inline"
"inline",
"staged_projection"
],
"adoption_outcome": "TransferredToRecovery"
},
@ -265,7 +271,8 @@
],
"requires_root_cas_contention": false,
"payload_kinds": [
"inline"
"inline",
"staged_projection"
],
"adoption_outcome": "Adopted"
},
@ -284,7 +291,8 @@
],
"requires_root_cas_contention": false,
"payload_kinds": [
"inline"
"inline",
"staged_projection"
],
"adoption_outcome": "Adopted"
},

364
crates/levcs-store/tests/projection_adoption.rs Normal file
View File

@ -0,0 +1,364 @@
//! Scope 6-B1 deliverable 3 and scope 6.2 item 9 — `adopt_projection` through
//! `StoreEngine::submit`, asserted through the public surface.
//!
//! The property under test is **ownership**, not visibility. An adopted
//! projection's objects are not in the frame that installed them: the frame
//! carries a descriptor, and the bytes stay in the artifacts staging sealed. So
//! two things have to be true at once, and each has a distinct way of being
//! wrong:
//!
//! 1. Each object is indexed at its **artifact's** logical generation, not at
//! the frame's. Indexing at the frame location would still make `locate`
//! return `Some`, and a reader following it would land at an offset inside a
//! journal frame that contains a descriptor rather than the object — a
//! plausible-looking answer that decodes into nothing.
//! 2. Those artifact generations survive the next retained generation. A
//! checkpoint publishes a successor and releases the predecessor's pins; an
//! adoption whose artifacts are not carried forward reads correctly exactly
//! once and then resolves to nothing, with no error at the moment the
//! mistake is made.
//!
//! One test catches both, which is why it is the first one written: locate and
//! resolve every object, checkpoint into a successor generation, then locate
//! and resolve all of them again. Failure 1 shows up in the first resolve,
//! failure 2 only in the second.
//!
//! **At least two chunks, always.** A single-chunk projection has exactly one
//! artifact, so "the index resolved every object to the right artifact" and
//! "the index resolved every object to the only artifact there is" are the same
//! observation, and B3 defines and resolves one `IndexLocation` per chunk. Two
//! chunks is the smallest arrangement where per-chunk ownership is load-bearing.
// The same three-feature gate as `namespace_snapshot.rs`, for the same reason:
// `engine_matrix` arms failpoints and reaches `drive.rs`, so it compiles only
// under the full set, and the Phase 1 gate runs exactly this combination.
#![cfg(all(
feature = "store-privileged",
feature = "store-internals",
feature = "failpoints"
))]
use levcs_core::{blake3_hash, ObjectHeader, ObjectId, ObjectType, FORMAT_VERSION};
use levcs_protocol::v2::{
ProjectionMode, ProjectionStageChunkV1, ProjectionStageManifestV1, ProjectionStageSessionV1,
StageSourceKindV1, StagedChunkObjectV1, StagedObjectV1,
};
use levcs_store::staging::{ProjectionStageBinding, StagedProjectionAdoption};
use levcs_store::types::NamespaceId;
use levcs_store::{RetainedObjectSource, StoreEngine, ValidatedTransaction};
#[path = "support/engine_matrix.rs"]
mod engine_matrix;
use engine_matrix::{
create_transaction, deadline, evidence, genesis_id, namespace_on_shard, now_micros,
open_absent_root, submit, DEFAULT_MAX_INDEX_RUNS,
};
const SHARD_COUNT: u16 = 2;
const CHUNKS: u32 = 2;
const PER_CHUNK: usize = 2;
const HOUR_MICROS: i64 = 3_600_000_000;
// ---------------------------------------------------------------------------
// A projection bound to a repository the engine actually created
// ---------------------------------------------------------------------------
struct Staged {
binding: ProjectionStageBinding,
chunks: Vec<ProjectionStageChunkV1>,
}
impl Staged {
/// Every object across every chunk, in no particular order — the test
/// asserts about all of them and never about their arrangement.
fn object_ids(&self) -> Vec<ObjectId> {
self.chunks
.iter()
.flat_map(|chunk| chunk.objects.iter())
.map(|object| object.descriptor.object_id)
.collect()
}
}
fn blob(body: &[u8]) -> StagedChunkObjectV1 {
let mut raw = ObjectHeader {
object_type: ObjectType::Blob,
format_version: FORMAT_VERSION,
body_len: body.len() as u64,
}
.encode()
.to_vec();
raw.extend_from_slice(body);
let id = blake3_hash(&raw);
StagedChunkObjectV1 {
descriptor: StagedObjectV1 {
object_id: id,
object_type: ObjectType::Blob as u8,
raw_len: raw.len() as u64,
raw_digest: id,
},
raw_bytes: raw,
}
}
/// A self-consistent projection bound to `namespace`.
///
/// The binding's `destination_repo`, `destination_genesis`, and
/// `expected_authority` are taken from the repository the engine created rather
/// than from constants. A projection bound to a repository that does not exist
/// is a different refusal entirely, and one bound to the wrong authority is
/// another; neither is what this file is about.
fn staged_projection(namespace: NamespaceId, session_id: [u8; 16], actor: [u8; 32]) -> Staged {
let total = CHUNKS as usize * PER_CHUNK;
let mut objects: Vec<StagedChunkObjectV1> = (0..total)
.map(|index| blob(format!("adopted-{}-{index}", hex::encode(session_id)).as_bytes()))
.collect();
// The manifest is the ordered concatenation of the chunks and must be
// strictly sorted, so the sort happens before the split, not inside each
// chunk.
objects.sort_by(|left, right| left.descriptor.cmp(&right.descriptor));
let total_object_bytes = objects
.iter()
.map(|object| object.descriptor.raw_len)
.sum::<u64>();
let descriptors: Vec<StagedObjectV1> = objects
.iter()
.map(|object| object.descriptor.clone())
.collect();
let chunks: Vec<ProjectionStageChunkV1> = (0..CHUNKS)
.map(|ordinal| {
let start = ordinal as usize * PER_CHUNK;
ProjectionStageChunkV1 {
session_id,
ordinal,
chunk_count: CHUNKS,
objects: objects[start..start + PER_CHUNK].to_vec(),
}
})
.collect();
let chunk_digests: Vec<ObjectId> = chunks
.iter()
.map(|chunk| chunk.chunk_digest().expect("chunk digest"))
.collect();
// The instance layer's commitment. B3 stores it and proves the manifest
// digest binds it; nothing in the store evaluates it.
let membership_root = blake3_hash(&session_id[..]);
let manifest = ProjectionStageManifestV1 {
session_id,
chunk_digests,
objects: descriptors,
membership_root,
};
let manifest_digest = manifest.manifest_digest().expect("manifest digest");
let authority = genesis_id(&namespace);
Staged {
binding: ProjectionStageBinding {
session: ProjectionStageSessionV1 {
session_id,
destination_repo: ObjectId(*namespace.as_bytes()),
destination_genesis: authority,
expected_authority: authority,
projection: ProjectionMode::Full,
source_kind: StageSourceKindV1::Mirror,
actor,
actor_key_epoch: 3,
source_generation_digest: ObjectId([12; 32]),
fork_proof: None,
final_operation_id: [13; 16],
final_operation_digest: ObjectId([14; 32]),
final_evidence_digest: ObjectId([15; 32]),
total_object_count: total as u64,
total_object_bytes,
chunk_count: CHUNKS,
manifest_digest,
expires_at_micros: now_micros() + HOUR_MICROS,
},
membership_root,
},
chunks,
}
}
/// Stage the projection through the engine's own staging and take the pin.
///
/// Through `engine.staging()` and not a second `ProjectionStaging`: staging's
/// ceilings are root-global and its constructor demands the root lock, so a
/// test that opened its own would be a second accountant for one root and would
/// prove the property against a staging the store does not use.
fn stage_and_finalize(engine: &StoreEngine, staged: &Staged) -> StagedProjectionAdoption {
let session = engine
.staging()
.begin(staged.binding.clone(), now_micros())
.expect("staging admits the session");
for chunk in &staged.chunks {
session.put_chunk(chunk, now_micros()).expect("chunk lands");
}
session.seal(now_micros()).expect("the session seals");
session
.finalize(now_micros())
.expect("the sealed session yields an adoption pin")
}
fn adopting_transaction(
namespace: NamespaceId,
operation: u8,
adoption: StagedProjectionAdoption,
) -> ValidatedTransaction {
let authority = genesis_id(&namespace);
ValidatedTransaction::builder(levcs_store::types::PrivilegedConstruction::assert_validated())
.namespace(namespace)
.operation(
levcs_store::types::OperationId([operation; 16]),
ObjectId([operation; 32]),
deadline(),
)
// Empty, and not a convenience: `FrameObjectsV1` is an enum, so a frame
// carries inline objects or an install descriptor and never both. The
// projection *is* this transaction's payload.
.objects(Vec::new())
.refs(Vec::new())
.authority(Some(authority), Some(authority))
.evidence(evidence())
.adopt_projection(adoption)
.expect("the builder admits one adoption")
.build()
.expect("a complete adopting transaction")
}
// ---------------------------------------------------------------------------
// The test
// ---------------------------------------------------------------------------
#[test]
fn an_adopted_projection_is_owned_by_its_artifacts_and_survives_a_checkpoint() {
let directory = tempfile::TempDir::new().expect("a temporary root");
let engine = open_absent_root(directory.path(), SHARD_COUNT, DEFAULT_MAX_INDEX_RUNS);
let namespace = namespace_on_shard(0, SHARD_COUNT, 1);
submit(&engine, create_transaction(namespace, 1))
.receipt()
.expect("the repository is created");
let staged = staged_projection(namespace, [25; 16], [21; 32]);
let ids = staged.object_ids();
assert_eq!(
ids.len(),
CHUNKS as usize * PER_CHUNK,
"the fixture must produce one object per chunk slot"
);
// Nothing staged is visible before a submit adopts it. Asserted before the
// adoption so a later `Some` is evidence the adoption produced it, rather
// than something that was always there.
let before = engine.snapshot(namespace).expect("snapshot");
for id in &ids {
assert_eq!(
before.locate(*id).expect("locate"),
None,
"a sealed but unadopted object must not be visible"
);
}
let adoption = stage_and_finalize(&engine, &staged);
let committed = submit(&engine, adopting_transaction(namespace, 2, adoption));
committed
.receipt()
.unwrap_or_else(|| panic!("the adoption must commit: {:?}", committed.error()));
let adopted = engine.snapshot(namespace).expect("snapshot");
let generations = resolve_all(&adopted, &ids, "immediately after adoption");
// A checkpoint publishes a successor retained generation and releases the
// predecessor's pins. This is the moment an adoption that was never carried
// forward stops resolving.
engine.checkpoint().expect("the store checkpoints");
let after = engine.snapshot(namespace).expect("snapshot");
let survived = resolve_all(&after, &ids, "after a checkpoint");
assert_eq!(
generations, survived,
"an adopted object's generation must not move when a successor generation is published; \
the artifacts are the same files either side of the checkpoint"
);
}
/// Staging outlives the engine, so the root lock has to outlive it too.
///
/// `StoreEngine::staging` hands out a reference, but a reference is not a
/// lifetime bound on what a caller may keep: `Arc::clone` escapes it, and so do
/// a `ProjectionStageSession` and an adoption pin, each of which owns a clone.
/// If `LOCK` were released when the engine dropped, any of those could still
/// write into a root another process had since opened.
///
/// Asserted against `RecoverySession::open` rather than against a second
/// `StoreEngine::open`, deliberately. Staging keeps an in-process registry of
/// open roots, so an engine open would be refused by *that* whether or not the
/// lock were held — a passing test proving nothing about the lock. Taking the
/// lock directly is the only form of this assertion that fails when the lease
/// is not retained.
#[test]
fn a_retained_staging_holds_the_root_lock_past_the_engine() {
let directory = tempfile::TempDir::new().expect("a temporary root");
let engine = open_absent_root(directory.path(), SHARD_COUNT, DEFAULT_MAX_INDEX_RUNS);
let staging = std::sync::Arc::clone(engine.staging());
drop(engine);
let contested = levcs_store::recovery::RecoverySession::open(directory.path());
assert!(
contested.is_err(),
"the root lock must still be held while staging is retained, but a second session took \
it: {contested:?}"
);
drop(staging);
levcs_store::recovery::RecoverySession::open(directory.path())
.expect("the lock is released once the last staging reference is gone");
}
/// Locate every id, require an artifact source for each, and hand back the
/// generations so the caller can compare them across a checkpoint.
///
/// Requiring `ProjectionArtifact` and not merely `Some` is the whole point:
/// an object indexed at the installing frame's location resolves to a tail or a
/// segment and would satisfy a `Some` assertion perfectly.
fn resolve_all(
snapshot: &levcs_store::RepoSnapshot,
ids: &[ObjectId],
when: &str,
) -> Vec<(ObjectId, u64)> {
let mut generations = Vec::with_capacity(ids.len());
for id in ids {
let location = snapshot
.locate(*id)
.expect("locate")
.unwrap_or_else(|| panic!("object {} is not visible {when}", hex::encode(id.0)));
let source = snapshot
.object_source(&location)
.expect("resolving a location this snapshot produced")
.unwrap_or_else(|| {
panic!(
"object {} resolves to no retained source {when}: generation {} is not \
retained, so its artifact was not carried forward",
hex::encode(id.0),
location.segment_generation
)
});
match source {
RetainedObjectSource::ProjectionArtifact(_) => {}
other => panic!(
"object {} resolves to {:?} {when}, not a projection artifact: it was indexed at \
the installing frame's location rather than at its chunk's",
hex::encode(id.0),
std::mem::discriminant(&other)
),
}
generations.push((*id, location.segment_generation));
}
generations
}

35
crates/levcs-store/tests/staging_sessions.rs
View File

@ -42,7 +42,7 @@ const HOUR_MICROS: i64 = 3_600_000_000;
/// a constructor production cannot reach.
struct Root {
directory: TempDir,
lock: RecoverySession,
lock: Arc<RecoverySession>,
options: StoreOptions,
}
@ -64,7 +64,7 @@ impl Root {
&DurabilityCounters::default(),
)
.expect("v2 root layout");
let lock = RecoverySession::open(directory.path()).expect("root lock");
let lock = Arc::new(RecoverySession::open(directory.path()).expect("root lock"));
Self {
directory,
lock,
@ -83,7 +83,7 @@ impl Root {
}
fn open_with(&self, durability: Arc<DurabilityCounters>) -> Arc<ProjectionStaging> {
ProjectionStaging::open(&self.lock, self.options.clone(), durability)
ProjectionStaging::open(Arc::clone(&self.lock), self.options.clone(), durability)
.expect("staging root opens")
}
@ -1091,7 +1091,7 @@ fn one_root_admits_exactly_one_staging_accountant() {
let (_first, _durability) = root.open();
let result = ProjectionStaging::open(
&root.lock,
Arc::clone(&root.lock),
root.options.clone(),
Arc::new(DurabilityCounters::default()),
);
@ -1108,7 +1108,7 @@ fn a_root_lock_held_on_another_root_is_not_proof() {
let other = Root::new();
let result = ProjectionStaging::open(
&other.lock,
Arc::clone(&other.lock),
owner.options.clone(),
Arc::new(DurabilityCounters::default()),
);
@ -1189,20 +1189,21 @@ fn session_of(
/// that the staged objects are invisible, and that they become visible only
/// after a `submit` adopts the descriptor.
///
/// Still blocked on B1 NamespaceTxn, and blocked in one more place than before.
/// `RepoSnapshot::locate`, `StoreEngine::snapshot`, and the submit path are all
/// `NotImplemented`/`unimplemented!` today. On top of that, the P1-4 fix moved
/// staging inside the locked engine lifetime: this test can no longer open its
/// own `ProjectionStaging` beside a `StoreEngine`, because holding two root
/// locks is exactly what was made unrepresentable. It needs an accessor on the
/// engine — recorded as an interface request to B1 — to reach the staging the
/// engine owns.
/// Still blocked on B1 NamespaceTxn, but in one place now rather than three.
/// `RepoSnapshot::locate` and `StoreEngine::snapshot` are implemented (scope
/// 6-B1 deliverable 8), and the interface request this comment used to record —
/// an engine accessor for the staging the engine owns, needed because the P1-4
/// fix made holding two root locks unrepresentable — was granted as
/// `StoreEngine::staging()` under contract review 2026-08-09-A. What remains is
/// the submit path: it refuses an adopting transaction, settling the pin before
/// anything is queued, until deliverable 3 wires the descriptor payload, the
/// pre-append revalidation, and the post-fence artifact merge.
///
/// Asserting the property against staging's own state instead would be charter
/// item 8 exactly — a property proved against the helper rather than the path
/// that runs — so it is marked blocked rather than satisfied the wrong way.
#[test]
#[ignore = "blocked on B1 NamespaceTxn: needs RepoSnapshot::locate, StoreEngine::snapshot/submit (scope 6.4 deliverables 1, 3-8), and a StoreEngine accessor for the engine-owned ProjectionStaging"]
#[ignore = "blocked on B1 NamespaceTxn: StoreEngine::submit refuses an adopting transaction until scope 6-B1 deliverable 3 wires it (the descriptor payload, the pre-append revalidation, and the post-fence artifact merge)"]
fn sealed_objects_stay_invisible_until_a_submit_adopts_them() {
let root = Root::new();
let fixture = projection([25; 16], [21; 32], 1, 2, HOUR_MICROS);
@ -1228,8 +1229,8 @@ fn sealed_objects_stay_invisible_until_a_submit_adopts_them() {
}
unimplemented!(
"B1: expose the engine-owned ProjectionStaging, begin/put/seal a session through it, \
re-assert locate() is None for every staged object, then build a ValidatedTransaction \
with adopt_projection(install, handle), submit it, and assert locate() returns Some"
"B1: begin/put/seal a session through engine.staging(), re-assert locate() is None for \
every staged object, then finalize() it and build a ValidatedTransaction with \
adopt_projection(adoption), submit it, and assert locate() returns Some"
);
}

247
crates/levcs-store/tests/support/engine_matrix.rs
View File

@ -281,6 +281,163 @@ pub fn push_transaction(namespace: NamespaceId, operation: u8, blob: u8) -> Vali
.expect("a complete push transaction")
}
/// Stage and finalize a two-chunk projection through the engine's own staging.
///
/// Two chunks rather than one for the same reason the B1 acceptance test uses
/// two: it is the smallest projection where per-chunk ownership is load-bearing,
/// and a crash row that only ever adopted one chunk would not exercise the
/// partitioned run set at all.
pub fn stage_projection_for(
engine: &StoreEngine,
namespace: NamespaceId,
seed: u8,
) -> levcs_store::staging::StagedProjectionAdoption {
let (binding, chunks) = projection_binding(namespace, seed);
let session = engine
.staging()
.begin(binding, now_micros())
.expect("staging admits the row's session");
for chunk in &chunks {
session
.put_chunk(chunk, now_micros())
.expect("the row's chunk lands");
}
session.seal(now_micros()).expect("the row's session seals");
session
.finalize(now_micros())
.expect("the row's sealed session yields an adoption pin")
}
/// The victim transaction when the row is driving the staged-projection axis.
pub fn adopting_transaction(
namespace: NamespaceId,
operation: u8,
adoption: levcs_store::staging::StagedProjectionAdoption,
) -> ValidatedTransaction {
let authority = genesis_id(&namespace);
ValidatedTransaction::builder(privileged())
.namespace(namespace)
.operation(
OperationId([operation; 16]),
ObjectId([operation; 32]),
deadline(),
)
// Empty: a frame carries inline objects or an install descriptor, never
// both, so the projection is this transaction's whole payload.
.objects(Vec::new())
.refs(Vec::new())
.authority(Some(authority), Some(authority))
.evidence(evidence())
.adopt_projection(adoption)
.expect("one adoption is admitted")
.build()
.expect("a complete adopting transaction")
}
fn projection_binding(
namespace: NamespaceId,
seed: u8,
) -> (
levcs_store::staging::ProjectionStageBinding,
Vec<levcs_protocol::v2::ProjectionStageChunkV1>,
) {
use levcs_core::{blake3_hash, ObjectHeader, FORMAT_VERSION};
use levcs_protocol::v2::{
ProjectionMode, ProjectionStageChunkV1, ProjectionStageManifestV1,
ProjectionStageSessionV1, StageSourceKindV1, StagedChunkObjectV1, StagedObjectV1,
};
const CHUNKS: u32 = 2;
const PER_CHUNK: usize = 1;
let session_id = [seed; 16];
let staged = |index: usize| -> StagedChunkObjectV1 {
let body = format!("row-staged-{seed:02x}-{index}");
let mut raw = ObjectHeader {
object_type: ObjectType::Blob,
format_version: FORMAT_VERSION,
body_len: body.len() as u64,
}
.encode()
.to_vec();
raw.extend_from_slice(body.as_bytes());
let id = blake3_hash(&raw);
StagedChunkObjectV1 {
descriptor: StagedObjectV1 {
object_id: id,
object_type: ObjectType::Blob as u8,
raw_len: raw.len() as u64,
raw_digest: id,
},
raw_bytes: raw,
}
};
let mut objects: Vec<StagedChunkObjectV1> =
(0..CHUNKS as usize * PER_CHUNK).map(staged).collect();
// The manifest is the ordered concatenation of the chunks and must be
// strictly sorted, so the sort happens before the split.
objects.sort_by(|left, right| left.descriptor.cmp(&right.descriptor));
let total_object_bytes = objects
.iter()
.map(|object| object.descriptor.raw_len)
.sum::<u64>();
let descriptors: Vec<StagedObjectV1> = objects
.iter()
.map(|object| object.descriptor.clone())
.collect();
let chunks: Vec<ProjectionStageChunkV1> = (0..CHUNKS)
.map(|ordinal| ProjectionStageChunkV1 {
session_id,
ordinal,
chunk_count: CHUNKS,
objects: objects[ordinal as usize * PER_CHUNK..][..PER_CHUNK].to_vec(),
})
.collect();
let chunk_digests: Vec<ObjectId> = chunks
.iter()
.map(|chunk| chunk.chunk_digest().expect("chunk digest"))
.collect();
let membership_root = blake3_hash(&session_id[..]);
let manifest = ProjectionStageManifestV1 {
session_id,
chunk_digests,
objects: descriptors,
membership_root,
};
let manifest_digest = manifest.manifest_digest().expect("manifest digest");
let authority = genesis_id(&namespace);
(
levcs_store::staging::ProjectionStageBinding {
session: ProjectionStageSessionV1 {
session_id,
destination_repo: ObjectId(*namespace.as_bytes()),
destination_genesis: authority,
expected_authority: authority,
projection: ProjectionMode::Full,
source_kind: StageSourceKindV1::Mirror,
actor: [0x21; 32],
actor_key_epoch: 3,
source_generation_digest: ObjectId([12; 32]),
fork_proof: None,
final_operation_id: [13; 16],
final_operation_digest: ObjectId([14; 32]),
final_evidence_digest: ObjectId([15; 32]),
total_object_count: objects.len() as u64,
total_object_bytes,
chunk_count: CHUNKS,
manifest_digest,
expires_at_micros: now_micros() + 3_600_000_000,
},
membership_root,
},
chunks,
)
}
fn privileged() -> levcs_store::types::PrivilegedConstruction {
levcs_store::types::PrivilegedConstruction::assert_validated()
}
@ -588,6 +745,56 @@ pub struct RowObservation {
/// against a comment about where the fence sits.
pub fences_before: u64,
pub fences_after: u64,
/// Staging's four adoption-pin terminal counters, read on the live engine
/// **before** it is dropped. A pin settled by an unwind or by a slot's
/// `Drop` is recorded here and nowhere else: the counters live in the
/// engine-owned staging and go with it, and reopening builds fresh ones.
///
/// `None` for an inline row, which takes no pin. That is a different fact
/// from "took a pin and settled it zero times", and collapsing the two
/// would let a row that silently stopped adopting keep passing.
pub adoption: Option<AdoptionCounters>,
}
/// The four ways an admitted pin can end, as staging counts them.
#[derive(Copy, Clone, Debug, Default, PartialEq, Eq)]
pub struct AdoptionCounters {
pub adopted: u64,
pub released: u64,
pub transferred: u64,
pub dropped: u64,
}
impl AdoptionCounters {
/// Which terminal outcome this row's pin actually reached.
///
/// `None` when no pin settled at all, and an error when more than one did:
/// a single adoption has exactly one outcome, and two counters moving means
/// the pin was settled twice or two pins were admitted where one was
/// expected.
pub fn sole_outcome(&self) -> Result<Option<&'static str>, String> {
let mut seen: Vec<&'static str> = Vec::new();
for (count, name) in [
(self.adopted, "Adopted"),
(self.released, "DefinitivePreAppendFailure"),
(self.transferred, "TransferredToRecovery"),
(self.dropped, "DroppedWithoutOutcome"),
] {
if count > 1 {
return Err(format!(
"{name} was recorded {count} times for one adoption"
));
}
if count == 1 {
seen.push(name);
}
}
match seen.as_slice() {
[] => Ok(None),
[one] => Ok(Some(one)),
many => Err(format!("one adoption reached {many:?}")),
}
}
}
impl RowObservation {
@ -625,8 +832,18 @@ pub fn drive_submit_row(
serial: &Serial,
point: Failpoint,
action: FailpointAction,
adopts: bool,
) -> RowObservation {
let row = format!("{} [{}]", point.name(), action_name(action));
let row = format!(
"{} [{}] [{}]",
point.name(),
action_name(action),
if adopts {
"staged_projection"
} else {
"inline"
}
);
let directory = tempfile::tempdir().expect("tempdir");
let root = directory.path().join("root");
@ -649,8 +866,16 @@ pub fn drive_submit_row(
.expect("shard 0 has a writer")
.fdatasync;
// Staged *before* the failpoint is armed. Staging writes chunks through
// maintenance workers, and arming first would let the row's fault fire on
// staging's own I/O rather than on the submit under test.
let staged = adopts.then(|| stage_projection_for(&engine, namespace, 0x22));
arm(serial, point, action);
let victim = submit(&engine, push_transaction(namespace, 0x22, 0xb2));
let victim = match staged {
Some(adoption) => submit(&engine, adopting_transaction(namespace, 0x22, adoption)),
None => submit(&engine, push_transaction(namespace, 0x22, 0xb2)),
};
let fences_after = engine
.durability_counters(0)
.expect("shard 0 has a writer")
@ -659,6 +884,19 @@ pub fn drive_submit_row(
.transaction_status(namespace, victim_operation)
.expect("transaction_status is a read and never fails on an open engine");
// Read while the engine is alive: staging's counters are engine-owned and a
// reopen builds fresh ones, so a pin settled by an unwind is only visible
// here.
let adoption = adopts.then(|| {
let counters = engine.staging().counters().snapshot();
AdoptionCounters {
adopted: counters.sessions_adopted,
released: counters.adoption_pins_released,
transferred: counters.adoption_pins_transferred,
dropped: counters.adoption_pins_dropped,
}
});
let probe = submit(&engine, push_transaction(namespace, 0x33, 0xb3));
// Whatever happened, nothing may stay armed for the next row: the registry
// is a one-shot global and a row that did not fire would otherwise hand its
@ -685,6 +923,7 @@ pub fn drive_submit_row(
prior_recovered,
fences_before,
fences_after,
adoption,
}
}
@ -946,6 +1185,10 @@ pub fn drive_root_cas_retry_row(
prior_recovered,
fences_before,
fences_after,
// The contention driver races shards against each other for the
// root CAS and never adopts; the payload axis is driven by
// `drive_submit_row`.
adoption: None,
}),
}
}

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

@ -1839,6 +1839,73 @@ The integration suite's `cleanup_is_deferred_and_names_its_deliverable` is repla
deleted — the distinction it stood for, that "nothing to do" and "cannot answer yet" are
different results, is now asserted the other way round at the public surface.
##### Contract review 2026-08-09-A
B1 deliverable 3 — `adopt_projection` through submit. Requested by B1 on starting the wiring,
because the public entry point the deliverable is named after could not be called. Three
visibility changes, two accessors, one lifetime correction, and one signature amendment —
`adopt_projection` takes the adoption as a single value, for the reason recorded below. No wire
format moves.
**`ValidatedTransactionBuilder::adopt_projection` was a public function no external caller
could invoke.** It takes a `ProjectionAdoption`, whose only issuer is
`ProjectionStageSession::finalize`, which was `pub(crate)` and returned the `pub(crate)`
`StagedProjectionAdoption`. The pin has no other constructor by design — that unforgeability is
the point — so the two together made the signature readable and uncallable. **`finalize` and
`StagedProjectionAdoption` become `pub`.** `StagedProjectionAdoption` sits in D0-B's frozen
adoption seam; this changes who may name it and nothing about what it is.
**`adopt_projection` now takes the whole `StagedProjectionAdoption`, and its fields stay
private.** Review found that publishing the two fields and keeping the two-argument signature
admits a pairing that is *wrong* rather than merely incomplete: finalize session A, take the
descriptor from session B, submit `(descriptor_B, handle_A)`. Every field of `descriptor_B` is
internally valid, so nothing looks damaged — the frame would simply name B's artifacts while
A's are the ones held against deletion. The frozen two-argument form existed so neither half
could be *forgotten* independently; taking one value keeps that and additionally makes the
mismatch unrepresentable. A pre-append revalidation against the pin's canonical resolution
still runs, because staging's state can change between `finalize` and submit — but it is now a
recheck of a pairing the type guarantees rather than the only thing standing between a caller
and a mispaired frame. `descriptor()` is added for inspection; it borrows, because taking the
descriptor out is what the type exists to prevent.
**`StoreEngine::staging()` is added**, returning `&Arc<ProjectionStaging>`. Since the P1-4 fix
moved staging inside the locked engine lifetime, no caller can construct one beside a
`StoreEngine` — holding two root locks is what that fix made unrepresentable — so without an
accessor there is no reachable way to stage the projection an adoption adopts. This closes the
interface request recorded in the ignored test at `tests/staging_sessions.rs`.
**`ProjectionStaging` now retains the `RecoverySession` lease, and the first version of this
accessor was unsound without it.** Its justification claimed that returning a reference kept
staging from outliving the lock. It does not: `Arc::clone` escapes the borrow, and so do a
`ProjectionStageSession` and an adoption pin, each of which owns an `Arc<ProjectionStaging>`.
With `LOCK` released when the engine dropped, any of those could still write into a root
another process had since opened — a second accountant for one root, which is the exact
condition staging's construction rules exist to prevent. **`ProjectionStaging::open` therefore
takes `Arc<RecoverySession>` and keeps it**, so the lock is released when the last holder is
dropped rather than when the engine is. `EngineShared` shares the same lease.
`a_retained_staging_holds_the_root_lock_past_the_engine` asserts it by taking the lock
directly, not by attempting a second engine open — staging's in-process root registry would
refuse that whether or not the lock were held, so only the direct form fails when the lease is
not retained. Verified by mutation: replacing the retained `Arc` with a `Weak` makes the test
fail with a second session holding the root.
**`RepoSnapshot` gains `object_source()` and carries its shard index.** `locate` answers *which
generation*, which stopped being sufficient the moment adoption existed: a logical generation
is a segment, an active tail, or a projection artifact, and an adopted object's bytes are not in
the frame that installed it. Before adoption every location a reader could hold was a frame in
this shard's journal, so the distinction was invisible; now a reader that cannot ask "what
kind" cannot act on the answer at all. The shard index is passed to `capture` rather than
derived because deriving it needs the shard count, which lives in `StoreOptions` and not in the
root. `RetainedObjectSource` is already re-exported, so `lib.rs` does not move.
**One finding recorded and not acted on.** `checkpoint.rs` has no notion of projection
artifacts: a retained generation's `projection_artifacts` are in-memory pins, and their
durability comes from the journal frame plus staging's adoption marker, with recovery
rebuilding them by replaying the frame through `resolve_committed`. The successor-generation
paths clone them forward correctly within a process. Whether a manifest that outlives the
frames it was derived from needs to record them is a B3/checkpoint question, not a B1 one, and
is raised here rather than answered.
##### Contract review 2026-08-07-A
B1 deliverable 8 — `RepoSnapshot` and `StoreEngine::snapshot`. Requested by B1 before any body