From 4c68f0844605dca4c1f0ec61bab9ae0dae887f4e Mon Sep 17 00:00:00 2001 From: Levi Neuwirth Date: Tue, 28 Jul 2026 18:01:25 -0400 Subject: [PATCH] Implement the partial B3 staging-session slice MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Scope 6.5 deliverables 1-5: what a staging session is on disk, what it costs, when it dies. The instance-layer half of plan §8 — identity, policy, ProjectionCore, the v2 routes — is deliberately absent; B3 binds and exposes the fields those checks key on and evaluates none of them. Two findings from the B2 review shaped the result more than the original deliverables did. Restart was in-memory handle reuse. open() built a fresh registry and never read the filesystem, so after a real reopen an existing session ID was admitted as new -- and if materialization then found the old directory, error cleanup could unlink a durable session. That is data loss reachable from an ordinary restart plus one error. open() now scans, validates, and reconstructs sessions, chunk indexes, and the whole occupancy account before returning, so a reconstructed ID is occupied and refused as a conflict long before the error path; and that path no longer calls remove_dir_all. Either change alone closes the loss. A session is final iff its directory holds a valid session record for its own ID, installed by rename_noreplace over fenced, digest-checked bytes, so the final name only ever appears atomically over complete data. A directory without one is abandoned materialization -- a crash between mkdir and that rename, unaccounted and unreferenceable. The two states share no code path, and reclaimed abandonments count on their own counter so they can never be read as aborts or expiries. Global quotas were per-handle. Every open() built an independent registry outside the root LOCK, so two handles admitted twice the global limit and the atomic-insertion work bought nothing across them. Construction now requires proof of the held root lock and refuses a second in-process instance, making two accountants on one root inexpressible rather than discouraged. Also: bounds are enforced atomically with insertion under one mutex with no read-then-decide path, refused as typed LimitExceeded or Overloaded and never by eviction; the directory-sync test pinned two syncs when the first session in a shard needs three, a counter assertion that encoded the bug; cleanup now validates a whole directory before unlinking anything, rather than discovering a surprise midway through destroying a live session; and artifact I/O moved off the registry mutex onto maintenance workers, with the calling thread asserted to hold no guard rather than documented not to. Deliverables 6-8 are explicit NotImplemented naming themselves. StagedSessionState omits Finalizing, so deliverable 6 will fail to compile at exactly the expiry and abort sites that must learn about a pin. Carry-forwards recorded in §6.5, not closed: no production path begins a session, so the sealed-invisibility acceptance stays ignored with both blockers named; and expire() has no scheduler, so session age is a bound enforced when asked and never asked. 116 library tests, 27 staging tests, 1 intentionally ignored. Co-Authored-By: Claude Opus 5 (1M context) --- crates/levcs-store/src/staging.rs | 2857 ++++++++++++++++++ crates/levcs-store/tests/staging_sessions.rs | 1207 ++++++++ 2 files changed, 4064 insertions(+) create mode 100644 crates/levcs-store/tests/staging_sessions.rs 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" + ); +}