diff --git a/crates/levcs-store/src/options.rs b/crates/levcs-store/src/options.rs index 80026ce..8e70ae4 100644 --- a/crates/levcs-store/src/options.rs +++ b/crates/levcs-store/src/options.rs @@ -144,7 +144,7 @@ impl Default for StoreOptions { staging_max_objects_global: 800_000_000, staging_max_bytes_per_principal: 2 * 1024 * 1024 * 1024 * 1024, staging_max_bytes_global: 8 * 1024 * 1024 * 1024 * 1024, - staging_max_files_per_session: 1_000_002, + staging_max_files_per_session: 1_000_004, staging_max_files_per_principal: 2_000_000, staging_max_files_global: 8_000_000, // A maximal 1 TiB projection at the supported 16 MiB/s floor @@ -314,12 +314,13 @@ impl StoreOptions { per-principal must be <= global" ); require!( - self.staging_max_files_per_session >= u64::from(self.max_projection_chunks) + 2 + self.staging_max_files_per_session + >= u64::from(self.max_projection_chunks) + crate::staging::SESSION_FIXED_FILES && self.staging_max_files_per_session <= self.staging_max_files_per_principal && self.staging_max_files_per_principal <= self.staging_max_files_global, - "staging file limits must admit one maximal projection's chunks, \ - session record, and manifest, and \ - per-session <= per-principal <= global" + "staging file limits must admit one maximal projection's chunks plus its \ + peak fixed per-session files (session record, sealed manifest, adoption marker, \ + and the marker's temporary), and per-session <= per-principal <= global" ); require!( self.staging_session_max_age_micros > 0, @@ -535,7 +536,8 @@ mod tests { let mut o = valid(); o.max_projection_chunks = u32::try_from(items + 1).expect("ceiling fits u32"); - o.staging_max_files_per_session = u64::from(o.max_projection_chunks) + 2; + o.staging_max_files_per_session = + u64::from(o.max_projection_chunks) + crate::staging::SESSION_FIXED_FILES; o.staging_max_files_per_principal = o.staging_max_files_per_session * 2; o.staging_max_files_global = o.staging_max_files_per_session * 8; assert!( diff --git a/crates/levcs-store/src/recovery.rs b/crates/levcs-store/src/recovery.rs index 7939830..3fa4152 100644 --- a/crates/levcs-store/src/recovery.rs +++ b/crates/levcs-store/src/recovery.rs @@ -2317,7 +2317,14 @@ fn recover_shard_under_lock( hex::encode(descriptor.session_id) )) })?; - let artifacts = resolver.resolve_committed(frame.facts.namespace, descriptor)?; + // 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. + let artifacts = resolver.resolve_committed( + frame.facts.namespace, + descriptor, + frame.facts.shard_sequence, + )?; if artifacts.descriptor() != descriptor { return Err(StoreError::Corruption(format!( "staging recovery returned a different descriptor for committed session {}", @@ -3187,6 +3194,7 @@ mod production_session_tests { namespace: NamespaceId, descriptor: StagedProjectionInstallV1, artifacts: RecoveredProjectionArtifacts, + expected_adoption_sequence: u64, absent_session: [u8; 16], notifications: Mutex>, } @@ -3200,9 +3208,18 @@ mod production_session_tests { &self, namespace: NamespaceId, descriptor: &StagedProjectionInstallV1, + adoption_shard_sequence: u64, ) -> Result { assert_eq!(namespace, self.namespace); assert_eq!(descriptor, &self.descriptor); + // Recovery must hand over the adopting frame's own sequence, not a + // placeholder: it is the sole authority that creates a position for + // a pin that transferred across a process boundary, and every + // artifact's logical generation is derived from it. + assert_eq!( + adoption_shard_sequence, self.expected_adoption_sequence, + "the resolver was given a sequence that is not the adoption frame's" + ); Ok(self.artifacts.clone()) } @@ -3301,6 +3318,7 @@ mod production_session_tests { namespace, descriptor: descriptor.clone(), artifacts, + expected_adoption_sequence: 0, absent_session: [27; 16], notifications: Mutex::new(Vec::new()), }; diff --git a/crates/levcs-store/src/staging.rs b/crates/levcs-store/src/staging.rs index d400c66..b8dd481 100644 --- a/crates/levcs-store/src/staging.rs +++ b/crates/levcs-store/src/staging.rs @@ -12,8 +12,8 @@ use levcs_protocol::v2::{ ProjectionStageManifestV1, ProjectionStageSessionV1, StagedProjectionInstallV1, }; -use crate::index::{IndexDelta, IndexRun}; -use crate::roots::{CommittedRoot, RetainedIndexRun, RetainedProjectionArtifact}; +use crate::index::{IndexDelta, IndexKey, IndexLocation, IndexRun}; +use crate::roots::{CommittedRoot, PinnedFile, RetainedIndexRun, RetainedProjectionArtifact}; use crate::types::{NamespaceId, StoreError}; /// One immutable artifact offered for adoption. @@ -38,7 +38,26 @@ pub struct ProjectionAdoptionResolution { /// The only three ways ownership of an admitted adoption pin may end. #[derive(Copy, Clone, Debug, PartialEq, Eq)] pub enum ProjectionAdoptionOutcome { - Adopted, + /// The frame naming these artifacts committed at `committed_shard_sequence`. + /// + /// The sequence is carried because *when* an adoption happened is what makes + /// a later reference proof meaningful. Cleanup answers "does any committed + /// root still point at this directory?" against a root a caller supplies, + /// and a root captured **before** this adoption references none of these + /// artifacts for the trivial reason that it predates them — so absence + /// measured against it is not absence, and cleanup would delete a directory + /// the current root points into. Recording the position turns that into a + /// question staging can refuse: a root at or past this sequence necessarily + /// includes this adoption's effects, and an earlier one is not evidence. + /// + /// The value is B1's to supply because only B1 knows it — the append that + /// produced it has just returned — and staging is constructed before any + /// committed root exists (it is recovery's resolver), so there is no moment + /// at which it could observe the position itself. Frozen D0-B amendment, + /// contract review 2026-07-31-C. + Adopted { + committed_shard_sequence: u64, + }, DefinitivePreAppendFailure, TransferredToRecovery, } @@ -177,10 +196,25 @@ pub(crate) trait ProjectionRecoveryResolver: Send + Sync { /// Resolve a canonical committed descriptor into exact namespace-scoped /// membership and live physical ownership. + /// + /// `adoption_shard_sequence` is the shard sequence of the **complete + /// replayed adoption frame**, and is required rather than optional or + /// inferred. It is the identity every artifact's logical generation is + /// derived from, and staging cannot supply it: a transferred `Finalizing` + /// session has no durable position — that is exactly the state recovery is + /// resolving — and this frame is the sole authority that creates one. + /// + /// The recovery-direction counterpart of the D0-B + /// `ProjectionAdoptionOutcome::Adopted` amendment, and granted for the same + /// reason: an adoption's position is known only to the side that made the + /// frame authoritative. There it flowed B1 to B3 after a live append; here + /// it flows recovery to B3 after a replayed one. Contract review + /// 2026-07-31-D. fn resolve_committed( &self, namespace: NamespaceId, descriptor: &StagedProjectionInstallV1, + adoption_shard_sequence: u64, ) -> Result; /// Finish one transferred session after the physical-state proof is @@ -310,12 +344,38 @@ const STAGING_ARTIFACT_HEADER_LEN: usize = 8 + 2 + 2 + 16 + 4; const SESSION_RECORD_NAME: &str = "session"; const MANIFEST_NAME: &str = "manifest"; +/// Deliverable 6's durable pin state. +/// +/// One file rather than one per outcome, rewritten in place by a replacing +/// rename, so a session never holds two markers that disagree and the per +/// session file bound does not grow with the number of outcomes. +const ADOPTION_NAME: &str = "adoption"; -/// Files a session may create: one per chunk, plus the session record and the -/// sealed manifest. `options.rs` validates `staging_max_files_per_session >= -/// max_projection_chunks + 2` against exactly this layout, so the constant is -/// the shared definition of that `+ 2` rather than a second opinion about it. -const SESSION_FIXED_FILES: u64 = 2; +/// Files a session may hold at once: one per chunk, plus the session record, the +/// sealed manifest, the adoption marker, and the marker's temporary. `options.rs` +/// validates `staging_max_files_per_session >= max_projection_chunks + +/// SESSION_FIXED_FILES` against exactly this layout. +/// +/// It was 2 and became 4 with deliverable 6: the pin's durable marker, and the +/// `adoption.tmp` that exists beside it for the width of the +/// `Finalizing -> Adopted` replacement. A **peak**, not a total — the temporary +/// is gone the moment the rename returns — and the reservation is charged +/// against the peak because that is the instant the directory is widest. +/// +/// The constant's doc already claimed to be "the shared definition rather than a +/// second opinion" while `options.rs` carried a literal `+ 2`, so the claim was +/// false in the one way that matters; adding a file is exactly the edit that +/// finds that out. `options.rs` now reads the constant. +/// +/// Four is the true peak, and the marker is the only reason it is not three. +/// Every other artifact publishes through a `.tmp` as well, but each of +/// those temporaries stands in for a final name that is *absent* — a chunk or a +/// first manifest does not yet exist — so it occupies the slot it is about to +/// become rather than an extra one. State ordering keeps those writes from +/// overlapping a seal or a finalize. `adoption.tmp` is the single case that +/// coexists with a final file already on disk, because the transition it +/// publishes replaces a marker rather than creating one. +pub(crate) const SESSION_FIXED_FILES: u64 = 4; /// Which of the three staging artifact kinds a file is. /// @@ -328,6 +388,9 @@ enum StagingArtifactKind { SessionRecord = 1, Chunk = 2, Manifest = 3, + /// The adoption marker of deliverable 6. Carries the outcome so the pin's + /// fate is durable, not only its existence. + Adoption = 4, } impl StagingArtifactKind { @@ -340,6 +403,7 @@ impl StagingArtifactKind { 1 => Some(Self::SessionRecord), 2 => Some(Self::Chunk), 3 => Some(Self::Manifest), + 4 => Some(Self::Adoption), _ => None, } } @@ -378,6 +442,28 @@ pub struct StagingCounters { pub sessions_sealed: AtomicU64, pub sessions_aborted: AtomicU64, pub sessions_expired: AtomicU64, + /// Sessions that took an adoption pin, and the four ways one ends. Kept + /// apart because they are four different physical claims: `adopted` means a + /// committed root references the artifacts, `released` means nothing was + /// appended, and both `transferred` and `dropped` mean this process cannot + /// say — the second being the one nobody intended. + pub sessions_finalized: AtomicU64, + pub sessions_adopted: AtomicU64, + pub adoption_pins_released: AtomicU64, + pub adoption_pins_transferred: AtomicU64, + pub adoption_pins_dropped: AtomicU64, + /// Adopted sessions whose directories cleanup reclaimed after proving no + /// committed root referenced them, and the times it declined for the + /// opposite reason. Separate counters because "nothing to do" and "there + /// was something and it was still referenced" are different answers, and + /// only one of them says the reference proof did any work. + pub sessions_cleaned_up: AtomicU64, + pub cleanup_declined_referenced: AtomicU64, + /// Adopted sessions cleanup skipped because the root it was given had not + /// reached the adoption. Counted apart from a referenced decline because + /// they say different things about the caller: one is a healthy store + /// holding its artifacts, the other is a caller asking the wrong question. + pub cleanup_declined_stale_root: AtomicU64, /// Gauge: sessions currently holding budget. pub sessions_live: AtomicU64, /// Gauge: object bytes reserved by live sessions. @@ -433,6 +519,14 @@ impl StagingCounters { sessions_created: self.sessions_created.load(Relaxed), sessions_refused: self.sessions_refused.load(Relaxed), sessions_sealed: self.sessions_sealed.load(Relaxed), + sessions_finalized: self.sessions_finalized.load(Relaxed), + sessions_adopted: self.sessions_adopted.load(Relaxed), + adoption_pins_released: self.adoption_pins_released.load(Relaxed), + adoption_pins_transferred: self.adoption_pins_transferred.load(Relaxed), + adoption_pins_dropped: self.adoption_pins_dropped.load(Relaxed), + sessions_cleaned_up: self.sessions_cleaned_up.load(Relaxed), + cleanup_declined_referenced: self.cleanup_declined_referenced.load(Relaxed), + cleanup_declined_stale_root: self.cleanup_declined_stale_root.load(Relaxed), sessions_aborted: self.sessions_aborted.load(Relaxed), sessions_expired: self.sessions_expired.load(Relaxed), sessions_live: self.sessions_live.load(Relaxed), @@ -461,6 +555,14 @@ pub struct StagingCounterSnapshot { pub sessions_created: u64, pub sessions_refused: u64, pub sessions_sealed: u64, + pub sessions_finalized: u64, + pub sessions_adopted: u64, + pub adoption_pins_released: u64, + pub adoption_pins_transferred: u64, + pub adoption_pins_dropped: u64, + pub sessions_cleaned_up: u64, + pub cleanup_declined_referenced: u64, + pub cleanup_declined_stale_root: u64, pub sessions_aborted: u64, pub sessions_expired: u64, pub sessions_live: u64, @@ -488,16 +590,39 @@ pub enum ChunkPutOutcome { AlreadyPresent, } -/// The two states a session can hold in this pass. +/// Every state a session can hold, all four durable and all four +/// reconstructable. /// -/// Deliverable 6 adds `Finalizing`, which is the state that holds an adoption -/// pin. Expiry and abort match this exhaustively rather than defaulting, so -/// adding that variant will fail to compile at exactly the sites that must -/// learn about a pin instead of silently reclaiming a pinned session. +/// Reconstructability is the membership rule, and it is why `SessionBusy` is a +/// separate private enum: a value that a restart cannot produce has no business +/// on the wire. Each of these is decided by an artifact on disk — the session +/// record, the sealed manifest, and the adoption marker's recorded outcome — +/// never by a flag that only this process remembers. +/// +/// The previous pass predicted deliverable 6 would add one variant. It adds two. +/// `Adopted` is not in the deliverable's own text, and dropping the session on +/// adoption instead is what forced it: the artifacts stay in `staging/` and are +/// pinned by the committed root from then on, so removing the session record +/// would make the next reconstruction read the directory as an abandoned +/// materialization and reclaim committed content. Keeping the record without a +/// state, the other way out, re-charges the whole reservation for a session that +/// no longer occupies any staging budget — a bound that leaks a little on every +/// restart. A durable `Adopted` says the true thing instead: this directory is +/// store content now, and it is cleanup's business rather than staging's. #[derive(Copy, Clone, Debug, PartialEq, Eq)] pub enum StagedSessionState { Open, Sealed, + /// An adoption pin is outstanding. The artifacts may not be reclaimed by + /// expiry, abort, or cleanup while a session is here, and a pin that + /// outlives its process is resolved by recovery rather than dropped. + Finalizing, + /// The pin ended in `Adopted`: a committed transaction references these + /// artifacts, they are no longer staging occupancy, and the session holds no + /// reservation. It stays in the registry so cleanup can prove — against a + /// `CommittedRoot` rather than against this state — whether anything still + /// references them. + Adopted, } /// Read-only view of one live session. Never carries artifact bytes. @@ -706,6 +831,9 @@ impl ChunkSlot { enum SessionBusy { Idle, Sealing, + /// Between admitting the sole finalizer and the marker being durable. The + /// pin does not exist yet, so this is an exclusion and not a state. + Finalizing, Reclaiming, } @@ -737,6 +865,15 @@ struct SessionRecord { written_bytes: u64, reservation: Reservation, resolution: Option>, + /// The committed shard sequence this session's adoption frame reached, for + /// an `Adopted` session and nothing else. + /// + /// Cleanup's reference proof is only meaningful against a root that includes + /// this adoption; an earlier root omits every reference the adoption created + /// and would "prove" an absence that is really a date. Durable in the + /// adoption marker, so the proof survives the restart that separates an + /// adoption from the compaction that eventually drops its reference. + adopted_at_shard_sequence: Option, } impl SessionRecord { @@ -1013,6 +1150,7 @@ impl ProjectionStaging { let mut chunks: BTreeMap = BTreeMap::new(); let mut sealed_manifest: Option = None; + let mut adoption: Option<(AdoptionMark, u64)> = None; let mut written_objects = 0u64; let mut written_bytes = 0u64; for entry in std::fs::read_dir(directory)? { @@ -1034,6 +1172,10 @@ impl ProjectionStaging { sealed_manifest = Some(path); continue; } + if name == ADOPTION_NAME { + adoption = Some(read_adoption_marker(&path, session_id, &binding.session)?); + continue; + } let Some((ordinal, digest_from_name)) = parse_chunk_artifact_name(name) else { self.counters.unrecognized_entries.fetch_add(1, Relaxed); continue; @@ -1104,16 +1246,47 @@ impl ProjectionStaging { }; // A durable manifest artifact is the seal's own commit point, so its - // presence — not a flag — is what makes the session `Sealed` again. - let (state, resolution) = match sealed_manifest { - None => (StagedSessionState::Open, None), - Some(path) => { + // presence — not a flag — is what makes the session `Sealed` again, and + // the adoption marker beside it is what carries the pin across the + // process boundary. A marker without a manifest is not a state this + // store can produce: `finalize` only admits a sealed session, and the + // marker is written after the manifest is durable. Refusing it is the + // difference between reconstructing a pin and inventing one. + let mut adopted_at: Option = None; + let (state, resolution) = match (sealed_manifest, adoption) { + (None, None) => (StagedSessionState::Open, None), + (None, Some(_)) => { + return Err(StoreError::Corruption(format!( + "staging session {} carries an adoption marker with no sealed manifest", + hex::encode(session_id) + ))) + } + (Some(path), mark) => { let resolution = self.reconstruct_resolution(&path, &binding, session_id, &chunks)?; - (StagedSessionState::Sealed, Some(resolution)) + let state = match mark { + None => StagedSessionState::Sealed, + Some((AdoptionMark::Finalizing, _)) => StagedSessionState::Finalizing, + Some((AdoptionMark::Adopted, at)) => { + adopted_at = Some(at); + StagedSessionState::Adopted + } + }; + (state, Some(resolution)) } }; + // An adopted session's bytes are store content, not staging occupancy. + // Reconstructing its reservation would re-charge, on every restart, a + // budget the adoption released — a ceiling that quietly shrinks each + // time the process comes back. + let reservation = match state { + StagedSessionState::Open + | StagedSessionState::Sealed + | StagedSessionState::Finalizing => reservation, + StagedSessionState::Adopted => Reservation::default(), + }; + Ok(Some(SessionRecord { binding, shard_index, @@ -1125,6 +1298,7 @@ impl ProjectionStaging { written_bytes, reservation, resolution, + adopted_at_shard_sequence: adopted_at, })) } @@ -1399,15 +1573,29 @@ impl ProjectionStaging { let past_expiry = now_micros > record.binding.session.expires_at_micros; let idle = match record.busy { SessionBusy::Idle => true, - // A sealing session is mid-maintenance; the next sweep - // takes it. Deliverable 6 replaces this with the pinned - // case, which expiry may never reclaim at all. - SessionBusy::Sealing | SessionBusy::Reclaiming => false, + // Mid-maintenance; the next sweep takes it. + SessionBusy::Sealing | SessionBusy::Finalizing | SessionBusy::Reclaiming => { + false + } }; past_expiry && idle && match record.state { StagedSessionState::Open | StagedSessionState::Sealed => true, + // The named acceptance case of deliverable 6: expiry + // prevents a *new* finalizer but must never delete + // artifacts an already-admitted one holds. `finalize` + // takes `require_live` under the same registry lock this + // sweep uses, so the race has exactly two orderings and + // both are safe — expiry first reclaims a session no + // finalizer can then admit, finalize first leaves a + // pinned session this arm declines. + StagedSessionState::Finalizing => false, + // Not staging occupancy any more, and not expiry's to + // reclaim: a committed transaction may reference these + // artifacts, and only a proof against a `CommittedRoot` + // can say. That proof is cleanup's (deliverable 7). + StagedSessionState::Adopted => false, } }) .map(|(id, _)| *id) @@ -1423,19 +1611,188 @@ impl ProjectionStaging { Ok(reclaimed) } - /// Remove artifacts no committed manifest references. + /// Remove staged artifacts no committed root references. /// - /// Deferred: deliverable 7. The proof this owes is answered against a - /// `CommittedRoot`, and the state it must be able to observe — an adopted - /// session — cannot exist until deliverable 6 and B1's `adopt_projection` - /// land. A version that reclaimed everything unreferenced today would be - /// correct today and would silently become a reclamation of adopted - /// artifacts the moment adoption started working. - pub fn cleanup_unreferenced(&self, _root: &CommittedRoot) -> Result { - Err(StoreError::NotImplemented( - "ProjectionStaging::cleanup_unreferenced — B3 StagingSessions, scope 6.5 \ - deliverable 7 (reference proof against a CommittedRoot)", - )) + /// Deliverable 7. The candidate set is exactly the **adopted** sessions, and + /// that is the whole design rather than a filter on it: + /// + /// * `Open` and `Sealed` are live and belong to expiry and abort, which + /// already own them and already know when they are dead. + /// * `Finalizing` holds an outstanding pin. Nothing in this process knows + /// whether a frame naming its artifacts is about to be appended, so there + /// is nothing to prove absence *of* yet. + /// * `Adopted` is the one state where the artifacts are store content and + /// the question "does anything still point at them?" is both meaningful + /// and answerable. A checkpoint or a compaction that stops referencing an + /// adopted projection is what eventually makes its directory reclaimable, + /// and this is the only path that may remove it. + /// + /// # Whole directories, not individual artifacts + /// + /// The deliverable's wording is per-artifact and the unit here is the + /// session, deliberately. A session's chunks are not independent files: the + /// manifest names all of them, and reconstruction refuses a sealed session + /// missing any ordinal. Removing the unreferenced half of a directory would + /// leave a session that no longer reconstructs — trading a bounded leak for + /// a root that fails to open — so a session is reclaimed only when *nothing* + /// in it is referenced. `reclaim_session_directory` then applies the + /// per-artifact rule that clause is really about: it removes only files + /// carrying this session's own marker, and refuses the whole directory if it + /// finds anything else. + /// + /// # The supplied root has to be new enough to be evidence + /// + /// A caller passes the root it holds, and a root captured **before** a + /// session was adopted references none of that session's artifacts — for the + /// trivial reason that it predates every reference the adoption created. + /// Absence measured against it is a date, not an absence, and acting on it + /// deletes a directory the *current* root points into. + /// + /// So each adopted session carries the committed shard sequence its adoption + /// frame reached, and is skipped unless the supplied root has reached at + /// least that far. A root at or past it necessarily includes the adoption's + /// effects, which is what makes the absence real. + /// + /// I had this backwards in review and it is worth stating plainly: the + /// argument "a racing newer root can only *add* references" is an argument + /// for the hazard, not against it. Adding references is precisely what makes + /// an older root omit them. + pub fn cleanup_unreferenced(&self, root: &CommittedRoot) -> Result { + let candidates: Vec<([u8; 16], PathBuf, u16, Option)> = { + let mut registry = self.lock(); + let eligible: Vec<[u8; 16]> = registry + .sessions + .iter() + .filter(|(_, record)| match record.state { + StagedSessionState::Adopted => true, + StagedSessionState::Open + | StagedSessionState::Sealed + | StagedSessionState::Finalizing => false, + }) + .filter(|(_, record)| matches!(record.busy, SessionBusy::Idle)) + .map(|(session_id, _)| *session_id) + .collect(); + // Marked `Reclaiming` under the same acquisition that selected them, + // so nothing can begin operating on a session between the choice and + // the exclusion. + let mut candidates = Vec::with_capacity(eligible.len()); + for session_id in eligible { + if let Some(record) = registry.sessions.get_mut(&session_id) { + record.busy = SessionBusy::Reclaiming; + candidates.push(( + session_id, + record.directory.clone(), + record.shard_index, + record.adopted_at_shard_sequence, + )); + } + } + candidates + }; + + let mut reclaimed = 0u64; + for (session_id, directory, shard_index, recorded) in candidates { + // The position is what makes the absence proof below evidence rather + // than a date. A session missing one is not reclaimable at all: an + // adopted session always records where it was adopted, so its + // absence means this store cannot say when the reference it is about + // to disprove came into existence. + let adopted_at = match recorded { + Some(sequence) => sequence, + None => { + self.clear_busy(session_id); + continue; + } + }; + // `Some(0)` and `None` are different answers and only one of them is + // evidence. Zero is a valid committed sequence, so defaulting an + // absent entry to it makes a root that says *nothing* about this + // shard indistinguishable from one that has committed through its + // first frame — and an adoption at sequence 0 becomes reclaimable + // through a root that never mentioned the shard it lives on. + let reached = matches!( + root.shard_committed_sequence(shard_index), + Some(through) if through >= adopted_at + ); + if !reached { + self.counters + .cleanup_declined_stale_root + .fetch_add(1, Relaxed); + self.clear_busy(session_id); + continue; + } + match self.session_is_unreferenced(root, &directory, session_id, adopted_at) { + Ok(true) => {} + Ok(false) => { + self.clear_busy(session_id); + continue; + } + Err(error) => { + self.clear_busy(session_id); + return Err(error); + } + } + match self.reclaim_directory(directory, session_id) { + Ok(()) => { + let mut registry = self.lock(); + // The reservation was already released at adoption, so this + // only drops the record. Releasing twice is why + // `release_reservation_locked` zeroes as it goes. + self.release_locked(&mut registry, session_id); + drop(registry); + self.counters.sessions_cleaned_up.fetch_add(1, Relaxed); + reclaimed += 1; + } + Err(error) => { + self.clear_busy(session_id); + return Err(error); + } + } + } + Ok(reclaimed) + } + + /// Does the committed root reference nothing in this session's directory? + /// + /// Every file is checked, not only the chunks. A root that pinned a + /// session's manifest and nothing else would still be holding that + /// directory, and answering on chunks alone would delete the file it holds. + fn session_is_unreferenced( + &self, + root: &CommittedRoot, + directory: &Path, + session_id: [u8; 16], + adopted_at: u64, + ) -> Result { + let _ = adopted_at; + if !directory.exists() { + return Ok(true); + } + let directory = directory.to_path_buf(); + let entries = self.run_maintenance(move || { + let mut paths = Vec::new(); + for entry in std::fs::read_dir(&directory)? { + paths.push(entry?.path()); + } + Ok(paths) + })?; + for path in entries { + if root.references_artifact(&path) { + self.counters + .cleanup_declined_referenced + .fetch_add(1, Relaxed); + let _ = session_id; + return Ok(false); + } + } + Ok(true) + } + + fn clear_busy(&self, session_id: [u8; 16]) { + let mut registry = self.lock(); + if let Some(record) = registry.sessions.get_mut(&session_id) { + record.busy = SessionBusy::Idle; + } } // --- internals ------------------------------------------------------ @@ -1690,6 +2047,7 @@ impl ProjectionStaging { written_bytes: 0, reservation, resolution: None, + adopted_at_shard_sequence: None, }, ); @@ -1716,6 +2074,60 @@ impl ProjectionStaging { self.release_locked(&mut registry, session_id); } + /// Release a session's budget while leaving the record in place. + /// + /// Adoption ends staging occupancy without ending the directory: the + /// artifacts are store content from that point and a committed root points + /// into them, so charging them against staging's ceilings forever would make + /// every adoption permanently shrink the budget for the next one. The record + /// stays so cleanup can still identify the directory and prove, against a + /// `CommittedRoot`, whether anything references it. + /// + /// Idempotent by construction: the reservation is zeroed as it is released, + /// so a repeated call subtracts nothing. + fn release_reservation_locked(&self, registry: &mut Registry, session_id: [u8; 16]) { + let Some(record) = registry.sessions.get_mut(&session_id) else { + return; + }; + let reservation = std::mem::take(&mut record.reservation); + let written_bytes = std::mem::take(&mut record.written_bytes); + let principal = record.principal(); + if let Some(entry) = registry.principals.get_mut(&principal) { + entry.sessions = entry.sessions.saturating_sub(1); + entry.objects = entry.objects.saturating_sub(reservation.objects); + entry.bytes = entry.bytes.saturating_sub(reservation.bytes); + entry.files = entry.files.saturating_sub(reservation.files); + if entry.sessions == 0 { + registry.principals.remove(&principal); + } + } + registry.global.sessions = registry.global.sessions.saturating_sub(1); + registry.global.objects = registry.global.objects.saturating_sub(reservation.objects); + registry.global.bytes = registry.global.bytes.saturating_sub(reservation.bytes); + registry.global.files = registry.global.files.saturating_sub(reservation.files); + registry.debt_reserved_bytes = registry + .debt_reserved_bytes + .saturating_sub(reservation.bytes); + + let counters = &self.counters; + counters.sessions_live.fetch_sub(1, Relaxed); + counters + .reserved_objects + .fetch_sub(reservation.objects, Relaxed); + counters + .reserved_bytes + .fetch_sub(reservation.bytes, Relaxed); + counters + .reserved_files + .fetch_sub(reservation.files, Relaxed); + counters + .compaction_debt_reserved_bytes + .fetch_sub(reservation.bytes, Relaxed); + counters + .compaction_debt_written_bytes + .fetch_sub(written_bytes, Relaxed); + } + fn release_locked(&self, registry: &mut Registry, session_id: [u8; 16]) { let Some(record) = registry.sessions.remove(&session_id) else { return; @@ -1831,10 +2243,23 @@ impl ProjectionStaging { .ok_or_else(|| unknown_session(session_id))?; match record.state { StagedSessionState::Open | StagedSessionState::Sealed => {} + // Reclamation deletes the directory, so it may not run against + // a session whose artifacts something else is entitled to: an + // outstanding pin, or a committed root's reference. Both are + // refused here rather than filtered by every caller, because a + // caller that forgets is a caller that deletes committed data. + StagedSessionState::Finalizing | StagedSessionState::Adopted => { + return Err(StoreError::Conflict(format!( + "staging session {} holds an adoption outcome and may not be \ + reclaimed by expiry or abort; only a reference proof against a \ + committed root may remove it", + hex::encode(session_id) + ))) + } } match record.busy { SessionBusy::Idle => {} - SessionBusy::Sealing | SessionBusy::Reclaiming => { + SessionBusy::Sealing | SessionBusy::Finalizing | SessionBusy::Reclaiming => { return Err(StoreError::Overloaded { limit: "staging_session_maintenance_in_flight", retry_after_micros: 1, @@ -2217,7 +2642,12 @@ impl ProjectionStageSession { .get(&self.session_id) .ok_or_else(|| unknown_session(self.session_id))?; match record.state { - StagedSessionState::Sealed => {} + // A pinned or adopted session still has the manifest sealing wrote, + // and the adopting side revalidates against it. Refusing here would + // make a pin unusable by the caller holding it. + StagedSessionState::Sealed + | StagedSessionState::Finalizing + | StagedSessionState::Adopted => {} StagedSessionState::Open => { return Err(StoreError::Conflict(format!( "staging session {} is still open and has no sealed manifest to resolve", @@ -2379,52 +2809,276 @@ impl ProjectionStageSession { Ok(()) } - /// Move `Open -> Finalizing` for the sole bound operation/digest and take + /// Move `Sealed -> Finalizing` for the sole bound operation/digest and take /// the adoption pin. /// - /// Deferred: deliverable 6. This is where the `ProjectionAdoptionLifecycle` - /// implementation lands, and it is the live B1/B3 seam — the pin has to - /// survive a definitive pre-append failure, an expiry racing an admitted - /// finalizer, and a transfer to recovery, and none of those can be tested - /// against a `submit` that does not exist yet. A placeholder that returned - /// a handle would hand out an adoption capability with no pin behind it, - /// which is precisely the security property this package exists to hold. + /// From `Sealed`, not `Open`: the sealed manifest is what an adopter + /// revalidates against, so a pin on a session without one names state that + /// does not exist. Scope §6.5 and plan §8 both said `Open` and are amended + /// by contract review 2026-07-31-A. + /// + /// The pin is made durable **before** it is issued. A handle backed by an + /// in-memory flag is an adoption capability with nothing behind it — the + /// process stops, no marker is on the device, and the next open reconstructs + /// a plain sealed session while a committed frame may already reference its + /// artifacts. That is precisely the security property this package exists to + /// hold. #[allow(dead_code)] // B1's `adopt_projection` is the only legitimate caller. - pub(crate) fn finalize( - &self, - _now_micros: i64, - ) -> Result { - Err(StoreError::NotImplemented( - "ProjectionStageSession::finalize — B3 StagingSessions, scope 6.5 deliverable 6 \ - (finalize and the adoption pin)", + pub(crate) fn finalize(&self, now_micros: i64) -> Result { + let staging = &self.staging; + + // Phase 1, under the registry: admit exactly one finalizer, and do it + // in the same critical section that expiry evaluates. That shared lock + // is the whole of the expiry/finalize race: either this runs first and + // `expire` then declines a `Finalizing` session, or expiry runs first + // and `require_live` refuses here. There is no ordering in which a + // sweep deletes artifacts a pin already holds. + let (install, directory, operation, digest) = { + let mut registry = staging.lock(); + let record = registry + .sessions + .get(&self.session_id) + .ok_or_else(|| unknown_session(self.session_id))?; + require_finalizable(record, self.session_id)?; + require_live(record, self.session_id, now_micros)?; + require_idle(record, self.session_id)?; + + let resolution = record.resolution.clone().ok_or_else(|| { + StoreError::Corruption("sealed staging session lost its resolution".into()) + })?; + let install = install_from_resolution(&resolution)?; + let directory = record.directory.clone(); + let operation = record.binding.session.final_operation_id; + let digest = record.binding.session.final_operation_digest; + let record = registry + .sessions + .get_mut(&self.session_id) + .ok_or_else(|| unknown_session(self.session_id))?; + record.busy = SessionBusy::Finalizing; + (install, directory, operation, digest) + }; + + // Phase 2, on a maintenance worker: make the pin durable *before* + // handing it out. A handle issued against an in-memory flag is an + // adoption capability with nothing behind it — the process stops, the + // marker is not there, and the next open reconstructs a plain sealed + // session while a committed frame may already reference its artifacts. + let session_id = self.session_id; + let durability = Arc::clone(&staging.durability); + let counters = Arc::clone(&staging.counters); + let written = staging.run_maintenance(move || { + write_adoption_marker( + &directory, + session_id, + AdoptionMark::Finalizing, + operation, + digest, + &durability, + &counters, + ) + }); + + // Phase 3, under the registry: publish the state the marker now proves, + // or clear the exclusion and hand back the reason. + let mut registry = staging.lock(); + let record = registry + .sessions + .get_mut(&self.session_id) + .ok_or_else(|| unknown_session(self.session_id))?; + record.busy = SessionBusy::Idle; + written?; + record.state = StagedSessionState::Finalizing; + drop(registry); + staging.counters.sessions_finalized.fetch_add(1, Relaxed); + + Ok(StagedProjectionAdoption { + descriptor: install, + handle: ProjectionAdoption::new(Arc::new(SessionAdoptionPin { + staging: Arc::clone(staging), + session_id: self.session_id, + })), + }) + } +} + +/// The lifecycle behind one issued [`ProjectionAdoption`]. +/// +/// Holds an `Arc` rather than a borrow because the pin is +/// deliberately allowed to outlive the `ProjectionStageSession` that produced +/// it: B1 carries the handle into a `ValidatedTransaction` and finishes it after +/// the append decides, which is a different scope entirely. +struct SessionAdoptionPin { + staging: Arc, + session_id: [u8; 16], +} + +impl ProjectionAdoptionLifecycle for SessionAdoptionPin { + fn resolution(&self) -> Result, StoreError> { + let registry = self.staging.lock(); + let record = registry + .sessions + .get(&self.session_id) + .ok_or_else(|| unknown_session(self.session_id))?; + record.resolution.clone().ok_or_else(|| { + StoreError::Corruption("pinned staging session lost its resolution".into()) + }) + } + + /// Record the one terminal outcome, durably, before the pin is released. + /// + /// Each arm is a different physical claim and none of them is a flag: + /// + /// * `Adopted` — a committed transaction references these artifacts. The + /// marker is rewritten so the next reconstruction says so; the directory + /// stays exactly where it is, because a committed root now points into it. + /// The reservation is released: staged bytes that became store content are + /// not staging occupancy, and charging them forever would make every + /// adoption shrink the budget for the next one. + /// * `DefinitivePreAppendFailure` — nothing was appended, so the pin simply + /// ends. The marker is removed and the session returns to `Sealed`, which + /// is what the durable manifest already says it is. Deliverable 6's text + /// says "returns the session to `Open`"; `Open` here would contradict + /// reconstruction, which reads the sealed manifest's presence as the seal's + /// own commit point and would hand back `Sealed` on the next restart + /// regardless. Returning it to a state a restart cannot reproduce is the + /// defect `SessionBusy` exists to avoid, so it returns to `Sealed` — still + /// finalizable, which is the property the sentence is about. If the session + /// is no longer live, the marker still goes and expiry reclaims it on the + /// next sweep, which is "otherwise cleanup aborts it". + /// * `TransferredToRecovery` — nobody in this process knows whether the frame + /// reached the journal. The marker stays, so the next open reconstructs + /// `Finalizing` and `transferred_sessions` hands the question to recovery. + fn finish(&self, outcome: ProjectionAdoptionOutcome) -> Result<(), StoreError> { + match outcome { + ProjectionAdoptionOutcome::Adopted { + committed_shard_sequence, + } => { + self.adopt_mark(committed_shard_sequence)?; + let mut registry = self.staging.lock(); + if let Some(record) = registry.sessions.get_mut(&self.session_id) { + record.state = StagedSessionState::Adopted; + record.adopted_at_shard_sequence = Some(committed_shard_sequence); + } + self.staging + .release_reservation_locked(&mut registry, self.session_id); + drop(registry); + self.staging.counters.sessions_adopted.fetch_add(1, Relaxed); + Ok(()) + } + ProjectionAdoptionOutcome::DefinitivePreAppendFailure => { + self.clear_mark()?; + let mut registry = self.staging.lock(); + if let Some(record) = registry.sessions.get_mut(&self.session_id) { + record.state = StagedSessionState::Sealed; + } + drop(registry); + self.staging + .counters + .adoption_pins_released + .fetch_add(1, Relaxed); + Ok(()) + } + ProjectionAdoptionOutcome::TransferredToRecovery => { + self.staging + .counters + .adoption_pins_transferred + .fetch_add(1, Relaxed); + Ok(()) + } + } + } + + /// A handle dropped without an outcome is not an error to report — there is + /// nobody left to report it to — but it is emphatically not a release. + /// + /// The caller was somewhere between "about to append" and "knows the + /// answer", and unwound without saying which. That is exactly the state + /// `TransferredToRecovery` describes, so the marker stays and recovery + /// decides. Treating it as a release would let the next sweep delete + /// artifacts a committed frame may already reference. + fn dropped_without_outcome(&self) { + self.staging + .counters + .adoption_pins_dropped + .fetch_add(1, Relaxed); + } +} + +impl SessionAdoptionPin { + fn adopt_mark(&self, committed_shard_sequence: u64) -> Result<(), StoreError> { + let (directory, session) = { + let registry = self.staging.lock(); + let record = registry + .sessions + .get(&self.session_id) + .ok_or_else(|| unknown_session(self.session_id))?; + (record.directory.clone(), record.binding.session.clone()) + }; + let session_id = self.session_id; + let durability = Arc::clone(&self.staging.durability); + let counters = Arc::clone(&self.staging.counters); + self.staging.run_maintenance(move || { + adopt_adoption_marker( + &directory, + session_id, + &session, + committed_shard_sequence, + &durability, + &counters, + ) + }) + } + + fn clear_mark(&self) -> Result<(), StoreError> { + let (directory, _, _) = self.pin_identity()?; + let durability = Arc::clone(&self.staging.durability); + self.staging + .run_maintenance(move || remove_adoption_marker(&directory, &durability)) + } + + #[allow(clippy::type_complexity)] + fn pin_identity(&self) -> Result<(PathBuf, [u8; 16], ObjectId), StoreError> { + let registry = self.staging.lock(); + let record = registry + .sessions + .get(&self.session_id) + .ok_or_else(|| unknown_session(self.session_id))?; + Ok(( + record.directory.clone(), + record.binding.session.final_operation_id, + record.binding.session.final_operation_digest, )) } } -/// Deliverable 8, one method answered and two still deferred. +/// Deliverable 8, complete: all three methods answer. /// -/// The seam is implemented rather than absent so recovery's dependency on -/// staging is visible at the type level: a store that recovers a committed +/// The seam existed before it was implemented so recovery's dependency on +/// staging was visible at the type level — a store that recovers a committed /// staged install without resolving it would publish membership for objects it -/// cannot locate. The two deferred methods name the deliverable rather than -/// returning an empty result, because "nothing to resolve" and "cannot answer -/// yet" are different answers and only one of them is true. +/// cannot locate — and while two methods were deferred they named the +/// deliverable rather than returning an empty result, because "nothing to +/// resolve" and "cannot answer yet" are different answers and only one of them +/// was true. +/// +/// Both now answer. `resolve_committed` binds a descriptor to the bytes on disk +/// and resolves whole or not at all; `notify_recovered` ends a transferred pin +/// as an adoption or reclaims artifacts recovery proved no frame names. Contract +/// review 2026-07-31-D. impl ProjectionRecoveryResolver for ProjectionStaging { - /// The transferred set, **derived rather than asserted, and provably empty - /// today.** + /// The transferred set, **derived rather than asserted.** /// - /// A pin transfers to recovery only from `Finalizing`, and `Finalizing` is - /// deliverable 6: `finalize` returns `NotImplemented`, so no session has - /// ever entered that state, and — now that reconstruction exists — the only - /// states a durable session directory can come back in are `Open` and - /// `Sealed`. The empty answer is therefore a proof about the reachable - /// state space, not the "no transferred sessions" guess the deferred - /// version would have been making. + /// It was provably empty while `finalize` was deferred, because no session + /// could enter `Finalizing` at all. Deliverable 6 landed that state and made + /// it durable, so this now answers the question it was written to be able to + /// answer: a session reconstructed in `Finalizing` held a pin when its + /// process stopped, and nothing in *this* process knows whether the final + /// frame reached the journal. /// - /// It is computed by an exhaustive match over the state rather than - /// returned as a constant, so adding `Finalizing` fails to compile here — - /// at the one place that must learn a pin can now outlive the process — - /// instead of silently continuing to answer "none". + /// Still an exhaustive match over the state rather than a filter, which is + /// how `Finalizing` and `Adopted` arrived here as compile errors — at the one + /// place that had to learn a pin can outlive the process — rather than being + /// swept silently into "none". fn transferred_sessions(&self, shard_index: u16) -> Result, StoreError> { let registry = self.lock(); let transferred: Vec<[u8; 16]> = registry @@ -2433,36 +3087,436 @@ impl ProjectionRecoveryResolver for ProjectionStaging { .filter(|(_, record)| record.shard_index == shard_index) .filter(|(_, record)| match record.state { StagedSessionState::Open | StagedSessionState::Sealed => false, + // The answer this method was written to be able to give. A + // session reconstructed in `Finalizing` held a pin when its + // process stopped, so nothing in this process knows whether the + // final frame reached the journal — only recovery can say. + StagedSessionState::Finalizing => true, + // Already resolved. Reporting it would ask recovery to decide a + // question that has a durable answer, and `notify_recovered` + // would then be handed a session with nothing left to notify. + StagedSessionState::Adopted => false, }) .map(|(session_id, _)| *session_id) .collect(); Ok(Arc::from(transferred)) } + /// Resolve a committed descriptor into membership and live ownership. + /// + /// Read-only, and mechanical: this verifies the descriptor against the + /// session's own sealed state and hands back what it finds. Identity, graph, + /// policy, authority, and federation decisions are deliberately absent — the + /// complete frame is already the durable authority, and resolution may + /// inspect, open, hash, and pin its immutable artifacts but may not repair + /// them. + /// + /// # It can never expose a partial chunk set + /// + /// Every ordinal the manifest declares is read back, decoded, and checked + /// before a single location is produced, and a missing or unreadable one + /// fails the whole resolution. `RecoveredProjectionArtifacts::new` then + /// refuses unless the entry count it was given equals the descriptor's own + /// `object_count`. So a directory holding some of its chunks resolves to an + /// error rather than to a smaller projection — which is the deliverable's + /// central claim, and the one a "resolve what is present" implementation + /// would quietly violate. + /// + /// # One location per chunk + /// + /// An `IndexLocation` names the whole certified record, never the object + /// bytes inside it, so every object in a chunk shares that chunk's location + /// and a reader validates the artifact before extracting from its decoded + /// vector — exactly as several objects share one journal frame. This is why + /// no per-object byte offsets are needed and why none are computed here: + /// pointing at object bytes would let a reader return bytes from a record it + /// never proved complete. fn resolve_committed( &self, - _namespace: NamespaceId, - _descriptor: &StagedProjectionInstallV1, + namespace: NamespaceId, + descriptor: &StagedProjectionInstallV1, + adoption_shard_sequence: u64, ) -> Result { - Err(StoreError::NotImplemented( - "ProjectionStaging::resolve_committed — B3 StagingSessions, scope 6.5 \ - deliverable 8 (recovery treatment of unreferenced artifacts)", - )) + let session_id = descriptor.session_id; + let resolution = { + let registry = self.lock(); + let record = registry.sessions.get(&session_id).ok_or_else(|| { + StoreError::Corruption(format!( + "committed staged projection names session {}, which this root does not \ + hold; its artifacts cannot be resolved and the objects it published \ + would be unreadable", + hex::encode(session_id) + )) + })?; + record.resolution.clone().ok_or_else(|| { + StoreError::Corruption(format!( + "committed staged projection session {} has no sealed manifest to \ + resolve", + hex::encode(session_id) + )) + })? + }; + + // The descriptor a frame carries must be the one this session would + // install. Anything else means the frame and the artifacts on this device + // describe different projections, and adopting either would publish + // membership the other does not support. + let expected = install_from_resolution(&resolution)?; + if &expected != descriptor { + return Err(StoreError::Corruption(format!( + "committed staged projection session {} does not reconstruct the descriptor \ + its frame carries", + hex::encode(session_id) + ))); + } + + let artifacts = Arc::clone(&resolution.artifacts); + let session = resolution.session.clone(); + let sealed_digests = resolution.manifest.chunk_digests.clone(); + let expected_set = descriptor.artifact_set_digest; + let read = self.run_maintenance(move || { + let mut chunks = Vec::with_capacity(artifacts.len()); + let mut observed_set: Vec = Vec::with_capacity(artifacts.len()); + for (ordinal, artifact) in artifacts.iter().enumerate() { + let ordinal = u32::try_from(ordinal).map_err(|_| { + StoreError::Corruption("staged chunk ordinal does not fit u32".into()) + })?; + let bytes = + read_staging_artifact(&artifact.path, StagingArtifactKind::Chunk, session_id)?; + let chunk = ProjectionStageChunkV1::decode_canonical(&bytes).map_err(|e| { + StoreError::Corruption(format!( + "committed staged chunk {} no longer decodes: {e}", + artifact.path.display() + )) + })?; + if chunk.ordinal != ordinal + || chunk.session_id != session_id + || chunk.chunk_count != session.chunk_count + { + return Err(StoreError::Corruption(format!( + "committed staged chunk {} no longer matches the ordinal, session, or \ + chunk count its manifest binds", + artifact.path.display() + ))); + } + // Bind on the bytes just read, not on the sealed record beside + // them. Everything above is shape — session, ordinal, count — + // and a *different* valid chunk of the same shape satisfies all + // of it while carrying entirely different objects. What makes + // this artifact the one the descriptor commits to is its digest. + let observed = chunk.chunk_digest().map_err(|e| { + StoreError::Corruption(format!( + "committed staged chunk {} digest: {e}", + artifact.path.display() + )) + })?; + if observed != artifact.digest { + return Err(StoreError::Corruption(format!( + "committed staged chunk {} hashes to {} but its sealed manifest \ + records {}; the artifact on disk is not the one this projection \ + committed", + artifact.path.display(), + hex::encode(observed.0), + hex::encode(artifact.digest.0) + ))); + } + // And against the manifest's own ordered list, so a resolution + // cannot be satisfied by a set of chunks that individually match + // records which were themselves swapped. + let sealed = sealed_digests.get(ordinal as usize).ok_or_else(|| { + StoreError::Corruption(format!( + "committed staged chunk {} has ordinal {ordinal}, beyond the \ + manifest's chunk list", + artifact.path.display() + )) + })?; + if &observed != sealed { + return Err(StoreError::Corruption(format!( + "committed staged chunk {} does not match the digest its manifest \ + binds at ordinal {ordinal}", + artifact.path.display() + ))); + } + let file_bytes = std::fs::metadata(&artifact.path)?.len(); + if file_bytes != artifact.bytes { + return Err(StoreError::Corruption(format!( + "committed staged chunk {} is {file_bytes} bytes but its sealed \ + manifest records {}; a location naming the whole record would name \ + a different span than the one that was certified", + artifact.path.display(), + artifact.bytes + ))); + } + let pinned = PinnedFile::open(artifact.path.clone())?; + observed_set.push(ProjectionArtifact { + path: artifact.path.clone(), + digest: observed, + bytes: file_bytes, + }); + chunks.push((ordinal, chunk, file_bytes, pinned)); + } + // The descriptor's own binding over the whole set, recomputed from + // what is on disk. The per-chunk checks above prove each artifact + // against the sealed record; this proves the *set* against the frame, + // which is the only value the committed transaction actually signed. + let observed_digest = artifact_set_digest(&observed_set); + if observed_digest != expected_set { + return Err(StoreError::Corruption(format!( + "committed staged projection session {} resolves to artifact set {} but \ + its frame commits to {}", + hex::encode(session_id), + hex::encode(observed_digest.0), + hex::encode(expected_set.0) + ))); + } + Ok(chunks) + })?; + + if u32::try_from(read.len()).unwrap_or(u32::MAX) != session.chunk_count { + return Err(StoreError::Corruption(format!( + "committed staged projection session {} resolved {} of {} chunks; a partial \ + chunk set is never exposed", + hex::encode(session_id), + read.len(), + session.chunk_count + ))); + } + + // The configured active-index bounds, the same ones ordinary recovery + // rebuilds under. Sizing this from `max_projection_objects` and a + // synthetic byte limit would have let a resolution admit a projection + // the recovered root cannot hold: nothing requires the active-index + // limits to admit a maximal projection, so the two are independent + // configurations and only one of them governs what a reopen may rebuild. + let mut delta = IndexDelta::from_options(&self.options); + let mut retained = Vec::with_capacity(read.len()); + for (ordinal, chunk, file_bytes, pinned) in read { + let generation = projection_generation(adoption_shard_sequence, ordinal, session_id)?; + let frame_len = u32::try_from(file_bytes).map_err(|_| { + StoreError::Corruption(format!( + "committed staged chunk ordinal {ordinal} of session {} is {file_bytes} \ + bytes, beyond what a location can name", + hex::encode(session_id) + )) + })?; + for object in &chunk.objects { + delta.insert( + IndexKey::new(namespace, object.descriptor.object_id), + IndexLocation { + segment_generation: generation, + // The whole artifact is the certified record. + frame_offset: 0, + frame_len, + object_type: object.descriptor.object_type, + shard_sequence: adoption_shard_sequence, + }, + )?; + } + retained.push(RetainedProjectionArtifact::new( + generation, + crate::roots::ProjectionArtifactFormat::CanonicalStageChunkV1, + pinned, + )); + } + + // Remembered so `notify_recovered` can end the pin as an adoption at + // the position this frame established. It is not durable yet: only a + // `Committed` notification, which follows the physical-state proof, may + // write it into the marker. + { + let mut registry = self.lock(); + if let Some(record) = registry.sessions.get_mut(&session_id) { + record.adopted_at_shard_sequence = Some(adoption_shard_sequence); + } + } + + RecoveredProjectionArtifacts::new( + descriptor.clone(), + Arc::new(delta), + Arc::from([]), + retained.into(), + ) } + /// Finish one transferred session once recovery has proved its physical + /// state. + /// + /// This is the transition `finish` could not make: the process holding the + /// pin stopped without knowing whether its frame reached the journal, and + /// recovery has now either made that frame authoritative or scanned the + /// complete history and proved no such frame exists. + /// + /// * `Committed` ends the pin as an adoption, at the position + /// `resolve_committed` recorded for this session in the same recovery. + /// Without that position cleanup could never prove absence of reference + /// against a root, so a `Committed` notification for a session this + /// recovery did not resolve is refused rather than adopted at a guess. + /// * `ProvedAbsent` means no complete frame names these artifacts and none + /// ever will — recovery has read the whole authoritative history. They are + /// synced-but-unreferenced garbage, which is exactly what the deliverable + /// says recovery treats as invisible, so the session is aborted and its + /// directory reclaimed. + /// + /// **Idempotent**, because recovery repeats every notification on the next + /// attempt if a later one fails: a session already in the state being asked + /// for, or already gone, is success and not a conflict. fn notify_recovered( &self, - _resolution: RecoveredProjectionResolution, + resolution: RecoveredProjectionResolution, ) -> Result<(), StoreError> { - Err(StoreError::NotImplemented( - "ProjectionStaging::notify_recovered — B3 StagingSessions, scope 6.5 \ - deliverable 8 (recovery treatment of unreferenced artifacts)", - )) + let session_id = resolution.session_id; + match resolution.outcome { + RecoveredProjectionOutcome::Committed => { + let adopted_at = { + let registry = self.lock(); + let Some(record) = registry.sessions.get(&session_id) else { + return Ok(()); + }; + if matches!(record.state, StagedSessionState::Adopted) { + return Ok(()); + } + record.adopted_at_shard_sequence.ok_or_else(|| { + StoreError::Corruption(format!( + "staging session {} is reported committed but this recovery never \ + resolved it, so nothing recorded where its adoption committed", + hex::encode(session_id) + )) + })? + }; + let (directory, session) = { + let registry = self.lock(); + let record = registry + .sessions + .get(&session_id) + .ok_or_else(|| unknown_session(session_id))?; + (record.directory.clone(), record.binding.session.clone()) + }; + let durability = Arc::clone(&self.durability); + let counters = Arc::clone(&self.counters); + self.run_maintenance(move || { + adopt_adoption_marker( + &directory, + session_id, + &session, + adopted_at, + &durability, + &counters, + ) + })?; + let mut registry = self.lock(); + if let Some(record) = registry.sessions.get_mut(&session_id) { + record.state = StagedSessionState::Adopted; + } + self.release_reservation_locked(&mut registry, session_id); + drop(registry); + self.counters.sessions_adopted.fetch_add(1, Relaxed); + Ok(()) + } + RecoveredProjectionOutcome::ProvedAbsent => { + { + let mut registry = self.lock(); + let Some(record) = registry.sessions.get_mut(&session_id) else { + return Ok(()); + }; + // The pin is over: recovery proved no frame names these + // artifacts. Returning the record to a reclaimable state is + // what lets `reclaim_session` — which refuses a pinned or + // adopted session on purpose — take it. + record.state = StagedSessionState::Sealed; + record.adopted_at_shard_sequence = None; + } + let directory = { + let registry = self.lock(); + match registry.sessions.get(&session_id) { + Some(record) => record.directory.clone(), + None => return Ok(()), + } + }; + let durability = Arc::clone(&self.durability); + self.run_maintenance(move || remove_adoption_marker(&directory, &durability))?; + self.reclaim_session(session_id)?; + self.counters.sessions_aborted.fetch_add(1, Relaxed); + Ok(()) + } + } } } // --- free helpers --------------------------------------------------------- +/// The band every adopted projection artifact's logical generation lives in. +/// +/// Segments, active tails, and projection artifacts share **one** generation +/// space — `validate_retained_object_sources` inserts all three into a single +/// map and refuses a duplicate — so a projection generation that collided with a +/// segment's would be a `Corruption` at open. Reserving the top bit makes the +/// collision impossible by construction rather than unlikely: segment and tail +/// generations are a counter that advances once per seal, and a store that +/// reached 2^63 seals has arithmetic problems that this constant is not the +/// right place to discover. +/// +/// `projection_generation` refuses anything that would land outside the band, +/// so the disjointness is enforced at the one place generations are minted +/// rather than assumed everywhere they are read. +const PROJECTION_GENERATION_BAND: u64 = 1 << 63; + +/// Bits reserved for the chunk ordinal within a band entry. +/// +/// `max_projection_chunks` is capped at `codec::MAX_CANONICAL_ITEMS`, which is +/// 1,000,000 and therefore fits in 20 bits; 24 leaves room for that ceiling to +/// rise without the mapping silently starting to alias. +const PROJECTION_ORDINAL_BITS: u32 = 24; + +/// The logical generation of one adopted chunk artifact. +/// +/// **Stable and injective in `(adoption frame sequence, chunk ordinal)`**, which +/// is the requirement rather than a convenience. Stable because a sealed index +/// run persists `IndexLocation`s across sessions, so the generation a run names +/// must resolve to the same artifact at the next open — the rule contract review +/// 2026-07-30-A established for segments and the active tail, applied here. +/// Injective because a multi-chunk session has one file per chunk and the root +/// validator correctly refuses two projection files at one generation: deriving +/// from the frame sequence alone would make every multi-chunk adoption +/// unopenable. +/// +/// Both inputs are durable — the frame's own sequence and the ordinal the +/// manifest fixes — so this is a function of committed state and not of +/// anything a session remembers. +fn projection_generation( + adoption_shard_sequence: u64, + ordinal: u32, + session_id: [u8; 16], +) -> Result { + let ordinal_ceiling = 1u64 << PROJECTION_ORDINAL_BITS; + if u64::from(ordinal) >= ordinal_ceiling { + return Err(StoreError::Corruption(format!( + "staged projection session {} has chunk ordinal {ordinal}, beyond the {} the \ + generation mapping can distinguish", + hex::encode(session_id), + ordinal_ceiling - 1 + ))); + } + // The sequence must fit *below* the band bit once shifted, not merely + // survive the shift. A round-trip check is not enough and the boundary is + // exactly one value wide: `1 << 39` shifts to `1 << 63`, which is the band + // bit itself, loses nothing on the way, and round-trips perfectly — and then + // OR-ing the band is a no-op, so it produces the same generation as sequence + // 0 at the same ordinal. One collision, at the one input a shift check + // cannot see. + let sequence_ceiling = 1u64 << (63 - PROJECTION_ORDINAL_BITS); + if adoption_shard_sequence >= sequence_ceiling { + return Err(StoreError::Corruption(format!( + "staged projection session {} was adopted at shard sequence {}, at or beyond the \ + {sequence_ceiling} the projection generation mapping can distinguish", + hex::encode(session_id), + adoption_shard_sequence + ))); + } + let shifted = adoption_shard_sequence << PROJECTION_ORDINAL_BITS; + debug_assert_eq!(shifted & PROJECTION_GENERATION_BAND, 0); + Ok(PROJECTION_GENERATION_BAND | shifted | u64::from(ordinal)) +} + /// Create `//` and sync every directory entry the new /// session directory depends on, outermost first. fn create_session_directory( @@ -2499,6 +3553,264 @@ fn create_session_directory( /// deliverable 5 says artifacts are written by maintenance workers. Taking /// `&self` is what let this drift onto whatever thread happened to hold the /// registry lock. +/// What an adoption marker asserts. One file, one of these, rewritten in place. +#[derive(Copy, Clone, Debug, PartialEq, Eq)] +enum AdoptionMark { + /// A pin is outstanding and this process no longer gets to decide its fate. + Finalizing = 1, + /// A committed transaction references these artifacts. + Adopted = 2, +} + +impl AdoptionMark { + fn code(self) -> u8 { + self as u8 + } + + fn from_code(code: u8) -> Option { + match code { + 1 => Some(Self::Finalizing), + 2 => Some(Self::Adopted), + _ => None, + } + } +} + +/// `mark || final_operation_id || final_operation_digest || adopted_at`. +/// +/// The operation and digest travel with the mark so a reconstruction can check +/// them against the session record beside it. A marker naming a different +/// operation than the session it lives in is not a pin this store issued, and +/// the deliverable's "every different operation/digest rejects" has to survive a +/// restart to mean anything. +/// `adopted_at` is zero while the mark is `Finalizing` and carries the adoption's +/// committed shard sequence once it is `Adopted`. Fixed width either way, so the +/// two marks are the same size and a replacement never changes the file's shape. +const ADOPTION_MARKER_LEN: usize = 1 + 16 + 32 + 8; + +fn write_adoption_marker( + directory: &Path, + session_id: [u8; 16], + mark: AdoptionMark, + operation: [u8; 16], + digest: ObjectId, + durability: &DurabilityCounters, + counters: &StagingCounters, +) -> Result<(), StoreError> { + let bytes = encode_staging_artifact( + StagingArtifactKind::Adoption, + session_id, + &adoption_payload(mark, operation, digest, 0), + ); + write_artifact(directory, ADOPTION_NAME, &bytes, durability, counters)?; + Ok(()) +} + +/// Decode a marker and prove it belongs to the session it was found in. +fn adoption_payload( + mark: AdoptionMark, + operation: [u8; 16], + digest: ObjectId, + adopted_at: u64, +) -> Vec { + let mut payload = Vec::with_capacity(ADOPTION_MARKER_LEN); + payload.push(mark.code()); + payload.extend_from_slice(&operation); + payload.extend_from_slice(&digest.0); + payload.extend_from_slice(&adopted_at.to_le_bytes()); + debug_assert_eq!(payload.len(), ADOPTION_MARKER_LEN); + payload +} + +fn read_adoption_marker( + path: &Path, + session_id: [u8; 16], + session: &ProjectionStageSessionV1, +) -> Result<(AdoptionMark, u64), StoreError> { + let payload = read_staging_artifact(path, StagingArtifactKind::Adoption, session_id)?; + if payload.len() != ADOPTION_MARKER_LEN { + return Err(StoreError::Corruption(format!( + "staging adoption marker {} is {} bytes, not {ADOPTION_MARKER_LEN}", + path.display(), + payload.len() + ))); + } + let mark = AdoptionMark::from_code(payload[0]).ok_or_else(|| { + StoreError::Corruption(format!( + "staging adoption marker {} carries unknown outcome {}", + path.display(), + payload[0] + )) + })?; + let mut operation = [0u8; 16]; + operation.copy_from_slice(&payload[1..17]); + let mut digest = [0u8; 32]; + digest.copy_from_slice(&payload[17..49]); + let mut adopted_at = [0u8; 8]; + adopted_at.copy_from_slice(&payload[49..]); + let adopted_at = u64::from_le_bytes(adopted_at); + if operation != session.final_operation_id || ObjectId(digest) != session.final_operation_digest + { + return Err(StoreError::Corruption(format!( + "staging adoption marker {} names operation {} but its session binds {}", + path.display(), + hex::encode(operation), + hex::encode(session.final_operation_id) + ))); + } + // A `Finalizing` marker carrying a position would be claiming an adoption it + // does not record, which is exactly the confusion the position exists to + // prevent. + if matches!(mark, AdoptionMark::Finalizing) && adopted_at != 0 { + return Err(StoreError::Corruption(format!( + "staging adoption marker {} is still finalizing but records a committed \ + sequence of {adopted_at}", + path.display() + ))); + } + Ok((mark, adopted_at)) +} + +/// Replace this session's `Finalizing` marker with `Adopted`, atomically. +/// +/// A separate path from [`write_artifact`], and deliberately not a relaxation of +/// it. That writer publishes with `rename_noreplace` because every other staging +/// artifact is unique-by-name and must never be overwritten — a chunk or a +/// manifest arriving twice at one name is a fault, not an update. The adoption +/// marker is the one file in a session that legitimately changes, and routing it +/// through a "replace if you like" flag on the shared writer would hand that +/// permission to the artifacts the no-replace rule exists to protect. +/// +/// So the licence to overwrite is bounded by proof rather than by a parameter, +/// exactly once, here: the marker on disk must decode as a staging artifact of +/// this session, carry the operation and digest this session binds, and say +/// `Finalizing`. Anything else is refused with the file untouched. +/// +/// Idempotent on an already-`Adopted` marker. `finish` can be reached twice — +/// a retried outcome is not a second event — and re-adopting what is already +/// adopted has to be a no-op rather than a refusal, or the retry wedges the +/// session in `Finalizing` forever. +fn adopt_adoption_marker( + directory: &Path, + session_id: [u8; 16], + session: &ProjectionStageSessionV1, + committed_shard_sequence: u64, + durability: &DurabilityCounters, + counters: &StagingCounters, +) -> Result<(), StoreError> { + assert!( + ON_MAINTENANCE_WORKER.with(Cell::get), + "staging artifacts are written by maintenance workers (scope 6.5 deliverable 5); \ + {} was offered to a caller thread", + directory.join(ADOPTION_NAME).display() + ); + let final_path = directory.join(ADOPTION_NAME); + match read_adoption_marker(&final_path, session_id, session)? { + (AdoptionMark::Adopted, _) => return Ok(()), + (AdoptionMark::Finalizing, _) => {} + } + + let bytes = encode_staging_artifact( + StagingArtifactKind::Adoption, + session_id, + &adoption_payload( + AdoptionMark::Adopted, + session.final_operation_id, + session.final_operation_digest, + committed_shard_sequence, + ), + ); + + let tmp_path = directory.join(format!("{ADOPTION_NAME}.tmp")); + { + // Through the no-follow funnel. This path publishes by *replacing*, so a + // symlink planted at the temporary name would redirect the write and + // then the rename would publish whatever it pointed at over a marker the + // store still believes it owns. + let Some(mut file) = + crate::sys::open_or_create_regular_truncated_nofollow(&tmp_path, durability)? + else { + return Err(StoreError::UnrecognizedLayout(format!( + "{} is not a regular file; refusing to record an adoption through it", + tmp_path.display() + ))); + }; + let end = crate::sys::write_vectored_all(&mut file, &[IoSlice::new(&bytes)], durability)?; + if end != bytes.len() as u64 { + return Err(StoreError::from(std::io::Error::new( + std::io::ErrorKind::WriteZero, + format!( + "short write recording adoption for staging session {}: wrote {end} of {} \ + bytes; the temporary is left behind and never renamed", + hex::encode(session_id), + bytes.len() + ), + ))); + } + crate::sys::fdatasync(&file, durability)?; + } + // Replacing, and only ever over the marker just proved to be this session's + // outstanding pin. A crash on either side of it leaves a whole, valid marker + // — `Finalizing` before, `Adopted` after — and never a torn one, which is + // why this is a rename and not a fixed-size overwrite in place. + crate::sys::rename_replace(&tmp_path, &final_path)?; + crate::sys::fsync_dir(directory, durability)?; + counters + .artifact_bytes_written + .fetch_add(bytes.len() as u64, Relaxed); + counters.maintenance_artifact_writes.fetch_add(1, Relaxed); + Ok(()) +} + +fn remove_adoption_marker( + directory: &Path, + durability: &DurabilityCounters, +) -> Result<(), StoreError> { + let path = directory.join(ADOPTION_NAME); + match crate::sys::unlink(&path) { + Ok(()) => {} + // Already gone is the outcome this asked for. `finish` must be safe to + // reach twice — a retried definitive failure is not a new event. + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(()), + Err(error) => return Err(StoreError::from(error)), + } + crate::sys::fsync_dir(directory, durability)?; + Ok(()) +} + +/// The descriptor a sealed session installs, derived from its own resolution. +/// +/// One derivation, used by `compose_seal` and by `finalize`, so a reconstructed +/// session cannot produce a descriptor that differs from the one its seal +/// returned. Two constructions of the same value is how a restart starts +/// disagreeing with the session it restarted. +fn install_from_resolution( + resolution: &ProjectionAdoptionResolution, +) -> Result { + let manifest_digest = resolution + .manifest + .manifest_digest() + .map_err(|e| StoreError::Corruption(format!("sealed staging manifest digest: {e}")))?; + let object_count = u64::try_from(resolution.manifest.objects.len()).map_err(|_| { + StoreError::Corruption("sealed staging manifest object count does not fit u64".into()) + })?; + let mut object_bytes = 0u64; + for object in &resolution.manifest.objects { + object_bytes = object_bytes.checked_add(object.raw_len).ok_or_else(|| { + StoreError::Corruption("sealed staging manifest byte total overflowed".into()) + })?; + } + Ok(StagedProjectionInstallV1 { + session_id: resolution.session.session_id, + manifest_digest, + projection: resolution.session.projection, + object_count, + object_bytes, + membership_root: resolution.manifest.membership_root, + artifact_set_digest: artifact_set_digest(&resolution.artifacts), + }) +} + fn write_artifact( directory: &Path, name: &str, @@ -2615,6 +3927,7 @@ fn reclaim_session_directory( fn is_session_artifact_name(name: &str) -> bool { name == SESSION_RECORD_NAME || name == MANIFEST_NAME + || name == ADOPTION_NAME || parse_chunk_artifact_name(name).is_some() } @@ -2653,13 +3966,45 @@ fn unknown_session(session_id: [u8; 16]) -> StoreError { fn require_open(record: &SessionRecord, session_id: [u8; 16]) -> Result<(), StoreError> { match record.state { StagedSessionState::Open => Ok(()), - StagedSessionState::Sealed => Err(StoreError::Conflict(format!( + StagedSessionState::Sealed + | StagedSessionState::Finalizing + | StagedSessionState::Adopted => Err(StoreError::Conflict(format!( "staging session {} is sealed and immutable", hex::encode(session_id) ))), } } +/// Admit the sole finalizer: a session that has sealed and is not already +/// carrying an outcome. +/// +/// Deliverable 6 asks for "identical concurrent finalizers coalesce onto one", +/// and at this layer that is one pin, not two handles. [`ProjectionAdoption`] is +/// an unforgeable capability with exactly one terminal outcome and a `Drop` that +/// reports its absence; there is no second copy to hand a second caller. So the +/// coalescing an instance layer does across retries of one operation appears +/// here as this refusal, which names the pin rather than pretending to issue +/// another. The bound operation/digest half needs no test at all: the binding +/// carries the *sole* final operation and digest, fixed at `begin`, so a +/// different one cannot reach a session in the first place. +fn require_finalizable(record: &SessionRecord, session_id: [u8; 16]) -> Result<(), StoreError> { + match record.state { + StagedSessionState::Sealed => Ok(()), + StagedSessionState::Open => Err(StoreError::Conflict(format!( + "staging session {} has not sealed and has no descriptor to adopt", + hex::encode(session_id) + ))), + StagedSessionState::Finalizing => Err(StoreError::Conflict(format!( + "staging session {} already holds an adoption pin", + hex::encode(session_id) + ))), + StagedSessionState::Adopted => Err(StoreError::Conflict(format!( + "staging session {} has already been adopted", + hex::encode(session_id) + ))), + } +} + /// Exclude a session that is already inside a maintenance operation. /// /// Exhaustive rather than a `!= Idle` test, for the same reason the state match @@ -2672,6 +4017,10 @@ fn require_idle(record: &SessionRecord, session_id: [u8; 16]) -> Result<(), Stor "staging session {} is sealing", hex::encode(session_id) ))), + SessionBusy::Finalizing => Err(StoreError::Conflict(format!( + "staging session {} is taking an adoption pin", + hex::encode(session_id) + ))), SessionBusy::Reclaiming => Err(StoreError::Conflict(format!( "staging session {} is being reclaimed", hex::encode(session_id) @@ -2976,6 +4325,7 @@ mod tests { &self, _namespace: NamespaceId, descriptor: &StagedProjectionInstallV1, + _adoption_shard_sequence: u64, ) -> Result { if self.artifacts.descriptor() != descriptor { return Err(StoreError::Corruption( @@ -3006,7 +4356,9 @@ mod tests { #[test] fn each_terminal_outcome_suppresses_the_drop_bug() { for outcome in [ - ProjectionAdoptionOutcome::Adopted, + ProjectionAdoptionOutcome::Adopted { + committed_shard_sequence: 1, + }, ProjectionAdoptionOutcome::DefinitivePreAppendFailure, ProjectionAdoptionOutcome::TransferredToRecovery, ] { @@ -3033,7 +4385,7 @@ mod tests { let transferred = resolver.transferred_sessions(3).unwrap(); assert_eq!(&*transferred, &[[20; 16], [26; 16]]); let recovered = resolver - .resolve_committed(NamespaceId([24; 32]), &install()) + .resolve_committed(NamespaceId([24; 32]), &install(), 0) .unwrap(); assert_eq!(recovered.descriptor(), &install()); assert_eq!(recovered.index_delta().len(), 1); @@ -3179,6 +4531,1159 @@ mod b3_tests { } } + /// A self-consistent one-chunk projection and the binding that seals to it. + /// + /// [`binding`] carries a placeholder `manifest_digest` and so can never + /// reach `Sealed`, which is why every sealing test lived in the integration + /// file. `finalize` is `pub(crate)` — B1's `adopt_projection` is its only + /// legitimate caller — so its regressions cannot live there, and the fixture + /// has to exist on this side of the wall. + fn sealable(session_id: [u8; 16]) -> (ProjectionStageBinding, ProjectionStageChunkV1) { + use levcs_protocol::v2::StagedChunkObjectV1; + let mut objects: Vec = (0..2u8) + .map(|index| { + let body = [index; 24]; + let mut raw = levcs_core::ObjectHeader { + object_type: levcs_core::ObjectType::Blob, + format_version: levcs_core::FORMAT_VERSION, + body_len: body.len() as u64, + } + .encode() + .to_vec(); + raw.extend_from_slice(&body); + let id = levcs_core::blake3_hash(&raw); + StagedChunkObjectV1 { + descriptor: StagedObjectV1 { + object_id: id, + object_type: levcs_core::ObjectType::Blob as u8, + raw_len: raw.len() as u64, + raw_digest: id, + }, + raw_bytes: raw, + } + }) + .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(|o| o.descriptor.raw_len).sum::(); + let descriptors: Vec = + objects.iter().map(|o| o.descriptor.clone()).collect(); + let chunk = ProjectionStageChunkV1 { + session_id, + ordinal: 0, + chunk_count: 1, + objects, + }; + let membership_root = levcs_core::blake3_hash(&session_id[..]); + let manifest = ProjectionStageManifestV1 { + session_id, + chunk_digests: vec![chunk.chunk_digest().expect("chunk digest")], + objects: descriptors, + membership_root, + }; + let manifest_digest = manifest.manifest_digest().expect("manifest digest"); + + let mut session = binding(session_id, HOUR_MICROS).session; + session.total_object_count = 2; + session.total_object_bytes = total_object_bytes; + session.chunk_count = 1; + session.manifest_digest = manifest_digest; + ( + ProjectionStageBinding { + session, + membership_root, + }, + chunk, + ) + } + + /// Deliverable 6 end to end, and the one test that had to exist first. + /// + /// It catches three things that are only visible together. The pin has to be + /// *takeable*: `finalize` writes a marker that no marker precedes. It has to + /// be *finishable*: recording `Adopted` replaces that marker rather than + /// colliding with it — the shared artifact writer publishes with + /// `rename_noreplace`, so routing the second write through it wedges every + /// adoption in `Finalizing` with an `EEXIST` nobody sees. And the outcome has + /// to be *durable and accounted*: adoption releases the reservation because + /// the bytes are store content now, and the reopen must reconstruct that + /// without charging for them a second time — a budget that shrinks on every + /// restart is a ceiling nobody can reason about. + #[test] + fn an_adopted_pin_survives_the_reopen_without_recharging_its_budget() { + let directory = TempDir::new().unwrap(); + let (options, lock) = layout(&directory); + let (binding, chunk) = sealable([31; 16]); + let session_id = binding.session.session_id; + + let reserved = { + let staging = ProjectionStaging::open( + &lock, + options.clone(), + Arc::new(DurabilityCounters::default()), + ) + .expect("staging opens"); + let session = staging.begin(binding.clone(), 0).expect("begin"); + session.put_chunk(&chunk, 0).expect("put"); + session.seal(0).expect("seal"); + let reserved = staging.counters().snapshot().reserved_bytes; + assert!(reserved > 0, "a sealed session holds budget"); + + let adoption = session.finalize(0).expect("a sealed session may be pinned"); + assert_eq!( + staging.counters().snapshot().sessions_finalized, + 1, + "the pin was taken" + ); + assert_eq!( + describe_state(&staging, session_id), + StagedSessionState::Finalizing + ); + + adoption + .handle + .finish(ProjectionAdoptionOutcome::Adopted { + committed_shard_sequence: 7, + }) + .expect( + "recording adoption must replace the pin's own marker; the shared artifact \ + writer refuses to replace, which would leave every adoption stuck in \ + Finalizing", + ); + assert_eq!( + describe_state(&staging, session_id), + StagedSessionState::Adopted + ); + assert_eq!( + staging.counters().snapshot().reserved_bytes, + 0, + "adopted bytes are store content and stop being staging occupancy" + ); + reserved + }; + + let staging = + ProjectionStaging::open(&lock, options, Arc::new(DurabilityCounters::default())) + .expect("staging reopens"); + assert_eq!( + describe_state(&staging, session_id), + StagedSessionState::Adopted, + "the outcome is durable, or the next open would offer the pin again" + ); + assert_eq!( + staging.counters().snapshot().reserved_bytes, + 0, + "reconstructing an adopted session must not re-charge the {reserved} bytes its \ + adoption released; a ceiling that shrinks on every restart is not a ceiling" + ); + + // The adoption's position has to survive too. A restart separates an + // adoption from the compaction that eventually drops its reference, so a + // position held only in memory would leave every reconstructed session + // either permanently unreclaimable or reclaimable through a stale root — + // the defect this records against. + assert_eq!( + staging + .cleanup_unreferenced(&root_at(&[], 6)) + .expect("cleanup"), + 0, + "a root short of the reconstructed adoption is still not evidence" + ); + assert_eq!(staging.counters().snapshot().cleanup_declined_stale_root, 1); + assert_eq!( + staging + .cleanup_unreferenced(&root_at(&[], 7)) + .expect("cleanup"), + 1, + "and a root that has reached it reclaims, so the position reconstructed as \ + itself rather than as something unreachable" + ); + } + + /// A sealed session, its staging, and the on-disk path of its marker. + /// + /// Returned rather than rebuilt per test because "did the marker actually + /// move" is the question every one of these asks, and a test that asserted + /// only the in-memory state would pass against a pin that never reached the + /// device. + fn sealed( + directory: &TempDir, + lock: &RecoverySession, + options: StoreOptions, + session_id: [u8; 16], + ) -> ( + Arc, + ProjectionStageSession, + PathBuf, + StagedProjectionInstallV1, + ) { + let (binding, chunk) = sealable(session_id); + let shard = StoreOptions::shard_of( + &NamespaceId::from(binding.session.destination_repo), + options.shard_count, + ); + let staging = + ProjectionStaging::open(lock, options, Arc::new(DurabilityCounters::default())) + .expect("staging opens"); + let session = staging.begin(binding, 0).expect("begin"); + session.put_chunk(&chunk, 0).expect("put"); + // Returned rather than discarded: `seal` requires `Open`, so it is the + // only chance to observe the descriptor a later resolution must match. + let install = session.seal(0).expect("seal"); + let marker = directory + .path() + .join(STAGING_DIR) + .join(format!("{shard:02}")) + .join(hex::encode(session_id)) + .join(ADOPTION_NAME); + (staging, session, marker, install) + } + + /// One pin, and the second finalizer is told so by name. + /// + /// Scope §6.5 originally promised that identical concurrent finalizers + /// "coalesce onto one" here. They cannot: a `ProjectionAdoption` is an + /// unforgeable capability with exactly one terminal outcome and a `Drop` that + /// reports its absence, so there is no second copy to hand a second caller. + /// Coalescing retries of one request belongs to whoever owns the request + /// (contract review 2026-07-31-A). What this layer owes is that a second + /// finalizer never produces a second pin and never quietly succeeds. + #[test] + fn a_second_finalizer_is_refused_rather_than_issued_a_second_pin() { + let directory = TempDir::new().unwrap(); + let (options, lock) = layout(&directory); + let (staging, session, _marker, _install) = sealed(&directory, &lock, options, [41; 16]); + + let first = session + .finalize(0) + .expect("the first finalizer takes the pin"); + let Err(StoreError::Conflict(detail)) = session.finalize(0) else { + panic!("a second finalizer must be refused, not handed another pin"); + }; + assert!(detail.contains("already holds an adoption pin"), "{detail}"); + assert_eq!( + staging.counters().snapshot().sessions_finalized, + 1, + "a refused finalizer must not count as a pin" + ); + first + .handle + .finish(ProjectionAdoptionOutcome::DefinitivePreAppendFailure) + .expect("release"); + } + + /// The named acceptance case: expiry may stop a *new* finalizer, never an + /// admitted one. + /// + /// Both halves matter. The sweep must decline the pinned session, and it must + /// leave the artifacts where they are — an expiry that reclaimed here would + /// delete the objects a transaction may already be appending a frame about. + #[test] + fn expiry_after_finalization_reclaims_nothing() { + let directory = TempDir::new().unwrap(); + let (options, lock) = layout(&directory); + let (staging, session, marker, _install) = sealed(&directory, &lock, options, [42; 16]); + let adoption = session.finalize(0).expect("pin"); + + assert_eq!( + staging.expire(HOUR_MICROS + 1).expect("sweep"), + 0, + "a pinned session is not expiry's to reclaim" + ); + assert_eq!( + describe_state(&staging, [42; 16]), + StagedSessionState::Finalizing + ); + assert!( + marker.exists() && marker.parent().expect("session directory").exists(), + "the sweep must leave a pinned session's artifacts on the device" + ); + adoption + .handle + .finish(ProjectionAdoptionOutcome::DefinitivePreAppendFailure) + .expect("release"); + } + + /// The other ordering of the named race: expiry arrives while a finalizer is + /// mid-admission. + /// + /// `expiry_after_finalization_reclaims_nothing` covers the ordering where the + /// marker is already durable and the *state* says `Finalizing`. This is the + /// window before that — the finalizer has been admitted under the registry + /// lock and released it to write the marker, so the session still reads + /// `Sealed` with no marker on disk. An expiry sweep that looked only at state + /// would find an expired, sealed, unpinned session and reclaim it, deleting + /// the artifacts out from under a pin that is about to be issued. + /// + /// What prevents it is the `busy` exclusion, so that is what this drives + /// directly. The session is put in exactly the mid-flight shape rather than + /// raced into it, because a race that reproduces one time in a thousand is a + /// test that passes for the wrong reason the other nine hundred and ninety + /// nine. + /// + /// The second half is what makes it load-bearing: clearing `busy` and + /// sweeping again *does* reclaim. Without that, an assertion that nothing was + /// reclaimed proves only that something declined — not that the exclusion is + /// what declined it. + #[test] + fn expiry_during_the_finalize_admission_window_reclaims_nothing() { + let directory = TempDir::new().unwrap(); + let (options, lock) = layout(&directory); + let (staging, _session, marker, _install) = sealed(&directory, &lock, options, [45; 16]); + let session_directory = marker.parent().expect("session directory").to_path_buf(); + + // The shape `finalize` holds between phase 1 and phase 3. + { + let mut registry = staging.lock(); + let record = registry.sessions.get_mut(&[45; 16]).expect("session"); + assert_eq!(record.state, StagedSessionState::Sealed); + record.busy = SessionBusy::Finalizing; + } + assert!( + !marker.exists(), + "the window under test is the one before the marker is durable" + ); + + assert_eq!( + staging.expire(HOUR_MICROS + 1).expect("sweep"), + 0, + "an expiry sweep must not reclaim a session a finalizer has already been \ + admitted to, even though its state still reads Sealed" + ); + assert!( + session_directory.exists(), + "and it must leave the artifacts the pin is about to cover" + ); + + { + let mut registry = staging.lock(); + registry.sessions.get_mut(&[45; 16]).expect("session").busy = SessionBusy::Idle; + } + assert_eq!( + staging.expire(HOUR_MICROS + 1).expect("sweep"), + 1, + "with the exclusion cleared the same sweep does reclaim, so the decline above \ + was the exclusion and not some other refusal" + ); + } + + /// Nothing was appended, so the pin simply ends — and the session is + /// finalizable again. + /// + /// Refinalization is the assertion that matters. Deliverable 6 says a + /// definitive pre-append failure returns a live session to a state it can be + /// finalized from; a repair that only cleared an in-memory flag, or that left + /// the marker behind, would pass a state check and still refuse the retry. + #[test] + fn a_definitive_pre_append_failure_returns_a_live_session_to_sealed() { + let directory = TempDir::new().unwrap(); + let (options, lock) = layout(&directory); + let (staging, session, marker, _install) = sealed(&directory, &lock, options, [43; 16]); + + let adoption = session.finalize(0).expect("pin"); + assert!(marker.exists(), "the pin is durable before it is issued"); + + adoption + .handle + .finish(ProjectionAdoptionOutcome::DefinitivePreAppendFailure) + .expect("release"); + assert!(!marker.exists(), "the marker goes with the pin"); + assert_eq!( + describe_state(&staging, [43; 16]), + StagedSessionState::Sealed, + "the durable manifest is still the seal's commit point, so Sealed is the only \ + state a restart could reproduce here" + ); + assert_eq!( + staging.counters().snapshot().adoption_pins_released, + 1, + "a release is its own event, distinct from an adoption" + ); + + session + .finalize(0) + .expect("a released session is finalizable again, or the retry has nowhere to go") + .handle + .finish(ProjectionAdoptionOutcome::DefinitivePreAppendFailure) + .expect("release"); + } + + /// The pin outlives the process, which is the only reason `Finalizing` is + /// durable at all. + /// + /// `TransferredToRecovery` means nobody in this process knows whether the + /// final frame reached the journal. The marker therefore stays, the next open + /// reconstructs `Finalizing`, and `transferred_sessions` hands the question + /// to recovery — the answer that method was written to be able to give and + /// could only prove empty while `finalize` was deferred. + #[test] + fn a_transferred_pin_reconstructs_and_is_offered_to_recovery() { + let directory = TempDir::new().unwrap(); + let (options, lock) = layout(&directory); + let session_id = [44; 16]; + let shard = + StoreOptions::shard_of(&NamespaceId::from(ObjectId([7; 32])), options.shard_count); + let marker = { + let (staging, session, marker, _install) = + sealed(&directory, &lock, options.clone(), session_id); + session + .finalize(0) + .expect("pin") + .handle + .finish(ProjectionAdoptionOutcome::TransferredToRecovery) + .expect("transfer"); + assert_eq!(staging.counters().snapshot().adoption_pins_transferred, 1); + marker + }; + assert!( + marker.exists(), + "a transferred pin leaves its marker behind" + ); + + let staging = + ProjectionStaging::open(&lock, options, Arc::new(DurabilityCounters::default())) + .expect("staging reopens"); + assert_eq!( + describe_state(&staging, session_id), + StagedSessionState::Finalizing, + "the pin survived the process, or recovery is never told to resolve it" + ); + assert_eq!( + &*staging.transferred_sessions(shard).expect("transferred"), + &[session_id], + "the session must be offered to recovery for the shard it belongs to" + ); + assert!( + staging + .transferred_sessions(shard ^ 1) + .expect("transferred") + .is_empty(), + "and to no other shard" + ); + } + + /// A committed root that pins `paths` as staged projection artifacts. + /// + /// Built through the real `RetainedProjectionArtifact`, which holds an open + /// descriptor, so a root in a test references a file the same way a root in + /// production does — by owning it, not by naming it. + fn root_referencing(paths: &[PathBuf]) -> CommittedRoot { + root_at(paths, 7) + } + + /// A root pinning `paths` that records no committed sequence for any shard. + /// + /// Not the same as one recording zero. This is the shape a root has for a + /// shard it knows nothing about, and cleanup must treat it as no evidence + /// rather than as evidence of zero. + fn root_without_sequences(paths: &[PathBuf]) -> CommittedRoot { + let mut root = root_at(paths, 0); + root = CommittedRoot::new( + root.repositories().clone(), + crate::roots::LayeredObjectIndex::default(), + root.terminal_statuses().clone(), + crate::roots::ShardSequenceMap::new(), + root.retained_generations().clone(), + ); + root + } + + /// A root pinning `paths` whose shards have committed through `through`. + /// + /// The sequence is not decoration: cleanup skips an adopted session unless + /// the root it is given has reached the adoption, so a root built without one + /// proves nothing and a test using it would assert against a skip rather than + /// against the reference proof. + fn root_at(paths: &[PathBuf], through: u64) -> CommittedRoot { + let artifacts: Vec = paths + .iter() + .enumerate() + .map(|(index, path)| { + RetainedProjectionArtifact::new( + index as u64, + crate::roots::ProjectionArtifactFormat::CanonicalStageChunkV1, + crate::roots::PinnedFile::open(path.clone()).expect("pin the artifact"), + ) + }) + .collect(); + let mut generations = crate::roots::GenerationMap::new(); + generations.insert( + crate::roots::GenerationId::new(0, 0), + Arc::new(crate::roots::RetainedGeneration::new( + crate::roots::GenerationId::new(0, 0), + Vec::new().into(), + Vec::new().into(), + Vec::new().into(), + Vec::new().into(), + artifacts.into(), + )), + ); + // Every shard, because a real committed root carries a sequence per + // shard and the fixture's destination does not land on shard 0. A helper + // that populated only one would make every test measure a stale-root + // skip while claiming to measure the reference proof. + let mut sequences = crate::roots::ShardSequenceMap::new(); + for shard in 0..4u16 { + sequences.insert(shard, through); + } + CommittedRoot::new( + crate::roots::RepoMap::new(), + crate::roots::LayeredObjectIndex::default(), + crate::roots::TerminalStatusMap::new(), + sequences, + generations, + ) + } + + fn session_files(directory: &Path) -> Vec { + let mut paths: Vec = std::fs::read_dir(directory) + .expect("session directory") + .map(|entry| entry.expect("entry").path()) + .collect(); + paths.sort(); + paths + } + + /// Deliverable 7's named acceptance case, and its inverse in the same test + /// so neither half can pass alone. + /// + /// Cleanup declines a session a committed root still references, and the + /// same session against a root that references nothing is reclaimed. Running + /// both against one adopted session is what makes the first assertion mean + /// "the reference proof declined it" rather than "cleanup does nothing here"; + /// a cleanup that reclaimed nothing ever would pass the decline half + /// perfectly. + #[test] + fn cleanup_declines_a_referenced_session_and_reclaims_an_unreferenced_one() { + let directory = TempDir::new().unwrap(); + let (options, lock) = layout(&directory); + let (staging, session, marker, _install) = sealed(&directory, &lock, options, [51; 16]); + let session_directory = marker.parent().expect("session directory").to_path_buf(); + + session + .finalize(0) + .expect("pin") + .handle + .finish(ProjectionAdoptionOutcome::Adopted { + committed_shard_sequence: 7, + }) + .expect("adopt"); + let files = session_files(&session_directory); + assert!( + files.len() >= 3, + "an adopted session keeps its chunk, manifest, and marker: {files:?}" + ); + + // A root holding exactly one of the session's artifacts. That is the + // state immediately after adoption, and it must be enough to decline. + let chunk = files + .iter() + .find(|path| { + path.file_name() + .and_then(|name| name.to_str()) + .is_some_and(|name| parse_chunk_artifact_name(name).is_some()) + }) + .expect("a chunk artifact") + .clone(); + let referencing = root_referencing(&[chunk]); + assert_eq!( + staging + .cleanup_unreferenced(&referencing) + .expect("cleanup runs"), + 0, + "an artifact a committed root still references is not cleanup's to remove" + ); + assert_eq!( + session_files(&session_directory), + files, + "and it must leave every file where it found it, not only the referenced one" + ); + assert_eq!( + describe_state(&staging, [51; 16]), + StagedSessionState::Adopted, + "a declined session stays exactly as it was" + ); + assert_eq!( + staging.counters().snapshot().cleanup_declined_referenced, + 1, + "the decline is the reference proof doing work, and is counted as such" + ); + + // Nothing points at it any more — a later checkpoint or compaction + // dropped the pin — so now it goes. + let empty = root_referencing(&[]); + assert_eq!( + staging.cleanup_unreferenced(&empty).expect("cleanup runs"), + 1, + "an adopted session nothing references is exactly what cleanup exists to reclaim" + ); + assert!( + !session_directory.exists(), + "the directory goes with it, or the next open reconstructs a session that was \ + reclaimed" + ); + assert_eq!(staging.counters().snapshot().sessions_cleaned_up, 1); + } + + /// P1: a stale root proves nothing, and cleanup trusted whatever it was + /// handed. + /// + /// The sequence is the reviewer's. `R0` is captured before the projection is + /// published and therefore references none of its artifacts. `R1` is + /// published referencing them, and the session records `Adopted`. Cleanup is + /// then called with `R0` — an older root that is not wrong, merely earlier — + /// and every artifact in it is absent from that root's pins, so the absence + /// proof succeeds and the directory is deleted out from under a committed + /// root that still points into it. + /// + /// My review note had this backwards: I argued a racing newer root "can only + /// add references", and concluded the proof was safe. Adding references is + /// precisely the hazard — it means an older root omits references a newer one + /// holds, so absence measured against the older root is not absence. + #[test] + fn cleanup_will_not_delete_through_a_root_older_than_the_adoption() { + let directory = TempDir::new().unwrap(); + let (options, lock) = layout(&directory); + let (staging, session, marker, _install) = sealed(&directory, &lock, options, [53; 16]); + let session_directory = marker.parent().expect("session directory").to_path_buf(); + + // 1. A root captured before anything adopted this projection: it holds + // none of its artifacts, and its committed prefix stops short of the + // sequence the adoption frame will reach. + let before = root_at(&[], 6); + + // 2 and 3. The projection is published and the session records it. + let files = session_files(&session_directory); + let referenced = root_at(&files, 7); + session + .finalize(0) + .expect("pin") + .handle + .finish(ProjectionAdoptionOutcome::Adopted { + committed_shard_sequence: 7, + }) + .expect("adopt"); + assert!( + referenced.references_artifact(&files[0]), + "the newer root does reference the artifacts, or this proves nothing" + ); + + // 4. Cleanup with the older root. + let reclaimed = staging + .cleanup_unreferenced(&before) + .expect("cleanup answers"); + assert_eq!( + reclaimed, 0, + "a root older than the adoption cannot prove absence of reference: it predates \ + every reference the adoption created" + ); + assert!( + session_directory.exists(), + "the committed root published at step 2 still points into this directory" + ); + assert_eq!( + staging.counters().snapshot().cleanup_declined_stale_root, + 1, + "and the reason must be the root's age, not an incidental refusal" + ); + + // The same session against a root that *has* reached the adoption and + // references nothing is reclaimable, so the skip above is the position + // check and not cleanup declining everything. + assert_eq!( + staging + .cleanup_unreferenced(&root_at(&[], 7)) + .expect("cleanup"), + 1 + ); + } + + /// A committed projection resolves to complete membership, and a partial + /// chunk set never resolves at all. + /// + /// The second half is the deliverable's central claim and the one an + /// implementation drifts away from by being helpful: resolving whatever + /// chunks are present would publish a smaller projection than the frame + /// committed, and every object in the missing chunk would be unreadable + /// through a root that says it is there. + #[test] + fn a_committed_projection_resolves_whole_or_not_at_all() { + let directory = TempDir::new().unwrap(); + let (options, lock) = layout(&directory); + let (staging, _session, marker, install) = sealed(&directory, &lock, options, [63; 16]); + let session_directory = marker.parent().expect("session directory").to_path_buf(); + let namespace = NamespaceId::from(ObjectId([7; 32])); + + let resolved = staging + .resolve_committed(namespace, &install, 9) + .expect("a sealed session resolves its own descriptor"); + assert_eq!(resolved.descriptor(), &install); + assert_eq!( + resolved.retained_artifacts().len(), + 1, + "one pinned artifact per chunk" + ); + let generation = projection_generation(9, 0, [63; 16]).expect("in range"); + assert_eq!( + resolved.retained_artifacts()[0].logical_generation, + generation, + "the artifact is pinned at the generation its locations name" + ); + assert_eq!( + resolved.index_delta().len() as u64, + install.object_count, + "every object the descriptor declares is located, or the root would publish \ + membership it cannot resolve" + ); + + // Every object in a chunk shares that chunk's location: the location + // names the whole certified record, and a reader validates the artifact + // before extracting from its decoded vector. + let file_bytes = std::fs::metadata( + &session_files(&session_directory) + .into_iter() + .find(|path| { + path.file_name() + .and_then(|name| name.to_str()) + .is_some_and(|name| parse_chunk_artifact_name(name).is_some()) + }) + .expect("a chunk artifact"), + ) + .expect("metadata") + .len(); + for (_, location) in resolved.index_delta().iter() { + assert_eq!(location.segment_generation, generation); + assert_eq!(location.frame_offset, 0); + assert_eq!(u64::from(location.frame_len), file_bytes); + } + + // Now take one chunk away. Nothing partial may resolve. + let chunk = session_files(&session_directory) + .into_iter() + .find(|path| { + path.file_name() + .and_then(|name| name.to_str()) + .is_some_and(|name| parse_chunk_artifact_name(name).is_some()) + }) + .expect("a chunk artifact"); + std::fs::remove_file(&chunk).expect("remove one chunk"); + let refused = staging + .resolve_committed(namespace, &install, 9) + .expect_err("a missing chunk must fail the whole resolution"); + assert!( + matches!(refused, StoreError::Corruption(_) | StoreError::Io(_)), + "expected a named fault, got {refused:?}" + ); + } + + /// Resolution obeys the active-index ceilings, not a limit of its own. + /// + /// `max_projection_objects` bounds what a *session* may stage; + /// `max_active_index_entries` and `max_active_index_bytes` bound what a + /// reopen may rebuild. Nothing requires the second pair to admit a maximal + /// projection, so a resolution sized from the first would hand recovery a + /// delta the recovered root cannot hold — publishing an over-limit root + /// instead of refusing, which is the hole admission accounting exists to + /// keep shut. + /// + /// Both ceilings are driven separately because they are separate refusals: a + /// projection can be under one and over the other. + #[test] + fn resolution_refuses_a_projection_over_the_active_index_ceilings() { + for (limit, adjust) in [ + ( + "max_active_index_entries", + Box::new(|options: &mut StoreOptions| options.max_active_index_entries = 1) + as Box, + ), + ( + "max_active_index_bytes", + Box::new(|options: &mut StoreOptions| { + options.max_active_index_bytes = crate::index::encoded_bytes_for(1, 1) + }), + ), + ] { + let directory = TempDir::new().unwrap(); + let (mut options, lock) = layout(&directory); + adjust(&mut options); + let (staging, _session, _marker, install) = + sealed(&directory, &lock, options, [68; 16]); + assert_eq!( + install.object_count, 2, + "the fixture must exceed a ceiling of one, or this asserts nothing" + ); + + let refused = staging + .resolve_committed(NamespaceId::from(ObjectId([7; 32])), &install, 9) + .expect_err("a projection over the active-index ceiling must not resolve"); + match refused { + StoreError::LimitExceeded { limit: named, .. } => assert_eq!( + named, limit, + "the refusal must name the ceiling that stopped it" + ), + other => panic!("expected {limit}, got {other:?}"), + } + } + } + + /// A swapped chunk of the same shape must not resolve. + /// + /// Shape is not identity. A replacement chunk carrying the same session ID, + /// ordinal, chunk count and object count satisfies every structural check a + /// resolution can make, and carries entirely different objects — so a + /// resolution that verified the descriptor against its own cached sealed + /// state, rather than against the bytes it just read, would index the + /// impostor's objects under a committed frame that never named them. That is + /// what this resolution did before review caught it. + #[test] + fn a_chunk_swapped_for_another_of_the_same_shape_is_refused() { + let directory = TempDir::new().unwrap(); + let (options, lock) = layout(&directory); + let (staging, _session, marker, install) = sealed(&directory, &lock, options, [66; 16]); + let session_directory = marker.parent().expect("session directory").to_path_buf(); + let namespace = NamespaceId::from(ObjectId([7; 32])); + + let chunk_path = session_files(&session_directory) + .into_iter() + .find(|path| { + path.file_name() + .and_then(|name| name.to_str()) + .is_some_and(|name| parse_chunk_artifact_name(name).is_some()) + }) + .expect("a chunk artifact"); + + // A chunk claiming this session's identity: same session ID, same + // ordinal, same chunk count, same object count. Only the object bytes + // differ, which is exactly the difference no structural check can see. + // + // Built inline rather than from `sealable`, because that fixture derives + // object bodies from the index alone — a second session produces + // byte-identical objects, and an "impostor" equal to the original tests + // nothing at all. + use levcs_protocol::v2::StagedChunkObjectV1; + let mut objects: Vec = (0..2u8) + .map(|index| { + let body = [index.wrapping_add(0x80); 24]; + let mut raw = levcs_core::ObjectHeader { + object_type: levcs_core::ObjectType::Blob, + format_version: levcs_core::FORMAT_VERSION, + body_len: body.len() as u64, + } + .encode() + .to_vec(); + raw.extend_from_slice(&body); + let id = levcs_core::blake3_hash(&raw); + StagedChunkObjectV1 { + descriptor: StagedObjectV1 { + object_id: id, + object_type: levcs_core::ObjectType::Blob as u8, + raw_len: raw.len() as u64, + raw_digest: id, + }, + raw_bytes: raw, + } + }) + .collect(); + objects.sort_by(|left, right| left.descriptor.cmp(&right.descriptor)); + let impostor = ProjectionStageChunkV1 { + session_id: [66; 16], + ordinal: 0, + chunk_count: 1, + objects, + }; + assert_ne!( + impostor.chunk_digest().expect("digest"), + ProjectionStageChunkV1::decode_canonical( + &read_staging_artifact(&chunk_path, StagingArtifactKind::Chunk, [66; 16]) + .expect("read the original") + ) + .expect("decode") + .chunk_digest() + .expect("digest"), + "the replacement must actually differ, or this test proves nothing" + ); + let bytes = encode_staging_artifact( + StagingArtifactKind::Chunk, + [66; 16], + &impostor.encode_canonical().expect("encode"), + ); + std::fs::write(&chunk_path, &bytes).expect("swap the chunk"); + + let refused = staging + .resolve_committed(namespace, &install, 9) + .expect_err("a chunk of the right shape but the wrong content must not resolve"); + let StoreError::Corruption(detail) = refused else { + panic!("expected a named corruption, got {refused:?}"); + }; + assert!( + detail.contains("is not the one this projection committed") + || detail.contains("does not match the digest its manifest binds"), + "the refusal must name the digest mismatch rather than some incidental \ + difference: {detail}" + ); + } + + /// The two answers recovery can bring back, and both are idempotent. + /// + /// `Committed` ends a transferred pin as an adoption at the position the + /// resolution established. `ProvedAbsent` means recovery read the complete + /// authoritative history and no frame names these artifacts — they are + /// synced-but-unreferenced garbage, which is what the deliverable says + /// recovery treats as invisible, so the session goes. + #[test] + fn recovery_notifications_end_a_transferred_pin_either_way() { + let directory = TempDir::new().unwrap(); + let (options, lock) = layout(&directory); + let namespace = NamespaceId::from(ObjectId([7; 32])); + + // Committed. + { + let (staging, session, _marker, install) = + sealed(&directory, &lock, options.clone(), [64; 16]); + session + .finalize(0) + .expect("pin") + .handle + .finish(ProjectionAdoptionOutcome::TransferredToRecovery) + .expect("transfer"); + staging + .resolve_committed(namespace, &install, 11) + .expect("resolve"); + for _ in 0..2 { + staging + .notify_recovered(RecoveredProjectionResolution { + session_id: [64; 16], + outcome: RecoveredProjectionOutcome::Committed, + }) + .expect("notification is repeated when a later one fails, so it repeats"); + } + assert_eq!( + describe_state(&staging, [64; 16]), + StagedSessionState::Adopted + ); + // The position the resolution established is what cleanup will later + // measure a root against. + assert_eq!( + staging + .cleanup_unreferenced(&root_at(&[], 10)) + .expect("cleanup"), + 0, + "a root short of the adoption is not evidence" + ); + } + + // Proved absent. + { + let (staging, session, marker, _install) = sealed(&directory, &lock, options, [65; 16]); + let session_directory = marker.parent().expect("session directory").to_path_buf(); + session + .finalize(0) + .expect("pin") + .handle + .finish(ProjectionAdoptionOutcome::TransferredToRecovery) + .expect("transfer"); + for _ in 0..2 { + staging + .notify_recovered(RecoveredProjectionResolution { + session_id: [65; 16], + outcome: RecoveredProjectionOutcome::ProvedAbsent, + }) + .expect("idempotent"); + } + assert!( + !session_directory.exists(), + "artifacts no complete frame names are invisible garbage and are reclaimed" + ); + } + } + + /// The generation mapping is injective and stays inside its band. + /// + /// Injectivity is the requirement a multi-chunk session imposes: one file per + /// chunk, and the root validator refuses two projection files at one + /// generation, so deriving from the adoption's frame sequence alone would + /// make every multi-chunk adoption fail to open. Disjointness from segments + /// and tails is the other half — they share one generation space, and a + /// collision there is a `Corruption` at open rather than a silent + /// misresolution, but only because something refuses it. + #[test] + fn projection_generations_are_injective_and_stay_in_their_band() { + let session = [61; 16]; + let mut seen = std::collections::BTreeSet::new(); + for sequence in [0u64, 1, 2, 4095, 1 << 32] { + for ordinal in [0u32, 1, 2, 999_999] { + let generation = + projection_generation(sequence, ordinal, session).expect("in range"); + assert!( + generation >= PROJECTION_GENERATION_BAND, + "generation {generation} escaped the band reserved against segment and \ + tail generations" + ); + assert!( + seen.insert(generation), + "({sequence}, {ordinal}) collided with an earlier pair; two projection \ + files at one generation make the root unopenable" + ); + } + } + + // Stability is the other property a sealed run depends on: the same pair + // must map to the same number in a later session. + assert_eq!( + projection_generation(7, 3, session).expect("in range"), + projection_generation(7, 3, session).expect("in range"), + ); + + // And the mapping refuses rather than aliases when either input leaves + // the range it can distinguish. + assert!(projection_generation(0, 1 << PROJECTION_ORDINAL_BITS, session).is_err()); + assert!(projection_generation(u64::MAX, 0, session).is_err()); + assert!(projection_generation(1 << 40, 0, session).is_err()); + } + + /// The sequence boundary is one value wide, and a shift check cannot see it. + /// + /// `1 << 39` shifts left by the ordinal width onto the band bit itself. It + /// loses no bits, so a round-trip check accepts it — and then OR-ing the band + /// is a no-op, so it lands on exactly the generation sequence 0 produces at + /// the same ordinal. Two adopted projections at one generation make the root + /// unopenable, and this is the one input that reaches that state through a + /// check designed to prevent it. + /// + /// My first version tested `1 << 40` and believed it covered this. It does + /// not: that value fails because bits shift off the top, which is a different + /// mechanism reached from the far side of the boundary. An out-of-range case + /// is not a boundary case. + #[test] + fn the_sequence_boundary_refuses_the_value_that_would_alias_sequence_zero() { + let session = [62; 16]; + let boundary = 1u64 << (63 - PROJECTION_ORDINAL_BITS); + + for ordinal in [0u32, 1, 999_999] { + assert!( + projection_generation(boundary, ordinal, session).is_err(), + "sequence {boundary} shifts onto the band bit and aliases sequence 0" + ); + let below = projection_generation(boundary - 1, ordinal, session) + .expect("the value below the boundary is still representable"); + let zero = projection_generation(0, ordinal, session).expect("sequence zero"); + assert_ne!( + below, zero, + "the last representable sequence must not alias sequence 0 either" + ); + assert!(below >= PROJECTION_GENERATION_BAND); + } + } + + /// The zero boundary: `Some(0)` is evidence and `None` is not. + /// + /// Zero is a valid committed shard sequence — a shard that has committed its + /// first frame and nothing since — so a root recording `Some(0)` for the + /// shard has genuinely reached an adoption at zero. A root recording nothing + /// for that shard has not said anything at all, and reading it as zero is how + /// the position check quietly stops being a check for exactly the adoption + /// that needs it most: the earliest one. + /// + /// Both halves are asserted against one session, because the failure this + /// guards is the two answers becoming the same. A test that only checked the + /// absent case would also pass against a cleanup that declined every root. + #[test] + fn an_adoption_at_sequence_zero_needs_a_root_that_says_so() { + let directory = TempDir::new().unwrap(); + let (options, lock) = layout(&directory); + let (staging, session, marker, _install) = sealed(&directory, &lock, options, [54; 16]); + let session_directory = marker.parent().expect("session directory").to_path_buf(); + + session + .finalize(0) + .expect("pin") + .handle + .finish(ProjectionAdoptionOutcome::Adopted { + committed_shard_sequence: 0, + }) + .expect("adopt"); + + assert_eq!( + staging + .cleanup_unreferenced(&root_without_sequences(&[])) + .expect("cleanup"), + 0, + "a root holding no sequence for this shard is silent about it, not a witness \ + that it has committed through zero" + ); + assert!(session_directory.exists()); + assert_eq!( + staging.counters().snapshot().cleanup_declined_stale_root, + 1, + "and the decline must be the position check, not a referenced artifact" + ); + + assert_eq!( + staging + .cleanup_unreferenced(&root_at(&[], 0)) + .expect("cleanup"), + 1, + "an explicit zero has reached an adoption at zero, so the same session is \ + reclaimable and the decline above was about the absence and not the value" + ); + assert!(!session_directory.exists()); + } + + /// Cleanup is for adopted sessions and nothing else. + /// + /// A sealed session and a pinned one are both unreferenced by any root — + /// nothing has adopted them — so a cleanup that proved absence of reference + /// and stopped there would delete a session a client is still uploading to, + /// and one whose adoption frame may be mid-append. The reference proof is + /// necessary and is not sufficient. + #[test] + fn cleanup_leaves_live_and_pinned_sessions_alone() { + let directory = TempDir::new().unwrap(); + let (options, lock) = layout(&directory); + let (staging, session, marker, _install) = + sealed(&directory, &lock, options.clone(), [52; 16]); + let sealed_directory = marker.parent().expect("session directory").to_path_buf(); + + let empty = root_referencing(&[]); + assert_eq!( + staging.cleanup_unreferenced(&empty).expect("cleanup runs"), + 0, + "a sealed session is expiry's and abort's, not cleanup's" + ); + assert!(sealed_directory.exists()); + + let adoption = session.finalize(0).expect("pin"); + assert_eq!( + staging.cleanup_unreferenced(&empty).expect("cleanup runs"), + 0, + "a pinned session has no absence to prove: the frame naming its artifacts may \ + be mid-append" + ); + assert!(sealed_directory.exists()); + assert_eq!( + describe_state(&staging, [52; 16]), + StagedSessionState::Finalizing + ); + adoption + .handle + .finish(ProjectionAdoptionOutcome::DefinitivePreAppendFailure) + .expect("release"); + } + + fn describe_state( + staging: &Arc, + session_id: [u8; 16], + ) -> StagedSessionState { + staging + .session(session_id) + .expect("the session is known") + .describe() + .expect("describe") + .state + } + #[test] fn cross_device_staging_is_refused_at_session_creation() { let directory = TempDir::new().unwrap(); @@ -3303,11 +5808,18 @@ mod b3_tests { assert!(is_session_artifact_name(&name)); assert!(is_session_artifact_name(SESSION_RECORD_NAME)); assert!(is_session_artifact_name(MANIFEST_NAME)); + assert!(is_session_artifact_name(ADOPTION_NAME)); assert!(!is_session_artifact_name("not-ours")); } + /// An unsealed session has no descriptor, so there is nothing to pin. + /// + /// The first thing `finalize` must not do is issue a capability against a + /// session that has not reached its own commit point: the manifest is what + /// an adopter revalidates against, and a pin without one is an adoption + /// capability naming state that does not exist. #[test] - fn finalize_is_deferred_and_names_its_deliverable() { + fn an_unsealed_session_cannot_be_finalized() { let directory = TempDir::new().unwrap(); let (options, lock) = layout(&directory); let staging = @@ -3316,25 +5828,32 @@ mod b3_tests { let session = staging .begin(binding([3; 16], HOUR_MICROS), 0) .expect("session"); - let result = session.finalize(0); - let Err(StoreError::NotImplemented(detail)) = result else { - panic!("finalize must be an explicit stub, not a plausible default"); + let Err(StoreError::Conflict(detail)) = session.finalize(0) else { + panic!("finalizing an open session must be refused, not defaulted"); }; - assert!(detail.contains("deliverable 6"), "{detail}"); + assert!(detail.contains("has not sealed"), "{detail}"); + assert_eq!( + staging.counters().snapshot().sessions_finalized, + 0, + "a refused finalize must not count as one" + ); } #[test] - fn the_recovery_resolver_seam_is_deferred_and_names_its_deliverable() { + fn the_recovery_resolver_seam_answers_rather_than_deferring() { let directory = TempDir::new().unwrap(); let (options, lock) = layout(&directory); let staging = ProjectionStaging::open(&lock, options, Arc::new(DurabilityCounters::default())) .expect("staging opens"); - // `transferred_sessions` *is* answered: no session can reach the - // transferred state until deliverable 6 exists, and a sealed session - // is not a transferred one. The proof that this is derived and not a - // constant is the exhaustive state match it is computed from. + // `transferred_sessions` *is* answered, and answers emptily here for the + // reason it always did: a session that has not been finalized has no pin + // to transfer. Deliverable 6 made the non-empty answer reachable — see + // `a_transferred_pin_reconstructs_and_is_offered_to_recovery` — so this + // asserts the negative case rather than an impossibility. The proof that + // it is derived and not a constant is the exhaustive state match it is + // computed from. let session = staging .begin(binding([9; 16], HOUR_MICROS), 0) .expect("session"); @@ -3342,6 +5861,11 @@ mod b3_tests { assert!(staging.transferred_sessions(shard).unwrap().is_empty()); drop(session); + // Both methods answer now, and both refuse a session this root does not + // hold rather than inventing one. `Corruption` and not `NotImplemented`: + // a descriptor naming an unknown session means a committed frame + // published objects whose artifacts are not here, which is a fault about + // the store rather than about this build. let resolved = staging.resolve_committed( NamespaceId([1; 32]), &StagedProjectionInstallV1 { @@ -3353,19 +5877,28 @@ mod b3_tests { membership_root: ObjectId([6; 32]), artifact_set_digest: ObjectId([7; 32]), }, + 3, ); - let Err(StoreError::NotImplemented(detail)) = resolved else { - panic!("resolve_committed must not answer before deliverable 8"); + let Err(StoreError::Corruption(detail)) = resolved else { + panic!("resolving an unheld session must be a named fault"); }; - assert!(detail.contains("deliverable 8"), "{detail}"); + assert!(detail.contains("this root does not hold"), "{detail}"); - let notified = staging.notify_recovered(RecoveredProjectionResolution { - session_id: [4; 16], - outcome: RecoveredProjectionOutcome::Committed, - }); - let Err(StoreError::NotImplemented(detail)) = notified else { - panic!("notify_recovered must not silently succeed"); - }; - assert!(detail.contains("deliverable 8"), "{detail}"); + // Notification is idempotent, and a session this root does not hold is + // the terminal case of that: recovery repeats every notification on the + // next attempt, so "already gone" has to be success rather than a + // conflict that would keep the shard unready forever. + staging + .notify_recovered(RecoveredProjectionResolution { + session_id: [4; 16], + outcome: RecoveredProjectionOutcome::Committed, + }) + .expect("notifying an unheld session is a no-op, not a fault"); + staging + .notify_recovered(RecoveredProjectionResolution { + session_id: [4; 16], + outcome: RecoveredProjectionOutcome::ProvedAbsent, + }) + .expect("and so is proving one absent"); } } diff --git a/crates/levcs-store/src/sys.rs b/crates/levcs-store/src/sys.rs index 676c318..04b7499 100644 --- a/crates/levcs-store/src/sys.rs +++ b/crates/levcs-store/src/sys.rs @@ -340,7 +340,7 @@ pub(crate) fn rename_noreplace(from: &Path, to: &Path) -> io::Result<()> { .map_err(|e| io::Error::from_raw_os_error(e.raw_os_error())) } -/// Plain atomic replacing rename. Two legitimate call sites, and no others: +/// Plain atomic replacing rename. Three legitimate call sites, and no others: /// /// 1. The `CURRENT.tmp` -> `CURRENT` pointer install of scope 3.5, which by /// definition replaces. @@ -348,8 +348,14 @@ pub(crate) fn rename_noreplace(from: &Path, to: &Path) -> io::Result<()> { /// install left finalized at a name the retry cannot choose again — and only /// over an occupant that call has proved reconciles with its replacement. /// Contract review 2026-07-30-F; see [`rename_noreplace`]. +/// 3. `staging::adopt_adoption_marker`, moving one session's adoption marker +/// from `Finalizing` to `Adopted` — the one file in a staging session that +/// legitimately changes — and only after proving the marker on disk is that +/// session's own outstanding pin. Contract review 2026-07-31-A. /// -/// Every other site uses `rename_noreplace`. +/// The pattern in 2 and 3 is the same and is the only one that earns this call: +/// the licence to overwrite is bounded by a proof about the occupant, not by a +/// flag on a shared writer. Every other site uses `rename_noreplace`. pub(crate) fn rename_replace(from: &Path, to: &Path) -> io::Result<()> { std::fs::rename(from, to) } diff --git a/crates/levcs-store/tests/staging_sessions.rs b/crates/levcs-store/tests/staging_sessions.rs index 1370ed3..caf6bbe 100644 --- a/crates/levcs-store/tests/staging_sessions.rs +++ b/crates/levcs-store/tests/staging_sessions.rs @@ -867,15 +867,43 @@ fn abort_releases_the_budget_and_removes_every_artifact() { .expect("released budget is reusable"); } +/// Cleanup on a root with live sessions and nothing adopted reclaims nothing — +/// and says so by succeeding. +/// +/// The distinction this asserts is the one the deferred stub used to stand for: +/// "there was nothing to do" and "I cannot answer yet" are different results, +/// and only one of them is true now. Reaching it through the public surface also +/// pins that a caller can run cleanup against a root that references nothing at +/// all without it becoming a licence to delete the sessions in flight — which is +/// exactly what a reference proof alone, unqualified by state, would do. #[test] -fn cleanup_is_deferred_and_names_its_deliverable() { +fn cleanup_reclaims_nothing_when_no_session_has_been_adopted() { let root = Root::new(); let (staging, _durability) = root.open(); - let result = staging.cleanup_unreferenced(&CommittedRoot::default()); - let Err(StoreError::NotImplemented(detail)) = result else { - panic!("cleanup must be an explicit stub, not a silent success"); - }; - assert!(detail.contains("deliverable 7"), "{detail}"); + let fixture = projection([26; 16], [21; 32], 1, 2, HOUR_MICROS); + let session = staging.begin(fixture.binding.clone(), 0).expect("begin"); + for chunk in &fixture.chunks { + session.put_chunk(chunk, 0).expect("put"); + } + session.seal(0).expect("seal"); + + assert_eq!( + staging + .cleanup_unreferenced(&CommittedRoot::default()) + .expect("cleanup answers rather than deferring"), + 0, + "no session has been adopted, so there is nothing for the reference proof to remove" + ); + assert_eq!( + staging + .session(fixture.session_id()) + .expect("still there") + .describe() + .expect("describe") + .state, + StagedSessionState::Sealed, + "and a sealed session survives a cleanup against a root that references nothing" + ); } // --------------------------------------------------------------------------- diff --git a/doc/instance-throughput-rewrite-plan.md b/doc/instance-throughput-rewrite-plan.md index 79f3764..c8669ec 100644 --- a/doc/instance-throughput-rewrite-plan.md +++ b/doc/instance-throughput-rewrite-plan.md @@ -557,7 +557,7 @@ The identity crate gains a repository-scoped `VerificationSession` over an objec Large initial projections use one bounded `ProjectionStageSessionV1`; `PushKindV2::Normal` never does. The initiator generates the random session ID before computing any final operation/stable digest that contains `ProjectionStageRefV1`. Session creation verifies that `/staging` and its target shard are on the same `st_dev` and canonically binds that session ID, destination repo/genesis and expected empty/current state, projection, authenticated source kind and actor/key epoch, source snapshot generation or `ForkProofV2`, final operation ID/stable digest/evidence digest, total object/byte/chunk counts, ordered manifest digest, expiry, and hard per-session/principal/global staging budgets. Every numbered chunk has a canonical digest and bounded object/byte count; upload validates framing, embedded types, IDs, exact bytes, canonical order, declared manifest position, and same-session ordinal/digest idempotency before a maintenance worker writes and syncs a uniquely named unreferenced artifact. Chunks never enter namespace membership, object-existence answers, snapshots, refs, receipts, event feeds, dedupe state, or `CURRENT`; cross-device adoption and copy fallback are forbidden. -Finalize atomically moves the session from `Open` to `Finalizing` for its sole bound operation/digest and takes an adoption pin; identical concurrent finalizers coalesce and every different operation/digest rejects. Expiry prevents a new finalizer but cannot delete artifacts held by an admitted bounded finalizer. Finalize requires every chunk, reconstructs the exact ordered manifest, and runs the same complete `ProjectionCore`, identity, authority, policy, source-snapshot/Fork-proof, and destination precondition validation used by inline ingestion. It builds and syncs an immutable staged object/index generation and then submits one ordinary bounded transaction with the appropriate `ClientV2`, `MirrorSnapshotV1`, or authenticated network-migration administrative evidence plus a `StagedProjectionInstallV1` descriptor. The shard owner rechecks session/operation/digest, destination lifecycle/CAS, source cursor, config/policy epoch, manifest and artifact hashes; assigns manifest-visible names; directory-syncs them; and appends a small final frame that binds the complete manifest/membership root and resulting state digest. A definitive pre-append failure releases the adoption pin and returns the session to `Open` only if it remains live; otherwise cleanup aborts it. Only the final frame's fence and committed-root swap atomically grant membership and publish refs/authority/cursor/receipt/event. Recovery treats synced-but-unreferenced artifacts as invisible garbage and a complete final frame as authoritative adoption; it can never expose a partial chunk set. +Finalize atomically moves the session from `Sealed` to `Finalizing` for its sole bound operation/digest and takes an adoption pin; every different operation/digest rejects. **Amended by contract review 2026-07-31-A** in two places, and the superseded wording is not reachable: the pin is taken from `Sealed` rather than `Open`, because the sealed manifest is what an adopter revalidates against and a pin on a session without one names state that does not exist; and *coalescing identical concurrent finalizers is the request owner's, not the store's* — an adoption pin is an unforgeable capability with exactly one terminal outcome, so there is no second copy to hand a second caller, and the store's guarantee is one pin with a second finalizer refused by name. Expiry prevents a new finalizer but cannot delete artifacts held by an admitted bounded finalizer. Finalize requires every chunk, reconstructs the exact ordered manifest, and runs the same complete `ProjectionCore`, identity, authority, policy, source-snapshot/Fork-proof, and destination precondition validation used by inline ingestion. It builds and syncs an immutable staged object/index generation and then submits one ordinary bounded transaction with the appropriate `ClientV2`, `MirrorSnapshotV1`, or authenticated network-migration administrative evidence plus a `StagedProjectionInstallV1` descriptor. The shard owner rechecks session/operation/digest, destination lifecycle/CAS, source cursor, config/policy epoch, manifest and artifact hashes; assigns manifest-visible names; directory-syncs them; and appends a small final frame that binds the complete manifest/membership root and resulting state digest. A definitive pre-append failure releases the adoption pin and returns the session to `Sealed` only if it remains live; otherwise cleanup aborts it. **Amended by 2026-07-31-A**: `Open` is not a state a restart can reproduce for a session whose sealed manifest is durable — reconstruction reads that manifest's presence as the seal's own commit point — and returning a session to an unreconstructable state is the defect the private busy/state split exists to prevent. `Sealed` is still finalizable, which is the property the rule is about. The full state machine is `Open → Sealed → Finalizing → {Sealed, Adopted}`, all four durable. Only the final frame's fence and committed-root swap atomically grant membership and publish refs/authority/cursor/receipt/event. Recovery treats synced-but-unreferenced artifacts as invisible garbage and a complete final frame as authoritative adoption; it can never expose a partial chunk set. Sessions are restartable by ID and chunk digest, have no renewal beyond their advertised maximum, and are aborted on authentication mismatch, quota/debt/low-space breach, explicit cancellation, or expiry. A remote source reserves a corresponding bounded snapshot-export base/request lease for the signed generation; each source chunk is served from that generation, and source-lease expiry aborts the destination session rather than mixing generations. Configured maximum projection size/session age and the supported minimum transfer rate must make one complete transfer feasible; otherwise session creation rejects before pinning. Cleanup removes only artifacts carrying a valid session marker after proving that no committed manifest references them, then syncs affected directories. Compaction/GC pins a finalized session from shard adoption through root publication and otherwise may reclaim expired unreferenced sessions. Admission accounts staged bytes, objects, files, sessions, validation work, age, and compaction debt independently of ordinary receive spools. Full-fork and network-migration clients use the v2 staging routes; an in-process mirror uses the identical codecs/service API without loopback HTTP. @@ -1676,6 +1676,275 @@ B4 surface this commit touches — the emitter's remaining claims are unchanged. **Next, and in this dispatch:** recovery discarding an index run whose covered identity was not preserved, which turns 2026-07-30-B's refusal into reclamation. +##### Contract review 2026-07-31-D + +B3 deliverable 8 — recovery treats synced-but-unreferenced artifacts as invisible garbage and a +complete final frame as authoritative adoption. With this, deliverables 6, 7 and 8 are complete. + +**Frozen-seam amendment: `ProjectionRecoveryResolver::resolve_committed` gains +`adoption_shard_sequence`.** Recovery passes `frame.facts.shard_sequence` from the complete +replayed adoption frame. Required, never optional or inferred: every artifact's logical generation +derives from it, and staging cannot supply it — a transferred `Finalizing` session has no durable +position, which is exactly the state being resolved, and that frame is the sole authority that +creates one. This is the **recovery-direction counterpart** of the D0-B +`ProjectionAdoptionOutcome::Adopted` amendment in 2026-07-31-C: the same value, granted for the +same reason, flowing from whichever side made the frame authoritative. Both resolver test doubles +are updated, and recovery's asserts that it is handed the frame's own sequence rather than a +placeholder. + +**The generation mapping** is `BAND | (adoption_sequence << 24) | ordinal`, injective in the pair +and confined to a reserved top-bit band because segments, tails and projections share one +generation space. Its boundary is one value wide and closed by a ceiling rather than a shift +round-trip; see 2026-07-31-C's successor note in `staging.rs`. + +**No per-object offsets, and none were needed.** `IndexLocation` names the whole certified record +— its own doc says so and names adopted projections explicitly — so every object in a chunk shares +that chunk's location with `frame_offset: 0` and `frame_len` covering header, canonical payload and +digest. A reader validates the artifact and then selects from the decoded vector, exactly as +several objects share one journal frame. I had raised this as a suspected protocol gap; it was a +contract I had not read on the type I was populating. Pointing at object bytes would let a reader +return bytes from a record it never proved complete. + +**Whole or not at all.** Every declared ordinal is read back, decoded, and checked against the +manifest's session and chunk count before any location is produced, and +`RecoveredProjectionArtifacts::new` then refuses unless the entry count equals the descriptor's +`object_count`. A directory holding some of its chunks resolves to an error, never to a smaller +projection — the failure a "resolve what is present" implementation reaches by being helpful. + +**`notify_recovered`.** `Committed` ends the pin as an adoption at the position the resolution +recorded, and refuses a session this recovery never resolved rather than adopting at a guess — +without a position, cleanup could never prove absence of reference. `ProvedAbsent` means recovery +read the complete authoritative history and no frame names these artifacts, so the session returns +to a reclaimable state and its directory goes. Both are idempotent, including for a session that is +already gone, because recovery repeats every notification when a later one fails. + +**Evidence.** A committed projection resolving to complete membership at the expected generation, +with every object in a chunk sharing that chunk's location and `frame_len` equal to the artifact's +size on disk; the same session refusing once a chunk is removed. Both notification outcomes driven +twice each to assert idempotence, with the `Committed` case then measured by cleanup against a root +short of the adoption to confirm the position it established is real. The integration suite's +"resolver seam is deferred" test is replaced by one asserting it answers. + +##### Contract review 2026-07-31-C + +P1 against 2026-07-31-B: cleanup could delete committed artifacts through a stale root. Found in +review, reproduced, repaired. Deliverable 8 was held until the authority boundary below was +decided, and the decision is recorded here. + +**The defect.** `cleanup_unreferenced` selected candidates from current staging state and then +trusted whatever `CommittedRoot` the caller supplied. Capture `R0` before a projection is +published; publish `R1` referencing its artifacts; record the session `Adopted`; call cleanup with +`R0`. Every artifact is absent from `R0`'s pins — because `R0` predates them — so the absence +proof succeeds and the directory is deleted out from under a committed root that still points into +it. Reproduced exactly as reported: the session was reclaimed. + +**My reasoning in B was inverted, and that is the part worth keeping.** I wrote that a racing +newer root "can only add references" and concluded the proof was safe. Adding references is +precisely the hazard: it means an older root *omits* references a newer one holds. I checked the +direction in which references change and never checked the direction in which the root travels. + +**The authority boundary, and why there is no purely-B3 repair.** `ProjectionStaging` is +constructed *before* any committed root exists — it has to be, since it is recovery's +`ProjectionRecoveryResolver` — so there is no moment at which staging can observe a root position +on its own. Both candidate repairs need an authority minted outside B3. + +**Decision: a durable minimum recorded at adoption.** `ProjectionAdoptionOutcome::Adopted` now +carries the committed shard sequence its frame reached — a **frozen D0-B amendment**, granted and +recorded here. B1 supplies it because only B1 knows it: the append that produced it has just +returned. Staging writes it into the adoption marker, which is already rewritten by that exact +transition and already carries a payload, so durability costs nothing new. Cleanup then skips any +adopted session whose recorded sequence exceeds the supplied root's, and a root at or past the +adoption necessarily includes its effects — which is what makes the absence real. + +Engine-owned cleanup under the same synchronization as root publication was the alternative. It is +the stronger property but reaches for the `StoreEngine` staging façade that §6.5 still records as a +pending interface amendment, and it would not on its own cover a session adopted before a restart. +The two are not exclusive; if the façade lands, this check remains the restart half. + +**`Some(0)` and `None` are different answers.** The first cut compared +`shard_committed_sequence(shard).unwrap_or(0)`, which conflates them — zero is a valid committed +sequence, so a root that says *nothing* about a shard became indistinguishable from one that has +committed through its first frame, and an adoption at sequence 0 was reclaimable through a root +that never mentioned the shard it lives on. The check now requires an explicit +`Some(through) if through >= adopted_at`, so silence is treated as no evidence. The conflation +disarmed the position check for precisely the adoption that needs it most, the earliest one. + +**A session missing a position is not reclaimable at all.** An adopted session always records +where it was adopted, so the absence of one means the store cannot say when the reference it is +about to disprove came into existence. Skipping is the only answer that cannot lose data. + +**Evidence.** The reviewer's R0/R1 sequence as a regression, which failed by deleting the session +before the repair. It now asserts three things rather than one: the stale root reclaims nothing, +the decline is attributed to the root's age rather than to some incidental refusal, and the *same* +session against a root that has reached the adoption is reclaimed — without that last clause the +test would pass against a cleanup that declined everything. Durability is asserted in the reopen +test: after a restart, a root short of the reconstructed adoption still declines and one at it +reclaims. A zero-boundary regression covers the other half against one adopted-at-zero session: a +root with no sequence for the shard declines, and the same session against an explicit `Some(0)` +reclaims — both halves together, because a test that only checked the absent case would pass +against a cleanup that declined every root. + +Five mutations, each caught: trusting any supplied root, recording no position at adoption, +writing no position into the marker, restoring the `unwrap_or(0)` conflation, and tightening the +comparison to `>` so an exactly-reached root is refused. The last fails four tests rather than +one, which is the shape an over-strict boundary should have. + +A test-fixture defect surfaced on the way and is worth recording because it would have hidden the +repair: the helper root populated a sequence for shard 0 only, while the fixture's destination +repository hashes to another shard. Every cleanup test would have measured a stale-root skip while +claiming to measure the reference proof. The helper now populates every shard, as a real root does. + +##### Contract review 2026-07-31-B + +B3 deliverable 7 — cleanup proves absence of reference. Deliverable 8 follows separately. + +**Two clarifications to §6.5's wording**, both recorded in the scope and neither a relaxation. +The deliverable reads as a per-artifact rule over everything staging owns; implementing it that +way would have been wrong twice. + +*Absence of reference is necessary and not sufficient.* A `Sealed` session a client is still +finalizing and a `Finalizing` one whose adoption frame may be mid-append are both referenced by +nothing at all. A cleanup that proved absence and stopped there would delete them. The candidate +set is therefore the **adopted** sessions: the one state where the artifacts are store content +and "does anything still point at them?" is both meaningful and answerable. `Open` and `Sealed` +belong to expiry and abort, which already own them. + +*The unit of removal is the session, not the artifact.* A session's chunks are not independent +files — the manifest names all of them and reconstruction refuses a sealed session missing any +ordinal — so removing the unreferenced half of a directory trades a bounded leak for a root that +fails to open. A session is reclaimed only when nothing in it is referenced, and the per-artifact +rule the clause is really about still applies inside that: `reclaim_session_directory` removes +only files carrying the session's own marker and refuses the whole directory on anything else. + +**Every file is checked, not only the chunks.** A root that pinned a session's manifest and +nothing else would still be holding that directory, and answering on chunks alone would delete +the file it holds. + +**The supplied root must be new enough to be evidence — corrected by 2026-07-31-C.** The first +version of this argued that a racing newer root "can only add references" and concluded the proof +was safe. That is an argument *for* the hazard: adding references is exactly what makes an older +root omit them, so absence measured against a root captured before an adoption is a date rather +than an absence. See 2026-07-31-C for the defect and its repair. + +**Evidence.** The deliverable's named acceptance case and its inverse in one test, against one +adopted session: a root holding a single chunk declines the whole directory and leaves every file +in place, then a root holding nothing reclaims it. Both halves together are what make the first +assertion mean "the reference proof declined it" rather than "cleanup does nothing here" — a +cleanup that never reclaimed anything would pass the decline half perfectly. A second test pins +that a sealed session and a pinned one both survive a cleanup against a root referencing nothing. +Mutation-checked: skipping the reference proof fails the first, and treating every state as a +candidate fails the second, each and only each. + +The integration suite's `cleanup_is_deferred_and_names_its_deliverable` is replaced rather than +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-07-31-A + +B3 deliverable 6 — finalize and the adoption pin. Deliverables 7 and 8 are **not** in this +commit; review found a blocker in 6 that had to close first. The dispatch is 6-8 only: the +ignored sealed-invisibility test stays ignored, and its blocker list is corrected below. + +**Scope §6.5's state machine was unimplementable as written**, in two of three transitions, and +is amended to `Open → Sealed → Finalizing → {Sealed, Adopted}` with all four states durable and +reconstructable. The reasoning is recorded in the scope rather than repeated here; the short +version is that `Open` appears twice in the original wording where only `Sealed` can be +reconstructed, and `Adopted` is a state the wording needed and did not have. Reconstructability +is the membership rule for this enum — a value no restart can produce has no business on the +wire — and both amendments follow from applying it. + +**Coalescing moves to B1.** §6.5 promised that "identical concurrent finalizers coalesce onto +one" at this layer. They cannot: `ProjectionAdoption` is an unforgeable capability with exactly +one terminal outcome and a `Drop` that reports its absence, so there is no second copy to hand a +second caller. B3's guarantee is *one pin*, and a second finalizer is refused by name. +Request-level coalescing across retries of one operation belongs to whoever owns the request. No +B3 coalescing regression is claimed. + +**The blocker, found in review.** `finalize` published the pin marker through the shared artifact +writer, which publishes with `rename_noreplace` because every other staging artifact is +unique-by-name. Recording `Adopted` then wrote the same pathname through the same writer and +always failed `EEXIST`, wedging every adoption in `Finalizing` — silently, because nothing +inspected the error. The repair is **not** a replace flag on the shared writer: that would hand +overwrite permission to the chunk and manifest artifacts the no-replace rule exists to protect. +It is one marker-specific path whose licence to overwrite is bounded by proof — the marker on +disk must decode as a staging artifact of this session, carry the operation and digest the +session binds, and say `Finalizing` — and which publishes through the no-follow temporary-open +primitive, then a replacing rename, then a directory sync. Idempotent on an +already-`Adopted` marker, because a retried outcome is not a second event. + +**`SESSION_FIXED_FILES` is 4, and it is a peak.** Deliverable 6 adds the marker and, for the +width of the replacement, its temporary. The reservation is charged against the peak because +that is the instant the directory is widest. Two things surfaced on the way: the constant's doc +claimed to be "the shared definition rather than a second opinion" while `options.rs` carried a +literal `+ 2`, and the **default `staging_max_files_per_session` was sized to the old layout +exactly**, so every root became an invalid configuration until it was raised. Both are +configuration-visible. + +Four is the *true* peak, not a conservative one. I first recorded a general residual here — +that every artifact write peaks at `+1` because `write_artifact` also publishes through a +`.tmp` — and review corrected it: those temporaries stand in for final names that are +absent, so each occupies the slot it is about to become rather than an extra one, and state +ordering keeps chunk writes from overlapping a seal or a finalize. `adoption.tmp` is the only +temporary that coexists with a final file already on disk, because its transition replaces a +marker rather than creating one. The claim was generalised from "there is a temporary" without +checking whether the final name was occupied. + +**Recorded, not fixed.** `write_artifact` opens its temporary with `File::options()` and no +`O_NOFOLLOW` — the same redirection family closed in `checkpoint.rs` under 2026-07-30-D. The new +marker path uses the funnel; the shared writer still does not. + +**Plan §8 is amended in place**, not merely superseded by this record. It still specified +`Open -> Finalizing`, store-level coalescing, and return-to-`Open`; a contradiction left standing +in the plan is a contradiction, and the later review being right does not make the earlier prose +unread. + +**Evidence.** Six regressions, covering every transition of +`Open -> Sealed -> Finalizing -> {Sealed, Adopted}` against both the in-memory state and the +device. The adoption chain — pin taken, adoption recorded, reservation released, reopen +reconstructing `Adopted` without re-charging — plus one each for: a second finalizer refused +rather than issued a second pin; expiry declining a pinned session and leaving its artifacts in +place; a definitive pre-append failure removing the marker, returning `Sealed`, and *permitting +refinalization*; and a transferred pin whose marker survives, reconstructs as `Finalizing`, and +is offered to `transferred_sessions` for its own shard and no other. Each asserts the marker on +disk, not only the state in the registry — a test that checked the registry alone would pass +against a pin that never reached the device. + +The expiry/finalize race needs **both** orderings and originally had one. The covered ordering +was expiry meeting an already-durable pin, where the *state* reads `Finalizing`. The other is the +admission window: the finalizer has been admitted under the registry lock and released it to +write the marker, so the session still reads `Sealed` with nothing on disk, and a sweep that +looked only at state would find an expired, sealed, unpinned session and delete the artifacts out +from under a pin about to be issued. What prevents it is the `busy` exclusion, so the regression +drives that shape directly rather than racing into it — a race reproducing one time in a thousand +is a test that passes for the wrong reason the other nine hundred and ninety-nine. Its second +half clears `busy` and sweeps again, which *does* reclaim: without that, "nothing was reclaimed" +proves only that something declined, not that the exclusion is what declined it. + +Every one is mutation-checked: admitting a second finalizer, letting expiry reclaim a pinned +session, skipping the marker removal on release, and hiding `Finalizing` from +`transferred_sessions` each fail exactly one test and no others. + +Mutating the admission-window exclusion produced a result worth recording. The sweep does change +behaviour and the regression fails — but on `Overloaded` from `reclaim_session`'s own `busy` +check rather than on a deletion, because the guard is two independent layers. Removing the second +one as well **does not compile**: that match is exhaustive, so dropping the pinned case from the +reclamation guard is a compile error rather than an omission. Reaching artifact loss from here +requires deliberately writing a `Finalizing => {}` arm, which is no longer something a future edit +does by accident. + +And for the adoption chain, both halves: +routing the adoption write back through the shared writer reproduces +`Io(Os { code: 17, kind: AlreadyExists })` at the finish, and reconstructing an adopted session +with its reservation fails the accounting assertion. The fixture had to be built inside +`staging.rs`'s unit module: `finalize` is `pub(crate)` so its regressions cannot live in the +integration file where all the sealing machinery is. + +**The ignored sealed-invisibility test remains blocked, and its stated reason is stale.** It +names `submit` as a blocker; `submit` has been production-reachable since the B-wave work. The +real blockers are `RepoSnapshot::locate`, `StoreEngine::snapshot`, and +`ValidatedTransactionBuilder::adopt_projection` — all three B1 deliverable 8 — plus the staging +accessor, which stays a pending interface amendment. Finishing 6-8 will not unblock it. + ##### Contract review 2026-07-30-G Recovery run-discard, which 2026-07-30-B deferred and named as the thing that would turn its refusal diff --git a/doc/phase1-storage-spine-scope.md b/doc/phase1-storage-spine-scope.md index c5dda44..490a116 100644 --- a/doc/phase1-storage-spine-scope.md +++ b/doc/phase1-storage-spine-scope.md @@ -1788,16 +1788,70 @@ B3's deliverables: 5. **Artifacts are written by maintenance workers, synced, uniquely named, and unreferenced.** They never enter namespace membership, object-existence answers, snapshots, refs, receipts, event feeds, dedupe state, or `CURRENT`. -6. **Finalize and the adoption pin.** `Open → Finalizing` moves atomically for the sole - bound operation/digest; identical concurrent finalizers coalesce onto one, and every - different operation/digest rejects. Expiry prevents a *new* finalizer but must not delete - artifacts held by an already-admitted one — the expiry/finalize race is a named acceptance - case, not an incidental detail. A definitive pre-append failure releases the pin and - returns the session to `Open` only if it is still live; otherwise cleanup aborts it. +6. **Finalize and the adoption pin.** The state machine is + `Open → Sealed → Finalizing → {Sealed, Adopted}`, and all four states are durable and + reconstructable — decided by an artifact on disk, never by a flag one process remembers. + Amended from `Open → Finalizing` by contract review 2026-07-31-A, which found two of the + three transitions in the original wording unimplementable as written: + + - The pin is taken from **`Sealed`**, not `Open`. The sealed manifest is what an adopter + revalidates against, so a pin on a session without one names state that does not exist. + - A definitive pre-append failure returns the session to **`Sealed`**, not `Open`. + Reconstruction reads the sealed manifest's presence as the seal's own commit point and + would hand back `Sealed` on the next restart regardless; returning a session to a state + no restart can reproduce is the defect the private `SessionBusy` enum exists to avoid. + `Sealed` is still finalizable, which is the property the rule is about. If the session is + no longer live the marker still goes and expiry reclaims it, which is "otherwise cleanup + aborts it". + - **`Adopted` is durable**, and is the transition the original wording had no name for. + Adopted artifacts stay in `staging/` with a committed root pointing into them, so + dropping the session record would make the next reconstruction read the directory as an + abandoned materialization and reclaim committed content; keeping the record with no state + re-charges the whole reservation on every restart. `Adopted` releases the reservation — + those bytes are store content, not staging occupancy — and leaves the directory to + cleanup's reference proof. + + The pin becomes durable **before** it is issued. A handle backed by an in-memory flag is an + adoption capability with nothing behind it. + + Only the sole bound operation/digest may finalize, which the binding enforces by carrying + it from `begin`; a different one cannot reach a session at all. **Coalescing identical + concurrent finalizers is B1's, not B3's** — also amended by 2026-07-31-A. A + `ProjectionAdoption` is an unforgeable capability with exactly one terminal outcome and a + `Drop` that reports its absence, so there is no second copy to hand a second caller: at this + layer the guarantee is *one pin*, and a second finalizer is refused by name. Request-level + coalescing across retries of one operation belongs to whoever owns the request. + + Expiry prevents a *new* finalizer but must not delete artifacts held by an already-admitted + one — the expiry/finalize race is a named acceptance case, not an incidental detail. 7. **Cleanup proves absence of reference.** It removes only artifacts carrying a valid session marker, and only after proving no committed manifest references them, then syncs the affected directories. *Accept:* a test that cleanup declines to remove an artifact a committed manifest still references. + + Two clarifications from contract review 2026-07-31-B, neither a relaxation: + + - **The supplied root must have reached the adoption.** A root captured before a session was + adopted references none of its artifacts because it predates them, so absence measured + against it is a date and not an absence; acting on it deletes a directory the current root + points into. Each adopted session records the committed shard sequence its adoption frame + reached — carried on `ProjectionAdoptionOutcome::Adopted` and made durable in the adoption + marker — and is skipped unless the supplied root has reached at least that far. The root + must say so *explicitly*: a root recording no sequence for that shard is silent, not a + witness that the shard has committed through zero, and zero is a valid sequence. Contract + review 2026-07-31-C. + - **The candidate set is the adopted sessions, and only those.** Absence of reference is + necessary and is not sufficient: a `Sealed` session a client is still finalizing, and a + `Finalizing` one whose adoption frame may be mid-append, are both referenced by nothing, + and a cleanup that proved absence and stopped there would delete them. `Open` and + `Sealed` belong to expiry and abort; `Finalizing` has no absence to prove yet. + - **The unit of removal is the session directory, not the individual artifact.** A + session's chunks are not independent: the manifest names all of them and reconstruction + refuses a sealed session missing any ordinal, so removing the unreferenced half of a + directory trades a bounded leak for a root that fails to open. A session is reclaimed + only when nothing in it is referenced. The per-artifact rule this clause is really about + still applies inside that: reclamation removes only files carrying the session's own + marker and refuses the whole directory if it finds anything else. 8. **Recovery treats synced-but-unreferenced artifacts as invisible garbage**, and a complete final frame as authoritative adoption. It can never expose a partial chunk set.