diff --git a/crates/levcs-store/src/staging.rs b/crates/levcs-store/src/staging.rs index a8d1548..d400c66 100644 --- a/crates/levcs-store/src/staging.rs +++ b/crates/levcs-store/src/staging.rs @@ -259,6 +259,2595 @@ pub(crate) struct StagedProjectionAdoption { pub(crate) handle: ProjectionAdoption, } +// =========================================================================== +// B3 StagingSessions — the storage mechanism and its bounds (scope 6.5, +// deliverables 1-5). Everything above this line is D0-B's frozen adoption +// seam and is not edited here. +// +// What this half decides: what a session is on disk, what it costs, when it +// dies, and that none of it is visible until a transaction adopts it. What it +// deliberately does not decide: `ProjectionCore` validation, identity and +// authority proofs, policy, source-snapshot and `ForkProofV2` checks, session +// authentication, and the export lease. Plan §5.1 puts those in Phase 2's +// `levcs-instance`. Every field those checks key on is bound, stored, and +// exposed here; none of them is evaluated here. +// =========================================================================== + +// A second `use` block rather than an addition to the frozen one at the top of +// the file, so the ownership boundary is visible in the diff. +use std::cell::Cell; +use std::collections::{BTreeMap, BTreeSet}; +use std::fs::File; +use std::io::IoSlice; +use std::ops::{Deref, DerefMut}; +use std::os::unix::fs::MetadataExt; +use std::sync::atomic::{AtomicU64, Ordering::Relaxed}; +use std::sync::{Mutex, OnceLock}; +use std::thread::JoinHandle; + +use levcs_protocol::v2::{ProjectionStageChunkV1, StagedObjectV1}; +use levcs_protocol::CanonicalCodec; + +use crate::format::digest; +use crate::options::StoreOptions; +use crate::recovery::RecoverySession; +use crate::types::DurabilityCounters; + +/// Subdirectory of the store root holding every staging session (scope 3.1). +/// +/// It is a sibling of `shards/`, not a child of one, and that placement is the +/// structural half of deliverable 5: nothing a manifest, `CURRENT`, checkpoint, +/// or index run can name lives under this path, so a staged artifact cannot +/// enter committed state by being mistaken for a shard file. +const STAGING_DIR: &str = "staging"; + +const STAGING_ARTIFACT_MAGIC: [u8; 8] = *b"LVCSSTG\0"; +const STAGING_ARTIFACT_VERSION: u16 = 1; +const STAGING_ARTIFACT_DIGEST_DOMAIN: &[u8] = b"levcs-staging-artifact/v1\0"; +const STAGING_ARTIFACT_SET_DIGEST_DOMAIN: &[u8] = b"levcs-staging-artifact-set/v1\0"; +/// `magic || version || kind || session_id || payload_len`. +const STAGING_ARTIFACT_HEADER_LEN: usize = 8 + 2 + 2 + 16 + 4; + +const SESSION_RECORD_NAME: &str = "session"; +const MANIFEST_NAME: &str = "manifest"; + +/// 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; + +/// Which of the three staging artifact kinds a file is. +/// +/// The kind travels in the header beside the session ID because cleanup +/// (deliverable 7) must be able to decide, from the bytes alone, that a file +/// belongs to a session it is reclaiming. A directory name is not a marker: a +/// file moved or left behind by a partial reclamation would keep it. +#[derive(Copy, Clone, Debug, PartialEq, Eq)] +enum StagingArtifactKind { + SessionRecord = 1, + Chunk = 2, + Manifest = 3, +} + +impl StagingArtifactKind { + fn code(self) -> u16 { + self as u16 + } + + fn from_code(code: u16) -> Option { + match code { + 1 => Some(Self::SessionRecord), + 2 => Some(Self::Chunk), + 3 => Some(Self::Manifest), + _ => None, + } + } +} + +/// The immutable creation binding of one staging session. +/// +/// `session` is the frozen wire binding: destination repo/genesis and expected +/// authority, projection mode, authenticated source kind and actor/key epoch, +/// source generation or `ForkProofV2`, final operation ID/digest/evidence +/// digest, total object/byte/chunk counts, the ordered manifest digest, and +/// expiry. B3 stores all of it and enforces the mechanical parts; it evaluates +/// none of the identity or policy fields. +/// +/// `membership_root` is carried separately because it is not a field of +/// `ProjectionStageSessionV1` and yet is an input to the manifest digest the +/// session already binds. Without it the seal could not reconstruct the exact +/// manifest the creator committed to — and inventing one would make +/// `manifest_digest` unverifiable rather than binding. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ProjectionStageBinding { + pub session: ProjectionStageSessionV1, + pub membership_root: ObjectId, +} + +/// Observable staging occupancy and work. +/// +/// Charter item 7: the bounds below are asserted against these counters and +/// against [`DurabilityCounters`], not read out of the code. The gauges are +/// the reserved budget, not the bytes on disk, because the budget is what the +/// ceilings are enforced against. +#[derive(Debug, Default)] +pub struct StagingCounters { + pub sessions_created: AtomicU64, + pub sessions_refused: AtomicU64, + pub sessions_sealed: AtomicU64, + pub sessions_aborted: AtomicU64, + pub sessions_expired: AtomicU64, + /// Gauge: sessions currently holding budget. + pub sessions_live: AtomicU64, + /// Gauge: object bytes reserved by live sessions. + pub reserved_bytes: AtomicU64, + /// Gauge: objects reserved by live sessions. + pub reserved_objects: AtomicU64, + /// Gauge: files reserved by live sessions. + pub reserved_files: AtomicU64, + /// Gauge: reclaimable staged object bytes charged to staging alone. + /// Deliberately never shares an accumulator with receive spools, journal + /// bytes, or segment bytes (plan §8: "independently of ordinary receive + /// spools"); a shared counter would let unrelated traffic close staging's + /// ceiling, or hide it. + pub compaction_debt_reserved_bytes: AtomicU64, + /// Gauge: staged object bytes charged by an accepted chunk put. Charged + /// when the ordinal is reserved and released if the artifact write fails, + /// so the declared-total bound stays atomic even though the write itself + /// happens on a maintenance worker with the registry unlocked. + pub compaction_debt_written_bytes: AtomicU64, + pub chunks_written: AtomicU64, + /// Chunk puts satisfied by an identical ordinal/digest already present. + /// A restart re-uploading a chunk must cost no second write; this is how + /// that is asserted rather than asserted about. + pub chunks_deduplicated: AtomicU64, + pub artifact_bytes_written: AtomicU64, + pub artifacts_unlinked: AtomicU64, + /// Durable sessions reconstructed from disk by [`ProjectionStaging::open`]. + /// A root-global ceiling is only meaningful if it counts what is already + /// durable, so this is how "the accounting survived the restart" is + /// asserted rather than asserted about. + pub sessions_reconstructed: AtomicU64, + /// Session directories reclaimed at open because they never acquired a + /// durable session record. Counted separately from `sessions_aborted` and + /// `sessions_expired` because an abandoned materialization was never a + /// session: nothing ever charged budget for it, and confusing the two is + /// how a reopen deletes a durable session. + pub abandoned_materializations_reclaimed: AtomicU64, + /// Artifact writes performed on a maintenance worker. Deliverable 5 says + /// artifacts are written by maintenance workers; this is the counter that + /// says so, and `write_artifact` refuses to run anywhere else. + pub maintenance_artifact_writes: AtomicU64, + /// Jobs dispatched to the maintenance pool and completed. + pub maintenance_jobs: AtomicU64, + /// Entries under `staging/` that are not a shard directory, a session + /// directory, or a recognized artifact. Never deleted, never interpreted; + /// counted so a surprise in the tree is visible instead of silent. + pub unrecognized_entries: AtomicU64, +} + +impl StagingCounters { + pub fn snapshot(&self) -> StagingCounterSnapshot { + StagingCounterSnapshot { + sessions_created: self.sessions_created.load(Relaxed), + sessions_refused: self.sessions_refused.load(Relaxed), + sessions_sealed: self.sessions_sealed.load(Relaxed), + sessions_aborted: self.sessions_aborted.load(Relaxed), + sessions_expired: self.sessions_expired.load(Relaxed), + sessions_live: self.sessions_live.load(Relaxed), + reserved_bytes: self.reserved_bytes.load(Relaxed), + reserved_objects: self.reserved_objects.load(Relaxed), + reserved_files: self.reserved_files.load(Relaxed), + compaction_debt_reserved_bytes: self.compaction_debt_reserved_bytes.load(Relaxed), + compaction_debt_written_bytes: self.compaction_debt_written_bytes.load(Relaxed), + chunks_written: self.chunks_written.load(Relaxed), + chunks_deduplicated: self.chunks_deduplicated.load(Relaxed), + artifact_bytes_written: self.artifact_bytes_written.load(Relaxed), + artifacts_unlinked: self.artifacts_unlinked.load(Relaxed), + sessions_reconstructed: self.sessions_reconstructed.load(Relaxed), + abandoned_materializations_reclaimed: self + .abandoned_materializations_reclaimed + .load(Relaxed), + maintenance_artifact_writes: self.maintenance_artifact_writes.load(Relaxed), + maintenance_jobs: self.maintenance_jobs.load(Relaxed), + unrecognized_entries: self.unrecognized_entries.load(Relaxed), + } + } +} + +#[derive(Copy, Clone, Debug, Default, PartialEq, Eq)] +pub struct StagingCounterSnapshot { + pub sessions_created: u64, + pub sessions_refused: u64, + pub sessions_sealed: u64, + pub sessions_aborted: u64, + pub sessions_expired: u64, + pub sessions_live: u64, + pub reserved_bytes: u64, + pub reserved_objects: u64, + pub reserved_files: u64, + pub compaction_debt_reserved_bytes: u64, + pub compaction_debt_written_bytes: u64, + pub chunks_written: u64, + pub chunks_deduplicated: u64, + pub artifact_bytes_written: u64, + pub artifacts_unlinked: u64, + pub sessions_reconstructed: u64, + pub abandoned_materializations_reclaimed: u64, + pub maintenance_artifact_writes: u64, + pub maintenance_jobs: u64, + pub unrecognized_entries: u64, +} + +/// Outcome of one numbered chunk put. Both arms are terminal successes; the +/// distinction exists so a restart can be asserted to cost no second write. +#[derive(Copy, Clone, Debug, PartialEq, Eq)] +pub enum ChunkPutOutcome { + Stored, + AlreadyPresent, +} + +/// The two states a session can hold in this pass. +/// +/// 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. +#[derive(Copy, Clone, Debug, PartialEq, Eq)] +pub enum StagedSessionState { + Open, + Sealed, +} + +/// Read-only view of one live session. Never carries artifact bytes. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct StagedSessionStatus { + pub session_id: [u8; 16], + pub state: StagedSessionState, + pub shard_index: u16, + pub chunks_present: u32, + pub chunks_expected: u32, + pub written_objects: u64, + pub written_bytes: u64, + pub expires_at_micros: i64, +} + +// --- maintenance workers (deliverable 5) ----------------------------------- +// +// Deliverable 5 requires artifacts to be *written by maintenance workers*. +// Two properties follow from that and neither is a matter of taste: +// +// * No filesystem work happens while the registry mutex is held. The +// registry is the one place every ceiling is evaluated, so holding it +// across a disk write makes admission latency a function of unrelated +// uploads — and makes the global bound's own critical section as slow as +// the slowest device in the root. +// * No filesystem work happens on the caller's thread. The caller is an +// instance-layer request thread (plan §5.2 forbids filesystem work on the +// Tokio pool); staging's own writes must not be the exception. +// +// Both are enforced mechanically rather than documented: `run_maintenance` +// refuses to dispatch while this thread holds the registry lock, and +// `write_artifact` refuses to run anywhere but a maintenance worker. + +thread_local! { + /// Set for the whole life of a maintenance worker thread. + static ON_MAINTENANCE_WORKER: Cell = const { Cell::new(false) }; + /// How many registry guards this thread currently holds. + static REGISTRY_LOCK_DEPTH: Cell = const { Cell::new(0) }; +} + +/// Maintenance workers per store root. +/// +/// A constant rather than a configured value: `options.rs` is frozen for this +/// pass and inventing a local knob that no operator can reach would be a +/// second opinion about configuration. Two is enough to keep one slow device +/// from serializing an unrelated session's chunk write, and small enough that +/// staging cannot become a source of I/O concurrency the capacity analysis did +/// not budget for. Raising it is an interface request against `options.rs`, +/// not an edit here. +const STAGING_MAINTENANCE_WORKERS: usize = 2; + +type MaintenanceJob = Box; + +struct MaintenancePool { + jobs: Option>, + workers: Vec>, +} + +impl MaintenancePool { + fn new(root: &Path) -> Self { + let (jobs, receiver) = crossbeam_channel::unbounded::(); + let workers = (0..STAGING_MAINTENANCE_WORKERS) + .map(|index| { + let receiver = receiver.clone(); + std::thread::Builder::new() + .name(format!("levcs-staging-maint-{index}")) + .spawn(move || { + ON_MAINTENANCE_WORKER.with(|flag| flag.set(true)); + for job in receiver { + job(); + } + }) + .unwrap_or_else(|error| { + panic!( + "staging maintenance worker for {} could not start: {error}", + root.display() + ) + }) + }) + .collect(); + Self { + jobs: Some(jobs), + workers, + } + } +} + +impl Drop for MaintenancePool { + fn drop(&mut self) { + // Closing the channel is the shutdown signal; the join is what makes + // "no artifact write outlives the staging instance" true rather than + // likely. + self.jobs = None; + for worker in self.workers.drain(..) { + let _ = worker.join(); + } + } +} + +/// Guard whose only extra job is to make "this thread holds the registry" +/// observable to [`ProjectionStaging::run_maintenance`]. +struct RegistryGuard<'a>(std::sync::MutexGuard<'a, Registry>); + +impl Deref for RegistryGuard<'_> { + type Target = Registry; + + fn deref(&self) -> &Registry { + &self.0 + } +} + +impl DerefMut for RegistryGuard<'_> { + fn deref_mut(&mut self) -> &mut Registry { + &mut self.0 + } +} + +impl Drop for RegistryGuard<'_> { + fn drop(&mut self) { + REGISTRY_LOCK_DEPTH.with(|depth| depth.set(depth.get().saturating_sub(1))); + } +} + +/// Every store root with a live [`ProjectionStaging`] in this process. +/// +/// The `RecoverySession` argument to `open` proves no *other* process holds +/// the root. This closes the remaining half: a second in-process instance for +/// one root would carry a second registry, and two registries each admit up to +/// the full global and per-principal ceilings, which makes a "global" bound a +/// per-handle bound. Membership is released when the instance is dropped. +static OPEN_STAGING_ROOTS: OnceLock>> = OnceLock::new(); + +fn open_staging_roots() -> &'static Mutex> { + OPEN_STAGING_ROOTS.get_or_init(|| Mutex::new(BTreeSet::new())) +} + +/// How staging learns which device a path is on. +/// +/// Behind a trait for exactly one reason: the cross-device refusal must be +/// asserted through `begin`, the entry point a consumer calls, and an +/// unprivileged test cannot create a second mount under a temporary root. The +/// production probe is the real `st_dev`, and a test asserts that it agrees +/// with `std::fs::metadata(..).dev()` so the substitute cannot come to mean +/// something the real one does not. +trait DeviceProbe: Send + Sync { + fn device_of(&self, path: &Path) -> Result; +} + +struct StatDeviceProbe; + +impl DeviceProbe for StatDeviceProbe { + fn device_of(&self, path: &Path) -> Result { + Ok(std::fs::metadata(path)?.dev()) + } +} + +/// One stored chunk artifact. +#[derive(Clone, Debug)] +struct StoredChunk { + digest: ObjectId, + path: PathBuf, + /// Bytes of the artifact file, header and all. Staged *object* bytes — + /// the unit every byte budget is denominated in — are accumulated on the + /// session record instead, so the running total has exactly one home and + /// cannot drift from the per-chunk copies. + file_bytes: u64, +} + +/// What the registry knows about one ordinal. +/// +/// `InFlight` exists because the artifact write happens on a maintenance +/// worker with the registry unlocked. Without it, two concurrent puts of the +/// same ordinal would both see an empty slot, both dispatch a write, and the +/// second `rename_noreplace` would fail with a filesystem error instead of the +/// typed answer this contract owes. +#[derive(Clone, Debug)] +enum ChunkSlot { + InFlight { digest: ObjectId }, + Stored(StoredChunk), +} + +impl ChunkSlot { + fn digest(&self) -> ObjectId { + match self { + Self::InFlight { digest } => *digest, + Self::Stored(stored) => stored.digest, + } + } + + fn stored(&self) -> Option<&StoredChunk> { + match self { + Self::InFlight { .. } => None, + Self::Stored(stored) => Some(stored), + } + } +} + +/// What long-running work a session is currently inside. +/// +/// Deliberately *not* a variant of the public [`StagedSessionState`]: sealing +/// and reclaiming are in-process exclusions, not durable states, and a caller +/// asking `describe()` about a session mid-seal is still being told the truth +/// when it hears `Open`. Making them public states would put a value on the +/// wire that no restart could ever reconstruct. +#[derive(Copy, Clone, Debug, PartialEq, Eq)] +enum SessionBusy { + Idle, + Sealing, + Reclaiming, +} + +/// Budget one session holds. Charged once at creation from the declared +/// totals and released exactly once when the session leaves the registry. +#[derive(Copy, Clone, Debug, Default, PartialEq, Eq)] +struct Reservation { + objects: u64, + bytes: u64, + files: u64, +} + +#[derive(Copy, Clone, Debug, Default)] +struct Usage { + sessions: u64, + objects: u64, + bytes: u64, + files: u64, +} + +struct SessionRecord { + binding: ProjectionStageBinding, + shard_index: u16, + directory: PathBuf, + state: StagedSessionState, + busy: SessionBusy, + chunks: BTreeMap, + written_objects: u64, + written_bytes: u64, + reservation: Reservation, + resolution: Option>, +} + +impl SessionRecord { + fn principal(&self) -> [u8; 32] { + self.binding.session.actor + } +} + +/// The one place staging occupancy lives. +/// +/// A single lock, and every ceiling is evaluated and the entry inserted under +/// one acquisition of it. Decision 9.8 is explicit that a check-then-insert +/// race admits entries past the ceiling under exactly the concurrency the +/// ceiling exists for, so there is deliberately no read-then-decide path here +/// at all — not even a fast one. +#[derive(Default)] +struct Registry { + sessions: BTreeMap<[u8; 16], SessionRecord>, + principals: BTreeMap<[u8; 32], Usage>, + global: Usage, + debt_reserved_bytes: u64, +} + +/// Bounded invisible projection staging for one store root. +/// +/// **Exactly one of these exists per root, for the life of the root lock.** +/// See [`ProjectionStaging::open`]. +pub struct ProjectionStaging { + options: StoreOptions, + staging_root: PathBuf, + /// Canonical root path, held so `Drop` releases exactly the entry `open` + /// claimed even if `options.root` was relative. + canonical_root: PathBuf, + counters: Arc, + durability: Arc, + device: Box, + registry: Mutex, + maintenance: MaintenancePool, +} + +impl Drop for ProjectionStaging { + fn drop(&mut self) { + if let Ok(mut roots) = open_staging_roots().lock() { + roots.remove(&self.canonical_root); + } + } +} + +impl fmt::Debug for ProjectionStaging { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("ProjectionStaging") + .field("staging_root", &self.staging_root) + .field("counters", &self.counters.snapshot()) + .finish_non_exhaustive() + } +} + +impl ProjectionStaging { + /// Open `/staging` under a held root lock and reconstruct every + /// durable session from disk. + /// + /// **The `lock` argument is the ownership proof, not a convenience.** The + /// ceilings this type enforces are advertised as root-global, and a global + /// bound is only global if exactly one accountant exists per root. A freely + /// callable constructor made "two handles on one root" expressible, and two + /// registries each admit up to the full global and per-principal limits — + /// so the bound silently became per-handle. [`RecoverySession`] holds the + /// root-wide `LOCK` and can only be obtained by taking it, which closes the + /// cross-process half; `OPEN_STAGING_ROOTS` closes the in-process half. + /// Together they make the second instance unrepresentable rather than + /// discouraged. + /// + /// The lock's root must be *this* root: a lock on a different root proves + /// nothing about this one, and accepting it would turn the proof back into + /// a convention. + /// + /// Construction is where accounting is rebuilt (see [`Self::reconstruct`]). + /// A root-global count that ignored what is already durable would be a + /// count of this process's uptime, not of the root. + /// + /// The directory is created and its parent synced here rather than lazily, + /// so the same-device check at session creation compares two directories + /// that both actually exist. A check against a path that is about to be + /// created would compare the device of whatever `metadata` happened to + /// resolve, which is how a same-device assertion becomes decorative. + pub fn open( + lock: &RecoverySession, + options: StoreOptions, + durability: Arc, + ) -> Result, StoreError> { + Self::open_with_device_probe(lock, options, durability, Box::new(StatDeviceProbe)) + } + + fn open_with_device_probe( + lock: &RecoverySession, + options: StoreOptions, + durability: Arc, + device: Box, + ) -> Result, StoreError> { + options.validate()?; + let canonical_root = std::fs::canonicalize(&options.root)?; + let locked_root = std::fs::canonicalize(lock.root())?; + if canonical_root != locked_root { + return Err(StoreError::InvalidConfiguration(format!( + "projection staging for {} was offered a root lock held on {}; a lock on \ + another root proves nothing about this one", + canonical_root.display(), + locked_root.display() + ))); + } + { + let mut roots = open_staging_roots() + .lock() + .expect("staging root registry mutex poisoned"); + if !roots.insert(canonical_root.clone()) { + return Err(StoreError::Conflict(format!( + "projection staging is already open for {}; its ceilings are root-global \ + and a second registry would admit a second full budget", + canonical_root.display() + ))); + } + } + + let staging_root = options.root.join(STAGING_DIR); + if !staging_root.exists() { + std::fs::create_dir_all(&staging_root)?; + crate::sys::fsync_dir(&options.root, &durability)?; + } + let maintenance = MaintenancePool::new(&options.root); + let staging = Arc::new(Self { + options, + staging_root, + canonical_root, + counters: Arc::new(StagingCounters::default()), + durability, + device, + registry: Mutex::new(Registry::default()), + maintenance, + }); + staging.reconstruct()?; + Ok(staging) + } + + pub fn counters(&self) -> &Arc { + &self.counters + } + + pub fn staging_root(&self) -> &Path { + &self.staging_root + } + + // --- restart reconstruction (deliverable 1: "restartable") ------------- + + /// Rebuild every durable session, its chunk index, and the whole occupancy + /// account from `/staging`. + /// + /// **A durable session directory and an abandoned materialization are + /// different states and must never share a cleanup path.** The commit point + /// of a session is the `rename_noreplace` that installs its `session` + /// record: the record is written to a temporary, fenced, and renamed, so + /// the final name only ever appears over complete, digest-checked bytes. + /// Therefore: + /// + /// * a directory holding a valid `session` record for its own ID is a + /// **final session**. It is reconstructed, it charges budget, and its + /// ID is thereafter occupied — which is what stops a reopen from + /// admitting it as new and then destroying it from an error path; + /// * a directory with no such record is an **abandoned materialization**: + /// a crash or an error between `create_dir` and that rename. Nothing + /// ever accounted for it and nothing can reference it, so it is + /// reclaimed here — through the same validate-the-whole-directory-first + /// path everything else uses, never `remove_dir_all`. + /// + /// Reconstruction charges budget without evaluating the ceilings. A root + /// whose durable occupancy exceeds a newly lowered configuration must still + /// open, or lowering a limit would strand the very sessions the operator + /// needs to expire; admission then refuses new work until the durable set + /// drains, which is the enforcement the ceiling actually owes. + fn reconstruct(self: &Arc) -> Result<(), StoreError> { + let mut abandoned: Vec<([u8; 16], PathBuf)> = Vec::new(); + for shard_entry in std::fs::read_dir(&self.staging_root)? { + let shard_path = shard_entry?.path(); + let Some(shard_index) = self.shard_index_of_directory(&shard_path) else { + self.counters.unrecognized_entries.fetch_add(1, Relaxed); + continue; + }; + for session_entry in std::fs::read_dir(&shard_path)? { + let session_path = session_entry?.path(); + let Some(session_id) = session_directory_id(&session_path) else { + self.counters.unrecognized_entries.fetch_add(1, Relaxed); + continue; + }; + match self.reconstruct_session(shard_index, &session_path, session_id)? { + Some(record) => { + self.insert_reconstructed(session_id, record); + self.counters.sessions_reconstructed.fetch_add(1, Relaxed); + } + None => abandoned.push((session_id, session_path)), + } + } + } + for (session_id, path) in abandoned { + self.reclaim_directory(path, session_id)?; + self.counters + .abandoned_materializations_reclaimed + .fetch_add(1, Relaxed); + } + Ok(()) + } + + fn shard_index_of_directory(&self, path: &Path) -> Option { + if !path.is_dir() { + return None; + } + let name = path.file_name()?.to_str()?; + if name.len() != 2 { + return None; + } + let index: u16 = name.parse().ok()?; + (index < self.options.shard_count).then_some(index) + } + + /// `Ok(None)` is "abandoned materialization", not "nothing to see". + fn reconstruct_session( + &self, + shard_index: u16, + directory: &Path, + session_id: [u8; 16], + ) -> Result, StoreError> { + let record_path = directory.join(SESSION_RECORD_NAME); + if !record_path.exists() { + return Ok(None); + } + let payload = + read_staging_artifact(&record_path, StagingArtifactKind::SessionRecord, session_id)?; + if payload.len() < 32 { + return Err(StoreError::Corruption(format!( + "staging session record {} is shorter than its membership root", + record_path.display() + ))); + } + let mut root_bytes = [0u8; 32]; + root_bytes.copy_from_slice(&payload[..32]); + let session = ProjectionStageSessionV1::decode_canonical(&payload[32..]).map_err(|e| { + StoreError::Corruption(format!( + "staging session record {} no longer decodes: {e}", + record_path.display() + )) + })?; + if session.session_id != session_id { + return Err(StoreError::Corruption(format!( + "staging session record {} names session {} but lives under {}", + record_path.display(), + hex::encode(session.session_id), + hex::encode(session_id) + ))); + } + let expected_shard = StoreOptions::shard_of( + &NamespaceId::from(session.destination_repo), + self.options.shard_count, + ); + if expected_shard != shard_index { + return Err(StoreError::Corruption(format!( + "staging session {} is stored under shard {shard_index} but its destination \ + repository belongs to shard {expected_shard}", + hex::encode(session_id) + ))); + } + let binding = ProjectionStageBinding { + session, + membership_root: ObjectId(root_bytes), + }; + + let mut chunks: BTreeMap = BTreeMap::new(); + let mut sealed_manifest: Option = None; + let mut written_objects = 0u64; + let mut written_bytes = 0u64; + for entry in std::fs::read_dir(directory)? { + let path = entry?.path(); + let Some(name) = path.file_name().and_then(|name| name.to_str()) else { + self.counters.unrecognized_entries.fetch_add(1, Relaxed); + continue; + }; + if !path.is_file() { + self.counters.unrecognized_entries.fetch_add(1, Relaxed); + continue; + } + // A `.tmp` never survived a rename, so it is by construction not + // durable state. It is left where it is; reclamation removes it. + if name.ends_with(".tmp") || name == SESSION_RECORD_NAME { + continue; + } + if name == MANIFEST_NAME { + sealed_manifest = Some(path); + continue; + } + let Some((ordinal, digest_from_name)) = parse_chunk_artifact_name(name) else { + self.counters.unrecognized_entries.fetch_add(1, Relaxed); + continue; + }; + let bytes = read_staging_artifact(&path, StagingArtifactKind::Chunk, session_id)?; + let chunk = ProjectionStageChunkV1::decode_canonical(&bytes).map_err(|e| { + StoreError::Corruption(format!( + "staged chunk artifact {} no longer decodes: {e}", + path.display() + )) + })?; + let digest = chunk.chunk_digest().map_err(|e| { + StoreError::Corruption(format!( + "staged chunk artifact {} digest: {e}", + path.display() + )) + })?; + if digest != digest_from_name + || chunk.ordinal != ordinal + || chunk.session_id != session_id + || chunk.chunk_count != binding.session.chunk_count + { + return Err(StoreError::Corruption(format!( + "staged chunk artifact {} no longer matches the ordinal, digest, session, \ + or chunk count it was stored under", + path.display() + ))); + } + written_objects = written_objects + .checked_add(u64::try_from(chunk.objects.len()).unwrap_or(u64::MAX)) + .ok_or_else(|| { + StoreError::Corruption("reconstructed staged object count overflowed".into()) + })?; + for object in &chunk.objects { + written_bytes = written_bytes + .checked_add(object.descriptor.raw_len) + .ok_or_else(|| { + StoreError::Corruption("reconstructed staged bytes overflowed".into()) + })?; + } + let file_bytes = std::fs::metadata(&path)?.len(); + if chunks + .insert( + ordinal, + ChunkSlot::Stored(StoredChunk { + digest, + path: path.clone(), + file_bytes, + }), + ) + .is_some() + { + return Err(StoreError::Corruption(format!( + "staging session {} has two durable artifacts for chunk ordinal {ordinal}", + hex::encode(session_id) + ))); + } + } + + let reservation = Reservation { + objects: binding.session.total_object_count, + bytes: binding.session.total_object_bytes, + files: u64::from(binding.session.chunk_count) + .checked_add(SESSION_FIXED_FILES) + .ok_or_else(|| { + StoreError::Corruption("reconstructed staging file count overflowed".into()) + })?, + }; + + // 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) => { + let resolution = + self.reconstruct_resolution(&path, &binding, session_id, &chunks)?; + (StagedSessionState::Sealed, Some(resolution)) + } + }; + + Ok(Some(SessionRecord { + binding, + shard_index, + directory: directory.to_path_buf(), + state, + busy: SessionBusy::Idle, + chunks, + written_objects, + written_bytes, + reservation, + resolution, + })) + } + + fn reconstruct_resolution( + &self, + manifest_path: &Path, + binding: &ProjectionStageBinding, + session_id: [u8; 16], + chunks: &BTreeMap, + ) -> Result, StoreError> { + let bytes = + read_staging_artifact(manifest_path, StagingArtifactKind::Manifest, session_id)?; + let manifest = ProjectionStageManifestV1::decode_canonical(&bytes).map_err(|e| { + StoreError::Corruption(format!( + "sealed staging manifest {} no longer decodes: {e}", + manifest_path.display() + )) + })?; + let manifest_digest = manifest.manifest_digest().map_err(|e| { + StoreError::Corruption(format!( + "sealed staging manifest {} digest: {e}", + manifest_path.display() + )) + })?; + if manifest.session_id != session_id + || manifest.membership_root != binding.membership_root + || manifest_digest != binding.session.manifest_digest + { + return Err(StoreError::Corruption(format!( + "sealed staging manifest {} no longer reconstructs the digest its session binds", + manifest_path.display() + ))); + } + let mut artifacts = Vec::with_capacity(chunks.len()); + for ordinal in 0..binding.session.chunk_count { + let stored = chunks + .get(&ordinal) + .and_then(ChunkSlot::stored) + .ok_or_else(|| { + StoreError::Corruption(format!( + "sealed staging session {} is missing durable chunk ordinal {ordinal}", + hex::encode(session_id) + )) + })?; + artifacts.push(ProjectionArtifact { + path: stored.path.clone(), + digest: stored.digest, + bytes: stored.file_bytes, + }); + } + Ok(Arc::new(ProjectionAdoptionResolution { + session: binding.session.clone(), + manifest, + artifacts: Arc::from(artifacts), + })) + } + + /// Charge a reconstructed session's budget. No ceiling is evaluated; see + /// [`Self::reconstruct`]. + fn insert_reconstructed(&self, session_id: [u8; 16], record: SessionRecord) { + let principal = record.principal(); + let reservation = record.reservation; + let written_bytes = record.written_bytes; + let mut registry = self.lock(); + let entry = registry.principals.entry(principal).or_default(); + entry.sessions += 1; + entry.objects += reservation.objects; + entry.bytes += reservation.bytes; + entry.files += reservation.files; + registry.global.sessions += 1; + registry.global.objects += reservation.objects; + registry.global.bytes += reservation.bytes; + registry.global.files += reservation.files; + registry.debt_reserved_bytes += reservation.bytes; + registry.sessions.insert(session_id, record); + drop(registry); + + let counters = &self.counters; + counters.sessions_live.fetch_add(1, Relaxed); + counters + .reserved_objects + .fetch_add(reservation.objects, Relaxed); + counters + .reserved_bytes + .fetch_add(reservation.bytes, Relaxed); + counters + .reserved_files + .fetch_add(reservation.files, Relaxed); + counters + .compaction_debt_reserved_bytes + .fetch_add(reservation.bytes, Relaxed); + counters + .compaction_debt_written_bytes + .fetch_add(written_bytes, Relaxed); + } + + /// Begin one session. Every refusal below happens before any budget is + /// charged and before any file exists. + pub fn begin( + self: &Arc, + binding: ProjectionStageBinding, + now_micros: i64, + ) -> Result { + match self.begin_inner(binding, now_micros) { + Ok(session) => Ok(session), + Err(error) => { + self.counters.sessions_refused.fetch_add(1, Relaxed); + Err(error) + } + } + } + + fn begin_inner( + self: &Arc, + binding: ProjectionStageBinding, + now_micros: i64, + ) -> Result { + // 1. The frozen binding must be internally well formed. `session_digest` + // runs `ProjectionStageSessionV1::validate`, which is where the + // nonzero-totals and Fork-proof-presence rules live. Restating them + // here would create a second opinion about a frozen contract. + let session_id = binding.session.session_id; + binding + .session + .session_digest() + .map_err(|e| StoreError::Conflict(format!("projection stage session binding: {e}")))?; + + // 2. Declared shape against the configured projection ceilings. + // + // The per-session `MAX_CANONICAL_ITEMS` checks that used to sit here + // are **deliberately gone.** Contract review 2026-07-28-A caps + // `max_projection_objects` and `max_projection_chunks` at that + // constant in `StoreOptions::validate`, so no configuration this + // store can open admits a declaration that reaches them. A typed + // refusal no caller can ever observe is not a belt: it advertises a + // limit name that will never appear in a real error, and it is a + // second opinion about a bound the configuration already owns — + // which is the exact defect the review deleted four copies of. + let options = &self.options; + require_at_most( + "max_projection_objects", + binding.session.total_object_count, + options.max_projection_objects, + )?; + require_at_most( + "max_projection_bytes", + binding.session.total_object_bytes, + options.max_projection_bytes, + )?; + require_at_most( + "max_projection_chunks", + u64::from(binding.session.chunk_count), + u64::from(options.max_projection_chunks), + )?; + // 3. Age and feasibility. `expires_at_micros` is the advertised + // maximum: there is no renewal entry point on this type, so the + // only way to outlive it is to create a new session, which pays for + // a new budget. + let lifetime = binding + .session + .expires_at_micros + .checked_sub(now_micros) + .ok_or_else(|| { + StoreError::Conflict("staging session lifetime arithmetic overflowed".into()) + })?; + if lifetime <= 0 { + return Err(StoreError::Conflict(format!( + "staging session {} expires at {} which is not after {now_micros}", + hex::encode(session_id), + binding.session.expires_at_micros + ))); + } + require_at_most( + "staging_session_max_age_micros", + u64::try_from(lifetime).expect("positive lifetime fits u64"), + u64::try_from(options.staging_session_max_age_micros) + .expect("options.rs refuses a non-positive maximum age"), + )?; + self.require_feasible(&binding, lifetime)?; + + // 4. Same device. Checked before any budget is charged, because a + // session on the wrong device can never be adopted at all. + let shard_index = StoreOptions::shard_of( + &NamespaceId::from(binding.session.destination_repo), + options.shard_count, + ); + self.require_same_device(shard_index)?; + + // 5. Occupancy. Ceilings and insertion under one lock acquisition. + let directory = self.session_directory(shard_index, &session_id); + let reservation = Reservation { + objects: binding.session.total_object_count, + bytes: binding.session.total_object_bytes, + files: u64::from(binding.session.chunk_count) + .checked_add(SESSION_FIXED_FILES) + .ok_or_else(|| { + StoreError::Conflict("staging session file count overflowed".into()) + })?, + }; + require_at_most( + "staging_max_files_per_session", + reservation.files, + options.staging_max_files_per_session, + )?; + self.admit( + &binding, + shard_index, + directory.clone(), + reservation, + now_micros, + )?; + + // 6. Only now does anything exist on disk. A failure here releases the + // budget rather than stranding it: a reservation nobody can abort + // is the leak the bound exists to prevent. + // + // The cleanup is the validated reclaim, never `remove_dir_all`. A + // recursive delete keyed only on a path is how an ordinary restart + // plus a materialization error destroys a session that was already + // durable; reclamation here removes only files this exact session + // could have written, and leaves the directory standing if it finds + // anything else. + if let Err(error) = self.materialize(&binding, &directory) { + self.release(session_id); + let _ = self.reclaim_directory(directory, session_id); + return Err(error); + } + + self.counters.sessions_created.fetch_add(1, Relaxed); + Ok(ProjectionStageSession { + staging: Arc::clone(self), + session_id, + }) + } + + /// Reattach to a live session by ID. This is what makes a session + /// restartable: the handle carries no state, so a caller that lost it — + /// or a fresh request for the same transfer — resumes with the same + /// budget, the same artifacts, and the same idempotent ordinals. + pub fn session( + self: &Arc, + session_id: [u8; 16], + ) -> Result { + let registry = self.lock(); + if !registry.sessions.contains_key(&session_id) { + return Err(unknown_session(session_id)); + } + drop(registry); + Ok(ProjectionStageSession { + staging: Arc::clone(self), + session_id, + }) + } + + /// Reclaim every session whose advertised expiry has passed. + /// + /// Legal in this pass only because no state reachable here can hold an + /// adoption pin: `StagedSessionState` has exactly `Open` and `Sealed`, and + /// neither has been handed to `submit`. Deliverable 6 adds `Finalizing`, + /// and deliverable 7 adds the committed-root reference proof; the match + /// below is exhaustive so both arrive as compile errors here rather than + /// as a silent reclamation of pinned artifacts. + pub fn expire(&self, now_micros: i64) -> Result { + // The candidate set is computed under the lock; every reclamation then + // runs on a maintenance worker with the lock released, so an expiry + // sweep never makes admission wait on a directory walk. + let registry = self.lock(); + let expired: Vec<[u8; 16]> = registry + .sessions + .iter() + .filter(|(_, record)| { + 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, + }; + past_expiry + && idle + && match record.state { + StagedSessionState::Open | StagedSessionState::Sealed => true, + } + }) + .map(|(id, _)| *id) + .collect(); + drop(registry); + + let mut reclaimed = 0u64; + for session_id in expired { + self.reclaim_session(session_id)?; + self.counters.sessions_expired.fetch_add(1, Relaxed); + reclaimed += 1; + } + Ok(reclaimed) + } + + /// Remove artifacts no committed manifest 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)", + )) + } + + // --- internals ------------------------------------------------------ + + fn lock(&self) -> RegistryGuard<'_> { + // Poisoning means a panic left the accounting half-applied. Continuing + // on it would spend a budget nobody can release, so this fails loudly. + let guard = self + .registry + .lock() + .expect("staging registry mutex poisoned by a panic mid-accounting"); + REGISTRY_LOCK_DEPTH.with(|depth| depth.set(depth.get() + 1)); + RegistryGuard(guard) + } + + /// Run one unit of filesystem work on a maintenance worker. + /// + /// The assertion is the deliverable. It is checked on every dispatch rather + /// than reviewed once, because "no disk I/O under the registry lock" is a + /// property a future edit removes by accident — moving a `write_artifact` + /// call three lines up is all it takes — and nothing else in the crate + /// would notice. + fn run_maintenance( + &self, + job: impl FnOnce() -> Result + Send + 'static, + ) -> Result { + assert_eq!( + REGISTRY_LOCK_DEPTH.with(Cell::get), + 0, + "staging dispatched filesystem work while holding the registry mutex; the \ + registry is where every root-global ceiling is evaluated and it may not be \ + held across a disk write (scope 6.5 deliverable 5)" + ); + let sender = self + .maintenance + .jobs + .as_ref() + .expect("the maintenance channel is closed only by Drop"); + let (reply, answer) = crossbeam_channel::bounded(1); + let counters = Arc::clone(&self.counters); + sender + .send(Box::new(move || { + let result = job(); + counters.maintenance_jobs.fetch_add(1, Relaxed); + let _ = reply.send(result); + })) + .map_err(|_| { + StoreError::Corruption("staging maintenance workers are not running".into()) + })?; + answer.recv().map_err(|_| { + StoreError::Corruption("a staging maintenance worker died mid-artifact".into()) + })? + } + + fn session_directory(&self, shard_index: u16, session_id: &[u8; 16]) -> PathBuf { + self.staging_root + .join(format!("{shard_index:02}")) + .join(hex::encode(session_id)) + } + + fn shard_directory(&self, shard_index: u16) -> PathBuf { + self.options + .root + .join("shards") + .join(format!("{shard_index:02}")) + } + + /// `/staging` and the target shard must share an `st_dev`. + /// + /// Adoption is a `link` into the shard's generation, exactly as scope 3.4's + /// seal is, and `link(2)` across devices is `EXDEV`. A copy fallback is + /// forbidden rather than merely unimplemented: a copy is not a rename, so + /// it reintroduces a window in which the adopted bytes exist under neither + /// a durable staging name nor a manifest-referenced one. Refusing here + /// costs one `stat`; discovering it at adoption costs a full transfer. + fn require_same_device(&self, shard_index: u16) -> Result<(), StoreError> { + let shard = self.shard_directory(shard_index); + let staging_device = self.device.device_of(&self.staging_root)?; + let shard_device = self.device.device_of(&shard)?; + if staging_device != shard_device { + return Err(StoreError::InvalidConfiguration(format!( + "{} is on device {staging_device} and {} is on device {shard_device}; \ + adoption is a link and a link across devices is not a rename, and a \ + copy fallback is forbidden", + self.staging_root.display(), + shard.display() + ))); + } + Ok(()) + } + + /// One complete transfer must fit in the session's own lifetime at the + /// supported floor rate, with the finalize margin left over. + /// + /// `options.rs` already proves the *configured maximum* is feasible at + /// startup. This is the per-session instance of the same arithmetic + /// against the bytes this session actually declares and the expiry it + /// actually asks for, which startup cannot see. A session that cannot + /// finish consumes budget for its whole life and then fails. + fn require_feasible( + &self, + binding: &ProjectionStageBinding, + lifetime_micros: i64, + ) -> Result<(), StoreError> { + let floor = self.options.minimum_projection_transfer_bytes_per_second; + let seconds = binding + .session + .total_object_bytes + .checked_add(floor - 1) + .map(|rounded| rounded / floor) + .ok_or_else(|| { + StoreError::Conflict("staged transfer duration arithmetic overflowed".into()) + })?; + let transfer_micros = seconds + .checked_mul(1_000_000) + .and_then(|micros| i64::try_from(micros).ok()) + .ok_or_else(|| { + StoreError::Conflict("staged transfer duration does not fit i64 micros".into()) + })?; + let required = transfer_micros + .checked_add(self.options.staging_finalize_margin_micros) + .ok_or_else(|| { + StoreError::Conflict("staged transfer plus finalize margin overflowed".into()) + })?; + if required > lifetime_micros { + return Err(StoreError::LimitExceeded { + limit: "staging_session_transfer_feasibility_micros", + observed: u64::try_from(required).unwrap_or(u64::MAX), + allowed: u64::try_from(lifetime_micros).unwrap_or(0), + }); + } + Ok(()) + } + + /// Evaluate every occupancy ceiling and insert, under one lock. + fn admit( + &self, + binding: &ProjectionStageBinding, + shard_index: u16, + directory: PathBuf, + reservation: Reservation, + now_micros: i64, + ) -> Result<(), StoreError> { + let options = &self.options; + let principal = binding.session.actor; + let session_id = binding.session.session_id; + let mut registry = self.lock(); + + if registry.sessions.contains_key(&session_id) { + return Err(StoreError::Conflict(format!( + "staging session {} already exists; a session is restartable by ID, \ + not re-creatable", + hex::encode(session_id) + ))); + } + + let used = registry + .principals + .get(&principal) + .copied() + .unwrap_or_default(); + let global = registry.global; + let retry_after = registry.retry_after_micros(now_micros, options); + + overload_at_most( + "staging_max_sessions_per_principal", + used.sessions + 1, + u64::from(options.staging_max_sessions_per_principal), + retry_after, + )?; + overload_at_most( + "staging_max_sessions_global", + global.sessions + 1, + u64::from(options.staging_max_sessions_global), + retry_after, + )?; + overload_sum( + "staging_max_objects_per_principal", + used.objects, + reservation.objects, + options.staging_max_objects_per_principal, + retry_after, + )?; + overload_sum( + "staging_max_objects_global", + global.objects, + reservation.objects, + options.staging_max_objects_global, + retry_after, + )?; + overload_sum( + "staging_max_bytes_per_principal", + used.bytes, + reservation.bytes, + options.staging_max_bytes_per_principal, + retry_after, + )?; + overload_sum( + "staging_max_bytes_global", + global.bytes, + reservation.bytes, + options.staging_max_bytes_global, + retry_after, + )?; + overload_sum( + "staging_max_files_per_principal", + used.files, + reservation.files, + options.staging_max_files_per_principal, + retry_after, + )?; + overload_sum( + "staging_max_files_global", + global.files, + reservation.files, + options.staging_max_files_global, + retry_after, + )?; + // Compaction debt is reserved from the declared bytes, not accrued as + // chunks land. Charging it on arrival would let a session be admitted + // whose completion is already guaranteed to breach the ceiling, which + // is the same defect the feasibility check exists to prevent one axis + // over. + overload_sum( + "staging_max_compaction_debt_bytes", + registry.debt_reserved_bytes, + reservation.bytes, + options.staging_max_compaction_debt_bytes, + retry_after, + )?; + + let entry = registry.principals.entry(principal).or_default(); + entry.sessions += 1; + entry.objects += reservation.objects; + entry.bytes += reservation.bytes; + entry.files += reservation.files; + registry.global.sessions += 1; + registry.global.objects += reservation.objects; + registry.global.bytes += reservation.bytes; + registry.global.files += reservation.files; + registry.debt_reserved_bytes += reservation.bytes; + registry.sessions.insert( + session_id, + SessionRecord { + binding: binding.clone(), + shard_index, + directory, + state: StagedSessionState::Open, + busy: SessionBusy::Idle, + chunks: BTreeMap::new(), + written_objects: 0, + written_bytes: 0, + reservation, + resolution: None, + }, + ); + + let counters = &self.counters; + counters.sessions_live.fetch_add(1, Relaxed); + counters + .reserved_objects + .fetch_add(reservation.objects, Relaxed); + counters + .reserved_bytes + .fetch_add(reservation.bytes, Relaxed); + counters + .reserved_files + .fetch_add(reservation.files, Relaxed); + counters + .compaction_debt_reserved_bytes + .fetch_add(reservation.bytes, Relaxed); + Ok(()) + } + + /// Release one session's budget. Idempotent by absence. + fn release(&self, session_id: [u8; 16]) { + let mut registry = self.lock(); + self.release_locked(&mut registry, session_id); + } + + fn release_locked(&self, registry: &mut Registry, session_id: [u8; 16]) { + let Some(record) = registry.sessions.remove(&session_id) else { + return; + }; + let principal = record.principal(); + let reservation = record.reservation; + 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(record.written_bytes, Relaxed); + } + + /// Create the session directory and write the durable session record, on a + /// maintenance worker. + /// + /// The renamed `session` record is this session's durability commit point, + /// so **every directory entry on the path to it must be synced first.** + /// Syncing only the immediate parent was a real gap: when + /// `staging/` was itself created by this call, its own entry under + /// `staging/` was never synced, so the first session in a shard could + /// survive as fenced bytes under a directory that no longer exists. + fn materialize( + &self, + binding: &ProjectionStageBinding, + directory: &Path, + ) -> Result<(), StoreError> { + let mut payload = binding.membership_root.0.to_vec(); + payload.extend_from_slice( + &binding + .session + .encode_canonical() + .map_err(|e| StoreError::Conflict(format!("staging session record: {e}")))?, + ); + let bytes = encode_staging_artifact( + StagingArtifactKind::SessionRecord, + binding.session.session_id, + &payload, + ); + let directory = directory.to_path_buf(); + let staging_root = self.staging_root.clone(); + let durability = Arc::clone(&self.durability); + let counters = Arc::clone(&self.counters); + self.run_maintenance(move || { + create_session_directory(&staging_root, &directory, &durability)?; + write_artifact( + &directory, + SESSION_RECORD_NAME, + &bytes, + &durability, + &counters, + )?; + Ok(()) + }) + } + + /// Validate a session directory completely, then reclaim it, on a + /// maintenance worker. + fn reclaim_directory( + &self, + directory: PathBuf, + session_id: [u8; 16], + ) -> Result<(), StoreError> { + let durability = Arc::clone(&self.durability); + let counters = Arc::clone(&self.counters); + self.run_maintenance(move || { + reclaim_session_directory(&directory, session_id, &durability, &counters) + }) + } + + /// End one accounted session: exclude it, reclaim its directory off the + /// registry lock, and release its budget only if the reclamation succeeded. + /// + /// The budget is released last on purpose. Releasing first and then failing + /// would leave files on disk that nothing accounts for and no later call + /// can name — the leak the bound exists to prevent — while this ordering + /// leaves a live session and a named error. + fn reclaim_session(&self, session_id: [u8; 16]) -> Result<(), StoreError> { + let directory = { + let mut registry = self.lock(); + let record = registry + .sessions + .get_mut(&session_id) + .ok_or_else(|| unknown_session(session_id))?; + match record.state { + StagedSessionState::Open | StagedSessionState::Sealed => {} + } + match record.busy { + SessionBusy::Idle => {} + SessionBusy::Sealing | SessionBusy::Reclaiming => { + return Err(StoreError::Overloaded { + limit: "staging_session_maintenance_in_flight", + retry_after_micros: 1, + }) + } + } + record.busy = SessionBusy::Reclaiming; + record.directory.clone() + }; + + let result = self.reclaim_directory(directory, session_id); + let mut registry = self.lock(); + match result { + Ok(()) => { + self.release_locked(&mut registry, session_id); + Ok(()) + } + Err(error) => { + if let Some(record) = registry.sessions.get_mut(&session_id) { + record.busy = SessionBusy::Idle; + } + Err(error) + } + } + } +} + +impl Registry { + /// Honest retry guidance: the soonest a live session can free budget is + /// its own advertised expiry. Derived from the sessions actually holding + /// the budget, never a constant — a fixed retry hint is a guess that gets + /// stale in exactly the overloaded state it is issued in. + fn retry_after_micros(&self, now_micros: i64, options: &StoreOptions) -> u64 { + self.sessions + .values() + .map(|record| record.binding.session.expires_at_micros) + .map(|expires| expires.saturating_sub(now_micros).max(1)) + .min() + .map(|micros| u64::try_from(micros).unwrap_or(u64::MAX)) + .unwrap_or_else(|| { + u64::try_from(options.staging_finalize_margin_micros) + .unwrap_or(0) + .max(1) + }) + } +} + +/// One staging session. Opaque: begin, idempotent numbered chunk-put, +/// read-only resolver, seal, abort (plan §5.1). +/// +/// Dropping this is deliberately not an abort. Sessions are restartable by ID +/// and survive the process; a handle is a way to name one, not ownership of +/// it. The only things that end a session are abort, expiry, and — from +/// deliverable 6 — adoption. +pub struct ProjectionStageSession { + staging: Arc, + session_id: [u8; 16], +} + +impl fmt::Debug for ProjectionStageSession { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("ProjectionStageSession") + .field("session_id", &hex::encode(self.session_id)) + .finish_non_exhaustive() + } +} + +impl ProjectionStageSession { + pub fn session_id(&self) -> [u8; 16] { + self.session_id + } + + /// Read-only status. Carries no artifact bytes and no adoption capability. + pub fn describe(&self) -> Result { + let registry = self.staging.lock(); + let record = registry + .sessions + .get(&self.session_id) + .ok_or_else(|| unknown_session(self.session_id))?; + Ok(StagedSessionStatus { + session_id: self.session_id, + state: record.state, + shard_index: record.shard_index, + // Durable chunks only. An ordinal whose artifact is still being + // written by a maintenance worker is reserved, not present, and + // reporting it as present would let a caller conclude a seal would + // succeed when the bytes are not fenced yet. + chunks_present: u32::try_from( + record + .chunks + .values() + .filter(|slot| slot.stored().is_some()) + .count(), + ) + .unwrap_or(u32::MAX), + chunks_expected: record.binding.session.chunk_count, + written_objects: record.written_objects, + written_bytes: record.written_bytes, + expires_at_micros: record.binding.session.expires_at_micros, + }) + } + + /// Store one numbered chunk, idempotently by ordinal and digest. + /// + /// The chunk is validated by the frozen codec before anything is written: + /// `chunk_digest` runs `ProjectionStageChunkV1::validate`, which checks + /// framing, embedded types, IDs, exact bytes, canonical order, and the + /// declared ordinal against the declared chunk count. None of that is + /// restated here — a second implementation of a frozen validation rule is + /// a second opinion, and the two only have to disagree once. + pub fn put_chunk( + &self, + chunk: &ProjectionStageChunkV1, + now_micros: i64, + ) -> Result { + let canonical = chunk + .encode_canonical() + .map_err(|e| StoreError::Conflict(format!("staged chunk: {e}")))?; + let chunk_digest = chunk + .chunk_digest() + .map_err(|e| StoreError::Conflict(format!("staged chunk digest: {e}")))?; + + let staging = &self.staging; + let mut registry = staging.lock(); + let record = registry + .sessions + .get(&self.session_id) + .ok_or_else(|| unknown_session(self.session_id))?; + require_open(record, self.session_id)?; + require_live(record, self.session_id, now_micros)?; + require_idle(record, self.session_id)?; + + if chunk.session_id != self.session_id { + return Err(StoreError::Conflict(format!( + "staged chunk names session {} but was offered to session {}", + hex::encode(chunk.session_id), + hex::encode(self.session_id) + ))); + } + if chunk.chunk_count != record.binding.session.chunk_count { + return Err(StoreError::Conflict(format!( + "staged chunk declares {} chunks but session {} is bound to {}", + chunk.chunk_count, + hex::encode(self.session_id), + record.binding.session.chunk_count + ))); + } + + // Restartability: an identical ordinal/digest is a repeat of work + // already durable, so it costs no write. A different digest at a + // bound ordinal is a different projection wearing the same session, + // and is refused rather than overwritten. + if let Some(existing) = record.chunks.get(&chunk.ordinal) { + if existing.digest() != chunk_digest { + return Err(StoreError::Conflict(format!( + "staged chunk ordinal {} of session {} is already bound to digest {}", + chunk.ordinal, + hex::encode(self.session_id), + hex::encode(existing.digest().0) + ))); + } + return match existing { + ChunkSlot::Stored(_) => { + staging.counters.chunks_deduplicated.fetch_add(1, Relaxed); + Ok(ChunkPutOutcome::AlreadyPresent) + } + // A concurrent identical put is already writing this ordinal. + // Answering `AlreadyPresent` here would claim durability the + // artifact has not reached yet, which is the one thing this + // outcome is relied on to mean. + ChunkSlot::InFlight { .. } => Err(StoreError::Overloaded { + limit: "staging_chunk_ordinal_in_flight", + retry_after_micros: 1, + }), + }; + } + + let object_count = u64::try_from(chunk.objects.len()) + .map_err(|_| StoreError::Conflict("staged chunk object count overflowed".into()))?; + let object_bytes = chunk + .objects + .iter() + .try_fold(0u64, |total, object| { + total.checked_add(object.descriptor.raw_len) + }) + .ok_or_else(|| StoreError::Conflict("staged chunk byte count overflowed".into()))?; + + // The uploaded totals may never exceed what creation reserved. This is + // why the reservation is taken from the *declared* totals rather than + // accrued as bytes arrive: the global and per-principal ceilings are + // then unreachable by uploading at all, and the only place they can be + // breached is admission, where they are enforced atomically. + let session = &record.binding.session; + require_at_most( + "projection_stage_session_objects", + record + .written_objects + .checked_add(object_count) + .ok_or_else(|| StoreError::Conflict("staged object count overflowed".into()))?, + session.total_object_count, + )?; + require_at_most( + "projection_stage_session_bytes", + record + .written_bytes + .checked_add(object_bytes) + .ok_or_else(|| StoreError::Conflict("staged byte count overflowed".into()))?, + session.total_object_bytes, + )?; + + // Reserve the ordinal *and its share of the declared totals*, then + // release the registry before touching the disk. The reservation is + // what makes the write safe to perform unlocked: a second put of this + // ordinal now sees `InFlight` and is answered typed instead of racing + // to the same `rename_noreplace`, and two concurrent puts of different + // ordinals cannot both pass the totals check against the same + // pre-write figure. + let directory = record.directory.clone(); + let name = chunk_artifact_name(chunk.ordinal, &chunk_digest); + let ordinal = chunk.ordinal; + let record = registry + .sessions + .get_mut(&self.session_id) + .ok_or_else(|| unknown_session(self.session_id))?; + record.chunks.insert( + ordinal, + ChunkSlot::InFlight { + digest: chunk_digest, + }, + ); + record.written_objects += object_count; + record.written_bytes += object_bytes; + drop(registry); + staging + .counters + .compaction_debt_written_bytes + .fetch_add(object_bytes, Relaxed); + + let bytes = + encode_staging_artifact(StagingArtifactKind::Chunk, self.session_id, &canonical); + let written = { + let directory = directory.clone(); + let name = name.clone(); + let durability = Arc::clone(&staging.durability); + let counters = Arc::clone(&staging.counters); + staging.run_maintenance(move || { + write_artifact(&directory, &name, &bytes, &durability, &counters) + }) + }; + + let mut registry = staging.lock(); + let record = registry + .sessions + .get_mut(&self.session_id) + .ok_or_else(|| unknown_session(self.session_id))?; + let file_bytes = match written { + Ok(file_bytes) => file_bytes, + Err(error) => { + // The ordinal and its totals are released, not left reserved: + // a reservation nothing can retire would make the chunk + // unputtable for the rest of the session's life. + record.chunks.remove(&ordinal); + record.written_objects -= object_count; + record.written_bytes -= object_bytes; + drop(registry); + staging + .counters + .compaction_debt_written_bytes + .fetch_sub(object_bytes, Relaxed); + return Err(error); + } + }; + record.chunks.insert( + ordinal, + ChunkSlot::Stored(StoredChunk { + digest: chunk_digest, + path: directory.join(&name), + file_bytes, + }), + ); + drop(registry); + + staging.counters.chunks_written.fetch_add(1, Relaxed); + Ok(ChunkPutOutcome::Stored) + } + + /// Reconstruct the exact ordered manifest and produce the adoption + /// descriptor. + /// + /// **Sealing publishes nothing.** It returns a canonical descriptor and no + /// capability: `ProjectionAdoption` is issued only by finalize, and only + /// `submit` may consume it. That is plan §4 identity invariant 7 — normal + /// possession of a session ID never grants membership to pre-uploaded + /// bytes. + pub fn seal(&self, now_micros: i64) -> Result { + let staging = &self.staging; + + // Phase 1, under the registry: admit exactly one sealer and take the + // physical inputs. Everything after this — reading every chunk + // artifact back, rehashing it, and writing the manifest — is disk work + // and runs on a maintenance worker with the registry released. + let (binding, stored, directory) = { + let mut registry = staging.lock(); + let record = registry + .sessions + .get(&self.session_id) + .ok_or_else(|| unknown_session(self.session_id))?; + require_open(record, self.session_id)?; + require_live(record, self.session_id, now_micros)?; + require_idle(record, self.session_id)?; + + let expected_chunks = record.binding.session.chunk_count; + let mut stored = Vec::with_capacity(usize::try_from(expected_chunks).unwrap_or(0)); + for ordinal in 0..expected_chunks { + match record.chunks.get(&ordinal).and_then(ChunkSlot::stored) { + Some(chunk) => stored.push(chunk.clone()), + None => { + let present = record + .chunks + .values() + .filter(|slot| slot.stored().is_some()) + .count(); + return Err(StoreError::Conflict(format!( + "staging session {} holds {present} of {expected_chunks} chunks", + hex::encode(self.session_id) + ))); + } + } + } + let binding = record.binding.clone(); + let directory = record.directory.clone(); + let record = registry + .sessions + .get_mut(&self.session_id) + .ok_or_else(|| unknown_session(self.session_id))?; + record.busy = SessionBusy::Sealing; + (binding, stored, directory) + }; + + let session_id = self.session_id; + let durability = Arc::clone(&staging.durability); + let counters = Arc::clone(&staging.counters); + let composed = staging.run_maintenance(move || { + compose_seal( + session_id, + binding, + stored, + directory, + &durability, + &counters, + ) + }); + + // Phase 3, under the registry: publish the sealed state, or clear the + // exclusion and hand the caller 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; + let (install, resolution) = composed?; + record.state = StagedSessionState::Sealed; + record.resolution = Some(Arc::new(resolution)); + drop(registry); + staging.counters.sessions_sealed.fetch_add(1, Relaxed); + Ok(install) + } + + /// Read-only resolution of a sealed session. + /// + /// Read-only is load-bearing: the adopting side revalidates, and a view + /// that could mutate would let adoption repair what it was meant to + /// reject. + pub fn resolve(&self) -> Result, StoreError> { + let registry = self.staging.lock(); + let record = registry + .sessions + .get(&self.session_id) + .ok_or_else(|| unknown_session(self.session_id))?; + match record.state { + StagedSessionState::Sealed => {} + StagedSessionState::Open => { + return Err(StoreError::Conflict(format!( + "staging session {} is still open and has no sealed manifest to resolve", + hex::encode(self.session_id) + ))) + } + } + record.resolution.clone().ok_or_else(|| { + StoreError::Corruption("sealed staging session lost its resolution".into()) + }) + } +} + +/// The whole of sealing that touches the disk, on a maintenance worker. +/// +/// Split out as a free function for the same reason `write_artifact` is: taking +/// `&self` here is what let a full re-read of every staged chunk happen while +/// the one lock every ceiling is evaluated under was held. +fn compose_seal( + session_id: [u8; 16], + binding: ProjectionStageBinding, + stored: Vec, + directory: PathBuf, + durability: &DurabilityCounters, + counters: &StagingCounters, +) -> Result<(StagedProjectionInstallV1, ProjectionAdoptionResolution), StoreError> { + { + let session = &binding.session; + let expected_chunks = session.chunk_count; + + // Ordinal order is manifest order. The frozen binding validator reads + // chunk `i` at manifest position `i`, so composing in any other order + // would produce a manifest that no adopter could reproduce. + let mut chunk_digests = Vec::with_capacity(stored.len()); + let mut objects: Vec = Vec::new(); + let mut artifacts = Vec::with_capacity(stored.len()); + let mut total_bytes = 0u64; + for (ordinal, stored) in (0..expected_chunks).zip(stored.iter()) { + let payload = + read_staging_artifact(&stored.path, StagingArtifactKind::Chunk, session_id)?; + let chunk = ProjectionStageChunkV1::decode_canonical(&payload).map_err(|e| { + // Our own fenced artifact failed a decode it passed on the way + // in. That is store state gone bad, not a caller's mistake. + StoreError::Corruption(format!( + "staged chunk artifact {} no longer decodes: {e}", + stored.path.display() + )) + })?; + let digest = chunk.chunk_digest().map_err(|e| { + StoreError::Corruption(format!( + "staged chunk artifact {} digest: {e}", + stored.path.display() + )) + })?; + if digest != stored.digest || chunk.ordinal != ordinal { + return Err(StoreError::Corruption(format!( + "staged chunk artifact {} no longer matches the ordinal/digest it was \ + stored under", + stored.path.display() + ))); + } + for object in &chunk.objects { + total_bytes = total_bytes + .checked_add(object.descriptor.raw_len) + .ok_or_else(|| { + StoreError::Conflict("staged object byte count overflowed".into()) + })?; + objects.push(object.descriptor.clone()); + } + chunk_digests.push(digest); + artifacts.push(ProjectionArtifact { + path: stored.path.clone(), + digest, + bytes: stored.file_bytes, + }); + } + + let manifest = ProjectionStageManifestV1 { + session_id, + chunk_digests, + objects, + membership_root: binding.membership_root, + }; + // `manifest_digest` runs the frozen manifest validation: nonempty, + // bounded, strictly sorted and unique. Concatenating chunks in ordinal + // order therefore has to produce a globally ordered object list or the + // seal fails here, which is the ordering rule and not a copy of it. + let manifest_digest = manifest + .manifest_digest() + .map_err(|e| StoreError::Conflict(format!("staged manifest: {e}")))?; + let object_count = u64::try_from(manifest.objects.len()) + .map_err(|_| StoreError::Conflict("staged object count overflowed".into()))?; + if manifest_digest != session.manifest_digest { + return Err(StoreError::Conflict(format!( + "staging session {} binds manifest digest {} but its chunks and membership \ + root reconstruct to {}", + hex::encode(session_id), + hex::encode(session.manifest_digest.0), + hex::encode(manifest_digest.0) + ))); + } + if object_count != session.total_object_count || total_bytes != session.total_object_bytes { + return Err(StoreError::Conflict(format!( + "staging session {} binds {}/{} objects/bytes but its chunks carry {}/{}", + hex::encode(session_id), + session.total_object_count, + session.total_object_bytes, + object_count, + total_bytes + ))); + } + + let install = StagedProjectionInstallV1 { + session_id, + manifest_digest, + projection: session.projection, + object_count, + object_bytes: total_bytes, + membership_root: manifest.membership_root, + artifact_set_digest: artifact_set_digest(&artifacts), + }; + // The descriptor must be encodable, because a final frame carries it. + install + .encode_canonical() + .map_err(|e| StoreError::Conflict(format!("staged projection install: {e}")))?; + + let manifest_bytes = encode_staging_artifact( + StagingArtifactKind::Manifest, + session_id, + &manifest + .encode_canonical() + .map_err(|e| StoreError::Conflict(format!("staged manifest: {e}")))?, + ); + write_artifact( + &directory, + MANIFEST_NAME, + &manifest_bytes, + durability, + counters, + )?; + + Ok(( + install, + ProjectionAdoptionResolution { + session: session.clone(), + manifest, + artifacts: Arc::from(artifacts), + }, + )) + } +} + +impl ProjectionStageSession { + /// Abort and reclaim. Explicit cancellation, per plan §8. + pub fn abort(self) -> Result<(), StoreError> { + let staging = Arc::clone(&self.staging); + staging.reclaim_session(self.session_id)?; + staging.counters.sessions_aborted.fetch_add(1, Relaxed); + Ok(()) + } + + /// Move `Open -> 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. + #[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)", + )) + } +} + +/// Deliverable 8, one method answered and two still deferred. +/// +/// 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 +/// 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. +impl ProjectionRecoveryResolver for ProjectionStaging { + /// The transferred set, **derived rather than asserted, and provably empty + /// today.** + /// + /// 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 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". + fn transferred_sessions(&self, shard_index: u16) -> Result, StoreError> { + let registry = self.lock(); + let transferred: Vec<[u8; 16]> = registry + .sessions + .iter() + .filter(|(_, record)| record.shard_index == shard_index) + .filter(|(_, record)| match record.state { + StagedSessionState::Open | StagedSessionState::Sealed => false, + }) + .map(|(session_id, _)| *session_id) + .collect(); + Ok(Arc::from(transferred)) + } + + fn resolve_committed( + &self, + _namespace: NamespaceId, + _descriptor: &StagedProjectionInstallV1, + ) -> Result { + Err(StoreError::NotImplemented( + "ProjectionStaging::resolve_committed — B3 StagingSessions, scope 6.5 \ + deliverable 8 (recovery treatment of unreferenced artifacts)", + )) + } + + fn notify_recovered( + &self, + _resolution: RecoveredProjectionResolution, + ) -> Result<(), StoreError> { + Err(StoreError::NotImplemented( + "ProjectionStaging::notify_recovered — B3 StagingSessions, scope 6.5 \ + deliverable 8 (recovery treatment of unreferenced artifacts)", + )) + } +} + +// --- free helpers --------------------------------------------------------- + +/// Create `//` and sync every directory entry the new +/// session directory depends on, outermost first. +fn create_session_directory( + staging_root: &Path, + directory: &Path, + durability: &DurabilityCounters, +) -> Result<(), StoreError> { + let parent = directory + .parent() + .expect("a session directory always has a shard parent"); + let parent_is_new = !parent.exists(); + std::fs::create_dir_all(directory)?; + if parent_is_new { + // `staging/`'s own entry, in `staging/`. Without this the shard + // directory can vanish on a crash and take a fully fenced session + // with it. + crate::sys::fsync_dir(staging_root, durability)?; + } + // ``'s entry, in `staging/`. + crate::sys::fsync_dir(parent, durability)?; + Ok(()) +} + +/// Write one uniquely named, unreferenced, fenced artifact. +/// +/// Ordering matches `checkpoint::install`: temporary, fence, then a no-replace +/// rename, then a directory fsync. The final name therefore only ever appears +/// over complete, fenced bytes. `rename_noreplace` is the load-bearing half of +/// "uniquely named": if a name were ever reused the rename fails rather than +/// quietly replacing an artifact another ordinal or another session still +/// accounts for. +/// +/// A free function, and asserted to run on a maintenance worker, because +/// 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. +fn write_artifact( + directory: &Path, + name: &str, + bytes: &[u8], + durability: &DurabilityCounters, + counters: &StagingCounters, +) -> Result { + 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(name).display() + ); + let final_path = directory.join(name); + let tmp_path = directory.join(format!("{name}.tmp")); + { + let mut file = File::options() + .create(true) + .write(true) + .truncate(true) + .open(&tmp_path)?; + 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 staging artifact {}: wrote {end} of {} bytes; the \ + temporary is left behind and never renamed", + final_path.display(), + bytes.len() + ), + ))); + } + crate::sys::fdatasync(&file, durability)?; + } + crate::sys::rename_noreplace(&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(bytes.len() as u64) +} + +/// Prove the whole directory is reclaimable, and only then unlink anything. +/// +/// The two-phase shape is the point. Unlinking as the scan walks the directory +/// means a valid marked artifact can be destroyed and *then* an unexpected file +/// encountered, which leaves a live session partially reclaimed: neither +/// removed nor usable, and no longer able to seal. Reclamation either takes the +/// whole directory or takes nothing and says which file stopped it. +fn reclaim_session_directory( + directory: &Path, + session_id: [u8; 16], + durability: &DurabilityCounters, + counters: &StagingCounters, +) -> Result<(), StoreError> { + if !directory.exists() { + return Ok(()); + } + let mut plan: Vec = Vec::new(); + for entry in std::fs::read_dir(directory)? { + let path = entry?.path(); + let name = path + .file_name() + .and_then(|name| name.to_str()) + .unwrap_or_default() + .to_owned(); + if !path.is_file() { + return Err(StoreError::Corruption(format!( + "staging session directory {} holds {}, which is not a file; reclamation \ + removes nothing rather than deleting around a surprise", + directory.display(), + path.display() + ))); + } + // A `.tmp` under one of this session's own artifact names never + // survived a rename, so it is ours by construction and may carry a + // torn header. Nothing else in the directory is taken on faith. + if let Some(stem) = name.strip_suffix(".tmp") { + if is_session_artifact_name(stem) { + plan.push(path); + continue; + } + } + if artifact_session_marker(&path)? == Some(session_id) { + plan.push(path); + continue; + } + return Err(StoreError::Corruption(format!( + "staging session directory {} holds {}, which carries no valid marker for \ + session {}; reclamation is not a licence to delete whatever is in the path", + directory.display(), + path.display(), + hex::encode(session_id) + ))); + } + + let mut unlinked = 0u64; + for path in &plan { + crate::sys::unlink(path)?; + unlinked += 1; + } + crate::sys::fsync_dir(directory, durability)?; + std::fs::remove_dir(directory)?; + if let Some(parent) = directory.parent() { + crate::sys::fsync_dir(parent, durability)?; + } + counters.artifacts_unlinked.fetch_add(unlinked, Relaxed); + Ok(()) +} + +/// A name this session's own writer could have produced. +fn is_session_artifact_name(name: &str) -> bool { + name == SESSION_RECORD_NAME + || name == MANIFEST_NAME + || parse_chunk_artifact_name(name).is_some() +} + +/// The inverse of [`chunk_artifact_name`], so reconstruction reads the ordinal +/// and digest a name binds instead of trusting the file's contents about them. +fn parse_chunk_artifact_name(name: &str) -> Option<(u32, ObjectId)> { + let rest = name.strip_prefix("chunk-")?; + let (ordinal, digest) = rest.split_once('-')?; + if ordinal.len() != 10 || digest.len() != 64 { + return None; + } + let ordinal: u32 = ordinal.parse().ok()?; + let mut bytes = [0u8; 32]; + hex::decode_to_slice(digest, &mut bytes).ok()?; + Some((ordinal, ObjectId(bytes))) +} + +/// The session ID a directory name binds, or `None` if the name is not one. +fn session_directory_id(path: &Path) -> Option<[u8; 16]> { + if !path.is_dir() { + return None; + } + let name = path.file_name()?.to_str()?; + if name.len() != 32 { + return None; + } + let mut session_id = [0u8; 16]; + hex::decode_to_slice(name, &mut session_id).ok()?; + Some(session_id) +} + +fn unknown_session(session_id: [u8; 16]) -> StoreError { + StoreError::Conflict(format!("no staging session {}", hex::encode(session_id))) +} + +fn require_open(record: &SessionRecord, session_id: [u8; 16]) -> Result<(), StoreError> { + match record.state { + StagedSessionState::Open => Ok(()), + StagedSessionState::Sealed => Err(StoreError::Conflict(format!( + "staging session {} is sealed and immutable", + 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 +/// is: deliverable 6 adds a pinned case, and it must arrive here as a compile +/// error rather than be silently swept into "busy". +fn require_idle(record: &SessionRecord, session_id: [u8; 16]) -> Result<(), StoreError> { + match record.busy { + SessionBusy::Idle => Ok(()), + SessionBusy::Sealing => Err(StoreError::Conflict(format!( + "staging session {} is sealing", + hex::encode(session_id) + ))), + SessionBusy::Reclaiming => Err(StoreError::Conflict(format!( + "staging session {} is being reclaimed", + hex::encode(session_id) + ))), + } +} + +/// Expiry comparison, matching the frozen `validate_projection_stage_finalize` +/// exactly: a session is usable *at* its expiry instant and not after it. A +/// stricter or looser comparison here would make the store and the protocol +/// disagree about a one-microsecond boundary, which is the kind of divergence +/// that only shows up as an unreproducible failure at the deadline. +fn require_live( + record: &SessionRecord, + session_id: [u8; 16], + now_micros: i64, +) -> Result<(), StoreError> { + if now_micros > record.binding.session.expires_at_micros { + return Err(StoreError::Conflict(format!( + "staging session {} expired at {}", + hex::encode(session_id), + record.binding.session.expires_at_micros + ))); + } + Ok(()) +} + +/// A bound the request violates by itself, regardless of load. Typed refusal, +/// never a clamp and never an eviction (decision 9.8). +fn require_at_most(limit: &'static str, observed: u64, allowed: u64) -> Result<(), StoreError> { + if observed > allowed { + return Err(StoreError::LimitExceeded { + limit, + observed, + allowed, + }); + } + Ok(()) +} + +/// A bound the request violates only because of concurrent occupancy. Plan +/// §5.1 reserves `StoreError` for inability to answer, and a store whose +/// staging budget is spent genuinely cannot answer *yet* — which is what +/// distinguishes this from `LimitExceeded`, where retrying changes nothing. +fn overload_at_most( + limit: &'static str, + observed: u64, + allowed: u64, + retry_after_micros: u64, +) -> Result<(), StoreError> { + if observed > allowed { + return Err(StoreError::Overloaded { + limit, + retry_after_micros, + }); + } + Ok(()) +} + +fn overload_sum( + limit: &'static str, + used: u64, + requested: u64, + allowed: u64, + retry_after_micros: u64, +) -> Result<(), StoreError> { + let total = used.checked_add(requested).ok_or(StoreError::Overloaded { + limit, + retry_after_micros, + })?; + overload_at_most(limit, total, allowed, retry_after_micros) +} + +/// Chunk artifact name: ordinal for order, digest for content. +/// +/// Both halves are required. The ordinal alone would let a retry with +/// different bytes reuse a name; the digest alone would let two ordinals +/// carrying identical bytes collide onto one file and make the manifest +/// position ambiguous. +fn chunk_artifact_name(ordinal: u32, digest: &ObjectId) -> String { + format!("chunk-{ordinal:010}-{}", hex::encode(digest.0)) +} + +/// Digest over the artifact set, in ordinal order, by content and size only. +/// +/// Paths are deliberately excluded: the shard owner assigns manifest-visible +/// names at adoption, so a digest over paths would change at exactly the +/// moment it is supposed to prove nothing changed. +fn artifact_set_digest(artifacts: &[ProjectionArtifact]) -> ObjectId { + let mut bytes = Vec::with_capacity(artifacts.len() * 40); + for (ordinal, artifact) in artifacts.iter().enumerate() { + bytes.extend_from_slice(&(ordinal as u32).to_le_bytes()); + bytes.extend_from_slice(&artifact.digest.0); + bytes.extend_from_slice(&artifact.bytes.to_le_bytes()); + } + digest(STAGING_ARTIFACT_SET_DIGEST_DOMAIN, &bytes) +} + +fn encode_staging_artifact( + kind: StagingArtifactKind, + session_id: [u8; 16], + payload: &[u8], +) -> Vec { + let mut bytes = Vec::with_capacity(STAGING_ARTIFACT_HEADER_LEN + payload.len() + 32); + bytes.extend_from_slice(&STAGING_ARTIFACT_MAGIC); + bytes.extend_from_slice(&STAGING_ARTIFACT_VERSION.to_le_bytes()); + bytes.extend_from_slice(&kind.code().to_le_bytes()); + bytes.extend_from_slice(&session_id); + bytes.extend_from_slice(&(payload.len() as u32).to_le_bytes()); + bytes.extend_from_slice(payload); + let trailer = digest(STAGING_ARTIFACT_DIGEST_DOMAIN, &bytes); + bytes.extend_from_slice(&trailer.0); + bytes +} + +/// The session marker a file carries, or `None` if it carries none. +/// +/// Deliverable 7's "valid session marker" in its cheapest form. A file whose +/// header is absent, truncated, wrong-magic, wrong-version, or fails its own +/// trailer digest has no marker, so reclamation leaves it alone rather than +/// guessing from its name or its directory. +fn artifact_session_marker(path: &Path) -> Result, StoreError> { + let bytes = std::fs::read(path)?; + Ok(parse_staging_artifact(&bytes).map(|(_, session_id, _)| session_id)) +} + +fn parse_staging_artifact(bytes: &[u8]) -> Option<(StagingArtifactKind, [u8; 16], &[u8])> { + if bytes.len() < STAGING_ARTIFACT_HEADER_LEN + 32 { + return None; + } + if bytes[..8] != STAGING_ARTIFACT_MAGIC { + return None; + } + if u16::from_le_bytes([bytes[8], bytes[9]]) != STAGING_ARTIFACT_VERSION { + return None; + } + let kind = StagingArtifactKind::from_code(u16::from_le_bytes([bytes[10], bytes[11]]))?; + let mut session_id = [0u8; 16]; + session_id.copy_from_slice(&bytes[12..28]); + let payload_len = u32::from_le_bytes([bytes[28], bytes[29], bytes[30], bytes[31]]) as usize; + let payload_end = STAGING_ARTIFACT_HEADER_LEN.checked_add(payload_len)?; + if bytes.len() != payload_end + 32 { + return None; + } + if digest(STAGING_ARTIFACT_DIGEST_DOMAIN, &bytes[..payload_end]).0 != bytes[payload_end..] { + return None; + } + Some(( + kind, + session_id, + &bytes[STAGING_ARTIFACT_HEADER_LEN..payload_end], + )) +} + +fn read_staging_artifact( + path: &Path, + kind: StagingArtifactKind, + session_id: [u8; 16], +) -> Result, StoreError> { + let bytes = std::fs::read(path)?; + let (actual_kind, actual_session, payload) = + parse_staging_artifact(&bytes).ok_or_else(|| { + StoreError::Corruption(format!( + "staging artifact {} has no valid session marker", + path.display() + )) + })?; + if actual_kind != kind || actual_session != session_id { + return Err(StoreError::Corruption(format!( + "staging artifact {} carries a marker for a different kind or session", + path.display() + ))); + } + Ok(payload.to_vec()) +} + #[cfg(test)] mod tests { use std::fs::File; @@ -512,3 +3101,271 @@ mod tests { assert!(matches!(error, StoreError::Corruption(message) if message.contains("ownership"))); } } + +/// B3-owned unit tests. +/// +/// These cover the two seams an integration test cannot reach: the device +/// probe, which needs a second `st_dev` under one temporary root, and the +/// artifact marker, which is the physical representation cleanup will key on. +/// Everything else about the session lifecycle is asserted through the public +/// entry points in `tests/staging_sessions.rs`, because a bound proved only +/// against an internal helper is charter item 8's decoy. +#[cfg(test)] +mod b3_tests { + use std::collections::BTreeMap; + + use levcs_protocol::v2::{ProjectionMode, StageSourceKindV1}; + use tempfile::TempDir; + + use super::*; + + const HOUR_MICROS: i64 = 3_600_000_000; + + struct FixedDeviceProbe { + devices: BTreeMap, + } + + impl DeviceProbe for FixedDeviceProbe { + fn device_of(&self, path: &Path) -> Result { + self.devices.get(path).copied().ok_or_else(|| { + StoreError::Corruption(format!("no device configured for {}", path.display())) + }) + } + } + + /// A real v2 root plus the held root lock staging now requires. + /// + /// The lock is returned, not dropped: it is the ownership proof, and a + /// fixture that let it die would be testing a constructor production can + /// never reach. + fn layout(directory: &TempDir) -> (StoreOptions, RecoverySession) { + let mut options = StoreOptions::new(directory.path()); + options.shard_count = 4; + crate::segment::initialize_root( + &crate::segment::RootLayout::new(directory.path()), + options.shard_count, + [42; 16], + 0, + &DurabilityCounters::default(), + ) + .expect("root layout"); + let lock = RecoverySession::open(directory.path()).expect("root lock"); + (options, lock) + } + + fn binding(session_id: [u8; 16], expires_at_micros: i64) -> ProjectionStageBinding { + ProjectionStageBinding { + session: ProjectionStageSessionV1 { + session_id, + destination_repo: ObjectId([7; 32]), + destination_genesis: ObjectId([8; 32]), + expected_authority: ObjectId([9; 32]), + projection: ProjectionMode::Full, + source_kind: StageSourceKindV1::Mirror, + actor: [11; 32], + actor_key_epoch: 3, + source_generation_digest: ObjectId([12; 32]), + fork_proof: None, + final_operation_id: [13; 16], + final_operation_digest: ObjectId([14; 32]), + final_evidence_digest: ObjectId([15; 32]), + total_object_count: 2, + total_object_bytes: 64, + chunk_count: 1, + manifest_digest: ObjectId([16; 32]), + expires_at_micros, + }, + membership_root: ObjectId([17; 32]), + } + } + + #[test] + fn cross_device_staging_is_refused_at_session_creation() { + let directory = TempDir::new().unwrap(); + let (options, lock) = layout(&directory); + let staging_root = directory.path().join(STAGING_DIR); + let shard = StoreOptions::shard_of(&NamespaceId([7; 32]), options.shard_count); + let shard_directory = directory.path().join("shards").join(format!("{shard:02}")); + + let staging = ProjectionStaging::open_with_device_probe( + &lock, + options, + Arc::new(DurabilityCounters::default()), + Box::new(FixedDeviceProbe { + devices: BTreeMap::from([(staging_root, 64), (shard_directory, 65)]), + }), + ) + .expect("staging opens"); + + let result = staging.begin(binding([1; 16], HOUR_MICROS), 0); + let Err(StoreError::InvalidConfiguration(message)) = result else { + panic!("a cross-device session must be refused, got {result:?}"); + }; + assert!( + message.contains("link across devices"), + "the refusal must say why a copy fallback is not an option: {message}" + ); + + // Refused before pinning: no budget charged and no session directory. + let counters = staging.counters().snapshot(); + assert_eq!(counters.sessions_live, 0); + assert_eq!(counters.reserved_bytes, 0); + assert_eq!(counters.compaction_debt_reserved_bytes, 0); + assert_eq!(counters.sessions_refused, 1); + assert!(!directory + .path() + .join(STAGING_DIR) + .join(format!("{shard:02}")) + .exists()); + } + + #[test] + fn same_device_staging_is_admitted_through_the_production_probe() { + let directory = TempDir::new().unwrap(); + let (options, lock) = layout(&directory); + let staging = + ProjectionStaging::open(&lock, options, Arc::new(DurabilityCounters::default())) + .expect("staging opens"); + staging + .begin(binding([2; 16], HOUR_MICROS), 0) + .expect("a same-device session is admitted by the real st_dev probe"); + } + + #[test] + fn the_production_device_probe_reports_the_real_st_dev() { + // The substituted probe is only honest if the real one means exactly + // this. Without this assertion the cross-device test could pass while + // production compared something else entirely. + let directory = TempDir::new().unwrap(); + let probe = StatDeviceProbe; + let observed = probe.device_of(directory.path()).expect("probe"); + let expected = std::fs::metadata(directory.path()).unwrap().dev(); + assert_eq!(observed, expected); + } + + #[test] + fn a_foreign_or_tampered_file_carries_no_session_marker() { + let directory = TempDir::new().unwrap(); + let session_id = [21u8; 16]; + let encoded = + encode_staging_artifact(StagingArtifactKind::Chunk, session_id, b"payload bytes"); + + let good = directory.path().join("good"); + std::fs::write(&good, &encoded).unwrap(); + assert_eq!(artifact_session_marker(&good).unwrap(), Some(session_id)); + assert_eq!( + read_staging_artifact(&good, StagingArtifactKind::Chunk, session_id).unwrap(), + b"payload bytes" + ); + + // A file that is not ours at all. + let foreign = directory.path().join("foreign"); + std::fs::write(&foreign, b"not a staging artifact").unwrap(); + assert_eq!(artifact_session_marker(&foreign).unwrap(), None); + + // One flipped payload byte breaks the trailer digest, so the marker is + // gone rather than merely suspect. Reclamation must not delete a file + // it cannot prove it owns. + let mut tampered = encoded.clone(); + let last_payload = STAGING_ARTIFACT_HEADER_LEN; + tampered[last_payload] ^= 0xff; + let tampered_path = directory.path().join("tampered"); + std::fs::write(&tampered_path, &tampered).unwrap(); + assert_eq!(artifact_session_marker(&tampered_path).unwrap(), None); + + // A valid artifact belonging to another session is refused by kind or + // session, not accepted because it parsed. + let other = read_staging_artifact(&good, StagingArtifactKind::Manifest, session_id); + let Err(StoreError::Corruption(message)) = other else { + panic!("a kind mismatch must be refused, got {other:?}"); + }; + assert!(message.contains("different kind or session")); + } + + /// Reconstruction reads a chunk's ordinal and digest out of its *name* + /// and then checks the file against them, so the name parser has to be + /// the exact inverse of the name builder. If it were not, a reopen would + /// either drop durable chunks (name unrecognized) or accept a file under + /// an ordinal it was never stored at. + #[test] + fn the_chunk_artifact_name_round_trips_and_rejects_everything_else() { + let digest = ObjectId([9; 32]); + let name = chunk_artifact_name(4_294_967_295, &digest); + assert_eq!( + parse_chunk_artifact_name(&name), + Some((4_294_967_295, digest)) + ); + assert_eq!(parse_chunk_artifact_name("chunk-0000000000"), None); + assert_eq!(parse_chunk_artifact_name("chunk-4-abc"), None); + assert_eq!(parse_chunk_artifact_name(SESSION_RECORD_NAME), None); + assert_eq!(parse_chunk_artifact_name(MANIFEST_NAME), None); + assert_eq!(parse_chunk_artifact_name(&format!("{name}.tmp")), None); + 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("not-ours")); + } + + #[test] + fn finalize_is_deferred_and_names_its_deliverable() { + let directory = TempDir::new().unwrap(); + let (options, lock) = layout(&directory); + let staging = + ProjectionStaging::open(&lock, options, Arc::new(DurabilityCounters::default())) + .expect("staging opens"); + 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"); + }; + assert!(detail.contains("deliverable 6"), "{detail}"); + } + + #[test] + fn the_recovery_resolver_seam_is_deferred_and_names_its_deliverable() { + 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. + let session = staging + .begin(binding([9; 16], HOUR_MICROS), 0) + .expect("session"); + let shard = StoreOptions::shard_of(&NamespaceId([7; 32]), 4); + assert!(staging.transferred_sessions(shard).unwrap().is_empty()); + drop(session); + + let resolved = staging.resolve_committed( + NamespaceId([1; 32]), + &StagedProjectionInstallV1 { + session_id: [4; 16], + manifest_digest: ObjectId([5; 32]), + projection: ProjectionMode::Full, + object_count: 1, + object_bytes: 1, + membership_root: ObjectId([6; 32]), + artifact_set_digest: ObjectId([7; 32]), + }, + ); + let Err(StoreError::NotImplemented(detail)) = resolved else { + panic!("resolve_committed must not answer before deliverable 8"); + }; + assert!(detail.contains("deliverable 8"), "{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}"); + } +} diff --git a/crates/levcs-store/tests/staging_sessions.rs b/crates/levcs-store/tests/staging_sessions.rs new file mode 100644 index 0000000..1370ed3 --- /dev/null +++ b/crates/levcs-store/tests/staging_sessions.rs @@ -0,0 +1,1207 @@ +//! B3 StagingSessions — scope 6.5 deliverables 1-5, asserted through the +//! entry points a consumer actually calls. +//! +//! Charter item 8 is the organizing rule here. Every bound, every refusal, and +//! every durability claim below is driven through `ProjectionStaging::begin`, +//! `ProjectionStageSession::put_chunk`, `::seal`, `::abort`, or +//! `ProjectionStaging::expire` — never through an internal helper. A helper +//! that enforces a ceiling correctly and is not on the path `begin` takes is +//! not a bound; it is a decoy that makes this file look finished. +//! +//! Charter item 7 is the second rule: occupancy and sync behavior are read out +//! of `StagingCounters` and `DurabilityCounters`, not asserted about. + +use std::sync::atomic::Ordering::Relaxed; +use std::sync::Arc; + +use levcs_core::{blake3_hash, ObjectHeader, ObjectId, ObjectType, FORMAT_VERSION}; +use levcs_protocol::v2::{ + validate_projection_stage_binding, ProjectionMode, ProjectionStageChunkV1, + ProjectionStageSessionV1, StageSourceKindV1, StagedChunkObjectV1, StagedObjectV1, +}; +use levcs_store::recovery::RecoverySession; +use levcs_store::segment::{initialize_root, RootLayout}; +use levcs_store::staging::{ + ChunkPutOutcome, ProjectionStageBinding, ProjectionStaging, StagedSessionState, +}; +use levcs_store::{CommittedRoot, DurabilityCounters, NamespaceId, StoreError, StoreOptions}; +use tempfile::TempDir; + +const HOUR_MICROS: i64 = 3_600_000_000; + +// --------------------------------------------------------------------------- +// Fixtures +// --------------------------------------------------------------------------- + +/// A real v2 store root plus the held root lock. +/// +/// The lock is part of the fixture because it is part of the constructor: +/// staging's ceilings are root-global, and `ProjectionStaging::open` requires +/// proof that this process holds `/LOCK` so a second accountant for one +/// root is not expressible. A fixture that forged its way past that would test +/// a constructor production cannot reach. +struct Root { + directory: TempDir, + lock: RecoverySession, + options: StoreOptions, +} + +impl Root { + fn new() -> Self { + Self::with(|_| {}) + } + + fn with(mutate: impl FnOnce(&mut StoreOptions)) -> Self { + let directory = TempDir::new().unwrap(); + let mut options = StoreOptions::new(directory.path()); + options.shard_count = 4; + mutate(&mut options); + initialize_root( + &RootLayout::new(directory.path()), + options.shard_count, + [42; 16], + 0, + &DurabilityCounters::default(), + ) + .expect("v2 root layout"); + let lock = RecoverySession::open(directory.path()).expect("root lock"); + Self { + directory, + lock, + options, + } + } + + fn path(&self) -> &std::path::Path { + self.directory.path() + } + + fn open(&self) -> (Arc, Arc) { + let durability = Arc::new(DurabilityCounters::default()); + let staging = self.open_with(Arc::clone(&durability)); + (staging, durability) + } + + fn open_with(&self, durability: Arc) -> Arc { + ProjectionStaging::open(&self.lock, self.options.clone(), durability) + .expect("staging root opens") + } + + fn session_directory(&self, session_id: [u8; 16]) -> std::path::PathBuf { + let shard = StoreOptions::shard_of(&NamespaceId([7; 32]), self.options.shard_count); + self.path() + .join("staging") + .join(format!("{shard:02}")) + .join(hex::encode(session_id)) + } +} + +/// One canonical unsigned object. Real bytes, because the frozen chunk codec +/// parses them and checks the embedded type against the outer descriptor. +fn blob(body: &[u8]) -> StagedChunkObjectV1 { + let mut raw = ObjectHeader { + object_type: ObjectType::Blob, + format_version: FORMAT_VERSION, + body_len: body.len() as u64, + } + .encode() + .to_vec(); + raw.extend_from_slice(body); + let id = blake3_hash(&raw); + StagedChunkObjectV1 { + descriptor: StagedObjectV1 { + object_id: id, + object_type: ObjectType::Blob as u8, + raw_len: raw.len() as u64, + raw_digest: id, + }, + raw_bytes: raw, + } +} + +struct Fixture { + binding: ProjectionStageBinding, + chunks: Vec, +} + +impl Fixture { + fn session_id(&self) -> [u8; 16] { + self.binding.session.session_id + } +} + +/// Build a self-consistent projection: globally sorted objects split into +/// `chunk_count` chunks, the manifest they reconstruct to, and a session bound +/// to that manifest's digest. +fn projection( + session_id: [u8; 16], + actor: [u8; 32], + chunk_count: u32, + per_chunk: usize, + expires_at_micros: i64, +) -> Fixture { + assert!(chunk_count >= 1 && per_chunk >= 1); + let total = chunk_count as usize * per_chunk; + let mut objects: Vec = (0..total) + .map(|index| blob(format!("staged-object-{}-{index}", hex::encode(session_id)).as_bytes())) + .collect(); + // The manifest must be strictly sorted, and the manifest is the ordered + // concatenation of the chunks, so the global sort has to happen before the + // split rather than inside each chunk. + objects.sort_by(|left, right| left.descriptor.cmp(&right.descriptor)); + + let total_object_bytes = objects + .iter() + .map(|object| object.descriptor.raw_len) + .sum::(); + let descriptors: Vec = objects + .iter() + .map(|object| object.descriptor.clone()) + .collect(); + + let mut chunks = Vec::with_capacity(chunk_count as usize); + for ordinal in 0..chunk_count { + let start = ordinal as usize * per_chunk; + chunks.push(ProjectionStageChunkV1 { + session_id, + ordinal, + chunk_count, + objects: objects[start..start + per_chunk].to_vec(), + }); + } + let chunk_digests: Vec = chunks + .iter() + .map(|chunk| chunk.chunk_digest().expect("chunk digest")) + .collect(); + + // The membership root is the instance layer's commitment, not the store's. + // Any stable value works here; B3 stores it and proves the manifest digest + // binds it, and evaluates nothing about it. + let membership_root = blake3_hash(&session_id[..]); + let manifest = levcs_protocol::v2::ProjectionStageManifestV1 { + session_id, + chunk_digests, + objects: descriptors, + membership_root, + }; + let manifest_digest = manifest.manifest_digest().expect("manifest digest"); + + Fixture { + binding: ProjectionStageBinding { + session: ProjectionStageSessionV1 { + session_id, + destination_repo: ObjectId([7; 32]), + destination_genesis: ObjectId([8; 32]), + expected_authority: ObjectId([9; 32]), + projection: ProjectionMode::Full, + source_kind: StageSourceKindV1::Mirror, + actor, + actor_key_epoch: 3, + source_generation_digest: ObjectId([12; 32]), + fork_proof: None, + final_operation_id: [13; 16], + final_operation_digest: ObjectId([14; 32]), + final_evidence_digest: ObjectId([15; 32]), + total_object_count: total as u64, + total_object_bytes, + chunk_count, + manifest_digest, + expires_at_micros, + }, + membership_root, + }, + chunks, + } +} + +// --------------------------------------------------------------------------- +// Deliverable 1 and 2 — lifecycle and the canonical binding +// --------------------------------------------------------------------------- + +#[test] +fn seal_reconstructs_a_manifest_the_frozen_binding_validator_accepts() { + let root = Root::new(); + let (staging, _durability) = root.open(); + let fixture = projection([1; 16], [21; 32], 3, 2, HOUR_MICROS); + + let session = staging.begin(fixture.binding.clone(), 0).expect("begin"); + for chunk in &fixture.chunks { + assert_eq!( + session.put_chunk(chunk, 0).expect("put"), + ChunkPutOutcome::Stored + ); + } + let install = session.seal(0).expect("seal"); + let resolution = session.resolve().expect("resolve"); + + // The frozen validator is the oracle, not a second opinion assembled here. + // Seal composes chunk digests, manifest position, totals, and the install + // descriptor from its own on-disk state; this proves that composition is + // exactly what the contract accepts. + validate_projection_stage_binding( + &fixture.binding.session, + &fixture.chunks, + &resolution.manifest, + &install, + ) + .expect("the sealed descriptor must satisfy the frozen binding contract"); + + assert_eq!(install.session_id, fixture.session_id()); + assert_eq!(install.projection, fixture.binding.session.projection); + assert_eq!( + install.object_count, + fixture.binding.session.total_object_count + ); + assert_eq!( + install.object_bytes, + fixture.binding.session.total_object_bytes + ); + assert_eq!( + install.manifest_digest, + fixture.binding.session.manifest_digest + ); + assert_eq!(install.membership_root, fixture.binding.membership_root); + assert_eq!(resolution.artifacts.len(), fixture.chunks.len()); + + let status = session.describe().expect("describe"); + assert_eq!(status.state, StagedSessionState::Sealed); + assert_eq!(status.chunks_present, status.chunks_expected); + assert_eq!(staging.counters().sessions_sealed.load(Relaxed), 1); +} + +/// The security property, from the side this pass can prove. +/// +/// A sealed session yields a *descriptor* and a read-only resolution. It does +/// not yield an adoption capability: `ProjectionAdoption` has no public +/// constructor and `finalize` is crate-private, so nothing outside the crate +/// can turn possession of a session ID into something `submit` would accept. +/// The other half — that the objects are invisible to a reader until a +/// `submit` adopts the descriptor — is +/// `sealed_objects_stay_invisible_until_a_submit_adopts_them` below, and is +/// blocked on B1. +#[test] +fn sealing_yields_a_descriptor_and_no_adoption_capability() { + let root = Root::new(); + let (staging, _durability) = root.open(); + let fixture = projection([2; 16], [21; 32], 1, 2, HOUR_MICROS); + let session = staging.begin(fixture.binding.clone(), 0).expect("begin"); + session.put_chunk(&fixture.chunks[0], 0).expect("put"); + session.seal(0).expect("seal"); + + // Resolution is read-only: it hands back shared immutable state, so an + // adopter can reject it but cannot repair it. + let first = session.resolve().expect("resolve"); + let second = session.resolve().expect("resolve"); + assert!(Arc::ptr_eq(&first, &second)); + + // A sealed session is immutable: a further chunk put is refused rather + // than quietly extending a projection somebody has already been handed a + // descriptor for. + let result = session.put_chunk(&fixture.chunks[0], 0); + let Err(StoreError::Conflict(message)) = result else { + panic!("a sealed session must refuse further chunks, got {result:?}"); + }; + assert!(message.contains("sealed"), "{message}"); +} + +#[test] +fn chunk_put_is_idempotent_by_ordinal_and_digest_and_costs_no_second_write() { + let root = Root::new(); + let (staging, durability) = root.open(); + let fixture = projection([3; 16], [21; 32], 2, 1, HOUR_MICROS); + let session = staging.begin(fixture.binding.clone(), 0).expect("begin"); + + assert_eq!( + session.put_chunk(&fixture.chunks[0], 0).expect("put"), + ChunkPutOutcome::Stored + ); + let after_first = durability.snapshot(); + + assert_eq!( + session.put_chunk(&fixture.chunks[0], 0).expect("repeat"), + ChunkPutOutcome::AlreadyPresent + ); + // Counters, not claims: a restart re-offering a chunk performs no write, + // no fence, and no directory sync. + assert_eq!(durability.snapshot(), after_first); + assert_eq!(staging.counters().chunks_deduplicated.load(Relaxed), 1); + assert_eq!(staging.counters().chunks_written.load(Relaxed), 1); +} + +#[test] +fn a_different_digest_at_a_bound_ordinal_is_refused() { + let root = Root::new(); + let (staging, _durability) = root.open(); + let fixture = projection([4; 16], [21; 32], 2, 1, HOUR_MICROS); + let session = staging.begin(fixture.binding.clone(), 0).expect("begin"); + session.put_chunk(&fixture.chunks[0], 0).expect("put"); + + // Chunk 1's objects offered at ordinal 0: same session, same declared + // chunk count, different bytes. + let impostor = ProjectionStageChunkV1 { + session_id: fixture.session_id(), + ordinal: 0, + chunk_count: fixture.binding.session.chunk_count, + objects: fixture.chunks[1].objects.clone(), + }; + let result = session.put_chunk(&impostor, 0); + let Err(StoreError::Conflict(message)) = result else { + panic!("a rebound ordinal must be refused, got {result:?}"); + }; + assert!(message.contains("already bound to digest"), "{message}"); + assert_eq!(staging.counters().chunks_written.load(Relaxed), 1); +} + +/// Handle reuse only. This proves a *dropped handle* is not an abort; it +/// deliberately does not claim anything about a process restart, which is +/// `a_real_reopen_reconstructs_sessions_chunks_and_accounting_from_disk`. +#[test] +fn dropping_a_session_handle_is_not_an_abort() { + let root = Root::new(); + let (staging, _durability) = root.open(); + let fixture = projection([5; 16], [21; 32], 2, 1, HOUR_MICROS); + let session_id = fixture.session_id(); + + { + let session = staging.begin(fixture.binding.clone(), 0).expect("begin"); + session.put_chunk(&fixture.chunks[0], 0).expect("put"); + // Dropping the handle must not abort: sessions outlive the party + // holding them, which is what "restartable" means. + } + + let resumed = staging.session(session_id).expect("resume by ID"); + assert_eq!( + resumed.put_chunk(&fixture.chunks[0], 0).expect("repeat"), + ChunkPutOutcome::AlreadyPresent + ); + resumed.put_chunk(&fixture.chunks[1], 0).expect("finish"); + resumed.seal(0).expect("seal"); +} + +#[test] +fn seal_refuses_an_incomplete_chunk_set() { + let root = Root::new(); + let (staging, _durability) = root.open(); + let fixture = projection([6; 16], [21; 32], 3, 1, HOUR_MICROS); + let session = staging.begin(fixture.binding.clone(), 0).expect("begin"); + session.put_chunk(&fixture.chunks[0], 0).expect("put"); + session.put_chunk(&fixture.chunks[2], 0).expect("put"); + + let result = session.seal(0); + let Err(StoreError::Conflict(message)) = result else { + panic!("a partial chunk set must not seal, got {result:?}"); + }; + assert!(message.contains("2 of 3 chunks"), "{message}"); +} + +/// A binding is only binding if a mismatch is refused, and refused for the +/// same reason the frozen validator would refuse it. +#[test] +fn seal_refuses_a_projection_the_session_did_not_bind() { + let root = Root::new(); + let (staging, _durability) = root.open(); + let fixture = projection([7; 16], [21; 32], 1, 2, HOUR_MICROS); + + // Same chunks, a membership root the session's manifest digest does not + // commit to. Nothing about the uploaded bytes changed. + let mut tampered = fixture.binding.clone(); + tampered.membership_root = ObjectId([99; 32]); + + let session = staging.begin(tampered, 0).expect("begin"); + session.put_chunk(&fixture.chunks[0], 0).expect("put"); + let result = session.seal(0); + let Err(StoreError::Conflict(message)) = result else { + panic!("an unbound manifest must not seal, got {result:?}"); + }; + assert!(message.contains("binds manifest digest"), "{message}"); +} + +// --------------------------------------------------------------------------- +// Deliverable 4 — bounds +// --------------------------------------------------------------------------- + +#[test] +fn a_declared_projection_over_a_configured_ceiling_is_refused_before_pinning() { + // Contract review 2026-07-28-A caps `max_projection_bytes` at + // `max_projection_chunks * MAX_CANONICAL_BYTES`, so the chunk ceiling and + // the byte ceiling can no longer be varied one at a time: lowering chunks + // alone is now an unsatisfiable configuration that `StoreOptions::validate` + // refuses at open, and the fixture would die before reaching `begin`. Each + // case therefore sets a configuration the store will actually accept, and + // the assertion is unchanged — the *declaration* is what must be refused, + // and refused before anything is pinned. + const CANONICAL_BYTES: u64 = levcs_protocol::codec::MAX_CANONICAL_BYTES as u64; + for (limit, mutate) in [ + ( + "max_projection_objects", + (|options: &mut StoreOptions| options.max_projection_objects = 1) + as fn(&mut StoreOptions), + ), + ("max_projection_bytes", |options| { + options.max_projection_bytes = 1 + }), + ("max_projection_chunks", |options| { + options.max_projection_chunks = 1; + options.max_projection_bytes = CANONICAL_BYTES; + options.staging_max_bytes_per_principal = CANONICAL_BYTES; + options.staging_max_bytes_global = CANONICAL_BYTES; + options.staging_max_compaction_debt_bytes = CANONICAL_BYTES; + }), + ] { + let root = Root::with(mutate); + let (staging, durability) = root.open(); + let before = durability.snapshot(); + + let fixture = projection([8; 16], [21; 32], 2, 2, HOUR_MICROS); + let result = staging.begin(fixture.binding, 0); + let Err(StoreError::LimitExceeded { + limit: observed_limit, + .. + }) = result + else { + panic!("{limit} must be a typed refusal, got {result:?}"); + }; + assert_eq!(observed_limit, limit); + + // "Before pinning" is the whole point: no budget, no file, no fence. + let counters = staging.counters().snapshot(); + assert_eq!(counters.sessions_live, 0); + assert_eq!(counters.reserved_bytes, 0); + assert_eq!(counters.reserved_files, 0); + assert_eq!(counters.compaction_debt_reserved_bytes, 0); + assert_eq!(durability.snapshot(), before); + } +} + +#[test] +fn a_session_older_than_the_advertised_maximum_is_refused() { + let root = Root::new(); + let maximum = root.options.staging_session_max_age_micros; + let (staging, _durability) = root.open(); + + let fixture = projection([9; 16], [21; 32], 1, 1, maximum + 1); + let result = staging.begin(fixture.binding, 0); + let Err(StoreError::LimitExceeded { + limit, + observed, + allowed, + }) = result + else { + panic!("an over-long session must be refused, got {result:?}"); + }; + assert_eq!(limit, "staging_session_max_age_micros"); + assert_eq!(observed, maximum as u64 + 1); + assert_eq!(allowed, maximum as u64); + + // And exactly at the maximum it is admitted, so the bound is the bound and + // not an off-by-one that happens to reject. + let fixture = projection([10; 16], [21; 32], 1, 1, maximum); + staging + .begin(fixture.binding, 0) + .expect("the boundary value is legal"); +} + +/// The feasibility check: configured floor rate, finalize margin, and the +/// session's own expiry must make one complete transfer possible. +#[test] +fn a_session_that_cannot_finish_is_refused_before_pinning() { + let root = Root::new(); + // One second of transfer at the floor rate plus the configured margin. + let margin = root.options.staging_finalize_margin_micros; + let (staging, durability) = root.open(); + let before = durability.snapshot(); + + let fixture = projection([11; 16], [21; 32], 1, 1, margin); + let result = staging.begin(fixture.binding.clone(), 0); + let Err(StoreError::LimitExceeded { + limit, + observed, + allowed, + }) = result + else { + panic!("an infeasible session must be refused, got {result:?}"); + }; + assert_eq!(limit, "staging_session_transfer_feasibility_micros"); + assert!( + observed > allowed, + "the refusal must report the shortfall it computed: {observed} vs {allowed}" + ); + + let counters = staging.counters().snapshot(); + assert_eq!(counters.sessions_live, 0); + assert_eq!(counters.reserved_bytes, 0); + assert_eq!(durability.snapshot(), before); + + // The same projection with one microsecond more than the computed + // requirement is admitted, so the refusal is arithmetic and not a blanket + // rejection of short sessions. + let mut feasible = fixture.binding; + feasible.session.expires_at_micros = observed as i64; + staging + .begin(feasible, 0) + .expect("a session with exactly the required lifetime is feasible"); +} + +#[test] +fn the_global_session_ceiling_is_enforced_atomically_with_insertion() { + // Per-principal is 1 and every thread uses its own principal, so the only + // bound in play is the global one. + let root = Root::with(|options| { + options.staging_max_sessions_per_principal = 1; + options.staging_max_sessions_global = 4; + }); + let (staging, _durability) = root.open(); + + let threads = 32u8; + let admitted = std::thread::scope(|scope| { + let handles: Vec<_> = (0..threads) + .map(|index| { + let staging = Arc::clone(&staging); + scope.spawn(move || { + // Dropping the handle is not an abort, so the budget stays + // held for the whole race whatever order the threads end in. + let fixture = projection([index; 16], [index; 32], 1, 1, HOUR_MICROS); + match staging.begin(fixture.binding, 0) { + Ok(_session) => true, + Err(error) => { + let StoreError::Overloaded { + limit, + retry_after_micros, + } = &error + else { + panic!("a ceiling must refuse by overload, got {error:?}"); + }; + assert_eq!(*limit, "staging_max_sessions_global"); + assert!( + *retry_after_micros > 0, + "an overload must carry usable retry guidance" + ); + false + } + } + }) + }) + .collect(); + handles + .into_iter() + .map(|handle| handle.join().expect("thread")) + .filter(|admitted| *admitted) + .count() + }); + + // A check-then-insert race admits entries past the ceiling under exactly + // this concurrency (decision 9.8). The count is the assertion. + assert_eq!(admitted, 4); + assert_eq!(staging.counters().sessions_live.load(Relaxed), 4); + assert_eq!( + staging.counters().sessions_refused.load(Relaxed), + u64::from(threads) - 4 + ); +} + +#[test] +fn the_per_principal_session_ceiling_refuses_the_same_principal_only() { + let root = Root::with(|options| { + options.staging_max_sessions_per_principal = 1; + options.staging_max_sessions_global = 8; + }); + let (staging, _durability) = root.open(); + + staging + .begin(projection([12; 16], [21; 32], 1, 1, HOUR_MICROS).binding, 0) + .expect("first session for this principal"); + let result = staging.begin(projection([13; 16], [21; 32], 1, 1, HOUR_MICROS).binding, 0); + let Err(StoreError::Overloaded { limit, .. }) = result else { + panic!("a second session for one principal must be refused, got {result:?}"); + }; + assert_eq!(limit, "staging_max_sessions_per_principal"); + + staging + .begin(projection([14; 16], [22; 32], 1, 1, HOUR_MICROS).binding, 0) + .expect("a different principal has its own budget"); +} + +#[test] +fn compaction_debt_is_bounded_independently_of_the_byte_budget() { + let fixture = projection([15; 16], [21; 32], 1, 2, HOUR_MICROS); + let session_bytes = fixture.binding.session.total_object_bytes; + + // Byte budget generous, debt ceiling exactly one session. The refusal must + // therefore name debt, which is only possible because staging accounts it + // on its own counter rather than sharing one with the byte budget. + let root = Root::with(|options| { + options.max_projection_bytes = session_bytes; + options.staging_max_bytes_per_principal = session_bytes * 8; + options.staging_max_bytes_global = session_bytes * 8; + options.staging_max_compaction_debt_bytes = session_bytes; + }); + let (staging, _durability) = root.open(); + + staging.begin(fixture.binding, 0).expect("first session"); + assert_eq!( + staging + .counters() + .compaction_debt_reserved_bytes + .load(Relaxed), + session_bytes + ); + + let second = projection([16; 16], [22; 32], 1, 2, HOUR_MICROS); + let result = staging.begin(second.binding, 0); + let Err(StoreError::Overloaded { limit, .. }) = result else { + panic!("the debt ceiling must refuse, got {result:?}"); + }; + assert_eq!(limit, "staging_max_compaction_debt_bytes"); +} + +#[test] +fn uploaded_totals_may_not_exceed_the_declared_binding() { + let root = Root::new(); + let (staging, _durability) = root.open(); + let fixture = projection([17; 16], [21; 32], 2, 1, HOUR_MICROS); + + // Declare one chunk's worth of objects, then upload both chunks. + let mut understated = fixture.binding.clone(); + understated.session.total_object_count = 1; + understated.session.total_object_bytes = fixture.chunks[0].objects[0].descriptor.raw_len; + + let session = staging.begin(understated, 0).expect("begin"); + session + .put_chunk(&fixture.chunks[0], 0) + .expect("first fits"); + let result = session.put_chunk(&fixture.chunks[1], 0); + let Err(StoreError::LimitExceeded { limit, .. }) = result else { + panic!("an over-declaration upload must be refused, got {result:?}"); + }; + assert_eq!(limit, "projection_stage_session_objects"); +} + +// --------------------------------------------------------------------------- +// Deliverable 5 — artifacts are written by maintenance workers, synced, +// uniquely named, and unreferenced +// --------------------------------------------------------------------------- + +/// Charter item 7, with the correction it needed: **pin the count that is +/// correct, not the count you observe.** +/// +/// The previous version of this test asserted two directory syncs for the first +/// session in a shard. Two is what the code did, and two is wrong: creating +/// `staging//` when `staging/` did not exist creates +/// *two* directory entries, and syncing only the innermost one leaves a fully +/// fenced session under a shard directory that a crash can take away. A counter +/// assertion pinned to the observed number preserved the omission instead of +/// exposing it. +/// +/// So the count is now stated by nesting depth: the first session in a shard +/// pays three (`staging/`, `staging//`, and the session directory the +/// record lands in), every later session in that shard pays two. +#[test] +fn every_artifact_is_fenced_and_every_new_directory_entry_is_synced() { + let root = Root::new(); + let (staging, durability) = root.open(); + let fixture = projection([18; 16], [21; 32], 2, 1, HOUR_MICROS); + + let before = durability.snapshot(); + let session = staging.begin(fixture.binding.clone(), 0).expect("begin"); + let after_begin = durability.snapshot(); + assert_eq!(after_begin.fdatasync - before.fdatasync, 1); + assert_eq!( + after_begin.fsync_dir - before.fsync_dir, + 3, + "the first session in a shard creates the shard directory too, and its entry \ + under staging/ must be synced before the record that depends on it" + ); + + for chunk in &fixture.chunks { + session.put_chunk(chunk, 0).expect("put"); + } + let after_chunks = durability.snapshot(); + assert_eq!(after_chunks.fdatasync - after_begin.fdatasync, 2); + assert_eq!(after_chunks.fsync_dir - after_begin.fsync_dir, 2); + + session.seal(0).expect("seal"); + let after_seal = durability.snapshot(); + assert_eq!(after_seal.fdatasync - after_chunks.fdatasync, 1); + assert_eq!(after_seal.fsync_dir - after_chunks.fsync_dir, 1); + assert!(after_seal.bytes_written > after_chunks.bytes_written); + + // A second session in the *same* shard creates one directory entry, so it + // pays two syncs. Without both halves the three above could be a constant + // that happens to be right. + let sibling = projection([28; 16], [22; 32], 1, 1, HOUR_MICROS); + let before_sibling = durability.snapshot(); + staging.begin(sibling.binding, 0).expect("begin"); + let after_sibling = durability.snapshot(); + assert_eq!(after_sibling.fsync_dir - before_sibling.fsync_dir, 2); + + // Charter item 7 for deliverable 5's other half: every artifact byte on + // disk was written by a maintenance worker, not by the calling thread. + let counters = staging.counters().snapshot(); + assert_eq!( + counters.maintenance_artifact_writes, + 2 + 2 + 1, + "two session records, two chunks, and one manifest" + ); + assert!(counters.maintenance_jobs >= counters.maintenance_artifact_writes); +} + +#[test] +fn staged_artifacts_are_uniquely_named_and_unreferenced_by_committed_state() { + let root = Root::new(); + let (staging, _durability) = root.open(); + let fixture = projection([19; 16], [21; 32], 2, 1, 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"); + let resolution = session.resolve().expect("resolve"); + + // The reference proof is answered against a committed root, never against + // staging's own bookkeeping. An empty root references nothing, and a + // staged artifact is exactly the thing that has not been adopted. + let committed = CommittedRoot::default(); + for artifact in resolution.artifacts.iter() { + assert!( + !committed.references_artifact(&artifact.path), + "a sealed artifact must not be referenced by committed state" + ); + // Structural half: staging is a sibling of shards/, so nothing a + // manifest, CURRENT, checkpoint, or index run can name lives here. + assert!(artifact.path.starts_with(staging.staging_root())); + assert!(!artifact.path.starts_with(root.path().join("shards"))); + } + + let names: std::collections::BTreeSet<_> = resolution + .artifacts + .iter() + .map(|artifact| artifact.path.clone()) + .collect(); + assert_eq!(names.len(), resolution.artifacts.len(), "names are unique"); +} + +// --------------------------------------------------------------------------- +// Deliverable 1 — expiry, abort, and the deferred cleanup entry point +// --------------------------------------------------------------------------- + +#[test] +fn expiry_reclaims_a_dead_session_and_leaves_a_live_one() { + let root = Root::new(); + let (staging, _durability) = root.open(); + let short = projection([20; 16], [21; 32], 1, 1, HOUR_MICROS); + let long = projection([21; 16], [22; 32], 1, 1, HOUR_MICROS * 2); + + let short_session = staging.begin(short.binding.clone(), 0).expect("begin"); + short_session.put_chunk(&short.chunks[0], 0).expect("put"); + staging.begin(long.binding.clone(), 0).expect("begin"); + let live = staging.counters().snapshot(); + assert_eq!(live.sessions_live, 2); + + // One microsecond past the short session's advertised expiry. + assert_eq!(staging.expire(HOUR_MICROS + 1).expect("expire"), 1); + let after = staging.counters().snapshot(); + assert_eq!(after.sessions_live, 1); + assert_eq!(after.sessions_expired, 1); + assert_eq!(after.artifacts_unlinked, 2, "session record and one chunk"); + assert_eq!( + after.reserved_bytes, long.binding.session.total_object_bytes, + "only the dead session's budget is released" + ); + assert_eq!( + after.compaction_debt_written_bytes, 0, + "reclaimed bytes stop being debt" + ); + + let result = short_session.describe(); + let Err(StoreError::Conflict(message)) = result else { + panic!("an expired session must be gone, got {result:?}"); + }; + assert!(message.contains("no staging session"), "{message}"); + staging + .session(long.binding.session.session_id) + .expect("the live session is untouched"); +} + +#[test] +fn expiry_at_the_advertised_instant_does_not_reclaim() { + let root = Root::new(); + let (staging, _durability) = root.open(); + let fixture = projection([22; 16], [21; 32], 1, 1, HOUR_MICROS); + staging.begin(fixture.binding, 0).expect("begin"); + + // The frozen `validate_projection_stage_finalize` treats `now > expires_at` + // as expired, so the store must agree at exactly the boundary microsecond. + assert_eq!(staging.expire(HOUR_MICROS).expect("expire"), 0); + assert_eq!(staging.counters().sessions_live.load(Relaxed), 1); + assert_eq!(staging.expire(HOUR_MICROS + 1).expect("expire"), 1); +} + +#[test] +fn abort_releases_the_budget_and_removes_every_artifact() { + let root = Root::new(); + let (staging, _durability) = root.open(); + let fixture = projection([23; 16], [21; 32], 2, 1, HOUR_MICROS); + let session = staging.begin(fixture.binding.clone(), 0).expect("begin"); + for chunk in &fixture.chunks { + session.put_chunk(chunk, 0).expect("put"); + } + let session_directory = root.session_directory(fixture.session_id()); + assert!(session_directory.exists()); + + session.abort().expect("abort"); + + assert!(!session_directory.exists()); + let counters = staging.counters().snapshot(); + assert_eq!(counters.sessions_live, 0); + assert_eq!(counters.sessions_aborted, 1); + assert_eq!(counters.reserved_bytes, 0); + assert_eq!(counters.reserved_objects, 0); + assert_eq!(counters.reserved_files, 0); + assert_eq!(counters.compaction_debt_reserved_bytes, 0); + assert_eq!(counters.compaction_debt_written_bytes, 0); + assert_eq!(counters.artifacts_unlinked, 3, "record plus two chunks"); + + // The budget is genuinely back: the same principal may open again. + staging + .begin(projection([24; 16], [21; 32], 1, 1, HOUR_MICROS).binding, 0) + .expect("released budget is reusable"); +} + +#[test] +fn cleanup_is_deferred_and_names_its_deliverable() { + 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}"); +} + +// --------------------------------------------------------------------------- +// Deliverable 1 — restartable, meaning a process restart +// --------------------------------------------------------------------------- + +/// Drop the whole `ProjectionStaging` and rebuild it from the filesystem. +/// +/// This is what "restartable" has to mean. The earlier version of this test +/// dropped a *handle* and reacquired one from the same live registry, which +/// proves nothing about a reopen: every fact it checked was still in memory. +/// Nothing in-memory survives this one. +#[test] +fn a_real_reopen_reconstructs_sessions_chunks_and_accounting_from_disk() { + let root = Root::new(); + let fixture = projection([30; 16], [21; 32], 2, 1, HOUR_MICROS); + let session_id = fixture.session_id(); + + { + let (staging, _durability) = root.open(); + let session = staging.begin(fixture.binding.clone(), 0).expect("begin"); + session.put_chunk(&fixture.chunks[0], 0).expect("put"); + } + + let (staging, durability) = root.open(); + let counters = staging.counters().snapshot(); + assert_eq!(counters.sessions_reconstructed, 1); + assert_eq!(counters.sessions_live, 1); + assert_eq!( + counters.reserved_bytes, fixture.binding.session.total_object_bytes, + "a root-global byte budget that ignored durable sessions would be a count of \ + this process's uptime" + ); + assert_eq!( + counters.reserved_objects, + fixture.binding.session.total_object_count + ); + assert_eq!( + counters.compaction_debt_reserved_bytes, + counters.reserved_bytes + ); + assert_eq!(counters.abandoned_materializations_reclaimed, 0); + + // The chunk index came back too: re-offering ordinal 0 costs no write. + let resumed = staging.session(session_id).expect("resume after reopen"); + let status = resumed.describe().expect("describe"); + assert_eq!(status.state, StagedSessionState::Open); + assert_eq!(status.chunks_present, 1); + assert_eq!(status.chunks_expected, 2); + assert_eq!( + status.written_bytes, + fixture.chunks[0].objects[0].descriptor.raw_len + ); + + let before = durability.snapshot(); + assert_eq!( + resumed.put_chunk(&fixture.chunks[0], 0).expect("repeat"), + ChunkPutOutcome::AlreadyPresent + ); + assert_eq!(durability.snapshot(), before); + + resumed.put_chunk(&fixture.chunks[1], 0).expect("finish"); + resumed.seal(0).expect("a reconstructed session seals"); +} + +/// The compounding half of the restart defect, and the reason it was a P1. +/// +/// Before the fix, a reopen carried no sessions, so `begin` admitted a durable +/// session ID as new; if materialization then failed, the error path deleted +/// the directory recursively and took the previously durable session with it. +/// Data loss from an ordinary restart plus one error. +#[test] +fn a_reopen_refuses_to_admit_a_durable_session_id_as_new() { + let root = Root::new(); + let fixture = projection([31; 16], [21; 32], 1, 1, HOUR_MICROS); + let directory = root.session_directory(fixture.session_id()); + + { + let (staging, _durability) = root.open(); + let session = staging.begin(fixture.binding.clone(), 0).expect("begin"); + session.put_chunk(&fixture.chunks[0], 0).expect("put"); + } + + let (staging, _durability) = root.open(); + let result = staging.begin(fixture.binding.clone(), 0); + let Err(StoreError::Conflict(message)) = result else { + panic!("a durable session ID must not be admitted as new, got {result:?}"); + }; + assert!(message.contains("already exists"), "{message}"); + + // And the durable session is untouched: the refusal happens before the + // error path that used to delete it could ever be reached. + assert!(directory.join("session").exists()); + assert_eq!(staging.counters().snapshot().sessions_live, 1); + staging + .session(fixture.session_id()) + .expect("the durable session is still resumable"); +} + +#[test] +fn a_sealed_session_survives_a_reopen_and_still_resolves() { + let root = Root::new(); + let fixture = projection([32; 16], [21; 32], 2, 1, HOUR_MICROS); + + let install = { + let (staging, _durability) = root.open(); + 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") + }; + + let (staging, _durability) = root.open(); + let resumed = staging.session(fixture.session_id()).expect("resume"); + assert_eq!( + resumed.describe().expect("describe").state, + StagedSessionState::Sealed, + "a durable manifest artifact is the seal's commit point, so it is what makes a \ + reopened session sealed again" + ); + let resolution = resumed.resolve().expect("a reopened seal still resolves"); + assert_eq!(resolution.manifest.chunk_digests.len(), 2); + assert_eq!( + resolution.manifest.manifest_digest().expect("digest"), + install.manifest_digest + ); + + // Sealed is immutable across the restart too. + let result = resumed.put_chunk(&fixture.chunks[0], 0); + let Err(StoreError::Conflict(message)) = result else { + panic!("a reopened sealed session must refuse chunks, got {result:?}"); + }; + assert!(message.contains("sealed"), "{message}"); +} + +/// A directory that never acquired a `session` record is not a session. +/// +/// It is what a crash between `mkdir` and the record's rename leaves behind. +/// Nothing accounted for it, nothing can reference it, and — the point — it +/// must not be confused with the final state, which is why the two do not +/// share a code path. +#[test] +fn an_abandoned_materialization_is_reclaimed_and_a_final_session_is_not() { + let root = Root::new(); + let fixture = projection([33; 16], [21; 32], 1, 1, HOUR_MICROS); + { + let (staging, _durability) = root.open(); + let session = staging.begin(fixture.binding.clone(), 0).expect("begin"); + session.put_chunk(&fixture.chunks[0], 0).expect("put"); + } + + // Hand-built: the directory exists, the record never landed. + let abandoned = root.session_directory([34; 16]); + std::fs::create_dir_all(&abandoned).expect("abandoned directory"); + std::fs::write(abandoned.join("session.tmp"), b"torn").expect("torn temporary"); + + let (staging, _durability) = root.open(); + let counters = staging.counters().snapshot(); + assert_eq!(counters.abandoned_materializations_reclaimed, 1); + assert_eq!(counters.sessions_reconstructed, 1); + assert_eq!(counters.sessions_live, 1); + assert!(!abandoned.exists(), "the abandoned directory is reclaimed"); + assert!( + root.session_directory(fixture.session_id()) + .join("session") + .exists(), + "the final session in the same shard is untouched" + ); +} + +// --------------------------------------------------------------------------- +// Deliverable 4 — the ceilings are root-global, which requires one accountant +// --------------------------------------------------------------------------- + +/// A second `ProjectionStaging` for one root is not expressible. +/// +/// Two registries each admit up to the full global and per-principal budget, so +/// a freely callable constructor turned every "global" ceiling into a +/// per-handle ceiling. The cross-process half is closed by requiring the root +/// lock; this is the in-process half. +#[test] +fn one_root_admits_exactly_one_staging_accountant() { + let root = Root::new(); + let (_first, _durability) = root.open(); + + let result = ProjectionStaging::open( + &root.lock, + root.options.clone(), + Arc::new(DurabilityCounters::default()), + ); + let Err(StoreError::Conflict(message)) = result else { + panic!("a second staging instance for one root must be refused, got {result:?}"); + }; + assert!(message.contains("already open"), "{message}"); +} + +/// And the lock has to be a lock on *this* root. +#[test] +fn a_root_lock_held_on_another_root_is_not_proof() { + let owner = Root::new(); + let other = Root::new(); + + let result = ProjectionStaging::open( + &other.lock, + owner.options.clone(), + Arc::new(DurabilityCounters::default()), + ); + let Err(StoreError::InvalidConfiguration(message)) = result else { + panic!("a foreign root lock must not open staging, got {result:?}"); + }; + assert!(message.contains("proves nothing"), "{message}"); +} + +// --------------------------------------------------------------------------- +// Deliverable 7's discipline, applied to reclamation +// --------------------------------------------------------------------------- + +/// Reclamation proves the whole directory before it unlinks anything. +/// +/// Unlinking as the scan walks means a valid marked artifact can be destroyed +/// and *then* an unexpected file encountered, leaving a live session partially +/// destroyed: not removed, and no longer sealable. Either the whole directory +/// goes or nothing does. +#[test] +fn reclamation_removes_nothing_when_the_directory_holds_a_surprise() { + let root = Root::new(); + let (staging, _durability) = root.open(); + let fixture = projection([35; 16], [21; 32], 2, 1, HOUR_MICROS); + let session = staging.begin(fixture.binding.clone(), 0).expect("begin"); + for chunk in &fixture.chunks { + session.put_chunk(chunk, 0).expect("put"); + } + let directory = root.session_directory(fixture.session_id()); + let before: Vec<_> = std::fs::read_dir(&directory) + .expect("read") + .map(|entry| entry.expect("entry").path()) + .collect(); + assert_eq!(before.len(), 3, "record plus two chunks"); + std::fs::write(directory.join("not-ours"), b"someone else's bytes").expect("foreign file"); + + let result = session.abort(); + let Err(StoreError::Corruption(message)) = result else { + panic!("an unexplained file must stop reclamation, got {result:?}"); + }; + assert!(message.contains("not a licence to delete"), "{message}"); + + // Counters, not claims: nothing was unlinked, and the session still holds + // its budget rather than being half gone. + let counters = staging.counters().snapshot(); + assert_eq!(counters.artifacts_unlinked, 0); + assert_eq!(counters.sessions_live, 1); + assert_eq!(counters.sessions_aborted, 0); + for path in &before { + assert!( + path.exists(), + "{} was destroyed before the refusal", + path.display() + ); + } + // And it is still a usable session, not a wreck. + session_of(&staging, fixture.session_id()) + .seal(0) + .expect("a session that survived a refused reclamation still seals"); +} + +fn session_of( + staging: &Arc, + session_id: [u8; 16], +) -> levcs_store::staging::ProjectionStageSession { + staging.session(session_id).expect("session") +} + +// --------------------------------------------------------------------------- +// The security property, asserted through the reader API +// --------------------------------------------------------------------------- + +/// **Sealing cannot publish membership** (plan §4 identity invariant 7, §8). +/// +/// This is the acceptance test for the whole package and it is deliberately +/// written in the shape it must finally take: seal a session, then prove +/// through `RepoSnapshot::locate` — the entry point a reader actually calls — +/// that the staged objects are invisible, and that they become visible only +/// after a `submit` adopts the descriptor. +/// +/// Still blocked on B1 NamespaceTxn, and blocked in one more place than before. +/// `RepoSnapshot::locate`, `StoreEngine::snapshot`, and the submit path are all +/// `NotImplemented`/`unimplemented!` today. On top of that, the P1-4 fix moved +/// staging inside the locked engine lifetime: this test can no longer open its +/// own `ProjectionStaging` beside a `StoreEngine`, because holding two root +/// locks is exactly what was made unrepresentable. It needs an accessor on the +/// engine — recorded as an interface request to B1 — to reach the staging the +/// engine owns. +/// +/// Asserting the property against staging's own state instead would be charter +/// item 8 exactly — a property proved against the helper rather than the path +/// that runs — so it is marked blocked rather than satisfied the wrong way. +#[test] +#[ignore = "blocked on B1 NamespaceTxn: needs RepoSnapshot::locate, StoreEngine::snapshot/submit (scope 6.4 deliverables 1, 3-8), and a StoreEngine accessor for the engine-owned ProjectionStaging"] +fn sealed_objects_stay_invisible_until_a_submit_adopts_them() { + let root = Root::new(); + let fixture = projection([25; 16], [21; 32], 1, 2, HOUR_MICROS); + let namespace = NamespaceId::from(fixture.binding.session.destination_repo); + let staged_ids: Vec = fixture.chunks[0] + .objects + .iter() + .map(|object| object.descriptor.object_id) + .collect(); + + // The engine takes the root lock, so the fixture's own lock is released + // first. That is the P1-4 shape: one lock, one engine, one staging. + let options = root.options.clone(); + drop(root); + let engine = levcs_store::StoreEngine::open(options).expect("engine opens"); + let snapshot = engine.snapshot(namespace).expect("snapshot"); + for id in &staged_ids { + assert_eq!( + snapshot.locate(*id).expect("locate"), + None, + "nothing is visible before anything is staged" + ); + } + + unimplemented!( + "B1: expose the engine-owned ProjectionStaging, begin/put/seal a session through it, \ + re-assert locate() is None for every staged object, then build a ValidatedTransaction \ + with adopt_projection(install, handle), submit it, and assert locate() returns Some" + ); +}