optionchain_simulator 0.2.29

//! Lifecycle for v2 rolling simulations.
//!
//! The v2 counterpart of [`crate::session::SessionManager`]: it owns the
//! [`SimulationStore`], the per-simulation factor tapes, and the bounded
//! snapshot cache, and it is the only thing the API layer talks to. Handlers
//! never reach into `domain` — that module is private, and the layering says
//! api → session → domain.
//!
//! # What it guarantees
//!
//! - **Serve-then-advance.** `advance` serves the snapshot at the current
//!   cursor and *then* moves it, so a simulation with `steps = N` serves
//!   exactly indices `0..N-1` over `N` advances. This is v1's semantics,
//!   deliberately carried forward.
//! - **A peek changes nothing.** `peek` builds the same snapshot and writes
//!   nothing back, so calling it repeatedly is safe and returns the same
//!   answer until an advance moves the cursor.
//! - **No lost advance.** Every advance persists through a compare-and-swap on
//!   the revision it read, so two concurrent advances cannot both commit: the
//!   loser gets a `Conflict` and retries.
//! - **Caches are never authoritative.** A factor tape or a snapshot can be
//!   dropped at any time; both rebuild identically from the effective
//!   parameters, so eviction changes latency and nothing else.

use crate::domain::factors::FactorTape;
use crate::domain::series::{SeriesBuilder, SeriesSnapshot, SnapshotCache};
use crate::infrastructure::{
    MetricsCollector, SimulationSnapshotRepository, SimulationV2Config, SnapshotRecord,
};
use crate::session::model::SessionState;
use crate::session::snapshot_record::{snapshot_quote_count, snapshot_record};
use crate::session::store::{BuildClaim, SharedTapeCache, SimulationStore, tape_key};
use crate::session::{SessionV2, SimulationParametersV2};
use crate::utils::ChainError;
use std::collections::HashMap;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use tokio::sync::{broadcast, mpsc};
use tracing::{debug, info, instrument, warn};
use uuid::Uuid;

/// How long a caller waits for the instance that claimed a build.
///
/// Long enough for a tape of the maximum step count, short enough that an
/// owner which died leaves the others building rather than waiting out the
/// claim's own expiry.
const SHARED_BUILD_WAIT: Duration = Duration::from_secs(30);

/// How often a waiter looks for the tape the owner is building.
const SHARED_BUILD_POLL: Duration = Duration::from_millis(50);

/// What asking for the deployment-wide build claim produced.
enum SharedBuild {
    /// This caller builds. The token, when there is one, is what releases the
    /// claim; there is none when no cache is configured, when the cache could
    /// not answer, or when the wait for another instance ran out.
    Owner(Option<String>),
    /// Another instance built it, and here it is.
    Waited(FactorTape),
}

/// Everything the owner of a tape build needs, owned rather than borrowed.
///
/// Owned because the workflow runs in a task that outlives the request that
/// started it: the build cannot be cancelled once it is on the blocking pool,
/// and the waiters, the shared cache and the deployment-wide claim must be
/// settled whether the caller is still there or not.
struct OwnedBuild {
    id: Uuid,
    /// The tape's shared-cache key, computed once by the caller.
    key: String,
    parameters: SimulationParametersV2,
    tapes: Arc<Mutex<HashMap<Uuid, TapeEntry>>>,
    builds: Arc<Mutex<HashMap<Uuid, TapeBuilds>>>,
    shared: Option<Arc<dyn SharedTapeCache>>,
    max_cached_tapes: usize,
    shared_build_wait: Duration,
}

impl OwnedBuild {
    /// Builds the tape (or takes another instance's), publishes it, and gives
    /// back every claim it took.
    async fn run(self, sender: TapeBuilds) {
        let claim = self.claim().await;
        let built_here = !matches!(claim, SharedBuild::Waited(_));
        // Kept out of the match, which consumes the claim, because the token
        // is what releases it at the end.
        let token = match &claim {
            SharedBuild::Owner(token) => token.clone(),
            SharedBuild::Waited(_) => None,
        };

        let result = match claim {
            SharedBuild::Waited(tape) => {
                SimulationManager::cache_tape(
                    &self.tapes,
                    self.max_cached_tapes,
                    self.id,
                    tape.clone(),
                );
                Ok(tape)
            }
            SharedBuild::Owner(_) => {
                SimulationManager::build_tape_into(
                    self.parameters.clone(),
                    self.id,
                    Arc::clone(&self.tapes),
                    self.max_cached_tapes,
                )
                .await
            }
        };

        // Publish to whoever is waiting and stop being the owner, in that
        // order: a caller that subscribes after the removal misses the
        // broadcast, retries, and finds the tape in the cache.
        //
        // Before the shared write, deliberately. A slow Redis would otherwise
        // hold every local waiter behind a round trip they do not need.
        {
            let mut builds = match self.builds.lock() {
                Ok(builds) => builds,
                Err(poisoned) => poisoned.into_inner(),
            };
            builds.remove(&self.id);
        }
        let published = match &result {
            Ok(tape) => Ok(tape.clone()),
            Err(error) => Err(error.to_string()),
        };
        // An error means nobody was waiting, which happens whenever the
        // request that started this went away.
        let _ = sender.send(published);

        // Only now offer it to the other instances: this process is already
        // served, so a slow or failing write costs the deployment a rebuild
        // elsewhere and costs this request nothing. A tape another instance
        // published is already there, so only one built here is worth writing.
        if let (Some(shared), true, Ok(tape)) = (self.shared.as_ref(), built_here, &result) {
            SimulationManager::share_tape_to(shared.as_ref(), &self.key, self.id, tape).await;
        }

        // The claim goes back after the shared write, not before: an instance
        // that wakes on the release must find the tape rather than an empty
        // key. Released whether the build succeeded or not, so a failure does
        // not make the others wait out the expiry for a tape nobody wrote.
        if let (Some(shared), Some(token)) = (self.shared.as_ref(), &token) {
            shared.release_build(&self.key, token).await;
        }
    }

    /// Claims the deployment-wide right to build this tape, or waits for the
    /// instance that holds it.
    ///
    /// The wait is bounded and polls the shared cache rather than a channel:
    /// there is no channel between processes, and an owner that dies would
    /// otherwise leave every waiter hanging until its claim expired. When the
    /// wait runs out this caller builds, which is duplicated work rather than
    /// a request that never answers.
    async fn claim(&self) -> SharedBuild {
        let Some(shared) = self.shared.as_ref() else {
            return SharedBuild::Owner(None);
        };

        match shared.claim_build(&self.key).await {
            // Nothing to release: without a claim there is nothing this
            // instance holds, and deleting the key would take a claim another
            // instance owns.
            BuildClaim::Unclaimed => return SharedBuild::Owner(None),
            BuildClaim::Held(token) => return SharedBuild::Owner(Some(token)),
            BuildClaim::Taken => {}
        }

        debug!(
            simulation_id = %self.id,
            "another instance is building this tape; waiting for it"
        );
        let deadline = Instant::now() + self.shared_build_wait;
        while Instant::now() < deadline {
            tokio::time::sleep(SHARED_BUILD_POLL).await;
            if let Some(tape) =
                SimulationManager::shared_tape_from(shared.as_ref(), &self.key, self.id).await
            {
                return SharedBuild::Waited(tape);
            }
        }

        warn!(
            simulation_id = %self.id,
            waited_secs = self.shared_build_wait.as_secs(),
            "the instance building this tape did not publish in time; building it here"
        );
        SharedBuild::Owner(None)
    }
}

/// One cached factor tape and the last time it was used.
struct TapeEntry {
    tape: FactorTape,
    last_access: Instant,
}

/// A snapshot's identity in the in-flight map: which simulation, which step.
type SnapshotKey = (Uuid, usize);

/// How a snapshot build publishes its result to the callers waiting on it.
///
/// The error is a `String` rather than a `ChainError` because `broadcast`
/// requires `Clone` and `ChainError` is not; the waiter rebuilds it as an
/// internal error, which is what the owner would have reported anyway.
type SnapshotBuilds = broadcast::Sender<Result<SeriesSnapshot, String>>;

/// How a tape build publishes its result to the callers waiting on it.
type TapeBuilds = broadcast::Sender<Result<FactorTape, String>>;

/// Owns the lifecycle of v2 rolling simulations.
pub struct SimulationManager {
    store: Arc<dyn SimulationStore>,
    config: SimulationV2Config,
    tapes: Arc<Mutex<HashMap<Uuid, TapeEntry>>>,
    /// Tape builds currently running, one entry per simulation.
    ///
    /// Without it, N concurrent first reads of one simulation start N identical
    /// builds — and a build is the one place a v2 request does seconds of CPU,
    /// so the duplicates are not a wasted allocation, they are the machine.
    /// `spawn_blocking` does not bound that: its pool grows to hundreds of
    /// threads, so the cache would still be cold while every core was busy
    /// filling it with the same answer.
    builds: Arc<Mutex<HashMap<Uuid, TapeBuilds>>>,
    /// Snapshot builds currently running, one entry per `(simulation, step)`.
    ///
    /// The same argument as [`SimulationManager::builds`], one level down and
    /// sharper since issue #74: with a warehouse registered a snapshot build
    /// prices up to `OCS_MAX_SNAPSHOT_CONTRACTS` contracts WITH both greek
    /// snapshots, so N concurrent readers of one step used to commit the
    /// machine to N copies of the same seconds-long job. One owner builds, the
    /// rest wait on its result.
    snapshot_builds: Mutex<HashMap<SnapshotKey, SnapshotBuilds>>,
    snapshots: Arc<Mutex<SnapshotCache>>,
    /// Where a built tape is left for the OTHER instances, when one is
    /// configured.
    ///
    /// The local map above is the first level and stays: a hit there costs
    /// nothing, where this costs a round trip. What it removes is the rebuild
    /// a second instance would otherwise do for a tape its neighbour already
    /// walked, which is the expensive half of serving a step (issue #136).
    ///
    /// `None` is a deployment with no shared cache configured, and every
    /// failure of the shared one is treated as `None` for that call: a cache
    /// that cannot be reached is a miss, never a failed request.
    shared_tapes: Option<Arc<dyn SharedTapeCache>>,
    /// How long to wait for the instance that claimed a build before building
    /// it here.
    ///
    /// A field rather than a constant so a test can prove the give-up path
    /// without waiting out the production value, which is deliberately as long
    /// as the slowest build.
    shared_build_wait: Duration,
    /// Where the count of filed rows is reported, when the binary wired it.
    ///
    /// Set before [`Self::with_warehouse`], which is what the writer task
    /// captures. `None` in a test or a library caller that has no collector,
    /// and the writer then files without counting.
    snapshot_metrics: Option<Arc<MetricsCollector>>,
    /// Where served snapshots are queued for filing, when the operator turned
    /// persistence on. `None` is the default and the whole feature is then
    /// absent from the serving path — no connection, no latency, no failure
    /// mode.
    warehouse: Option<Warehouse>,
}

/// The queue in front of the warehouse, and what is currently in it.
struct Warehouse {
    /// The repository itself, so a reader — the export — can consult the same
    /// warehouse the writer fills without being handed a second handle to it.
    repository: Arc<dyn SimulationSnapshotRepository>,
    sender: mpsc::Sender<SnapshotRecord>,
    /// Quote rows queued but not yet written. Incremented before a send and
    /// decremented by the writer once the record leaves the queue, so it
    /// measures what is resident rather than what has been served.
    queued_contracts: Arc<AtomicUsize>,
}

/// How many snapshots may be waiting to be filed.
///
/// The queue is what keeps a degraded warehouse from becoming a memory leak. An
/// unbounded spawn-per-advance cannot delay a response, but at a sustained
/// advance rate against a warehouse that is timing out it accumulates records
/// until the process dies — which fails every request, not just the write.
const SNAPSHOT_QUEUE_DEPTH: usize = 1_024;

/// How many quote rows may be waiting to be filed, across every queued record.
///
/// A depth in *records* is the wrong unit for the same reason an entry count
/// was the wrong unit for the snapshot cache: a record is a few hundred quotes
/// in the reference configuration and up to the per-snapshot cap in a large
/// one, so 1 024 of them is anywhere from a hundred thousand to two hundred
/// million rows. This bounds what is actually resident.
///
/// Like the cache bound, the count is derived from a byte budget of roughly
/// 850 MB rather than chosen: `size_of::<QuoteRow>()` is 620 bytes since issue
/// #74 gave it both greek snapshots (212 bytes before), so 1 350 000 × 620 B ≈
/// 837 MB. The count came down from 4 000 000 in the same change, to hold the
/// budget the previous count expressed.
///
/// Neither bound is a knob. A deployment that needs to tune them is one whose
/// warehouse cannot keep up with its advance rate, and the answer there is the
/// warehouse, not a deeper buffer in front of it.
const SNAPSHOT_QUEUE_CONTRACTS: usize = 1_350_000;

impl SimulationManager {
    /// Creates a manager over a simulation store.
    ///
    /// The only method the binary needs; everything else is crate-internal,
    /// because it deals in `domain` types that are not part of this crate's
    /// public API. The v2 REST surface is the contract, not these signatures.
    #[must_use]
    pub fn new(store: Arc<dyn SimulationStore>, config: SimulationV2Config) -> Self {
        Self {
            store,
            config,
            tapes: Arc::new(Mutex::new(HashMap::new())),
            builds: Arc::new(Mutex::new(HashMap::new())),
            snapshot_builds: Mutex::new(HashMap::new()),
            snapshots: Arc::new(Mutex::new(SnapshotCache::with_bounds(
                config.max_cached_snapshots,
                config.max_cached_snapshot_contracts,
            ))),
            shared_tapes: None,
            shared_build_wait: SHARED_BUILD_WAIT,
            snapshot_metrics: None,
            warehouse: None,
        }
    }

    /// Shares built tapes with the other instances through `cache`.
    ///
    /// Opt-in for the same reason the warehouse is: a single-instance
    /// deployment should not have to name the feature to not use it, and the
    /// serving path should express "no shared cache" as a missing dependency
    /// rather than as a flag it branches on.
    #[must_use]
    pub fn with_shared_tapes(mut self, cache: Arc<dyn SharedTapeCache>) -> Self {
        self.shared_tapes = Some(cache);
        self
    }

    /// How long to wait for another instance's build before building here.
    ///
    /// Exists for tests: production wants the default, which is sized for the
    /// slowest build, and a test that has to reach the give-up path should not
    /// spend that long doing it.
    #[must_use]
    pub fn with_shared_build_wait(mut self, wait: Duration) -> Self {
        self.shared_build_wait = wait;
        self
    }

    /// Reports filed rows to `metrics`.
    ///
    /// Call it BEFORE [`Self::with_warehouse`]: that method spawns the writer,
    /// and the writer captures whatever this left behind. Separate from it
    /// rather than a parameter of it, because the warehouse is wired by
    /// library callers that have no collector and the signature is public.
    ///
    /// What it buys is a signal that does not exist otherwise: the write is
    /// detached from the advance that produced it, so "the step was served"
    /// says nothing about "the row is stored", and `v2_snapshot_rows_filed_total`
    /// is how anything outside the process can tell (issue #149).
    #[must_use]
    pub fn with_snapshot_metrics(mut self, metrics: Arc<MetricsCollector>) -> Self {
        self.snapshot_metrics = Some(metrics);
        self
    }

    /// Files every served snapshot in `warehouse`.
    ///
    /// Opt-in, and deliberately a separate constructor rather than an argument
    /// to [`SimulationManager::new`]: a deployment without ClickHouse should not
    /// have to name the feature to not use it, and the serving path should not
    /// branch on a config flag it can express as a missing dependency.
    #[must_use]
    pub fn with_warehouse(mut self, repository: Arc<dyn SimulationSnapshotRepository>) -> Self {
        let (sender, mut receiver) = mpsc::channel::<SnapshotRecord>(SNAPSHOT_QUEUE_DEPTH);
        let queued_contracts = Arc::new(AtomicUsize::new(0));
        let writer_contracts = Arc::clone(&queued_contracts);
        let warehouse = Arc::clone(&repository);
        let writer_metrics = self.snapshot_metrics.clone();

        // One writer, not one task per advance: the queue bounds what a slow
        // warehouse can accumulate, and serialising the writes means two steps
        // of one simulation reach the warehouse in the order they were served.
        tokio::spawn(async move {
            while let Some(record) = receiver.recv().await {
                let simulation = record.simulation;
                let step = record.step;
                let contracts = record.quote_count();

                let result = warehouse.persist(record).await;
                writer_contracts.fetch_sub(contracts, Ordering::SeqCst);

                // Counted on the write that SUCCEEDED: a queue that never
                // drains must not look like storage that is filling.
                if result.is_ok()
                    && let Some(metrics) = writer_metrics.as_ref()
                {
                    metrics.record_snapshot_rows_filed(contracts);
                }

                if let Err(error) = result {
                    warn!(
                        simulation_id = %simulation,
                        step,
                        error = %error,
                        "Could not file the snapshot; the step can be replayed and rewritten"
                    );
                }
            }
        });

        self.warehouse = Some(Warehouse {
            repository,
            sender,
            queued_contracts,
        });
        self
    }

    /// The warehouse this manager files into, if any.
    ///
    /// Exists so the export can prefer persisted snapshots over replay without
    /// the binary threading a second handle through the server: the manager
    /// already owns the one the writer uses, and two handles could drift to two
    /// different configurations.
    #[must_use]
    pub fn warehouse(&self) -> Option<Arc<dyn SimulationSnapshotRepository>> {
        self.warehouse
            .as_ref()
            .map(|warehouse| Arc::clone(&warehouse.repository))
    }

    /// The operational configuration this manager applies.
    #[must_use]
    pub fn config(&self) -> SimulationV2Config {
        self.config
    }

    /// Creates a simulation from resolved parameters.
    ///
    /// The factor tape is **not** built here. Creation stays cheap and
    /// predictable, and the first peek or advance pays for the tape — which it
    /// would have to be able to rebuild after an eviction anyway.
    ///
    /// # Errors
    ///
    /// Returns [`ChainError::AlreadyExists`] on an id collision, or any storage
    /// failure.
    #[instrument(skip(self, parameters), level = "debug")]
    pub(crate) async fn create(
        &self,
        parameters: SimulationParametersV2,
    ) -> Result<SessionV2, ChainError> {
        let simulation = SessionV2::new(parameters);
        self.store.create(simulation.clone()).await?;

        info!(
            simulation_id = %simulation.id,
            steps = simulation.total_steps,
            seed = simulation.parameters.seed,
            "Created a v2 rolling simulation"
        );
        Ok(simulation)
    }

    /// Reads a simulation's metadata without touching its cursor.
    ///
    /// # Errors
    ///
    /// Returns [`ChainError::NotFound`] when no simulation has that id.
    #[instrument(skip(self), level = "debug")]
    pub(crate) async fn get(&self, id: Uuid) -> Result<SessionV2, ChainError> {
        self.store.get(id).await
    }

    /// Builds the snapshot at the current cursor **without** advancing or
    /// persisting anything.
    ///
    /// Safe and repeatable: the same call returns the same snapshot until an
    /// advance moves the cursor. The only side effect is a warmer cache.
    ///
    /// # Errors
    ///
    /// Returns [`ChainError::NotFound`] for an unknown id,
    /// [`ChainError::SimulatorError`] when the simulation has already served
    /// every step (410 at the boundary, matching v1's exhausted path),
    /// [`ChainError::InvalidState`] for a simulation in the terminal error
    /// state, and whatever the tape or snapshot build surfaces.
    #[instrument(skip(self), level = "debug")]
    pub(crate) async fn peek(&self, id: Uuid) -> Result<(SessionV2, SeriesSnapshot), ChainError> {
        let simulation = self.store.get(id).await?;
        Self::reject_terminal(&simulation, "no current step")?;

        let snapshot = self
            .snapshot_at(&simulation, simulation.current_step)
            .await?;
        Ok((simulation, snapshot))
    }

    /// Serves the snapshot at the current cursor, then advances it exactly
    /// once.
    ///
    /// The advance that serves the last snapshot marks the simulation
    /// `Completed` and drops its cached state: its tape and snapshots can never
    /// be served again, and a re-created simulation would rebuild them
    /// identically anyway.
    ///
    /// # Errors
    ///
    /// As [`SimulationManager::peek`], plus [`ChainError::Conflict`] when a
    /// concurrent advance committed first — the caller re-reads and retries,
    /// and there is deliberately no silent retry loop here.
    #[instrument(skip(self), level = "debug")]
    pub(crate) async fn advance(
        &self,
        id: Uuid,
    ) -> Result<(SessionV2, SeriesSnapshot), ChainError> {
        let mut simulation = self.store.get(id).await?;

        // The revision read here is what the compare-and-swap below commits
        // against, so two concurrent advances that both read this snapshot
        // cannot both persist.
        let expected_version = simulation.version;
        Self::reject_terminal(&simulation, "no further steps")?;

        let snapshot = self
            .snapshot_at(&simulation, simulation.current_step)
            .await?;

        simulation.current_step = simulation
            .current_step
            .checked_add(1)
            .ok_or_else(|| ChainError::Internal("the cursor overflowed".to_string()))?;
        simulation.state = if simulation.is_complete() {
            SessionState::Completed
        } else {
            SessionState::InProgress
        };

        simulation.bump_version()?;
        self.store
            .save_cas(simulation.clone(), expected_version)
            .await?;

        // After the commit, never before: a snapshot is only real once the
        // cursor that served it is durable, and persisting first would leave a
        // row for a step a losing writer never served.
        self.file_snapshot(&simulation, &snapshot);

        if simulation.state == SessionState::Completed {
            self.evict(id);
            debug!(simulation_id = %id, "Simulation completed; cached state evicted");
        }

        Ok((simulation, snapshot))
    }

    /// Queues a served snapshot for filing, if a warehouse is configured.
    ///
    /// **Off the request's clock.** A failure cannot fail the advance — the
    /// cursor has already committed and the client already has its snapshot —
    /// and neither can a slow one delay it: the record goes into a bounded
    /// queue that one writer task drains.
    ///
    /// A full queue **drops** the record with a `WARN` naming the step. That is
    /// the same trade as a failed write, made explicit: the step stays
    /// reproducible, replay rebuilds it, and a retry writes the same rows. What
    /// it costs is a gap, and the honest way to find one is to compare a
    /// simulation's cursor against what `read_range` returns — a log line can be
    /// lost with the process, a missing row cannot.
    ///
    /// Filing is idempotent because both tables sort on
    /// `(simulation, generation, step, …)` and their `ReplacingMergeTree` engine
    /// collapses on that sorting key, so a retry of a step that did land
    /// replaces its rows rather than adding a second copy. The derived
    /// `snapshot_id` rides along as a payload column and is verified on read; it
    /// is not what does the replacing.
    ///
    /// The trade this accepts: a snapshot filed after the response means a
    /// client that advances and immediately queries the warehouse may not find
    /// the step yet. Deterministic replay is the read path that is always
    /// current; the warehouse is the one that is durable.
    fn file_snapshot(&self, simulation: &SessionV2, snapshot: &SeriesSnapshot) {
        let Some(warehouse) = &self.warehouse else {
            return;
        };

        // Decide before building anything. Materialising a record clones every
        // quote, so doing it and *then* discovering the queue is full would put
        // the cost of a degraded warehouse back on the advance — which is the
        // one thing this path exists to avoid.
        let incoming = snapshot_quote_count(snapshot);
        let queued = warehouse.queued_contracts.load(Ordering::SeqCst);
        if warehouse.sender.capacity() == 0
            || queued.saturating_add(incoming) > SNAPSHOT_QUEUE_CONTRACTS
        {
            warn!(
                simulation_id = %simulation.id,
                step = snapshot.step,
                queued,
                "The snapshot queue is full; the step was not filed and can be replayed"
            );
            return;
        }

        let record = snapshot_record(simulation.id, &simulation.parameters.symbol, snapshot);
        warehouse
            .queued_contracts
            .fetch_add(incoming, Ordering::SeqCst);

        if let Err(error) = warehouse.sender.try_send(record) {
            // Lost the race with another advance; undo the reservation.
            warehouse
                .queued_contracts
                .fetch_sub(incoming, Ordering::SeqCst);
            warn!(
                simulation_id = %simulation.id,
                step = snapshot.step,
                error = %error,
                "The snapshot queue is full; the step was not filed and can be replayed"
            );
        }
    }

    /// Deletes a simulation and everything cached for it.
    ///
    /// # Errors
    ///
    /// Returns any storage failure. A missing id is `Ok(false)`, not an error.
    #[instrument(skip(self), level = "debug")]
    pub(crate) async fn delete(&self, id: Uuid) -> Result<bool, ChainError> {
        let deleted = self.store.delete(id).await?;
        // Evict regardless: a delete that found nothing may still be cleaning
        // up after a simulation the store expired on its own.
        self.evict(id);
        self.forget_shared(id).await;
        Ok(deleted)
    }

    /// Expires idle simulations and evicts everything cached for them.
    ///
    /// Returns the ids that went, which is what makes the eviction possible at
    /// all — a count could not tell the caches which entries to drop.
    ///
    /// # Errors
    ///
    /// Returns any storage failure.
    #[instrument(skip(self), level = "debug")]
    pub async fn cleanup(&self) -> Result<Vec<Uuid>, ChainError> {
        let expired = self.store.cleanup().await?;
        for id in &expired {
            self.evict(*id);
            self.forget_shared(*id).await;
        }
        Ok(expired)
    }

    /// The number of factor tapes currently cached.
    #[must_use]
    pub fn cached_tapes(&self) -> usize {
        match self.tapes.lock() {
            Ok(tapes) => tapes.len(),
            Err(poisoned) => poisoned.into_inner().len(),
        }
    }

    /// The number of snapshots currently cached.
    #[must_use]
    pub fn cached_snapshots(&self) -> usize {
        match self.snapshots.lock() {
            Ok(snapshots) => snapshots.len(),
            Err(poisoned) => poisoned.into_inner().len(),
        }
    }

    /// Rejects a simulation that can no longer serve a snapshot.
    ///
    /// `Completed` maps to `410 Gone` at the boundary, matching v1's exhausted
    /// path; the terminal error state maps to `400`.
    fn reject_terminal(simulation: &SessionV2, what: &str) -> Result<(), ChainError> {
        if simulation.state == SessionState::Completed || simulation.is_complete() {
            return Err(ChainError::SimulatorError(format!(
                "simulation completed; {what}"
            )));
        }
        if simulation.state == SessionState::Error {
            return Err(ChainError::InvalidState(
                "simulation is in error state".to_string(),
            ));
        }
        Ok(())
    }

    /// Returns the snapshot at `step`, building whatever is missing.
    ///
    /// Locks are held only for the map operations, never across a build: the
    /// tape and the snapshot are produced outside any critical section, so a
    /// slow build cannot stall another simulation's request.
    async fn snapshot_at(
        &self,
        simulation: &SessionV2,
        step: usize,
    ) -> Result<SeriesSnapshot, ChainError> {
        if let Some(cached) = self.cached_snapshot(simulation.id, step) {
            return Ok(cached);
        }

        let key = (simulation.id, step);

        // Either this call owns the build or it waits on the one already
        // running, decided under the lock so two callers cannot both decide
        // they are the owner. Without this, N concurrent readers of the same
        // cold step each start the same priced build.
        let subscription = {
            let mut builds = match self.snapshot_builds.lock() {
                Ok(builds) => builds,
                Err(poisoned) => poisoned.into_inner(),
            };
            match builds.get(&key) {
                Some(running) => Some(running.subscribe()),
                None => {
                    let (sender, _) = broadcast::channel(1);
                    builds.insert(key, sender);
                    None
                }
            }
        };

        if let Some(mut waiting) = subscription {
            return match waiting.recv().await {
                Ok(Ok(snapshot)) => Ok(snapshot),
                // The owner failed; report what it reported rather than
                // starting a second build that would fail the same way.
                Ok(Err(reason)) => Err(ChainError::Internal(reason)),
                // The owner's task died without publishing. Rare, and the
                // honest answer is to build it here rather than hang.
                Err(_) => self.build_snapshot(simulation, step).await,
            };
        }

        let result = self.build_snapshot(simulation, step).await;

        // Publish, then stop being the owner — in that order, so a caller that
        // subscribes after the removal misses the broadcast, retries, and finds
        // the snapshot in the cache.
        let sender = {
            let mut builds = match self.snapshot_builds.lock() {
                Ok(builds) => builds,
                Err(poisoned) => poisoned.into_inner(),
            };
            builds.remove(&key)
        };
        if let Some(sender) = sender {
            let published = match &result {
                Ok(snapshot) => Ok(snapshot.clone()),
                Err(error) => Err(error.to_string()),
            };
            // An error means nobody was waiting, which is the common case.
            let _ = sender.send(published);
        }

        result
    }

    /// Prices one snapshot, off the runtime and under the shared bound.
    ///
    /// The greek snapshots are built only when something will read them: a
    /// registered warehouse files every step, and a filed step has to carry
    /// what a replayed one does (issue #74). Without a warehouse the API prices
    /// them per request instead, and only when asked.
    ///
    /// Pricing is real synchronous CPU — up to `OCS_MAX_SNAPSHOT_CONTRACTS`
    /// contracts, about 1.54x that again with the greek snapshots on — so it
    /// runs under the same bound the API renderers use rather than a second
    /// one: they compete for the same cores.
    ///
    /// The result is cached **inside** the job, exactly as [`Self::build_tape`]
    /// files a tape. A blocking task cannot be cancelled but awaiting it can,
    /// so a client that disconnects mid-build would otherwise throw away a
    /// snapshot that ran to completion anyway and leave the cache cold for the
    /// next reader.
    async fn build_snapshot(
        &self,
        simulation: &SessionV2,
        step: usize,
    ) -> Result<SeriesSnapshot, ChainError> {
        let tape = self.tape_for(simulation).await?;
        let greek_snapshots = self.warehouse.is_some();
        let parameters = simulation.parameters.clone();
        let snapshots = Arc::clone(&self.snapshots);
        let id = simulation.id;

        crate::utils::admission::admit_blocking(move || {
            let snapshot = SeriesBuilder::new(&parameters, &tape)?
                .with_greek_snapshots(greek_snapshots)
                .snapshot(step)?;
            Self::cache_snapshot_into(&snapshots, id, snapshot.clone());
            Ok(snapshot)
        })
        .await
    }

    /// Reads a cached snapshot, refreshing its recency.
    fn cached_snapshot(&self, id: Uuid, step: usize) -> Option<SeriesSnapshot> {
        let mut snapshots = match self.snapshots.lock() {
            Ok(snapshots) => snapshots,
            Err(poisoned) => poisoned.into_inner(),
        };
        snapshots.get(id, step).cloned()
    }

    /// Stores a built snapshot.
    ///
    /// Static so a blocking job can file its own result without borrowing the
    /// manager; see [`Self::build_snapshot`] for why filing happens there.
    fn cache_snapshot_into(snapshots: &Mutex<SnapshotCache>, id: Uuid, snapshot: SeriesSnapshot) {
        let mut snapshots = match snapshots.lock() {
            Ok(snapshots) => snapshots,
            Err(poisoned) => poisoned.into_inner(),
        };
        snapshots.insert(id, snapshot);
    }

    /// Returns the simulation's factor tape, building it on a miss.
    ///
    /// Built outside the lock — holding the map while it runs would serialise
    /// every other simulation behind it — and off the runtime. `FactorTape::build`
    /// is pure and synchronous, and it is the one place a v2 request does real
    /// CPU work up front: a historical walk estimates a volatility per step,
    /// which at the 10 000-step cap measures over three seconds. Left on a
    /// worker that would stall every other request the worker holds, so it goes
    /// to the blocking pool, exactly as the export path already does with the
    /// same call.
    ///
    /// The result is filed **inside** the blocking task rather than after the
    /// await. A `spawn_blocking` task cannot be cancelled, but awaiting it can:
    /// a client that disconnects or times out mid-build drops this future, and
    /// filing afterwards would throw away a build that ran to completion
    /// anyway. At three seconds a caller retrying under a shorter timeout would
    /// then never warm the cache and would pin a blocking thread on every
    /// attempt.
    async fn tape_for(&self, simulation: &SessionV2) -> Result<FactorTape, ChainError> {
        if let Some(tape) = self.cached_tape(simulation.id) {
            return Ok(tape);
        }

        let id = simulation.id;
        // Computed ONCE. The key serialises every parameter, and a historical
        // simulation carries its prices, so recomputing it per lookup — the
        // waiter below polls twenty times a second — would put a large
        // synchronous serialization on the runtime for every poll.
        let key = tape_key(id, &simulation.parameters);

        // Another instance may already have walked this. Consulted before the
        // build lock rather than after, so a hit costs one round trip instead
        // of queueing behind a build that is about to be redundant.
        if let Some(tape) = self.shared_tape(&key, id).await {
            Self::cache_tape(&self.tapes, self.config.max_cached_tapes, id, tape.clone());
            return Ok(tape);
        }

        // The lookup above is a round trip, and another request in THIS
        // process can finish and cache the tape while it is in flight. Without
        // this recheck that caller would elect itself owner and rebuild a tape
        // that is already warm.
        if let Some(tape) = self.cached_tape(id) {
            return Ok(tape);
        }

        // Either this call owns the build or it waits on the one already
        // running. Decided under the lock, so two callers cannot both decide
        // they are the owner.
        let (mut waiting, owned) = {
            let mut builds = match self.builds.lock() {
                Ok(builds) => builds,
                Err(poisoned) => poisoned.into_inner(),
            };

            match builds.get(&id) {
                Some(running) => (running.subscribe(), None),
                None => {
                    let (sender, receiver) = broadcast::channel(1);
                    builds.insert(id, sender.clone());
                    (receiver, Some(sender))
                }
            }
        };

        // The owner's work runs in a task that OUTLIVES this future, and the
        // owner waits on the same channel as everyone else. A blocking build
        // keeps running when the client that asked for it disconnects, so
        // doing this inline would let a cancelled owner leave its entry in
        // `builds` with nobody to publish it, its deployment-wide claim held
        // until expiry, and its tape unshared — local waiters hanging on a
        // tape that was in fact built.
        if let Some(sender) = owned {
            tokio::spawn(
                OwnedBuild {
                    id,
                    key,
                    parameters: simulation.parameters.clone(),
                    tapes: Arc::clone(&self.tapes),
                    builds: Arc::clone(&self.builds),
                    shared: self.shared_tapes.clone(),
                    max_cached_tapes: self.config.max_cached_tapes,
                    shared_build_wait: self.shared_build_wait,
                }
                .run(sender),
            );
        }

        match waiting.recv().await {
            Ok(Ok(tape)) => Ok(tape),
            // The owner failed; report what it reported rather than
            // starting a second build that would fail the same way.
            Ok(Err(reason)) => Err(ChainError::Internal(reason)),
            // The owner's task died without publishing. Rare, and the
            // honest answer is to build it here rather than hang.
            Err(_) => self.build_tape(simulation).await,
        }
    }

    /// Builds a tape off the runtime and files it.
    ///
    /// `FactorTape::build` is pure and synchronous, and it is the one place a
    /// v2 request does real CPU work up front: a historical walk estimates a
    /// volatility per step, which at the 10 000-step cap measures over three
    /// seconds. Left on a worker it would stall every other request that worker
    /// holds, so it goes to the blocking pool, exactly as the export path
    /// already does with the same call.
    ///
    /// The result is filed **inside** the blocking task rather than after the
    /// await. A `spawn_blocking` task cannot be cancelled, but awaiting it can:
    /// a client that disconnects or times out mid-build drops that future, and
    /// filing afterwards would throw away a build that ran to completion
    /// anyway.
    async fn build_tape(&self, simulation: &SessionV2) -> Result<FactorTape, ChainError> {
        Self::build_tape_into(
            simulation.parameters.clone(),
            simulation.id,
            Arc::clone(&self.tapes),
            self.config.max_cached_tapes,
        )
        .await
    }

    /// As [`Self::build_tape`], without borrowing the manager.
    ///
    /// Owned so the build can run in a task that outlives the request that
    /// asked for it; see [`OwnedBuild`].
    async fn build_tape_into(
        parameters: SimulationParametersV2,
        id: Uuid,
        tapes: Arc<Mutex<HashMap<Uuid, TapeEntry>>>,
        max_cached_tapes: usize,
    ) -> Result<FactorTape, ChainError> {
        tokio::task::spawn_blocking(move || {
            let tape = FactorTape::build(&parameters, &parameters.method)?;
            Self::cache_tape(&tapes, max_cached_tapes, id, tape.clone());
            Ok(tape)
        })
        .await
        .map_err(|e| ChainError::Internal(format!("the factor tape build did not finish: {e}")))?
    }

    /// The tape another instance left, decoded, or `None`.
    ///
    /// Takes the key rather than the simulation because computing it
    /// serialises every parameter, which is expensive enough that the caller
    /// does it once and reuses it.
    async fn shared_tape(&self, key: &str, id: Uuid) -> Option<FactorTape> {
        let shared = self.shared_tapes.as_ref()?;
        Self::shared_tape_from(shared.as_ref(), key, id).await
    }

    /// As [`Self::shared_tape`], against a cache the caller already has.
    ///
    /// A stored document this build cannot decode is a miss, not an error: the
    /// key carries the snapshot generation and a fingerprint of the
    /// parameters, so it should not happen, and rebuilding is the answer that
    /// still serves the request.
    async fn shared_tape_from(
        shared: &dyn SharedTapeCache,
        key: &str,
        id: Uuid,
    ) -> Option<FactorTape> {
        let encoded = shared.get(key).await?;

        match serde_json::from_str::<FactorTape>(&encoded) {
            Ok(tape) => Some(tape),
            Err(error) => {
                warn!(
                    %error,
                    simulation_id = %id,
                    "a shared tape did not decode; rebuilding it"
                );
                None
            }
        }
    }

    /// Leaves a built tape where the other instances can find it.
    async fn share_tape_to(shared: &dyn SharedTapeCache, key: &str, id: Uuid, tape: &FactorTape) {
        match serde_json::to_string(tape) {
            Ok(encoded) => shared.put(key, &encoded).await,
            Err(error) => warn!(
                %error,
                simulation_id = %id,
                "a built tape did not encode; it stays local to this instance"
            ),
        }
    }

    /// Reads a cached tape, refreshing its recency.
    fn cached_tape(&self, id: Uuid) -> Option<FactorTape> {
        let mut tapes = match self.tapes.lock() {
            Ok(tapes) => tapes,
            Err(poisoned) => poisoned.into_inner(),
        };
        let entry = tapes.get_mut(&id)?;
        entry.last_access = Instant::now();
        Some(entry.tape.clone())
    }

    /// Stores a built tape, evicting the least recently used first.
    ///
    /// Takes the map rather than `&self` so the builder can file its result
    /// from inside the blocking task, where no caller can drop it.
    ///
    /// One race that follows from filing there, recorded because it is benign
    /// only under the current routes: a build already running when the
    /// simulation is deleted, completed or reaped will file afterwards, leaving
    /// a tape for an id the store no longer knows. Nothing can serve it — every
    /// path reads the store before the cache — so it costs memory until the LRU
    /// pushes it out. It would stop being benign the day v2 gains a route that
    /// changes a simulation's parameters in place, because the stale tape would
    /// then be a tape of the *old* parameters under a live id.
    fn cache_tape(
        tapes: &Mutex<HashMap<Uuid, TapeEntry>>,
        max_cached_tapes: usize,
        id: Uuid,
        tape: FactorTape,
    ) {
        let mut tapes = match tapes.lock() {
            Ok(tapes) => tapes,
            Err(poisoned) => poisoned.into_inner(),
        };

        tapes.remove(&id);
        // The capacity is validated `>= 1` when the configuration loads, so
        // `- 1` cannot underflow. Evicting before the insert keeps the id being
        // inserted out of the running for victim.
        let max = max_cached_tapes;
        debug_assert!(
            max >= 1,
            "the configured capacity is validated >= 1 at load"
        );
        while tapes.len() > max - 1 {
            let victim = tapes
                .iter()
                .min_by_key(|(_, entry)| entry.last_access)
                .map(|(id, _)| *id);
            match victim {
                Some(victim) => {
                    tapes.remove(&victim);
                }
                None => break,
            }
        }

        tapes.insert(
            id,
            TapeEntry {
                tape,
                last_access: Instant::now(),
            },
        );
    }

    /// Drops everything cached for a simulation.
    /// Drops what this simulation left in the shared cache.
    ///
    /// Separate from [`SimulationManager::evict`] because it is I/O and that
    /// is not: the local eviction must stay callable from anywhere, including
    /// a blocking task.
    async fn forget_shared(&self, id: Uuid) {
        if let Some(shared) = self.shared_tapes.as_ref() {
            shared.forget_simulation(&id.to_string()).await;
        }
    }

    fn evict(&self, id: Uuid) {
        match self.tapes.lock() {
            Ok(mut tapes) => {
                tapes.remove(&id);
            }
            Err(poisoned) => {
                poisoned.into_inner().remove(&id);
            }
        }
        match self.snapshots.lock() {
            Ok(mut snapshots) => {
                snapshots.evict_simulation(id);
            }
            Err(poisoned) => {
                poisoned.into_inner().evict_simulation(id);
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::api::rest::models::{ApiTimeFrame, ApiWalkType};
    use crate::api::rest::requests_v2::CreateSimulationRequest;
    use crate::infrastructure::{ContractQuote, ContractSeriesQuery, SnapshotRecord};
    use crate::session::store::InMemorySimulationStore;
    use crate::session::{ExpiryRule, ExpiryRuleKind};
    use chrono::{TimeZone, Utc, Weekday};

    fn request(steps: usize) -> CreateSimulationRequest {
        let rules = vec![
            match ExpiryRule::new("zero_dte", ExpiryRuleKind::Daily, 1) {
                Ok(rule) => rule,
                Err(error) => panic!("the test rule must be valid: {error}"),
            },
            match ExpiryRule::new(
                "weeklies",
                ExpiryRuleKind::weekly([Weekday::Mon, Weekday::Fri]),
                2,
            ) {
                Ok(rule) => rule,
                Err(error) => panic!("the test rule must be valid: {error}"),
            },
        ];
        let start_at = match Utc.with_ymd_and_hms(2026, 1, 5, 14, 30, 0).single() {
            Some(instant) => instant,
            None => panic!("the test instant must be valid"),
        };

        CreateSimulationRequest {
            symbol: "SPX".to_string(),
            steps,
            start_at: Some(start_at),
            step_interval_seconds: Some(86_400),
            timezone: "America/New_York".to_string(),
            calendar: None,
            expiration_time: "17:00".to_string(),
            schedules: rules,
            initial_price: 5000.0,
            volatility: 0.18,
            risk_free_rate: 0.04,
            dividend_yield: 0.0,
            method: ApiWalkType::Brownian {
                dt: 1.0 / 252.0,
                drift: 0.0,
                volatility: 0.18,
            },
            time_frame: ApiTimeFrame::Day,
            chain_size: Some(3),
            strike_interval: Some(25.0),
            skew_slope: None,
            smile_curve: None,
            spread: Some(0.02),
            strike_ladder: Default::default(),
            spread_proportional: None,
            spread_moneyness_widening: None,
            spread_tenor_widening: None,
            spread_tick: None,
            seed: Some(42),
        }
    }

    fn parameters(steps: usize) -> SimulationParametersV2 {
        match SimulationParametersV2::try_from(request(steps)) {
            Ok(parameters) => parameters,
            Err(error) => panic!("the request must convert: {error}"),
        }
    }

    fn manager() -> SimulationManager {
        SimulationManager::new(
            Arc::new(InMemorySimulationStore::new()),
            SimulationV2Config::default(),
        )
    }

    /// A shared tape cache in memory, standing in for Redis, that counts what
    /// it was asked to do.
    ///
    /// The counts are the point: a manager that BUILT a tape always writes one,
    /// so "no write" is how a test sees that the second instance did not
    /// rebuild.
    #[derive(Default)]
    struct SharedTapes {
        entries: Mutex<HashMap<String, String>>,
        /// The keys some instance has claimed the right to build.
        building: Mutex<std::collections::HashSet<String>>,
        reads: AtomicUsize,
        writes: AtomicUsize,
        claims: AtomicUsize,
    }

    impl SharedTapes {
        fn writes(&self) -> usize {
            self.writes.load(Ordering::SeqCst)
        }

        fn reads(&self) -> usize {
            self.reads.load(Ordering::SeqCst)
        }

        fn claims(&self) -> usize {
            self.claims.load(Ordering::SeqCst)
        }

        fn keys(&self) -> Vec<String> {
            match self.entries.lock() {
                Ok(entries) => entries.keys().cloned().collect(),
                Err(poisoned) => poisoned.into_inner().keys().cloned().collect(),
            }
        }
    }

    #[async_trait::async_trait]
    impl SharedTapeCache for SharedTapes {
        async fn get(&self, key: &str) -> Option<String> {
            self.reads.fetch_add(1, Ordering::SeqCst);
            match self.entries.lock() {
                Ok(entries) => entries.get(key).cloned(),
                Err(poisoned) => poisoned.into_inner().get(key).cloned(),
            }
        }

        async fn put(&self, key: &str, encoded: &str) {
            self.writes.fetch_add(1, Ordering::SeqCst);
            match self.entries.lock() {
                Ok(mut entries) => entries.insert(key.to_string(), encoded.to_string()),
                Err(poisoned) => poisoned
                    .into_inner()
                    .insert(key.to_string(), encoded.to_string()),
            };
        }

        async fn forget_simulation(&self, id: &str) {
            let suffix = format!(":{id}");
            match self.entries.lock() {
                Ok(mut entries) => entries.retain(|key, _| !key.ends_with(&suffix)),
                Err(poisoned) => poisoned
                    .into_inner()
                    .retain(|key, _| !key.ends_with(&suffix)),
            }
        }

        async fn claim_build(&self, key: &str) -> BuildClaim {
            self.claims.fetch_add(1, Ordering::SeqCst);
            let mut held = match self.building.lock() {
                Ok(held) => held,
                Err(poisoned) => poisoned.into_inner(),
            };
            if held.insert(key.to_string()) {
                BuildClaim::Held(format!("token:{key}"))
            } else {
                BuildClaim::Taken
            }
        }

        async fn release_build(&self, key: &str, token: &str) {
            let mut held = match self.building.lock() {
                Ok(held) => held,
                Err(poisoned) => poisoned.into_inner(),
            };
            // Ownership-safe like the Redis script: a token that is not this
            // claim's releases nothing.
            if token == format!("token:{key}") {
                held.remove(key);
            }
        }
    }

    /// A shared cache that is always down, which is what an unreachable Redis
    /// looks like from here.
    struct BrokenTapes;

    #[async_trait::async_trait]
    impl SharedTapeCache for BrokenTapes {
        async fn get(&self, _key: &str) -> Option<String> {
            None
        }

        async fn put(&self, _key: &str, _encoded: &str) {}

        async fn forget_simulation(&self, _id: &str) {}

        /// A gate that cannot be reached must let the caller build, or an
        /// unreachable Redis becomes an unanswerable request.
        async fn claim_build(&self, _key: &str) -> BuildClaim {
            BuildClaim::Unclaimed
        }

        async fn release_build(&self, _key: &str, _token: &str) {}
    }

    /// A warehouse that records what it was asked to file, and can be told to
    /// fail — the two behaviours the wiring promises something about.
    #[derive(Default)]
    struct RecordingWarehouse {
        filed: Mutex<Vec<SnapshotRecord>>,
        fail: bool,
    }

    impl RecordingWarehouse {
        fn failing() -> Self {
            Self {
                filed: Mutex::new(Vec::new()),
                fail: true,
            }
        }

        /// The records it was handed, whole — so a test can assert on what is
        /// inside one, not just that one arrived.
        fn records(&self) -> Vec<SnapshotRecord> {
            match self.filed.lock() {
                Ok(filed) => filed.clone(),
                Err(poisoned) => poisoned.into_inner().clone(),
            }
        }

        fn filed(&self) -> Vec<(Uuid, usize)> {
            self.records()
                .iter()
                .map(|record| (record.simulation, record.step))
                .collect()
        }
    }

    #[async_trait::async_trait]
    impl SimulationSnapshotRepository for RecordingWarehouse {
        /// No server behind it; reachable exactly as long as the process is.
        async fn ping(&self) -> Result<(), ChainError> {
            Ok(())
        }

        async fn persist(&self, record: SnapshotRecord) -> Result<(), ChainError> {
            if self.fail {
                return Err(ChainError::Internal("the warehouse is down".to_string()));
            }
            match self.filed.lock() {
                Ok(mut filed) => filed.push(record),
                Err(poisoned) => poisoned.into_inner().push(record),
            }
            Ok(())
        }

        async fn get(
            &self,
            _simulation: Uuid,
            _generation: u64,
            _step: usize,
        ) -> Result<Option<SnapshotRecord>, ChainError> {
            Ok(None)
        }

        async fn read_range(
            &self,
            _simulation: Uuid,
            _generation: u64,
            _from_step: usize,
            _to_step: usize,
        ) -> Result<Vec<SnapshotRecord>, ChainError> {
            Ok(Vec::new())
        }

        async fn contract_series(
            &self,
            _query: ContractSeriesQuery,
        ) -> Result<Vec<ContractQuote>, ChainError> {
            Ok(Vec::new())
        }
    }

    /// A warehouse whose first write never completes — the shape a degraded
    /// deployment has, and the one an unbounded queue cannot survive.
    #[derive(Default)]
    struct StallingWarehouse {
        started: AtomicUsize,
    }

    impl StallingWarehouse {
        fn started(&self) -> usize {
            self.started.load(Ordering::SeqCst)
        }
    }

    #[async_trait::async_trait]
    impl SimulationSnapshotRepository for StallingWarehouse {
        /// No server behind it; reachable exactly as long as the process is.
        async fn ping(&self) -> Result<(), ChainError> {
            Ok(())
        }

        async fn persist(&self, _record: SnapshotRecord) -> Result<(), ChainError> {
            self.started.fetch_add(1, Ordering::SeqCst);
            std::future::pending::<()>().await;
            Ok(())
        }

        async fn get(
            &self,
            _simulation: Uuid,
            _generation: u64,
            _step: usize,
        ) -> Result<Option<SnapshotRecord>, ChainError> {
            Ok(None)
        }

        async fn read_range(
            &self,
            _simulation: Uuid,
            _generation: u64,
            _from_step: usize,
            _to_step: usize,
        ) -> Result<Vec<SnapshotRecord>, ChainError> {
            Ok(Vec::new())
        }

        async fn contract_series(
            &self,
            _query: ContractSeriesQuery,
        ) -> Result<Vec<ContractQuote>, ChainError> {
            Ok(Vec::new())
        }
    }

    /// Filing is detached, so a test has to let the spawned write run before it
    /// can observe it. One yield is enough on the current-thread runtime the
    /// tests use; the loop keeps it from being a race on a busier one.
    async fn settle() {
        for _ in 0..16 {
            tokio::task::yield_now().await;
        }
    }

    /// The simulation id is not an input to anything seeded.
    ///
    /// This is what the switch to random ids rests on. Two simulations built
    /// from one set of parameters have different ids and must still produce the
    /// same snapshots, strike for strike — `SeriesSnapshot`'s equality compares
    /// premiums, Greeks and the underlying price, not lengths. If the id ever
    /// leaked into the tape, the planner or the chain build, this fails.
    #[tokio::test]
    async fn test_the_simulation_id_does_not_reach_the_tape() {
        let manager = manager();

        let first = created(&manager, 3).await;
        let second = created(&manager, 3).await;
        assert_ne!(first.id, second.id, "ids are random, so two differ");
        assert_eq!(
            first.parameters.seed, second.parameters.seed,
            "the fixture must pin the seed, or this proves nothing"
        );

        for _ in 0..3 {
            let left = match manager.advance(first.id).await {
                Ok((_, snapshot)) => snapshot,
                Err(error) => panic!("the first simulation must advance: {error}"),
            };
            let right = match manager.advance(second.id).await {
                Ok((_, snapshot)) => snapshot,
                Err(error) => panic!("the second simulation must advance: {error}"),
            };

            assert_eq!(
                left, right,
                "step {} differs between two simulations that share every parameter",
                left.step
            );
        }
    }

    /// Concurrent readers of one cold STEP share one priced build.
    ///
    /// The tape has its own single-flight; the snapshot needs the same and for
    /// a sharper reason since issue #74. With a warehouse registered a snapshot
    /// build prices up to the per-snapshot cap with both greek sets on, so N
    /// concurrent readers of the same step used to commit the machine to N
    /// copies of the same seconds-long job.
    ///
    /// Every reader must get the same snapshot, one entry must be cached, and
    /// the owner must have cleaned up after itself — a stale entry in the
    /// in-flight map would make the NEXT reader of that step wait forever on a
    /// broadcast nobody will send.
    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
    async fn test_concurrent_readers_of_one_step_share_one_snapshot_build() {
        let manager = Arc::new(manager().with_warehouse(
            Arc::new(RecordingWarehouse::default()) as Arc<dyn SimulationSnapshotRepository>
        ));
        let simulation = created(&manager, 3).await;

        let mut readers = Vec::new();
        for _ in 0..8 {
            let manager = Arc::clone(&manager);
            let id = simulation.id;
            readers.push(tokio::spawn(async move { manager.peek(id).await }));
        }

        let mut snapshots = Vec::new();
        for reader in readers {
            match reader.await {
                Ok(Ok((_, snapshot))) => snapshots.push(snapshot),
                Ok(Err(error)) => panic!("every reader must be served: {error}"),
                Err(error) => panic!("a reader panicked: {error}"),
            }
        }

        assert_eq!(snapshots.len(), 8);
        for snapshot in &snapshots {
            assert_eq!(
                snapshot, &snapshots[0],
                "every reader must see the same snapshot"
            );
        }
        assert_eq!(
            manager.cached_snapshots(),
            1,
            "eight readers of one step must leave one snapshot"
        );
        assert!(
            match manager.snapshot_builds.lock() {
                Ok(builds) => builds.is_empty(),
                Err(poisoned) => poisoned.into_inner().is_empty(),
            },
            "the owner must stop being the owner once it has published"
        );
    }

    /// Concurrent first reads of one simulation share a single build.
    ///
    /// The build is the one place a v2 request does seconds of CPU, so N
    /// concurrent peeks starting N identical builds is not wasted allocation,
    /// it is the machine. What proves the sharing is the snapshots: every
    /// caller gets the same tape, and only one entry is cached.
    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
    async fn test_concurrent_first_reads_share_one_build() {
        let manager = Arc::new(manager());
        let simulation = created(&manager, 3).await;

        let mut readers = Vec::new();
        for _ in 0..8 {
            let manager = Arc::clone(&manager);
            let id = simulation.id;
            readers.push(tokio::spawn(async move { manager.peek(id).await }));
        }

        let mut snapshots = Vec::new();
        for reader in readers {
            match reader.await {
                Ok(Ok((_, snapshot))) => snapshots.push(snapshot),
                Ok(Err(error)) => panic!("every reader must be served: {error}"),
                Err(error) => panic!("a reader panicked: {error}"),
            }
        }

        assert_eq!(snapshots.len(), 8);
        for snapshot in &snapshots {
            assert_eq!(
                snapshot, &snapshots[0],
                "every reader must see the same tape"
            );
        }
        assert_eq!(
            manager.cached_tapes(),
            1,
            "eight readers of one simulation must leave one tape"
        );
    }

    /// An advance files exactly the step it served.
    #[tokio::test]
    async fn test_an_advance_files_the_step_it_served() {
        let warehouse = Arc::new(RecordingWarehouse::default());
        let manager = SimulationManager::new(
            Arc::new(InMemorySimulationStore::new()),
            SimulationV2Config::default(),
        )
        .with_warehouse(Arc::clone(&warehouse) as Arc<dyn SimulationSnapshotRepository>);

        let simulation = created(&manager, 3).await;
        match manager.advance(simulation.id).await {
            Ok(_) => {}
            Err(error) => panic!("the advance must serve: {error}"),
        }
        settle().await;

        assert_eq!(
            warehouse.filed(),
            vec![(simulation.id, 0)],
            "the step the advance served is the step that is filed"
        );
    }

    /// A registered warehouse is what turns the greek snapshots on.
    ///
    /// The wiring that makes issue #74 work: `SeriesBuilder` builds them only
    /// when asked, and the manager asks exactly when there is a warehouse to
    /// file them into. A filed record with empty snapshots would persist a tape
    /// strictly poorer than a replayed one — the asymmetry the issue removes —
    /// and every existing test would still pass, because none of them looks
    /// inside a filed record.
    #[tokio::test]
    async fn test_a_registered_warehouse_files_the_greeks() {
        let warehouse = Arc::new(RecordingWarehouse::default());
        let manager = SimulationManager::new(
            Arc::new(InMemorySimulationStore::new()),
            SimulationV2Config::default(),
        )
        .with_warehouse(Arc::clone(&warehouse) as Arc<dyn SimulationSnapshotRepository>);

        let simulation = created(&manager, 3).await;
        match manager.advance(simulation.id).await {
            Ok(_) => {}
            Err(error) => panic!("the advance must serve: {error}"),
        }
        settle().await;

        let records = warehouse.records();
        let record = match records.first() {
            Some(record) => record,
            None => panic!("the advance must file a record"),
        };
        let quotes: Vec<_> = record
            .expirations
            .iter()
            .flat_map(|expiration| expiration.quotes.iter())
            .collect();
        assert!(!quotes.is_empty(), "the filed record must carry quotes");
        assert!(
            quotes
                .iter()
                .all(|quote| quote.greeks_call.is_some() && quote.greeks_put.is_some()),
            "every filed quote must carry both snapshots"
        );
    }

    /// A registered warehouse does not change the tape.
    ///
    /// The flag it turns on selects a different upstream branch inside
    /// `OptionChain::build_chain`, so two deployments of the same build, same
    /// parameters and same seed take two different code paths to the same
    /// market. If they ever disagreed, whether a step was filed would change
    /// what a client was served — the worst regression this service can have,
    /// and one no other test would see, because every other comparison is
    /// between two managers configured the same way.
    #[tokio::test]
    async fn test_a_warehouse_does_not_change_the_served_market() {
        let filing = SimulationManager::new(
            Arc::new(InMemorySimulationStore::new()),
            SimulationV2Config::default(),
        )
        .with_warehouse(
            Arc::new(RecordingWarehouse::default()) as Arc<dyn SimulationSnapshotRepository>
        );
        let plain = SimulationManager::new(
            Arc::new(InMemorySimulationStore::new()),
            SimulationV2Config::default(),
        );

        let served = async |manager: &SimulationManager| {
            let simulation = created(manager, 3).await;
            match manager.peek(simulation.id).await {
                Ok((_, snapshot)) => snapshot,
                Err(error) => panic!("the peek must serve: {error}"),
            }
        };
        let with_warehouse = served(&filing).await;
        let without = served(&plain).await;

        assert_eq!(
            with_warehouse.spot, without.spot,
            "the seeded price path must not depend on the warehouse"
        );
        assert_eq!(with_warehouse.base_volatility, without.base_volatility);
        assert_eq!(with_warehouse.chains.len(), without.chains.len());

        // Every priced value, strike by strike. `PartialEq` on `ExpiryChain`
        // compares the whole chain including the greek snapshots, which DO
        // differ by design, so the comparison is over what is served.
        //
        // Counted, not just zipped: a `zip` over two chains of different
        // lengths truncates silently, and over two EMPTY ones it asserts
        // nothing at all — which is the outcome at the low volatilities where a
        // chain serves no valid strike.
        let mut compared = 0_usize;
        for (filed, replayed) in with_warehouse.chains.iter().zip(without.chains.iter()) {
            assert_eq!(filed.expires_at, replayed.expires_at);
            assert_eq!(filed.days_to_expiration, replayed.days_to_expiration);
            assert_eq!(
                filed.chain.iter().count(),
                replayed.chain.iter().count(),
                "the two deployments must quote the same strikes"
            );
            for (left, right) in filed.chain.iter().zip(replayed.chain.iter()) {
                compared += 1;
                assert_eq!(left.strike_price, right.strike_price);
                assert_eq!(left.implied_volatility, right.implied_volatility);
                assert_eq!(left.call_bid, right.call_bid);
                assert_eq!(left.call_ask, right.call_ask);
                assert_eq!(left.call_middle, right.call_middle);
                assert_eq!(left.put_bid, right.put_bid);
                assert_eq!(left.put_ask, right.put_ask);
                assert_eq!(left.put_middle, right.put_middle);
                assert_eq!(left.delta_call, right.delta_call);
                assert_eq!(left.delta_put, right.delta_put);
                assert_eq!(left.gamma, right.gamma);
            }
        }
        assert!(compared > 0, "the fixture must actually quote something");

        // And the one thing that IS meant to differ.
        assert!(
            with_warehouse
                .chains
                .iter()
                .flat_map(|chain| chain.chain.iter())
                .all(|data| data.greeks_call.is_some())
        );
        assert!(
            without
                .chains
                .iter()
                .flat_map(|chain| chain.chain.iter())
                .all(|data| data.greeks_call.is_none())
        );
    }

    /// Without a warehouse nothing pays for the greeks.
    ///
    /// The other half of the same wiring: a deployment that files nothing and
    /// whose clients never ask must not be charged about 1.5x a chain build on
    /// every advance. When a client does ask, the API prices them per request
    /// instead.
    #[tokio::test]
    async fn test_a_manager_without_a_warehouse_does_not_price_the_greeks() {
        let manager = SimulationManager::new(
            Arc::new(InMemorySimulationStore::new()),
            SimulationV2Config::default(),
        );

        let simulation = created(&manager, 3).await;
        let (_, snapshot) = match manager.peek(simulation.id).await {
            Ok(served) => served,
            Err(error) => panic!("the peek must serve: {error}"),
        };

        let contracts: Vec<_> = snapshot
            .chains
            .iter()
            .flat_map(|chain| chain.chain.iter())
            .collect();
        assert!(!contracts.is_empty(), "the snapshot must quote something");
        assert!(
            contracts
                .iter()
                .all(|data| data.greeks_call.is_none() && data.greeks_put.is_none()),
            "no snapshot should have been priced"
        );
    }

    /// A warehouse that never drains stops receiving, rather than accumulating
    /// records until the process dies.
    ///
    /// The bound that matters is rows, not records: a record is a few hundred
    /// quotes in this fixture and up to the per-snapshot cap in a large
    /// configuration, so a depth in records says nothing about what is
    /// resident.
    #[tokio::test]
    async fn test_a_stalled_warehouse_stops_being_queued() {
        let warehouse = Arc::new(StallingWarehouse::default());
        let manager = SimulationManager::new(
            Arc::new(InMemorySimulationStore::new()),
            SimulationV2Config::default(),
        )
        .with_warehouse(Arc::clone(&warehouse) as Arc<dyn SimulationSnapshotRepository>);

        // More advances than the queue can hold, against a warehouse whose
        // first write never returns.
        for _ in 0..(SNAPSHOT_QUEUE_DEPTH + 8) {
            let simulation = created(&manager, 2).await;
            match manager.advance(simulation.id).await {
                Ok(_) => {}
                Err(error) => panic!("the advance must serve regardless: {error}"),
            }
        }
        settle().await;

        assert!(
            warehouse.started() <= SNAPSHOT_QUEUE_DEPTH + 1,
            "a stalled warehouse must stop receiving, got {} starts",
            warehouse.started()
        );
    }

    /// A warehouse that is down does not fail the advance. This is the whole
    /// point of filing after the commit and off the request's clock.
    #[tokio::test]
    async fn test_a_failing_warehouse_does_not_fail_the_advance() {
        let warehouse = Arc::new(RecordingWarehouse::failing());
        let manager = SimulationManager::new(
            Arc::new(InMemorySimulationStore::new()),
            SimulationV2Config::default(),
        )
        .with_warehouse(warehouse as Arc<dyn SimulationSnapshotRepository>);

        let simulation = created(&manager, 3).await;

        match manager.advance(simulation.id).await {
            Ok((advanced, _)) => assert_eq!(advanced.current_step, 1, "the cursor still moved"),
            Err(error) => panic!("a warehouse failure must not fail the advance: {error}"),
        }
        settle().await;
    }

    /// A peek serves a snapshot and files nothing: it moves no cursor, so there
    /// is no step to file.
    #[tokio::test]
    async fn test_a_peek_files_nothing() {
        let warehouse = Arc::new(RecordingWarehouse::default());
        let manager = SimulationManager::new(
            Arc::new(InMemorySimulationStore::new()),
            SimulationV2Config::default(),
        )
        .with_warehouse(Arc::clone(&warehouse) as Arc<dyn SimulationSnapshotRepository>);

        let simulation = created(&manager, 3).await;
        match manager.peek(simulation.id).await {
            Ok(_) => {}
            Err(error) => panic!("the peek must serve: {error}"),
        }
        settle().await;

        assert!(warehouse.filed().is_empty(), "a peek persists nothing");
    }

    /// Without a warehouse the serving path is unchanged — there is nothing to
    /// call and nothing to fail.
    #[tokio::test]
    async fn test_a_manager_without_a_warehouse_serves_normally() {
        let manager = manager();
        let simulation = created(&manager, 2).await;

        match manager.advance(simulation.id).await {
            Ok((advanced, snapshot)) => {
                assert_eq!(advanced.current_step, 1);
                assert_eq!(snapshot.step, 0);
            }
            Err(error) => panic!("the advance must serve: {error}"),
        }
    }

    async fn created(manager: &SimulationManager, steps: usize) -> SessionV2 {
        match manager.create(parameters(steps)).await {
            Ok(simulation) => simulation,
            Err(error) => panic!("the simulation must be created: {error}"),
        }
    }

    /// A created simulation starts at cursor zero and is readable back.
    #[tokio::test]
    async fn test_create_then_get_returns_the_simulation() {
        let manager = manager();
        let created = created(&manager, 5).await;

        match manager.get(created.id).await {
            Ok(loaded) => {
                assert_eq!(loaded, created);
                assert_eq!(loaded.current_step, 0);
                assert_eq!(loaded.state, SessionState::Initialized);
            }
            Err(error) => panic!("the simulation must load: {error}"),
        }
    }

    /// Creation does not build the factor tape: it stays cheap and predictable,
    /// and the first peek pays for it.
    #[tokio::test]
    async fn test_creation_does_not_build_the_tape() {
        let manager = manager();
        let created = created(&manager, 5).await;

        assert_eq!(manager.cached_tapes(), 0);

        match manager.peek(created.id).await {
            Ok(_) => assert_eq!(manager.cached_tapes(), 1),
            Err(error) => panic!("the peek must succeed: {error}"),
        }
    }

    /// The tape cache still honours the configured capacity now that the build
    /// files its own result from inside the blocking task and the cap travels
    /// as a parameter rather than through `&self`.
    #[tokio::test]
    async fn test_the_tape_cache_still_honours_its_capacity() {
        let config = SimulationV2Config {
            max_cached_tapes: 2,
            ..SimulationV2Config::default()
        };
        let manager = SimulationManager::new(Arc::new(InMemorySimulationStore::new()), config);

        for _ in 0..4 {
            let created = created(&manager, 5).await;
            if let Err(error) = manager.peek(created.id).await {
                panic!("the peek must succeed: {error}");
            }
        }

        assert_eq!(
            manager.cached_tapes(),
            2,
            "four tapes were built under a cap of two"
        );
    }

    /// A peek is repeatable and changes nothing.
    #[tokio::test]
    async fn test_a_peek_is_repeatable_and_changes_nothing() {
        let manager = manager();
        let created = created(&manager, 5).await;

        let first = match manager.peek(created.id).await {
            Ok((_, snapshot)) => snapshot,
            Err(error) => panic!("the peek must succeed: {error}"),
        };
        let second = match manager.peek(created.id).await {
            Ok((_, snapshot)) => snapshot,
            Err(error) => panic!("the peek must succeed: {error}"),
        };

        assert_eq!(first, second);
        match manager.get(created.id).await {
            Ok(loaded) => {
                assert_eq!(loaded.current_step, 0, "a peek must not move the cursor");
                assert_eq!(loaded.version, created.version, "a peek must not persist");
                assert_eq!(loaded.state, SessionState::Initialized);
            }
            Err(error) => panic!("the simulation must load: {error}"),
        }
    }

    /// An advance serves the current snapshot and then moves the cursor.
    #[tokio::test]
    async fn test_an_advance_serves_then_advances() {
        let manager = manager();
        let created = created(&manager, 5).await;

        let peeked = match manager.peek(created.id).await {
            Ok((_, snapshot)) => snapshot,
            Err(error) => panic!("the peek must succeed: {error}"),
        };
        let (advanced, served) = match manager.advance(created.id).await {
            Ok(result) => result,
            Err(error) => panic!("the advance must succeed: {error}"),
        };

        assert_eq!(
            served, peeked,
            "the advance must serve the snapshot the peek showed"
        );
        assert_eq!(advanced.current_step, 1);
        assert_eq!(advanced.state, SessionState::InProgress);
    }

    /// Walking a simulation serves indices 0..N-1 and then completes.
    #[tokio::test]
    async fn test_walking_serves_every_index_once_then_completes() {
        let manager = manager();
        let created = created(&manager, 3).await;

        let mut served = Vec::new();
        for _ in 0..3 {
            match manager.advance(created.id).await {
                Ok((_, snapshot)) => served.push(snapshot.step),
                Err(error) => panic!("the advance must succeed: {error}"),
            }
        }
        assert_eq!(served, vec![0, 1, 2]);

        match manager.get(created.id).await {
            Ok(loaded) => assert_eq!(loaded.state, SessionState::Completed),
            Err(error) => panic!("the simulation must load: {error}"),
        }

        // A completed simulation has nothing left to serve, on either path.
        match manager.advance(created.id).await {
            Err(ChainError::SimulatorError(message)) => assert!(message.contains("completed")),
            other => panic!("expected the exhausted path, got {other:?}"),
        }
        match manager.peek(created.id).await {
            Err(ChainError::SimulatorError(message)) => assert!(message.contains("completed")),
            other => panic!("expected the exhausted path, got {other:?}"),
        }
    }

    /// Completing a simulation drops everything cached for it.
    #[tokio::test]
    async fn test_completion_evicts_the_cached_state() {
        let manager = manager();
        let created = created(&manager, 1).await;

        match manager.advance(created.id).await {
            Ok(_) => {}
            Err(error) => panic!("the advance must succeed: {error}"),
        }

        assert_eq!(manager.cached_tapes(), 0);
        assert_eq!(manager.cached_snapshots(), 0);
    }

    /// Two advances that read the same revision produce one winner.
    #[tokio::test]
    async fn test_a_lost_race_is_a_conflict() {
        let store = Arc::new(InMemorySimulationStore::new());
        let manager = SimulationManager::new(store.clone(), SimulationV2Config::default());
        let created = created(&manager, 5).await;

        // Advance once through the manager, then replay an advance built from
        // the pre-advance revision — exactly what a concurrent caller holds.
        match manager.advance(created.id).await {
            Ok(_) => {}
            Err(error) => panic!("the first advance must succeed: {error}"),
        }

        // The mutation a concurrent caller would hold: it read the simulation
        // before the advance, so it carries the pre-advance revision but an
        // otherwise valid post-advance state.
        let mut stale = created.clone();
        stale.current_step = 1;
        stale.state = SessionState::InProgress;
        let expected = match stale.bump_version() {
            Ok(expected) => expected,
            Err(error) => panic!("must bump: {error}"),
        };
        match store.save_cas(stale, expected).await {
            Err(ChainError::Conflict(_)) => {}
            other => panic!("expected Conflict, got {other:?}"),
        }
    }

    /// Deleting removes the simulation and its cached state.
    #[tokio::test]
    async fn test_delete_removes_the_simulation_and_its_caches() {
        let manager = manager();
        let created = created(&manager, 5).await;
        match manager.peek(created.id).await {
            Ok(_) => {}
            Err(error) => panic!("the peek must succeed: {error}"),
        }
        assert_eq!(manager.cached_tapes(), 1);

        match manager.delete(created.id).await {
            Ok(deleted) => assert!(deleted),
            Err(error) => panic!("the delete must succeed: {error}"),
        }

        assert_eq!(manager.cached_tapes(), 0);
        assert_eq!(manager.cached_snapshots(), 0);
        assert!(manager.get(created.id).await.is_err());
    }

    /// Deleting something that is not there is not an error, and still clears
    /// any cache left behind by a store that expired it on its own.
    #[tokio::test]
    async fn test_deleting_a_missing_simulation_is_not_an_error() {
        let manager = manager();

        match manager.delete(Uuid::new_v4()).await {
            Ok(deleted) => assert!(!deleted),
            Err(error) => panic!("a missing delete must not error: {error}"),
        }
    }

    /// An unknown id is not found, on every read path.
    #[tokio::test]
    async fn test_an_unknown_id_is_not_found() {
        let manager = manager();
        let missing = Uuid::new_v4();

        assert!(matches!(
            manager.get(missing).await,
            Err(ChainError::NotFound(_))
        ));
        assert!(matches!(
            manager.peek(missing).await,
            Err(ChainError::NotFound(_))
        ));
        assert!(matches!(
            manager.advance(missing).await,
            Err(ChainError::NotFound(_))
        ));
    }

    /// A second instance serves a step the first one built, without building
    /// it again.
    ///
    /// Two managers over one store and one shared cache is what a replicated
    /// deployment is, with the balancer's choice made explicit. The second
    /// manager's local cache is empty, so a rebuild is the only alternative to
    /// a shared hit — and a rebuild always writes, so the write count is what
    /// distinguishes them.
    #[tokio::test]
    async fn test_a_second_instance_serves_a_tape_it_did_not_build() {
        let store = Arc::new(InMemorySimulationStore::new());
        let shared = Arc::new(SharedTapes::default());
        let first = SimulationManager::new(
            Arc::clone(&store) as Arc<dyn SimulationStore>,
            SimulationV2Config::default(),
        )
        .with_shared_tapes(Arc::clone(&shared) as Arc<dyn SharedTapeCache>);
        let second = SimulationManager::new(
            Arc::clone(&store) as Arc<dyn SimulationStore>,
            SimulationV2Config::default(),
        )
        .with_shared_tapes(Arc::clone(&shared) as Arc<dyn SharedTapeCache>);

        let created = match first.create(parameters(4)).await {
            Ok(created) => created,
            Err(error) => panic!("the simulation must be created: {error}"),
        };

        // Peeked rather than advanced, so both instances describe the SAME
        // step: an advance would move the shared cursor and the second
        // instance would legitimately serve the next one.
        let served_first = match first.peek(created.id).await {
            Ok(snapshot) => snapshot,
            Err(error) => panic!("the first instance must serve: {error}"),
        };
        assert_eq!(shared.writes(), 1, "building a tape must share it");
        assert_eq!(
            shared.keys().len(),
            1,
            "one simulation is one shared entry: {:?}",
            shared.keys()
        );

        let writes_after_build = shared.writes();
        let served_second = match second.peek(created.id).await {
            Ok(snapshot) => snapshot,
            Err(error) => panic!("the second instance must serve: {error}"),
        };

        assert_eq!(
            shared.writes(),
            writes_after_build,
            "the second instance wrote a tape, so it built one rather than reading the shared \
             one"
        );
        assert!(shared.reads() >= 1, "the second instance must have looked");
        assert_eq!(
            second.cached_tapes(),
            1,
            "a shared hit must still populate the local cache, or every step pays the round trip"
        );

        // Step zero's spot is the initial price whatever the walk did, so
        // comparing what both instances served proves nothing about the round
        // trip. The stored tape itself is compared instead, every row of it,
        // against one built directly from the same parameters: that is the
        // same-seed identical-tape contract surviving encode and decode.
        let stored = match shared.entries.lock() {
            Ok(entries) => entries.values().next().cloned(),
            Err(poisoned) => poisoned.into_inner().values().next().cloned(),
        };
        let stored = match stored {
            Some(stored) => stored,
            None => panic!("the tape must be in the shared cache to be compared"),
        };
        let decoded: FactorTape = match serde_json::from_str(&stored) {
            Ok(decoded) => decoded,
            Err(error) => panic!("a shared tape must decode: {error}"),
        };
        let built = match FactorTape::build(&created.parameters, &created.parameters.method) {
            Ok(built) => built,
            Err(error) => panic!("the tape must build: {error}"),
        };
        assert_eq!(
            decoded, built,
            "the tape that came back from the shared cache is not the tape that was built"
        );

        // And it is a tape with something in it: a walk that never moved would
        // make the comparison above vacuous.
        assert!(decoded.len() > 1, "the tape must cover its steps");
        assert!(
            decoded
                .rows()
                .iter()
                .any(|row| row.spot != decoded.rows()[0].spot),
            "the walk must move, or comparing it proves nothing"
        );

        // The step both instances described is the same one, which is what a
        // client sees.
        assert_eq!(served_first.1.step, served_second.1.step);
        assert_eq!(served_first.1.spot, served_second.1.spot);
    }

    /// A deleted simulation takes its shared tape with it.
    ///
    /// Without this the cache keeps a tape nobody can use until its TTL, which
    /// on a busy deployment is most of what it holds.
    #[tokio::test]
    async fn test_deleting_a_simulation_drops_its_shared_tape() {
        let store = Arc::new(InMemorySimulationStore::new());
        let shared = Arc::new(SharedTapes::default());
        let manager = SimulationManager::new(
            Arc::clone(&store) as Arc<dyn SimulationStore>,
            SimulationV2Config::default(),
        )
        .with_shared_tapes(Arc::clone(&shared) as Arc<dyn SharedTapeCache>);

        let created = match manager.create(parameters(3)).await {
            Ok(created) => created,
            Err(error) => panic!("{error}"),
        };
        let _ = manager.peek(created.id).await;
        assert_eq!(shared.keys().len(), 1, "the tape must be shared first");

        match manager.delete(created.id).await {
            Ok(deleted) => assert!(deleted, "the simulation must have been there"),
            Err(error) => panic!("{error}"),
        }

        assert!(
            shared.keys().is_empty(),
            "the deleted simulation left {:?} behind in the shared cache",
            shared.keys()
        );
    }

    /// Only one instance builds a tape, even when both ask at once.
    ///
    /// Two managers over one store and one shared cache, both told to serve
    /// the same simulation at the same moment. The write count is what
    /// distinguishes a build from a read, and exactly one build may happen:
    /// this is the coalescing that exists inside a process, extended across
    /// them.
    #[tokio::test]
    async fn test_only_one_instance_builds_a_tape() {
        let store = Arc::new(InMemorySimulationStore::new());
        let shared = Arc::new(SharedTapes::default());
        let first = SimulationManager::new(
            Arc::clone(&store) as Arc<dyn SimulationStore>,
            SimulationV2Config::default(),
        )
        .with_shared_tapes(Arc::clone(&shared) as Arc<dyn SharedTapeCache>);
        let second = SimulationManager::new(
            Arc::clone(&store) as Arc<dyn SimulationStore>,
            SimulationV2Config::default(),
        )
        .with_shared_tapes(Arc::clone(&shared) as Arc<dyn SharedTapeCache>);

        let created = match first.create(parameters(4)).await {
            Ok(created) => created,
            Err(error) => panic!("the simulation must be created: {error}"),
        };

        // Both peek the same step at the same time, which is the shape a
        // balancer produces when two clients arrive together.
        let (one, two) = tokio::join!(first.peek(created.id), second.peek(created.id));
        let one = match one {
            Ok(snapshot) => snapshot,
            Err(error) => panic!("the first instance must serve: {error}"),
        };
        let two = match two {
            Ok(snapshot) => snapshot,
            Err(error) => panic!("the second instance must serve: {error}"),
        };

        assert_eq!(
            shared.writes(),
            1,
            "two instances wrote {} tapes for one simulation, so both built it",
            shared.writes()
        );
        assert!(
            shared.claims() >= 2,
            "both instances must have asked for the build claim, got {}",
            shared.claims()
        );
        assert_eq!(
            one.1.spot, two.1.spot,
            "the instance that waited must serve what the builder built"
        );
    }

    /// A claim that nobody releases does not wedge the next caller.
    ///
    /// The claim is held and never given back, which is what an instance
    /// killed mid-build leaves behind. The next caller waits, gives up, and
    /// builds rather than hanging.
    #[tokio::test]
    async fn test_an_abandoned_claim_does_not_block_a_build() {
        let store = Arc::new(InMemorySimulationStore::new());
        let shared = Arc::new(SharedTapes::default());
        let manager = SimulationManager::new(
            Arc::clone(&store) as Arc<dyn SimulationStore>,
            SimulationV2Config::default(),
        )
        .with_shared_tapes(Arc::clone(&shared) as Arc<dyn SharedTapeCache>)
        .with_shared_build_wait(Duration::from_millis(200));

        let created = match manager.create(parameters(3)).await {
            Ok(created) => created,
            Err(error) => panic!("{error}"),
        };

        // Somebody else claims the build and dies without publishing.
        let key = tape_key(created.id, &created.parameters);
        assert!(
            matches!(shared.claim_build(&key).await, BuildClaim::Held(_)),
            "the claim must be free first"
        );

        match manager.peek(created.id).await {
            Ok(_) => {}
            Err(error) => panic!("an abandoned claim must not stop a build: {error}"),
        }
    }

    /// A client that goes away mid-build strands nothing.
    ///
    /// The build is on the blocking pool and keeps running whatever the caller
    /// does, so the owner's workflow lives in a task of its own. Were it in the
    /// request's future, a disconnect would leave the local waiters with an
    /// entry in `builds` nobody will ever publish, the deployment-wide claim
    /// held until it expires, and the finished tape unshared.
    #[tokio::test]
    async fn test_an_abandoned_owner_still_publishes_and_releases() {
        let store = Arc::new(InMemorySimulationStore::new());
        let shared = Arc::new(SharedTapes::default());
        let manager = Arc::new(
            SimulationManager::new(
                Arc::clone(&store) as Arc<dyn SimulationStore>,
                SimulationV2Config::default(),
            )
            .with_shared_tapes(Arc::clone(&shared) as Arc<dyn SharedTapeCache>),
        );

        let created = match manager.create(parameters(11)).await {
            Ok(created) => created,
            Err(error) => panic!("the simulation must be created: {error}"),
        };

        // The owner starts the build and its client disconnects.
        let owner = {
            let manager = Arc::clone(&manager);
            let id = created.id;
            tokio::spawn(async move { manager.peek(id).await })
        };
        owner.abort();

        // Whoever is left must still be served.
        match tokio::time::timeout(Duration::from_secs(5), manager.peek(created.id)).await {
            Ok(Ok(_)) => {}
            other => panic!("an abandoned owner must not strand the request after it: {other:?}"),
        }

        // And the deployment-wide claim came back rather than being left to
        // expire, so another instance is not blocked behind a client that left.
        let held = match shared.building.lock() {
            Ok(held) => held.len(),
            Err(poisoned) => poisoned.into_inner().len(),
        };
        assert_eq!(held, 0, "the build claim must have been released");
        assert_eq!(
            shared.keys().len(),
            1,
            "the built tape must have been shared even though its caller left"
        );
    }

    /// A shared cache that cannot be reached costs a rebuild, not a failure.
    #[tokio::test]
    async fn test_an_unreachable_shared_cache_degrades_to_building() {
        let store = Arc::new(InMemorySimulationStore::new());
        let manager = SimulationManager::new(
            Arc::clone(&store) as Arc<dyn SimulationStore>,
            SimulationV2Config::default(),
        )
        .with_shared_tapes(Arc::new(BrokenTapes) as Arc<dyn SharedTapeCache>);

        let created = match manager.create(parameters(3)).await {
            Ok(created) => created,
            Err(error) => panic!("the simulation must be created: {error}"),
        };

        for step in 0..3 {
            match manager.advance(created.id).await {
                Ok(_) => {}
                Err(error) => panic!("step {step} must serve with the cache down: {error}"),
            }
        }
    }

    /// Two simulations, two entries: a shared cache must not let one tape be
    /// served for another.
    #[tokio::test]
    async fn test_the_shared_cache_keys_each_simulation_separately() {
        let store = Arc::new(InMemorySimulationStore::new());
        let shared = Arc::new(SharedTapes::default());
        let manager = SimulationManager::new(
            Arc::clone(&store) as Arc<dyn SimulationStore>,
            SimulationV2Config::default(),
        )
        .with_shared_tapes(Arc::clone(&shared) as Arc<dyn SharedTapeCache>);

        let one = match manager.create(parameters(3)).await {
            Ok(created) => created,
            Err(error) => panic!("{error}"),
        };
        let two = match manager.create(parameters(3)).await {
            Ok(created) => created,
            Err(error) => panic!("{error}"),
        };
        assert_ne!(one.id, two.id);

        let _ = manager.advance(one.id).await;
        let _ = manager.advance(two.id).await;

        assert_eq!(
            shared.keys().len(),
            2,
            "two simulations must occupy two entries, whatever their parameters: {:?}",
            shared.keys()
        );
    }

    /// Cleanup expires idle simulations and evicts what they left cached.
    #[tokio::test]
    async fn test_cleanup_expires_and_evicts() {
        let store = Arc::new(InMemorySimulationStore::with_idle_retention(
            std::time::Duration::from_secs(1),
        ));
        let manager = SimulationManager::new(store, SimulationV2Config::default());
        let created = created(&manager, 5).await;
        match manager.peek(created.id).await {
            Ok(_) => {}
            Err(error) => panic!("the peek must succeed: {error}"),
        }
        assert_eq!(manager.cached_tapes(), 1);

        // Age the stored document past its retention window.
        let mut aged = created.clone();
        aged.updated_at = std::time::SystemTime::now() - std::time::Duration::from_secs(3_600);
        // Save at the same revision it was read at: the point is to age the
        // document, not to change it.
        let expected = aged.version;
        match manager.store.save_cas(aged, expected).await {
            Ok(()) => {}
            Err(error) => panic!("the aged document must save: {error}"),
        }

        match manager.cleanup().await {
            Ok(expired) => assert_eq!(expired, vec![created.id]),
            Err(error) => panic!("the cleanup must succeed: {error}"),
        }
        assert_eq!(manager.cached_tapes(), 0);
        assert_eq!(manager.cached_snapshots(), 0);
    }

    /// A snapshot survives an eviction of its tape, because both rebuild.
    #[tokio::test]
    async fn test_an_evicted_tape_rebuilds_identically() {
        let manager = manager();
        let created = created(&manager, 4).await;

        let before = match manager.peek(created.id).await {
            Ok((_, snapshot)) => snapshot,
            Err(error) => panic!("the peek must succeed: {error}"),
        };

        manager.evict(created.id);
        assert_eq!(manager.cached_tapes(), 0);

        let after = match manager.peek(created.id).await {
            Ok((_, snapshot)) => snapshot,
            Err(error) => panic!("the peek must succeed: {error}"),
        };
        assert_eq!(before, after, "a rebuild must be indistinguishable");
    }
}