[{"_1":2,"_3":-5,"_4":-5},"loaderData",{"_5":6,"_7":8,"_9":10,"_11":12},"actionData","errors","root",{"_179":-5,"_182":183,"_184":185},"routes/shell",{"_41":42,"_178":-5,"_179":-5,"_180":181},"routes/repo",{"_34":-7,"_35":36,"_37":38,"_39":40,"_41":42,"_43":32,"_44":32,"_45":32,"_46":32,"_47":32,"_48":32,"_49":50,"_51":50,"_52":32,"_53":-7,"_54":-7,"_55":56,"_57":58,"_59":60,"_61":-5,"_62":63,"_64":65,"_66":67,"_68":69},"routes/repo.blob",{"_13":14,"_15":16,"_17":18,"_19":20,"_21":22,"_23":24,"_25":-5,"_26":27,"_28":29,"_29":30,"_31":32,"_33":14},"ref","main","path","crates/logic/lib.rs","oid","45b1e57305b119824eaec5c94005e447e8de741f","size",299128,"commit","90b43017241f81189453d326d05948f388b34652","mounts",[],"source","plain",{"_13":14,"_15":16},"kind","text","// Copyright 2026 Deno Land Inc. Apache-2.0 license.\n\n//! Clean-sheet celld decision core.\n//!\n//! [`on_event`] is the only way behavioral state advances. The production\n//! executor and deterministic simulator both feed it events and perform the\n//! returned effects. No adapter may mutate [`State`] directly.\n\nuse std::collections::{BTreeMap, BTreeSet, VecDeque};\n\npub mod alarm;\npub mod cache;\npub mod cell;\npub mod cron;\npub mod dead_node_reconciliation;\npub mod drain;\npub mod durability;\npub mod epoch_gc;\npub mod format;\npub mod gate;\npub mod http;\npub mod isolate;\npub mod kv;\npub mod log_evict;\npub mod log_tier;\nmod output_gate;\npub mod peer;\npub mod pressure;\npub mod queue;\npub mod rebalance;\npub mod restore;\npub mod routing;\npub mod schedule;\npub mod sqlite;\npub mod sweep;\npub mod wake;\n\nmod types;\npub use types::*;\n\n#[derive(Clone, Debug, PartialEq, Eq)]\nenum Activation {\n Claim(Claim),\n Restore(RestoreSpec),\n}\n\n#[derive(Clone, Debug, PartialEq, Eq)]\nenum ColdStart {\n ReadOwner,\n Restore(RestoreSpec),\n}\n\n/// What a node does for a cell that needs a residency slot it does not have.\n///\n/// Every caller of `State::activate_or_wait` states this, because the answer\n/// is not a property of the cell: it is a property of what the node already\n/// knows about the rest of the fleet.\n#[derive(Clone, Copy, Debug, PartialEq, Eq)]\nenum WhenFull {\n /// Cut an idle resident to admit the waiter. This is the answer whenever\n /// the local wait is the node's whole answer: a lease-disabled node has\n /// no fleet to ask, an alarm wake has no fleet transport, and a cell this\n /// node already owns is not placeable elsewhere. It is also the answer\n /// when a capacity lookup answered and named no peer with room, because\n /// a single-node fleet at its cap would otherwise never admit another\n /// cell again.\n ShedToAdmit,\n /// Park the waiter and keep every resident. A capacity lookup that could\n /// not answer says nothing about fleet room, so a node must not read it\n /// as \"the fleet is full\" and cut a live cell for a placement a peer\n /// would have taken. `State::has_capacity` states the rule the hard cap\n /// obeys: a node at its cap refuses and holds what it has. The waiter\n /// still expires through `watch_queued`, and `pump_capacity` admits it\n /// if a resident finishes first.\n HoldResidents,\n}\n\n/// Facts already decided by ownership resolution that select a safe restore\n/// source. The effect adapter must not rediscover or guess these.\n#[derive(Clone, Debug, PartialEq, Eq)]\npub struct RestoreSpec {\n pub epoch: Epoch,\n /// This activation conditionally created epoch one, so no replica can\n /// precede it.\n pub fresh: bool,\n /// Ownership was seized from a different node (or a released record), so\n /// a previous local eviction cache is not authoritative.\n pub took_over: bool,\n /// The node-level lease handoff proved that this exact local epoch remains\n /// authoritative. The adapter must not consult the remote replica.\n pub resume_local: bool,\n /// The owner the takeover displaced — the node named by the record\n /// version the acquire consumed — for the node-log takeover interlock.\n /// `None` when the record was released or absent, which the release\n /// path already proved durable. Carried here rather than memoized in\n /// the executor so an acquire confirmed by reconciliation names its\n /// prior exactly like one confirmed by the CAS response.\n pub prior: Option,\n}\n\n#[derive(Clone, Debug, PartialEq, Eq)]\npub struct Claim {\n pub guard: CasGuard,\n pub epoch: Epoch,\n pub takeover: bool,\n /// The node named by the record version `guard` matches — the prior\n /// owner this claim displaces if it applies. `None` for a released or\n /// absent record.\n pub prior: Option,\n /// Ambiguous acquires already reconciled for this claim.\n ///\n /// An ambiguous compare-and-swap is re-read rather than retried blindly,\n /// which is correct — it may have applied. But the re-read leads to\n /// another acquire, and if that is ambiguous too the cycle repeats. With\n /// no bound a persistently unanswered store turns a request that used to\n /// hang into one that spins, which is worse: it burns a slot and an\n /// object-store budget forever instead of merely waiting.\n pub reconciles: u32,\n}\n\n/// How many times one claim may reconcile an ambiguous acquire before the\n/// request is failed and the caller left to decide. Small: each pass is a full\n/// read plus a write, and a store answering ambiguously three times running is\n/// not about to start answering.\npub const MAX_ACQUIRE_RECONCILES: u32 = 3;\n\n/// The two local upper bounds on reusing an observed remote lease.\n#[derive(Clone, Copy, Debug, PartialEq, Eq)]\npub struct RemoteLeaseDeadline {\n expires_ms: u64,\n expires_mono_ms: u64,\n}\n\nimpl RemoteLeaseDeadline {\n fn observed(expires_ms: u64, issued_mono_ms: u64, received_ms: u64) -> Self {\n Self {\n expires_ms,\n // The read starts before its response samples wall time. Pairing\n // that wall sample with the earlier monotonic instant can only\n // shorten reuse. Using the core's latest monotonic observation\n // could instead extend a delayed snapshot beyond its expiry.\n expires_mono_ms: issued_mono_ms.saturating_add(expires_ms.saturating_sub(received_ms)),\n }\n }\n\n fn live(self, now_ms: u64, now_mono_ms: u64) -> bool {\n now_ms < self.expires_ms && now_mono_ms < self.expires_mono_ms\n }\n}\n\n/// The evidence that permits another request to reuse a remote route.\n#[derive(Clone, Copy, Debug, PartialEq, Eq)]\npub enum RemoteCache {\n /// The node lease observed while resolving this cell's foreign owner.\n OwnerLease(RemoteLeaseDeadline),\n /// An epoch-zero placement hint, not evidence of cell ownership.\n CapacitySample {\n sampled_ms: u64,\n deadline: RemoteLeaseDeadline,\n },\n /// An adoption response carries no lease. Resolve the next request.\n Uncached,\n}\n\nimpl RemoteCache {\n fn reusable(self, now_ms: u64, now_mono_ms: u64) -> bool {\n match self {\n Self::OwnerLease(deadline) | Self::CapacitySample { deadline, .. } => {\n deadline.live(now_ms, now_mono_ms)\n }\n Self::Uncached => false,\n }\n }\n}\n\n#[derive(Clone, Debug)]\nstruct CachedNodeLease {\n record: NodeLeaseRecord,\n deadline: RemoteLeaseDeadline,\n}\n\n/// Phases map onto the Durable Objects lifecycle states where Cloudflare has\n/// a name for them:\n///\n/// - `Resident` is active or idle in memory (states 1-3).\n/// - `Dormant` is out of memory but still owned by this node, so the cell has\n/// not been moved off its host. A dormant cell with surviving hibernatable\n/// sockets is *hibernated* (state 4); see [`Core::is_hibernated`].\n/// - `Inactive` is out of memory and owned by nobody: removed from the host,\n/// needing a cold start (state 5, and the initial state of every cell).\n///\n/// Note that celld and Cloudflare use \"evict\" differently. Cloudflare evicts a\n/// cell *off its host*, which produces `Inactive`. celld evicts a cell *out of\n/// memory*, which produces `Dormant`, and shedding is what then publishes it\n/// unowned.\n#[derive(Clone, Debug, PartialEq, Eq)]\npub enum Phase {\n Inactive,\n /// Cold demand queued behind `max_activations`, before any I/O begins.\n WaitingActivation,\n ReadingOwner {\n op: OpId,\n },\n ReadingNodeLease {\n op: OpId,\n owner: OwnerRecord,\n /// The wall instant when the read effect was issued. A returned\n /// snapshot can prove expiry only relative to its invocation; using\n /// response time lets transport delay age a previously live snapshot\n /// into false takeover evidence after its owner has renewed.\n issued_ms: u64,\n issued_mono_ms: u64,\n },\n /// The takeover interlock (lease-fold): the dead owner's folded log\n /// state was not sealed, so the executor is recovering its sessions\n /// before this cell may claim.\n RecoveringOwnerLog {\n op: OpId,\n owner: OwnerRecord,\n },\n /// A dead-owner log recovery failed; the cell holds its requests and\n /// re-enters resolution when the retry timer fires. It holds no\n /// activation permit while it waits, so the backoff of one dead\n /// owner's cells cannot starve unrelated activations.\n WaitingOwnerLogRecovery {\n generation: u64,\n },\n ReadingCapacity {\n op: OpId,\n claim: Claim,\n issued_mono_ms: u64,\n },\n WaitingCapacity,\n Acquiring {\n op: OpId,\n claim: Claim,\n /// The claim lands `Dormant` (every request on the cell is an\n /// ownership-only handoff), so it holds no residency slot and a\n /// node at its cap can still accept it.\n dormant: bool,\n },\n ReconcilingAcquire {\n op: OpId,\n claim: Claim,\n dormant: bool,\n },\n Restoring {\n op: OpId,\n spec: RestoreSpec,\n },\n Starting {\n op: OpId,\n epoch: Epoch,\n },\n Publishing {\n op: OpId,\n epoch: Epoch,\n },\n /// The runtime remains published and routable while replica durability is\n /// being proved. A failed or ambiguous proof returns to `Resident`.\n EnsuringDurability {\n op: OpId,\n epoch: Epoch,\n },\n Cleaning {\n op: OpId,\n epoch: Epoch,\n cause: StopCause,\n },\n /// Out of memory, still owned here. Hibernated when sockets survived.\n Dormant {\n epoch: Epoch,\n },\n /// The owner record is no longer this node's, and a signed peer request is\n /// waiting for a successor to restore and publish the cell.\n Adopting {\n op: OpId,\n released_epoch: Epoch,\n },\n Resident {\n epoch: Epoch,\n },\n Remote {\n node: NodeId,\n addr: String,\n epoch: Epoch,\n peer_protocol: u16,\n cache: RemoteCache,\n },\n Fenced,\n}\n\n/// One authority observation from one state-machine boundary.\n#[cfg(celld_internal_tests)]\n#[derive(Clone, Debug, PartialEq, Eq)]\npub struct AuthorityObserverSnapshot {\n pub node_authoritative: bool,\n pub fenced: bool,\n pub now_mono_ms: u64,\n pub phase: Option,\n}\n\n/// The independent inputs for one StepSafety census.\n#[cfg(celld_internal_tests)]\n#[derive(Clone, Debug, PartialEq, Eq)]\npub struct StepSafetyObserverSnapshot {\n pub occupied_counter: usize,\n pub capacity_holder_ids: BTreeSet,\n pub activation_permit_ids: BTreeSet,\n pub activation_holder_ids: BTreeSet,\n}\n\n/// The reported name of a phase. Stable across internal renames, because\n/// `/state` and `celld diagnose` publish it.\nfn phase_name(phase: &Phase) -> &'static str {\n match phase {\n Phase::Inactive => \"inactive\",\n Phase::WaitingActivation => \"waiting_activation\",\n Phase::ReadingOwner { .. } => \"reading_owner\",\n Phase::ReadingNodeLease { .. } => \"reading_node_lease\",\n Phase::RecoveringOwnerLog { .. } => \"recovering_owner_log\",\n Phase::WaitingOwnerLogRecovery { .. } => \"waiting_owner_log_recovery\",\n Phase::ReadingCapacity { .. } => \"reading_capacity\",\n Phase::WaitingCapacity => \"waiting_capacity\",\n Phase::Acquiring { .. } => \"acquiring\",\n Phase::ReconcilingAcquire { .. } => \"reconciling_acquire\",\n Phase::Restoring { .. } => \"restoring\",\n Phase::Starting { .. } => \"starting\",\n Phase::Publishing { .. } => \"publishing\",\n Phase::EnsuringDurability { .. } => \"ensuring_durability\",\n Phase::Cleaning {\n cause: StopCause::Swap,\n ..\n } => \"swapping\",\n Phase::Cleaning { .. } => \"cleaning\",\n Phase::Dormant { .. } => \"dormant\",\n Phase::Adopting { .. } => \"adopting\",\n Phase::Resident { .. } => \"resident\",\n Phase::Remote { .. } => \"remote\",\n Phase::Fenced => \"fenced\",\n }\n}\n\n/// What waits behind an output gate.\n///\n/// An output waits to leave on its channel. A fired alarm waits for its own\n/// settlement, because the core orders the consume-side wake-entry delete from\n/// the far side of the gate: deleting the entry while the consuming commit is\n/// still local would lose both the alarm and the record that could revive it.\n///\n/// The two settle differently and that is the whole reason this is an enum: an\n/// output is released to the shell, an alarm is replayed back into the core. A\n/// channel cannot express the difference -- an alarm has no request to route\n/// by and no adapter holding anything for it -- so it is a second kind of\n/// owner rather than a seventh [`Channel`].\n#[derive(Clone, Debug, PartialEq, Eq)]\nenum GateOwner {\n /// An output withheld until every write it can reveal is proven.\n Output(Held),\n /// The firing alarm's op, and the observation `fire_alarm` made. Both are\n /// replayed into `alarm_finished` when the gate settles.\n Alarm {\n alarm: OpId,\n snapshot: wake::AlarmSnapshot,\n covered: bool,\n },\n}\n\n/// One local write held open by the output gate.\n#[derive(Clone, Debug, PartialEq, Eq)]\nstruct Barrier {\n /// What opened it. An output's `channel` is returned verbatim in the\n /// release effect; nothing in the core reads it.\n owner: GateOwner,\n cell: CellId,\n epoch: Epoch,\n position: u64,\n /// Read-only outputs that finished after this write committed but before\n /// its durability proof completed. They reveal the same state, so they\n /// share the write's verdict -- one proof, one verdict, however many\n /// channels are waiting on it.\n followers: Vec,\n}\n\n/// One withheld output: which request raised it, and which way it leaves.\n#[derive(Clone, Copy, Debug, PartialEq, Eq)]\nstruct Held {\n request: RequestId,\n channel: Channel,\n}\n\n/// One reason that a quiescing cell cannot enter its handoff yet.\n///\n/// This is a read-only management projection. The core remains the authority\n/// for the request and WebSocket indexes that produce it.\n#[derive(Clone, Debug, PartialEq, Eq)]\npub struct DrainPin {\n pub cell: CellId,\n pub request: Option,\n pub active_request_count: usize,\n pub output_gate_count: usize,\n pub regular_websocket_count: usize,\n pub outbound_websocket_count: usize,\n}\n\n#[derive(Clone, Debug)]\nstruct Cell {\n phase: Phase,\n /// The highest committed-write position a verified proof has covered in\n /// this residency, and the epoch it belongs to. A read-only output that\n /// observed no more than this reveals nothing a crash can still lose, so\n /// it needs no proof of its own. `set_phase` clears it when a residency\n /// begins, because positions belong to one runtime's connection, and a\n /// proof at another epoch covers nothing here.\n proven_position: Option<(Epoch, u64)>,\n /// Failed dead-owner log recovery cycles in the current episode.\n /// Reset when a recovery succeeds or the episode fails its cap, so\n /// a later episode starts with the full budget.\n owner_log_recovery_attempts: u32,\n /// The shutdown drain has reserved this resident cell for its next\n /// handoff batch. New requests wait for the successor route while work\n /// which was already admitted finishes on this runtime.\n quiescing: bool,\n requests: BTreeSet,\n websockets: BTreeMap,\n waiting_for: Option,\n waiting_activation: Option,\n /// The in-flight release of this cell's ownership record, if any.\n releasing: Option,\n /// Whether the eviction now under way hands this cell to the fleet.\n evict_rebalance: bool,\n /// The epoch this node holds the record at, kept across the transient\n /// wake phases. A dormant cell that waits for an activation permit still\n /// routes every request here, and a fleet-wide wake wave otherwise hides\n /// hundreds of owned cells per node from the published count for a\n /// minute; the 2026-09-03 fleet moved cells on that dip.\n held_epoch: Option,\n /// A balancing move is in progress on a node that is not draining. It\n /// carries the drain's handoff semantics for this one dormant cell: a\n /// request waits for the successor, and the release continues into\n /// adoption.\n handoff: bool,\n alarm: Option,\n /// Source order survives queued notifications and same-time re-arms.\n /// It also suppresses repeated retirement work for unrelated cell writes.\n alarm_source: Option,\n alarm_wake: bool,\n /// Startup, rather than a request or alarm, demands this activation.\n resume_demand: bool,\n /// A restarted process is reading this cell's record to confirm what it\n /// owns. Like startup demand it keeps the read queued, but it wakes\n /// nothing: the read settles the cell to dormant or inactive.\n confirming: bool,\n /// When this cell's eviction was last refused, if it has been.\n ///\n /// A cell that cannot prove its replica durable goes back to residency,\n /// and being cold is exactly why nobody registered it -- so it settles at\n /// the head of the eviction order and is chosen again on the next pass,\n /// and the one after. Enough of those and the node has no shed candidate\n /// it will ever succeed with. Recording the refusal lets the order prefer\n /// cells that have not just failed, while still coming back to them when\n /// they are all that is left.\n eviction_refused_mono_ms: Option,\n /// The node-wide identity of the V8 heap holding this cell's realm.\n /// Eviction order reads it: taking a heap's last cell returns the heap,\n /// while taking one of thirty-two returns only a cell record.\n isolate: Option,\n /// The application generation whose isolate holds this cell's realm.\n /// Zero until the shell reports a start, and the swap pump reads zero as\n /// stale: a cell it cannot place is moved rather than trusted.\n generation: u64,\n /// This cell is moving to the current generation: quiescing for the\n /// swap, stopped, or starting again. Cleared when it publishes, and by\n /// any phase change that leaves the swap path.\n swapping: bool,\n /// A forced swap already cancelled this cell's activity. It asks once.\n swap_cancelled: bool,\n /// The queue deadline now watching this cell, if it is parked. A timer\n /// carrying any other generation has been outlived and expires nothing.\n queued_generation: Option,\n /// When this cell last did something, on the remembered monotonic clock.\n ///\n /// Eviction order is the whole reason this exists. Without it the shed\n /// candidate is whichever cell sorts first by id, so a node under\n /// sustained pressure evicts its alphabetically-first cell over and over\n /// -- restoring and shedding the busiest cell on the node while an idle\n /// one further down the alphabet is never touched.\n last_used_mono_ms: u64,\n /// The newest local request routed to this cell.\n ///\n /// A load sample updates the monotonic clock less often than requests\n /// arrive, so many busy cells can share one `last_used_mono_ms` and fall\n /// back to lexical drain order. Request IDs are monotonically increasing\n /// core inputs, so this field preserves their exact replayable order.\n last_used_request: RequestId,\n /// How many local requests this process routed to the cell.\n ///\n /// A bounded handoff protects a shared singleton before a newer one-shot\n /// cell. The count is process-local because the drain orders only this\n /// process's resident set.\n local_request_count: u64,\n}\n\n#[derive(Clone, Debug, PartialEq, Eq)]\nenum AlarmState {\n Armed {\n at_ms: i64,\n generation: u64,\n covered: bool,\n },\n Firing {\n op: OpId,\n at_ms: i64,\n generation: u64,\n covered: bool,\n /// The newest runtime observation that arrived before the firing's\n /// durability result. It cannot settle the alarm by itself.\n observed: Option<(wake::AlarmSnapshot, bool)>,\n },\n}\n\n#[derive(Clone, Copy, Debug, PartialEq, Eq)]\nstruct AlarmObservation {\n alarm: wake::AlarmSnapshot,\n covered: bool,\n firing_settled: bool,\n}\n\n#[derive(Clone, Debug)]\nstruct HeldNodeLease {\n spec: NodeLeaseSpec,\n record: NodeLeaseRecord,\n last_ok_mono_ms: u64,\n last_attempt_mono_ms: u64,\n timer_generation: u64,\n}\n\n#[derive(Clone, Debug)]\nstruct PendingNodeLease {\n spec: NodeLeaseSpec,\n desired: NodeLeaseRecord,\n prior: Option,\n /// This read follows a write whose result was ambiguous. It may prove the\n /// desired record landed, but must not turn a failed create into an\n /// unbounded read/write loop.\n readback_only: bool,\n /// The monotonic instant `desired.expires_ms` was computed from.\n ///\n /// Authority is bounded by what the *bucket* says, and the bucket says\n /// `expires_ms`, which is sampled here rather than when the write lands.\n /// Anchoring the local fence deadline to the write's completion instead\n /// would let this node serve for the whole duration of the round trip\n /// after every peer is entitled to declare it dead -- the store is under\n /// no obligation to answer quickly, so that window is unbounded.\n anchor_mono_ms: u64,\n /// This initial CAS replaces the exact live generation certified by the\n /// clean local shutdown. Renewals never set this bit.\n resume_local: bool,\n}\n\n#[derive(Clone, Debug)]\nenum NodeAuthority {\n Unstarted,\n Reading {\n op: OpId,\n pending: PendingNodeLease,\n },\n Writing {\n op: OpId,\n pending: PendingNodeLease,\n },\n Held(HeldNodeLease),\n /// A continuous node whose initial acquisition did not succeed, waiting to\n /// try again.\n ///\n /// `StartNodeLease` is emitted exactly once, at process start, so without\n /// this the first failed acquisition is permanent: the process stays up,\n /// holds no authority, and answers every request `NodeUnavailable` for as\n /// long as it runs. One slow or rate-limited bucket response at startup\n /// should not cost a node its whole lifetime.\n Retrying {\n spec: NodeLeaseSpec,\n generation: u64,\n },\n Failed,\n Fenced,\n}\n\nimpl Default for Cell {\n fn default() -> Self {\n Self {\n phase: Phase::Inactive,\n proven_position: None,\n owner_log_recovery_attempts: 0,\n quiescing: false,\n requests: BTreeSet::new(),\n websockets: BTreeMap::new(),\n waiting_for: None,\n waiting_activation: None,\n releasing: None,\n evict_rebalance: false,\n held_epoch: None,\n handoff: false,\n alarm: None,\n alarm_source: None,\n alarm_wake: false,\n resume_demand: false,\n confirming: false,\n isolate: None,\n generation: 0,\n swapping: false,\n swap_cancelled: false,\n queued_generation: None,\n eviction_refused_mono_ms: None,\n last_used_mono_ms: 0,\n last_used_request: 0,\n local_request_count: 0,\n }\n }\n}\n\n/// The last cut the walk down took, and what the node looked like when it did.\n///\n/// The measurement travels with the value because the two are not comparable.\n/// When the latches change, so does the number the walk down reads, and\n/// comparing a fresh resident set size against a stale in-use figure finds them\n/// flat by coincidence.\n#[derive(Clone, Copy, Debug)]\nstruct ShedCut {\n /// The measurement `bytes` was read from.\n metric: pressure::Metric,\n /// What that measurement said at the cut.\n bytes: u64,\n /// The residency the cut was taken at.\n cells: usize,\n}\n\n/// All authoritative coordination state for the first vertical slice.\npub struct State {\n node: NodeId,\n config: Config,\n fenced: bool,\n /// A shutdown handoff is under way: every resident cell is to be\n /// released, pumped at most `max_releases` at a time. Sticky -- a\n /// draining node never goes back to serving.\n draining: bool,\n /// Number of cell-table entries inspected by the shutdown release pump.\n /// Instrumentation uses this to enforce a structural work bound instead\n /// of timing a host whose speed is unrelated to the decision.\n #[cfg(celld_internal_tests)]\n drain_table_visits: u64,\n /// Cells reserved at the shutdown or generation-swap quiesce boundary.\n /// The set is bounded by `max_releases`, so an actor event can promote an\n /// idle reservation without walking the complete resident table.\n quiescing_cells: BTreeSet,\n /// Cells whose owner-record release is in flight. This index makes both\n /// drain pacing and the 50 ms shell status poll independent of fleet size.\n releasing_cells: BTreeSet,\n /// Released cells waiting for a successor acknowledgement. Like the other\n /// handoff indexes, this set is the exact bounded work queue.\n adopting_cells: BTreeSet,\n /// An exact clean predecessor was replaced and its local inventory is\n /// being scanned or materialized. Readiness remains false until every\n /// discovered cell either publishes or fails closed.\n resuming: bool,\n resuming_cells: BTreeSet,\n /// A restarted process is reading the ownership record of every cell it\n /// finds on its own disk, to count what it still owns without waking\n /// any of it. Until this finishes the published owned count is absent,\n /// so no peer balances against a node that only looks empty.\n confirming: bool,\n confirming_cells: BTreeSet,\n next_op: OpId,\n next_timer_generation: u64,\n cells: BTreeMap,\n request_cells: BTreeMap,\n /// Local routes handed to the executor but not yet reported complete.\n /// These are eviction pins: the adapter may still be running user code.\n active_requests: BTreeMap,\n /// Reverse index of `active_requests`: how many active requests target\n /// each cell. `is_active` reads it in O(1) where scanning the values was\n /// O(active_requests), and `worker_request` calls `is_active` once per\n /// resident cell — the product wedged cold-cell activation, the rate\n /// falling with the cell count (engine/pathological-load.md).\n active_cells: BTreeMap,\n /// The active requests that `gate_request` admitted past a quiescing\n /// cell, and how many each cell holds. A gate cannot admit the next gate:\n /// counted as the work that raises gates, a background chain of\n /// overlapping gates admitted its own successor after every handler had\n /// ended, so the cell never reached its swap safe point.\n admitted_gates: BTreeSet,\n admitted_gate_cells: BTreeMap,\n /// Local write responses withheld by the output gate until their cell is\n /// proven durable to the written position, keyed by the durability op. An\n /// open gate makes its cell active, so the cell cannot be evicted\n /// underneath it; a fence drains these to a failed response so a write is\n /// never acknowledged after the node loses authority.\n barriers: BTreeMap,\n /// Requests whose activity ended while a write of theirs was still on the\n /// output gate. The pin outlives the activity: it keeps the cell from being\n /// evicted under the gate, and keeps a later write of the same request able\n /// to open one. See `activity_finished`.\n gate_pinned: BTreeSet,\n /// Last resident selected for top-level Worker execution. Cell IDs are\n /// ordered, so this is enough to replay a fair round-robin without shell\n /// atomics or registry iteration order leaking into behavior.\n worker_cursor: Option,\n activity: ActivitySnapshot,\n activation_waiters: VecDeque,\n /// Cells holding a complete cold-route admission. Keeping this explicit,\n /// instead of inferring it from executor tasks, makes the concurrency bound\n /// part of the replayable state machine.\n activation_permits: BTreeSet,\n /// Cells with an eviction in flight -- from the durability proof through\n /// the runtime stop. Explicit for the same reason as the activation\n /// permits: the bound belongs in the replayable state machine rather than\n /// being implied by how many executor tasks happen to exist.\n eviction_permits: BTreeSet,\n /// Bind explicit callers to their operation in the same place that moves\n /// a proof to its stop. The executor must not reconstruct that binding\n /// from unrelated effects in the same batch. Policy evictions have none.\n eviction_notifications: BTreeMap>,\n /// Cells waiting for a residency slot, in arrival order. FIFO is the\n /// whole admission policy: waking every waiter on a release and letting\n /// them race is unfair by construction — under sustained eviction a\n /// waiter can time out while thousands of slots are freed around it.\n /// The queue converts \"eventually, probably\" into a bound: a waiter with\n /// `k` waiters ahead of it is admitted within `k` releases, so no\n /// arrival pattern can starve it.\n capacity_waiters: VecDeque,\n /// Requests received as epoch-zero capacity handoffs. Keeping the mode on\n /// the request rather than in the HTTP adapter makes the admit/refuse race\n /// atomic with every other lifecycle transition.\n capacity_requests: BTreeSet,\n /// Signed shutdown handoffs that need ownership but not a warm runtime.\n /// A real request joining the same acquisition still causes the normal\n /// restore, so demand always outranks the lazy handoff optimization.\n handoff_requests: BTreeSet,\n /// Reservations made against each unchanged advisory sample. Concurrent\n /// lookups are applied one event at a time, so projected load cannot pick\n /// the same advertised final slot twice by accident.\n capacity_reservations: BTreeMap,\n capacity_samples: BTreeMap,\n /// A refusal disproves exactly one load sample. The node becomes eligible\n /// again only after its lease advertises a newer sample.\n capacity_rejections: BTreeMap,\n /// Live routing authority is shared by every cell owned by the same node.\n /// Keeping that cache here makes expiry and invalidation deterministic,\n /// rather than an invisible executor optimization.\n node_lease_cache: BTreeMap,\n /// Start/publish effects invalidated by fencing may still commit. Their\n /// late completion must trigger compensating cleanup, not be ignored.\n retired_runtime_ops: BTreeMap,\n /// Runtime starts which exceeded their request deadline but still own a\n /// live executor task. A start cannot be cancelled safely, so the core\n /// keeps its phase and operation until the real completion arrives. New\n /// callers fail fast instead of starting a duplicate runtime for the same\n /// cell and epoch.\n timed_out_runtime_starts: BTreeSet,\n /// The cell each in-flight cell-scoped operation belongs to. Completion\n /// events name only the op; this index is what makes resolving one a\n /// lookup instead of a walk of every cell.\n cell_ops: BTreeMap,\n /// How many cells currently occupy capacity, maintained at every phase\n /// transition. `has_capacity` asks on every admission and `validate` on\n /// every debug event; counting by walking the map priced both by the\n /// total cell count rather than the answer.\n occupied: usize,\n node_authority: NodeAuthority,\n /// A NudgeNodeLease arrived while a lease write was already in flight.\n /// That write's body was serialized BEFORE the\n /// nudge's publish, so it cannot carry it. Drain the flag by renewing\n /// again the moment the in-flight write settles into Held.\n nudge_pending: bool,\n /// The most recent wall-clock instant any event carried. Alarms are wall\n /// clock, so judging whether one is imminent needs this rather than the\n /// monotonic reading next to it.\n now_ms: u64,\n /// The most recent monotonic instant any event carried. Not a clock read:\n /// the core never asks what time it is, it remembers what it was told, so\n /// a handler that was not handed a timestamp can still arm a deadline\n /// relative to the event being processed.\n now_mono_ms: u64,\n /// How far down the current sample asked the node to shed.\n ///\n /// The walk down runs at eviction speed rather than sample speed: each\n /// completed eviction starts the next while residency is still above this\n /// floor. Without one it is a cell per sample, so a node fifteen over its\n /// watermark refuses work for fifteen sampling periods. Recomputed every\n /// sample, so a resource trigger comes down by a proportion of what was\n /// last measured rather than aiming at a cell count that means nothing to\n /// it.\n shed_floor: usize,\n /// The cut the current shed floor was set from. A later latched sample\n /// compares against this: a completed cut that left the number flat makes\n /// another cut futile.\n shed_cut: Option,\n /// Which ceilings are latched. Held apart from `shed_reason`, which names\n /// only the more serious crossing.\n latches: pressure::Latches,\n /// The application generation the node serves, as last announced by\n /// `Event::GenerationChanged`. Zero before any announcement, when no cell\n /// is stale.\n current_generation: u64,\n /// When the current generation was announced, on the monotonic clock.\n /// The maximum age counts from here.\n generation_changed_mono_ms: u64,\n /// How long a resident cell may stay on an older generation before its\n /// swap is forced. Zero forces every stale cell at the announcement.\n swap_max_age_ms: u64,\n /// The classes whose cells are forced without waiting for the maximum\n /// age, from the last announcement.\n ///\n /// Held against the announcement rather than marked on each cell,\n /// because only a resident cell is ever nominated: a mark left a dormant\n /// reserved cell -- which is what a reserved cell is whenever it is idle\n /// -- carrying a forced flag that nothing could act on, and the core\n /// refuses that pair. Eagerness is a property of the class and the\n /// adoption, so the pump reads it where it decides, and no cell can hold\n /// a swap flag it is not swapping under.\n swap_eager_classes: BTreeSet,\n /// The shedding latch. celld kept this in the executor, which meant the\n /// hysteresis -- the part with actual behaviour -- was the one piece the\n /// simulation could not reach. It is carried here so a sample\n /// sequence is replayable.\n shedding: bool,\n /// Whether the last load sample is below every configured low watermark.\n /// This is stricter than `!shedding` and reserves room for a fleet handoff.\n memory_headroom: bool,\n /// Which resource is holding the latch, for the effect the shell logs.\n shed_reason: Option<&'static str>,\n}\n\nimpl State {\n pub fn new(node: impl Into, config: Config) -> Self {\n assert!(config.max_evictions > 0, \"max_evictions must be positive\");\n assert!(config.max_releases > 0, \"max_releases must be positive\");\n assert!(\n config.max_activations > 0,\n \"max_activations must be positive\"\n );\n Self {\n node: node.into(),\n config,\n fenced: false,\n draining: false,\n #[cfg(celld_internal_tests)]\n drain_table_visits: 0,\n quiescing_cells: BTreeSet::new(),\n releasing_cells: BTreeSet::new(),\n adopting_cells: BTreeSet::new(),\n current_generation: 0,\n generation_changed_mono_ms: 0,\n swap_max_age_ms: 0,\n swap_eager_classes: BTreeSet::new(),\n resuming: false,\n resuming_cells: BTreeSet::new(),\n confirming: false,\n confirming_cells: BTreeSet::new(),\n next_op: 1,\n next_timer_generation: 1,\n cells: BTreeMap::new(),\n request_cells: BTreeMap::new(),\n active_requests: BTreeMap::new(),\n active_cells: BTreeMap::new(),\n admitted_gates: BTreeSet::new(),\n admitted_gate_cells: BTreeMap::new(),\n barriers: BTreeMap::new(),\n gate_pinned: BTreeSet::new(),\n worker_cursor: None,\n activity: ActivitySnapshot::default(),\n activation_waiters: VecDeque::new(),\n activation_permits: BTreeSet::new(),\n eviction_permits: BTreeSet::new(),\n eviction_notifications: BTreeMap::new(),\n capacity_waiters: VecDeque::new(),\n capacity_requests: BTreeSet::new(),\n handoff_requests: BTreeSet::new(),\n capacity_reservations: BTreeMap::new(),\n capacity_samples: BTreeMap::new(),\n capacity_rejections: BTreeMap::new(),\n node_lease_cache: BTreeMap::new(),\n retired_runtime_ops: BTreeMap::new(),\n timed_out_runtime_starts: BTreeSet::new(),\n cell_ops: BTreeMap::new(),\n occupied: 0,\n node_authority: NodeAuthority::Unstarted,\n nudge_pending: false,\n now_ms: 0,\n now_mono_ms: 0,\n shed_floor: 0,\n shed_cut: None,\n latches: pressure::Latches::default(),\n shedding: false,\n memory_headroom: false,\n shed_reason: None,\n }\n }\n\n pub fn node(&self) -> &str {\n &self.node\n }\n\n pub fn is_fenced(&self) -> bool {\n self.fenced\n }\n\n /// Authority is the lease's published validity evaluated at ask time on\n /// the monotonic clock — not the `Held` variant. A peer treats the record\n /// as dead the instant `now_ms` reaches `expires_ms`, so a holder whose\n /// lease lapsed must answer `false` here even if `Timer::NodeLeaseFence`\n /// has not fired yet: the timer is the liveness backstop that halts the\n /// process, never the safety mechanism. A suspended VM's timer fires\n /// late, but a request evaluated after resume still refuses. The `<` /\n /// `>=` polarity matches the fence timer at `handle_timer`, so the\n /// predicate and the timer cannot disagree.\n pub fn node_authoritative(&self) -> bool {\n if !self.config.require_node_lease {\n return !self.fenced;\n }\n let fresh = |held: &HeldNodeLease| {\n self.now_mono_ms.saturating_sub(held.last_ok_mono_ms) < held.spec.ttl_ms\n };\n match &self.node_authority {\n NodeAuthority::Held(held) => fresh(held),\n NodeAuthority::Reading { pending, .. } | NodeAuthority::Writing { pending, .. } => {\n pending.prior.as_ref().is_some_and(fresh)\n }\n _ => false,\n }\n }\n\n /// Whether the owned-cell count describes every record this node holds.\n /// False while a restarted process confirms its local inventory.\n pub fn ownership_confirmed(&self) -> bool {\n !self.confirming\n }\n\n /// Whether this node is handing its cells away. Published in the lease,\n /// where it keeps a draining node out of the balancer's receivers.\n pub fn draining(&self) -> bool {\n self.draining\n }\n\n /// Whether the process can accept new traffic.\n pub fn ready_to_serve(&self) -> bool {\n if self.fenced || self.resuming {\n return false;\n }\n self.node_authoritative()\n }\n\n pub fn phase(&self, cell: &str) -> Option<&Phase> {\n self.cells.get(cell).map(|cell| &cell.phase)\n }\n\n /// Return every cell phase for an opt-in transition observer.\n ///\n /// An observer joins these snapshots by cell before it discards each cell\n /// identifier. A node-wide census cannot preserve that association.\n #[cfg(celld_internal_tests)]\n pub fn phase_snapshot_for_test(&self) -> BTreeMap {\n self.cells\n .iter()\n .map(|(id, cell)| (id.clone(), cell.phase.clone()))\n .collect()\n }\n\n /// Return the authority inputs from one observation boundary.\n #[cfg(celld_internal_tests)]\n pub fn authority_observer_snapshot(&self, cell: &str) -> AuthorityObserverSnapshot {\n AuthorityObserverSnapshot {\n node_authoritative: self.node_authoritative(),\n fenced: self.fenced,\n now_mono_ms: self.now_mono_ms,\n phase: self.phase(cell).cloned(),\n }\n }\n\n /// Return both sides of each structural count from the decision core.\n #[cfg(celld_internal_tests)]\n pub fn step_safety_observer_snapshot(&self) -> StepSafetyObserverSnapshot {\n StepSafetyObserverSnapshot {\n occupied_counter: self.occupied,\n capacity_holder_ids: self\n .cells\n .iter()\n .filter(|(_, cell)| phase_occupies_capacity(&cell.phase))\n .map(|(id, _)| id.clone())\n .collect(),\n activation_permit_ids: self.activation_permits.clone(),\n activation_holder_ids: self\n .cells\n .iter()\n .filter(|(_, cell)| phase_holds_activation(&cell.phase))\n .map(|(id, _)| id.clone())\n .collect(),\n }\n }\n\n /// Which resource is currently holding the shedding latch, if any.\n ///\n /// A node that is refusing work should be able to say why; without this\n /// the operator sees only that admissions stopped.\n pub fn shed_reason(&self) -> Option<&'static str> {\n self.shed_reason\n }\n\n /// Whether this node is currently shedding, for peers ranking it as a\n /// placement target. A node that says no while walking down invites the\n /// work it is trying to get rid of.\n pub fn shedding(&self) -> bool {\n self.shedding\n }\n\n /// Whether the last load sample reserves memory below every pressure\n /// resume line. False until the first `LoadSampled` event. The actor must\n /// fold a sample before it publishes a lease, or the lease reports no\n /// headroom.\n pub fn memory_headroom(&self) -> bool {\n self.memory_headroom\n }\n\n /// Host-side sockets open across every cell. Each one pins its cell\n /// against eviction, so a node holding many is a poor landing place\n /// even when its residency looks unremarkable.\n pub fn host_websockets(&self) -> usize {\n self.cells.values().map(|cell| cell.websockets.len()).sum()\n }\n\n /// Does this cell still hold that socket? False when the core declined\n /// it, so the shell can answer the opener instead of closing underneath.\n pub fn holds_websocket(&self, id: &str, websocket: WebSocketId) -> bool {\n self.cells\n .get(id)\n .is_some_and(|cell| cell.websockets.contains_key(&websocket))\n }\n\n pub fn occupied(&self) -> usize {\n self.occupied\n }\n\n /// Cells whose ownership acquisition this state machine has confirmed.\n ///\n /// This includes dormant cells, which consume no residency capacity but\n /// still route every later request to this node, and a confirmed cell\n /// that is waking through the transient phases. A first acquisition is\n /// excluded until reconciliation proves that its CAS landed.\n pub fn owned_cells(&self) -> usize {\n self.cells\n .values()\n .filter(|cell| cell.held_epoch.is_some())\n .count()\n }\n\n /// Are fewer than `ceiling` cells occupying capacity?\n fn occupied_below(&self, ceiling: usize) -> bool {\n self.occupied < ceiling\n }\n\n pub fn residents(&self) -> Vec {\n self.cells\n .iter()\n .filter(|(_, cell)| {\n matches!(\n cell.phase,\n Phase::Resident { .. } | Phase::EnsuringDurability { .. }\n )\n })\n .map(|(id, _)| id.clone())\n .collect()\n }\n\n /// Resident cells which have stopped admitting new local requests so the\n /// shutdown drain can reach a final durability boundary.\n pub fn quiescing(&self) -> usize {\n self.quiescing_cells.len()\n }\n\n /// Return the release pump's exact cell-table work for a bounded\n /// work assertion. This is not elapsed time, so the result is stable on\n /// every host and under every scheduler.\n #[cfg(celld_internal_tests)]\n pub fn drain_table_visits_for_test(&self) -> u64 {\n self.drain_table_visits\n }\n\n #[cfg(celld_internal_tests)]\n fn record_drain_table_visits(&mut self, visits: usize) {\n self.drain_table_visits = self.drain_table_visits.saturating_add(visits as u64);\n }\n\n /// Return every pin on every resident reserved for a shutdown handoff.\n /// A cell without an active request still gets one row, so a regular or\n /// outbound WebSocket cannot leave an otherwise empty stall invisible.\n pub fn drain_pins(&self) -> Vec {\n let mut pins = Vec::new();\n for (cell, state) in self.cells.iter().filter(|(_, cell)| cell.quiescing) {\n let active: Vec = self\n .active_requests\n .iter()\n .filter_map(|(request, request_cell)| (request_cell == cell).then_some(*request))\n .collect();\n let regular_websocket_count = state\n .websockets\n .values()\n .filter(|kind| matches!(kind, WebSocketKind::Regular))\n .count();\n let outbound_websocket_count = state\n .websockets\n .values()\n .filter(|kind| matches!(kind, WebSocketKind::Outbound))\n .count();\n if active.is_empty() {\n pins.push(DrainPin {\n cell: cell.clone(),\n request: None,\n active_request_count: 0,\n output_gate_count: 0,\n regular_websocket_count,\n outbound_websocket_count,\n });\n continue;\n }\n let active_request_count = active.len();\n for request in active {\n pins.push(DrainPin {\n cell: cell.clone(),\n request: Some(request),\n active_request_count,\n output_gate_count: self\n .barriers\n .values()\n .filter(|barrier| {\n matches!(&barrier.owner, GateOwner::Output(held) if held.request == request)\n || barrier.followers.iter().any(|held| held.request == request)\n })\n .count(),\n regular_websocket_count,\n outbound_websocket_count,\n });\n }\n }\n pins\n }\n\n /// Return the exact management projection for the current event boundary.\n /// Activations are not visible until publication; a runtime being held for\n /// durability remains visible because it is still published and routable.\n pub fn presence_snapshot(&self) -> PresenceSnapshot {\n let cells = self\n .cells\n .iter()\n .filter_map(|(id, cell)| match cell.phase {\n Phase::Resident { epoch } | Phase::EnsuringDurability { epoch, .. } => {\n Some(PresenceCell {\n id: id.clone(),\n epoch,\n })\n }\n _ => None,\n })\n .collect();\n PresenceSnapshot {\n serving: self.ready_to_serve(),\n cells,\n activity: self.activity,\n }\n }\n\n /// The isolate each resident cell sits in. A node holds one V8 heap per\n /// distinct value here, so the spread of this is the memory a walk down\n /// can still return.\n pub fn resident_isolates(&self) -> Vec {\n self.cells\n .values()\n .filter(|cell| matches!(cell.phase, Phase::Resident { .. }))\n .filter_map(|cell| cell.isolate)\n .collect()\n }\n\n pub fn waiting(&self) -> Vec {\n self.capacity_waiters.iter().cloned().collect()\n }\n\n pub fn activation_waiting(&self) -> Vec {\n self.activation_waiters.iter().cloned().collect()\n }\n\n pub fn activating(&self) -> usize {\n self.activation_permits.len()\n }\n\n /// Evictions in flight, from the durability proof through the runtime\n /// stop. During a drain this saturates at `max_releases`.\n pub fn evicting(&self) -> usize {\n self.eviction_permits.len()\n }\n\n /// How many cells sit in each phase, by a stable short name, omitting the\n /// phases with no cells.\n ///\n /// A node that refuses cells is diagnosed by where its cells are, and\n /// `occupied` cannot say: it counts residency, so a node holding thousands\n /// of cells part-way through a cold start reports almost none. A fleet held\n /// that state for fifteen minutes and the record could not say what the\n /// cells were doing.\n ///\n /// The names are part of the operator interface. A chart and a human read\n /// them, so they do not change with an internal rename.\n pub fn phase_census(&self) -> Vec<(&'static str, usize)> {\n let mut counts: BTreeMap<&'static str, usize> = BTreeMap::new();\n for cell in self.cells.values() {\n *counts.entry(phase_name(&cell.phase)).or_default() += 1;\n }\n counts.into_iter().collect()\n }\n\n /// Cold routes that have not finished: cells that hold an activation\n /// permit plus cells queued behind the activation ceiling. A capacity\n /// waiter already holds a permit, so this counts every cell once. This is\n /// capacity telemetry; a graceful handoff is paced by successor ownership\n /// acknowledgements instead.\n pub fn activation_backlog(&self) -> usize {\n self.activation_permits.len() + self.activation_waiters.len()\n }\n\n /// Ownership releases still in flight. The shutdown drain waits for\n /// zero: a released cell stops occupying capacity before its record\n /// write commits, so an exit gated on occupancy alone can outrun its\n /// own releases and leave records a successor must wait out the node\n /// lease for -- the takeover-at-once promise broken on every real\n /// store, where the write loses a race against a 50ms drain poll that\n /// a local fixture always wins.\n pub fn releasing(&self) -> usize {\n self.releasing_cells.len()\n }\n\n /// Released cells still waiting for a successor ownership acknowledgement.\n ///\n /// This count holds the shutdown handoff permit after the owner write has\n /// completed. Without it the release pump can blank the complete resident\n /// set before any successor reserves restoration capacity.\n pub fn adopting(&self) -> usize {\n self.adopting_cells.len()\n }\n\n /// Cell movements past nomination and before a successor acknowledgement.\n pub fn handoffs_in_flight(&self) -> usize {\n self.quiescing()\n .saturating_add(self.eviction_permits.len())\n .saturating_add(self.releasing())\n .saturating_add(self.adopting())\n }\n\n /// Completed successor ownership acknowledgements during this process lifetime.\n /// Shutdown uses this monotonic value to distinguish a slow, advancing\n /// handoff from one that has stopped making progress.\n pub fn handed_off(&self) -> u64 {\n self.activity.handed_off\n }\n\n /// Return an epoch this node has acquired for a dormant, warming, or\n /// published cell. A shutdown handoff acknowledges this authority without\n /// forcing the cell runtime into memory.\n pub fn accepted_epoch(&self, id: &str) -> Option {\n self.cells\n .get(id)\n .and_then(|cell| phase_owned_epoch(&cell.phase))\n }\n\n /// Cheap internal consistency gate run by both executors after every event.\n /// Check one withheld output: pinned on the barrier's cell, and the only\n /// output this request holds on that channel.\n fn validate_held(\n held: &Held,\n op: &OpId,\n barrier: &Barrier,\n active_requests: &BTreeMap,\n seen: &mut BTreeSet<(RequestId, &'static str)>,\n ) -> Result<(), String> {\n if active_requests.get(&held.request) != Some(&barrier.cell) {\n return Err(format!(\n \"held output {} on {} behind op {op} is not pinned on {:?}\",\n held.request,\n held.channel.as_str(),\n barrier.cell\n ));\n }\n if !seen.insert((held.request, held.channel.as_str())) {\n return Err(format!(\n \"request {} has two outputs held on {}\",\n held.request,\n held.channel.as_str()\n ));\n }\n Ok(())\n }\n\n pub fn validate(&self) -> Result<(), String> {\n if !self.occupied_below(self.config.max_resident.saturating_add(1)) {\n return Err(format!(\n \"occupied {} exceeds ceiling {}\",\n self.occupied(),\n self.config.max_resident\n ));\n }\n let eviction_ceiling = self.eviction_ceiling();\n for op in self.eviction_notifications.keys() {\n let waiting = self.cell_ops.get(op).and_then(|id| self.cells.get(id));\n if !waiting.is_some_and(|cell| {\n matches!(cell.phase,\n Phase::EnsuringDurability { op: current, .. }\n | Phase::Cleaning { op: current, cause: StopCause::Evict { .. }, .. }\n if current == *op\n )\n }) {\n return Err(format!(\"eviction notification {op} has no live operation\"));\n }\n }\n if self.eviction_permits.len() > eviction_ceiling {\n return Err(format!(\n \"evicting {} exceeds ceiling {}\",\n self.eviction_permits.len(),\n eviction_ceiling\n ));\n }\n let handoffs = self.handoffs_in_flight();\n if self.draining && handoffs > self.config.max_releases {\n return Err(format!(\n \"handoffs {handoffs} exceeds release ceiling {}\",\n self.config.max_releases\n ));\n }\n if self.activation_permits.len() > self.config.max_activations {\n return Err(format!(\n \"activating {} exceeds ceiling {}\",\n self.activation_permits.len(),\n self.config.max_activations\n ));\n }\n\n let mut activation_queued = BTreeSet::new();\n for id in &self.activation_waiters {\n if !activation_queued.insert(id) {\n return Err(format!(\"activation waiter {id:?} is queued twice\"));\n }\n let Some(cell) = self.cells.get(id) else {\n return Err(format!(\"activation waiter {id:?} has no cell state\"));\n };\n if cell.phase != Phase::WaitingActivation || cell.waiting_activation.is_none() {\n return Err(format!(\"activation waiter {id:?} is not waiting\"));\n }\n if self.activation_permits.contains(id) {\n return Err(format!(\"activation waiter {id:?} also holds a permit\"));\n }\n }\n for id in &self.activation_permits {\n let Some(cell) = self.cells.get(id) else {\n return Err(format!(\"activation permit {id:?} has no cell state\"));\n };\n if !phase_holds_activation(&cell.phase) {\n return Err(format!(\n \"activation permit {id:?} is held in terminal phase {:?}\",\n cell.phase\n ));\n }\n }\n\n for (id, cell) in &self.cells {\n let in_pipeline = match cell.phase {\n Phase::Dormant { .. } => cell.releasing.is_some(),\n Phase::Adopting { .. } => true,\n _ => false,\n };\n if cell.handoff && !in_pipeline {\n return Err(format!(\n \"handoff cell {id:?} is outside the pipeline in {:?}\",\n cell.phase\n ));\n }\n }\n let actual_quiescing: BTreeSet<_> = self\n .cells\n .iter()\n .filter(|(_, cell)| cell.quiescing)\n .map(|(id, _)| id.clone())\n .collect();\n if actual_quiescing != self.quiescing_cells {\n return Err(format!(\n \"quiescing index {:?} differs from cells {:?}\",\n self.quiescing_cells, actual_quiescing\n ));\n }\n let actual_releasing: BTreeSet<_> = self\n .cells\n .iter()\n .filter(|(_, cell)| cell.releasing.is_some())\n .map(|(id, _)| id.clone())\n .collect();\n if actual_releasing != self.releasing_cells {\n return Err(format!(\n \"releasing index {:?} differs from cells {:?}\",\n self.releasing_cells, actual_releasing\n ));\n }\n let actual_adopting: BTreeSet<_> = self\n .cells\n .iter()\n .filter(|(_, cell)| matches!(cell.phase, Phase::Adopting { .. }))\n .map(|(id, _)| id.clone())\n .collect();\n if actual_adopting != self.adopting_cells {\n return Err(format!(\n \"adopting index {:?} differs from cells {:?}\",\n self.adopting_cells, actual_adopting\n ));\n }\n let actual_confirming: BTreeSet<_> = self\n .cells\n .iter()\n .filter(|(_, cell)| cell.confirming)\n .map(|(id, _)| id.clone())\n .collect();\n if actual_confirming != self.confirming_cells {\n return Err(format!(\n \"confirming index {:?} differs from cells {:?}\",\n self.confirming_cells, actual_confirming\n ));\n }\n for id in &self.confirming_cells {\n let cell = &self.cells[id];\n if !matches!(\n cell.phase,\n Phase::ReadingOwner { .. } | Phase::WaitingActivation\n ) {\n return Err(format!(\n \"confirming cell {id:?} left the owner read in {:?}\",\n cell.phase\n ));\n }\n }\n if !self.confirming && !self.confirming_cells.is_empty() {\n return Err(\"confirmation finished with cells still confirming\".into());\n }\n let actual_resuming: BTreeSet<_> = self\n .cells\n .iter()\n .filter(|(_, cell)| cell.resume_demand)\n .map(|(id, _)| id.clone())\n .collect();\n if actual_resuming != self.resuming_cells {\n return Err(format!(\n \"resuming index {:?} differs from cells {:?}\",\n self.resuming_cells, actual_resuming\n ));\n }\n if !self.resuming && !actual_resuming.is_empty() {\n return Err(format!(\n \"settled node retains resume demand for {actual_resuming:?}\"\n ));\n }\n\n let mut queued = BTreeSet::new();\n for id in &self.capacity_waiters {\n if !queued.insert(id) {\n return Err(format!(\"capacity waiter {id:?} is queued twice\"));\n }\n let Some(cell) = self.cells.get(id) else {\n return Err(format!(\"capacity waiter {id:?} has no cell state\"));\n };\n if cell.phase != Phase::WaitingCapacity || cell.waiting_for.is_none() {\n return Err(format!(\"capacity waiter {id:?} is not waiting\"));\n }\n }\n for (id, cell) in &self.cells {\n if cell.resume_demand\n && !matches!(\n cell.phase,\n Phase::WaitingActivation\n | Phase::WaitingCapacity\n | Phase::Restoring { .. }\n | Phase::Starting { .. }\n | Phase::Publishing { .. }\n | Phase::Fenced\n )\n {\n return Err(format!(\n \"cell {id:?} retains resume demand in terminal phase {:?}\",\n cell.phase\n ));\n }\n if cell.quiescing && !matches!(cell.phase, Phase::Resident { .. }) {\n return Err(format!(\n \"cell {id:?} is quiescing in non-resident phase {:?}\",\n cell.phase\n ));\n }\n if cell.swapping\n && !matches!(\n cell.phase,\n Phase::Resident { .. }\n | Phase::Cleaning {\n cause: StopCause::Swap,\n ..\n }\n | Phase::Starting { .. }\n | Phase::Publishing { .. }\n )\n {\n return Err(format!(\n \"cell {id:?} is swapping in phase {:?}, outside the swap path\",\n cell.phase\n ));\n }\n if cell.swap_cancelled && !cell.swapping {\n return Err(format!(\"cell {id:?} carries a swap flag without swapping\"));\n }\n let is_activation_queued = activation_queued.contains(id);\n if (cell.phase == Phase::WaitingActivation) != is_activation_queued {\n return Err(format!(\"cell {id:?} activation queue and phase disagree\"));\n }\n if phase_requires_activation_permit(&cell.phase, cell.swapping)\n && !self.activation_permits.contains(id)\n {\n return Err(format!(\n \"cell {id:?} is in activation phase {:?} without a permit\",\n cell.phase\n ));\n }\n let is_queued = queued.contains(id);\n if (cell.phase == Phase::WaitingCapacity) != is_queued {\n return Err(format!(\"cell {id:?} queue and phase disagree\"));\n }\n for request in &cell.requests {\n if self.request_cells.get(request) != Some(id) {\n return Err(format!(\n \"request {request} index disagrees with cell {id:?}\"\n ));\n }\n }\n // Every in-flight cell op must be resolvable through the index,\n // or its completion event will never find this cell.\n let indexed = |op: OpId| self.cell_ops.get(&op) == Some(id);\n if let Some(op) = phase_op(&cell.phase) {\n if !indexed(op) {\n return Err(format!(\"cell {id:?} phase op {op} is not indexed\"));\n }\n }\n if let Some(AlarmState::Firing { op, .. }) = cell.alarm {\n if !indexed(op) {\n return Err(format!(\"cell {id:?} firing alarm op {op} is not indexed\"));\n }\n }\n if let Some(op) = cell.releasing {\n if !indexed(op) {\n return Err(format!(\"cell {id:?} release op {op} is not indexed\"));\n }\n }\n }\n let walked = self\n .cells\n .values()\n .filter(|cell| phase_occupies_capacity(&cell.phase))\n .count();\n if walked != self.occupied {\n return Err(format!(\n \"occupied counter {} disagrees with the walk {walked}\",\n self.occupied\n ));\n }\n for (request, id) in &self.request_cells {\n if !self\n .cells\n .get(id)\n .is_some_and(|cell| cell.requests.contains(request))\n {\n return Err(format!(\"request index {request} has no matching waiter\"));\n }\n }\n for (op, id) in &self.cell_ops {\n let Some(cell) = self.cells.get(id) else {\n return Err(format!(\"cell op {op} points to missing cell {id:?}\"));\n };\n if phase_op(&cell.phase) != Some(*op)\n && cell.releasing != Some(*op)\n && !matches!(cell.alarm, Some(AlarmState::Firing { op: current, .. }) if current == *op)\n {\n return Err(format!(\"indexed cell op {op} has no owning slot in {id:?}\"));\n }\n }\n for op in &self.timed_out_runtime_starts {\n let Some(id) = self.cell_ops.get(op) else {\n return Err(format!(\"timed-out runtime start op {op} is not indexed\"));\n };\n if !self.cells.get(id).is_some_and(\n |cell| matches!(cell.phase, Phase::Starting { op: current, .. } if current == *op),\n ) {\n return Err(format!(\n \"timed-out runtime start op {op} has no matching phase\"\n ));\n }\n }\n // The reverse index must agree with the map it summarizes, exactly.\n let mut recount: BTreeMap = BTreeMap::new();\n for id in self.active_requests.values() {\n *recount.entry(id.clone()).or_insert(0) += 1;\n }\n if recount != self.active_cells {\n return Err(format!(\n \"active_cells index {:?} disagrees with active_requests {:?}\",\n self.active_cells, recount\n ));\n }\n let mut gate_recount: BTreeMap = BTreeMap::new();\n for request in &self.admitted_gates {\n let Some(id) = self.active_requests.get(request) else {\n return Err(format!(\"admitted gate {request} is not active\"));\n };\n *gate_recount.entry(id.clone()).or_insert(0) += 1;\n }\n if gate_recount != self.admitted_gate_cells {\n return Err(format!(\n \"admitted_gate_cells index {:?} disagrees with admitted_gates {:?}\",\n self.admitted_gate_cells, gate_recount\n ));\n }\n for (request, id) in &self.active_requests {\n if self.request_cells.contains_key(request) {\n return Err(format!(\"request {request} is both pending and active\"));\n }\n if !self.cells.get(id).is_some_and(|cell| {\n matches!(\n cell.phase,\n Phase::Resident { .. } | Phase::EnsuringDurability { .. }\n )\n }) {\n return Err(format!(\n \"active request {request} has no resident cell {id:?}\"\n ));\n }\n }\n // `(request, channel)` identifies a withheld output. The shell keys its\n // adapters on that pair, so a second output for the same pair would be\n // released once and stranded once.\n let mut held_outputs: BTreeSet<(RequestId, &'static str)> = BTreeSet::new();\n for (op, barrier) in &self.barriers {\n match &barrier.owner {\n // A held output keeps its request pinned, so the cell cannot be\n // evicted underneath the barrier and a later write of the same\n // request can still find it.\n GateOwner::Output(held) => {\n Self::validate_held(\n held,\n op,\n barrier,\n &self.active_requests,\n &mut held_outputs,\n )?;\n }\n // An alarm pins nothing: it is not a request, so it has no\n // activity to hold. What keeps the cell resident is the firing\n // state itself, which `shed_candidate` refuses to evict. The\n // phase is deliberately not asserted here — an activity fold\n // can retire the firing state while this proof is still in\n // flight, and the barrier that outlives it is then settled by\n // whichever teardown takes the cell.\n GateOwner::Alarm { .. } => {\n if !self.cells.contains_key(&barrier.cell) {\n return Err(format!(\n \"gated alarm write op {op} names a cell that is gone: {:?}\",\n barrier.cell\n ));\n }\n }\n }\n // A follower is always an output: an alarm opens a barrier and\n // never trails one.\n for held in &barrier.followers {\n Self::validate_held(held, op, barrier, &self.active_requests, &mut held_outputs)?;\n }\n }\n for (id, cell) in &self.cells {\n if matches!(cell.alarm, Some(AlarmState::Firing { .. }))\n && !matches!(cell.phase, Phase::Resident { .. })\n {\n return Err(format!(\"cell {id:?} is firing an alarm while not resident\"));\n }\n if cell\n .websockets\n .values()\n .any(|kind| matches!(kind, WebSocketKind::Regular | WebSocketKind::Outbound))\n && !matches!(\n cell.phase,\n Phase::Resident { .. } | Phase::EnsuringDurability { .. }\n )\n {\n return Err(format!(\n \"cell {id:?} has a live transport while not resident\"\n ));\n }\n }\n Ok(())\n }\n\n fn op(&mut self) -> OpId {\n let op = self.next_op;\n self.next_op = self.next_op.checked_add(1).expect(\"operation id exhausted\");\n op\n }\n\n /// Allocate an op that will live in `cell`'s phase, alarm, or release\n /// slot, and index it.\n fn cell_op(&mut self, cell: &str) -> OpId {\n let op = self.op();\n self.cell_ops.insert(op, cell.to_string());\n op\n }\n\n /// Resolve a completion's op to its cell, consuming the index entry —\n /// but only when `predicate` confirms the cell still holds the op. On a\n /// mismatch the entry stays: the op's real completion has not arrived\n /// yet (an expiry can probe the wrong handler), and it must still find\n /// its way here later.\n fn take_cell_op(&mut self, op: OpId, predicate: impl Fn(&Cell) -> bool) -> Option {\n let id = self.cell_ops.get(&op)?.clone();\n if !self.cells.get(&id).is_some_and(predicate) {\n return None;\n }\n self.cell_ops.remove(&op);\n Some(id)\n }\n\n fn has_capacity(&self) -> bool {\n // Residency is a hard cap, known exactly and counted -- never sampled.\n // A node at its cell cap is at capacity, not overloaded: it refuses\n // more and holds what it has, rather than shedding a live cell it must\n // then place again elsewhere. The only sampled fact that refuses\n // admission is genuine memory pressure, which a cell count\n // cannot see.\n self.occupied_below(self.config.max_resident) && !self.shedding\n }\n\n pub fn is_active(&self, id: &str) -> bool {\n self.active_cells.contains_key(id)\n || self.cells.get(id).is_some_and(|cell| {\n cell.websockets\n .values()\n .any(|kind| matches!(kind, WebSocketKind::Regular | WebSocketKind::Outbound))\n })\n }\n\n /// Record `request` as active on `cell`, keeping the `active_cells`\n /// reverse index in step with `active_requests`. Every insert into\n /// `active_requests` goes through here so the two never disagree.\n fn activate_request(&mut self, request: RequestId, cell: CellId) {\n *self.active_cells.entry(cell.clone()).or_insert(0) += 1;\n self.active_requests.insert(request, cell);\n }\n\n /// Drop `request` from the active set, decrementing its cell's index and\n /// forgetting the cell once no request targets it. Returns the cell the\n /// request held, as `active_requests.remove` did.\n fn deactivate_request(&mut self, request: RequestId) -> Option {\n let cell = self.active_requests.remove(&request)?;\n if let Some(count) = self.active_cells.get_mut(&cell) {\n *count -= 1;\n if *count == 0 {\n self.active_cells.remove(&cell);\n }\n }\n if self.admitted_gates.remove(&request) {\n if let Some(count) = self.admitted_gate_cells.get_mut(&cell) {\n *count -= 1;\n if *count == 0 {\n self.admitted_gate_cells.remove(&cell);\n }\n }\n }\n Some(cell)\n }\n\n /// Whether `id` runs active work other than an admitted gate.\n fn runs_gate_source(&self, id: &str) -> bool {\n let active = self.active_cells.get(id).copied().unwrap_or(0);\n let gates = self.admitted_gate_cells.get(id).copied().unwrap_or(0);\n active > gates\n }\n\n /// Hibernated (Durable Objects state 4): out of memory, still on this\n /// host, with WebSocket clients parked at the network layer. Nothing in\n /// memory survives -- the next event runs the constructor again, exactly\n /// as a cold start does -- so the surviving sockets are the whole of the\n /// difference from `Inactive`.\n pub fn is_hibernated(&self, id: &str) -> bool {\n self.cells.get(id).is_some_and(|cell| {\n matches!(cell.phase, Phase::Dormant { .. })\n && cell\n .websockets\n .values()\n .any(|kind| matches!(kind, WebSocketKind::Hibernatable))\n })\n }\n\n /// Hibernatable (Durable Objects state 2): resident, doing no work, and\n /// holding nothing that must survive in memory. State 3 is the negation --\n /// resident and idle, but pinned by a `Regular` or `Outbound` socket that\n /// cannot outlive the isolate.\n ///\n /// This is the cell's own criteria, not a decision to evict. celld also\n /// refuses eviction for an imminent or uncovered alarm and for state the\n /// bucket cannot restore; those are policy and are checked separately.\n pub fn is_hibernatable(&self, id: &str) -> bool {\n self.cells\n .get(id)\n .is_some_and(|cell| matches!(cell.phase, Phase::Resident { .. }))\n && !self.is_active(id)\n }\n\n pub fn websocket_count(&self, id: &str) -> usize {\n self.cells.get(id).map_or(0, |cell| cell.websockets.len())\n }\n\n fn record_local_request(cell: &mut Cell, request: RequestId, now_mono_ms: u64) {\n cell.last_used_mono_ms = now_mono_ms;\n cell.last_used_request = cell.last_used_request.max(request);\n cell.local_request_count = cell\n .local_request_count\n .checked_add(1)\n .expect(\"local request count exhausted\");\n }\n\n fn complete_request(\n &mut self,\n id: &str,\n request: RequestId,\n result: Result,\n effects: &mut Vec,\n ) {\n self.capacity_requests.remove(&request);\n self.handoff_requests.remove(&request);\n match &result {\n Ok(Route::Local) => {\n self.activate_request(request, id.to_string());\n let now = self.now_mono_ms;\n if let Some(cell) = self.cells.get_mut(id) {\n Self::record_local_request(cell, request, now);\n }\n }\n Ok(Route::Remote { .. }) => {\n self.activity.proxied = self.activity.proxied.saturating_add(1);\n }\n Err(_) => {}\n }\n effects.push(Effect::Complete { request, result });\n }\n\n // Saturation would reopen the previous writer's identity space after\n // rollback. Exhaustion must fail closed on every acquisition path.\n fn next_writer_epoch(\n &mut self,\n id: &str,\n cell: &mut Cell,\n previous: Epoch,\n effects: &mut Vec,\n ) -> Option {\n match previous.checked_add(1) {\n Some(epoch) => Some(epoch),\n None => {\n set_phase(&mut self.occupied, cell, Phase::Inactive);\n self.finish_requests(id, cell, Err(RequestError::ResolveFailed), effects);\n None\n }\n }\n }\n\n fn record_acquisition(&mut self, spec: &RestoreSpec) {\n self.activity.acquired = self.activity.acquired.saturating_add(1);\n if spec.took_over {\n self.activity.expired_owner_leases =\n self.activity.expired_owner_leases.saturating_add(1);\n }\n if spec.epoch > 1 {\n self.activity.advanced_epochs = self.activity.advanced_epochs.saturating_add(1);\n }\n }\n\n /// Whether a cell takes the handoff path: the node is draining, or a\n /// balancing move reserved this one cell.\n fn hands_off(&self, cell: &Cell) -> bool {\n self.draining || cell.handoff\n }\n\n /// Return a voluntary eviction to service as one transition. Every\n /// rescue returns the permit with the phase, or the cell looks resident\n /// while a permit stays spent.\n fn cancel_eviction(\n &mut self,\n id: &str,\n cell: &mut Cell,\n epoch: Epoch,\n reason: EvictCancellation,\n effects: &mut Vec,\n ) {\n if let Some(op) = phase_op(&cell.phase) {\n self.cell_ops.remove(&op);\n self.finish_eviction(op, Err(EvictError::Cancelled(reason)), effects);\n }\n set_phase(&mut self.occupied, cell, Phase::Resident { epoch });\n self.eviction_permits.remove(id);\n }\n\n fn worker_request(&mut self, request: RequestId, effects: &mut Vec) {\n let route = self\n .cells\n .iter()\n .filter_map(|(id, cell)| {\n let epoch = match cell.phase {\n Phase::Resident { epoch } | Phase::EnsuringDurability { epoch, .. } => epoch,\n _ => return None,\n };\n if cell.quiescing\n || self.is_active(id)\n || matches!(cell.alarm, Some(AlarmState::Firing { .. }))\n {\n return None;\n }\n Some((id.clone(), epoch))\n })\n .find(|(id, _)| self.worker_cursor.as_ref().is_none_or(|cursor| id > cursor))\n .or_else(|| {\n self.cells.iter().find_map(|(id, cell)| {\n let epoch = match cell.phase {\n Phase::Resident { epoch } | Phase::EnsuringDurability { epoch, .. } => {\n epoch\n }\n _ => return None,\n };\n (!cell.quiescing\n && !self.is_active(id)\n && !matches!(cell.alarm, Some(AlarmState::Firing { .. })))\n .then(|| (id.clone(), epoch))\n })\n })\n .map(|(cell, epoch)| WorkerRoute { cell, epoch });\n\n if let Some(route) = &route {\n self.worker_cursor = Some(route.cell.clone());\n self.activate_request(request, route.cell.clone());\n if let Some(mut cell) = self.cells.remove(&route.cell) {\n if matches!(cell.phase, Phase::EnsuringDurability { .. }) {\n self.cancel_eviction(\n &route.cell,\n &mut cell,\n route.epoch,\n EvictCancellation::Activity,\n effects,\n );\n }\n Self::record_local_request(&mut cell, request, self.now_mono_ms);\n self.cells.insert(route.cell.clone(), cell);\n }\n }\n effects.push(Effect::CompleteWorker { request, route });\n }\n\n fn finish_requests(\n &mut self,\n id: &str,\n cell: &mut Cell,\n result: Result,\n effects: &mut Vec,\n ) {\n for request in std::mem::take(&mut cell.requests) {\n self.request_cells.remove(&request);\n if result == Ok(Route::Local) {\n Self::record_local_request(cell, request, self.now_mono_ms);\n }\n self.complete_request(id, request, result.clone(), effects);\n }\n }\n\n fn begin_cold_route(\n &mut self,\n id: &str,\n cell: &mut Cell,\n start: ColdStart,\n effects: &mut Vec,\n ) {\n debug_assert!(self.activation_permits.contains(id));\n cell.waiting_activation = None;\n match start {\n ColdStart::ReadOwner => {\n let op = self.cell_op(id);\n set_phase(&mut self.occupied, cell, Phase::ReadingOwner { op });\n effects.push(Effect::ReadOwner {\n op,\n cell: id.to_string(),\n });\n }\n ColdStart::Restore(spec) => self.activate_or_wait(\n id,\n cell,\n Activation::Restore(spec),\n WhenFull::ShedToAdmit,\n effects,\n ),\n }\n }\n\n fn admit_or_queue_activation(\n &mut self,\n id: &str,\n cell: &mut Cell,\n start: ColdStart,\n effects: &mut Vec,\n ) {\n if self.activation_permits.contains(id) {\n self.begin_cold_route(id, cell, start, effects);\n } else if self.activation_permits.len() < self.config.max_activations {\n self.activation_permits.insert(id.to_string());\n self.begin_cold_route(id, cell, start, effects);\n } else {\n set_phase(&mut self.occupied, cell, Phase::WaitingActivation);\n cell.waiting_activation = Some(start);\n self.activation_waiters.push_back(id.to_string());\n self.watch_queued(id, cell, effects);\n }\n }\n\n fn pump_activations(&mut self, effects: &mut Vec) {\n self.activation_permits.retain(|id| {\n self.cells\n .get(id)\n .is_some_and(|cell| phase_holds_activation(&cell.phase))\n });\n\n while self.activation_permits.len() < self.config.max_activations {\n let Some(id) = self.activation_waiters.pop_front() else {\n break;\n };\n let Some(mut cell) = self.cells.remove(&id) else {\n continue;\n };\n if cell.phase != Phase::WaitingActivation {\n self.cells.insert(id, cell);\n continue;\n }\n let Some(start) = cell.waiting_activation.take() else {\n self.cells.insert(id, cell);\n continue;\n };\n if cell.requests.is_empty()\n && !cell.alarm_wake\n && !cell.resume_demand\n && !cell.confirming\n {\n set_phase(\n &mut self.occupied,\n &mut cell,\n match start {\n ColdStart::ReadOwner => Phase::Inactive,\n ColdStart::Restore(spec) => Phase::Dormant { epoch: spec.epoch },\n },\n );\n } else {\n self.activation_permits.insert(id.clone());\n self.begin_cold_route(&id, &mut cell, start, effects);\n }\n self.cells.insert(id, cell);\n }\n }\n\n /// Adopt the exact epochs encoded by a clean predecessor's local paths.\n /// The node lease CAS that caused this event is the ownership proof, so\n /// this path deliberately emits no per-cell ownership effects.\n fn local_cells_read(\n &mut self,\n result: Result, Failure>,\n effects: &mut Vec,\n ) {\n if self.confirming {\n self.confirm_local_cells(result, effects);\n return;\n }\n if !self.resuming {\n return;\n }\n let Ok(mut cells) = result else {\n self.resuming = false;\n return;\n };\n // A preserved eviction copy is not a resumable epoch.\n cells.retain(|cell| cell.live);\n cells.sort();\n cells.dedup();\n if cells.len() > self.config.max_resident || !self.cells.is_empty() {\n self.resuming = false;\n return;\n }\n for local in cells {\n let mut cell = Cell {\n resume_demand: true,\n ..Cell::default()\n };\n self.resuming_cells.insert(local.id.clone());\n self.admit_or_queue_activation(\n &local.id,\n &mut cell,\n ColdStart::Restore(RestoreSpec {\n epoch: local.epoch,\n fresh: false,\n took_over: false,\n resume_local: true,\n prior: None,\n }),\n effects,\n );\n self.cells.insert(local.id, cell);\n }\n if self.resuming_cells.is_empty() {\n self.resuming = false;\n }\n }\n\n /// Read the ownership record of every cell on this node's disk that the\n /// process does not know yet, and settle each one to dormant if the\n /// record still names this node, without waking it. A crash restart\n /// otherwise publishes only the cells that traffic has touched, and its\n /// peers balance against a node that looks empty. The startup demand\n /// marker keeps the read queued behind the activation ceiling; a request\n /// that arrives first turns the read into an ordinary wake.\n fn confirm_local_cells(\n &mut self,\n result: Result, Failure>,\n effects: &mut Vec,\n ) {\n let Ok(mut cells) = result else {\n self.confirming = false;\n return;\n };\n cells.sort();\n cells.dedup();\n for local in cells {\n if self.cells.contains_key(&local.id) || self.confirming_cells.contains(&local.id) {\n continue;\n }\n let mut cell = Cell {\n confirming: true,\n ..Cell::default()\n };\n self.confirming_cells.insert(local.id.clone());\n self.admit_or_queue_activation(&local.id, &mut cell, ColdStart::ReadOwner, effects);\n self.cells.insert(local.id, cell);\n }\n if self.confirming_cells.is_empty() {\n self.confirming = false;\n }\n }\n\n fn settle_confirmation(&mut self, id: &str) {\n if self.confirming_cells.remove(id) && self.confirming_cells.is_empty() {\n self.confirming = false;\n }\n }\n\n fn settle_local_resume(&mut self, id: &str) {\n if self.resuming_cells.remove(id) && self.resuming_cells.is_empty() {\n self.resuming = false;\n }\n if let Some(cell) = self.cells.get_mut(id) {\n cell.resume_demand = false;\n }\n }\n\n /// A leaving node can still discover a live successor, but it must not\n /// acquire somebody else's or an unowned cell just to hand it off again.\n /// Keep this check at claim admission as well as before recovery and\n /// placement: queued work or a delayed recovery can cross ReleaseAll.\n /// Restores after an issued CAS and work on authority already held here\n /// still settle, so draining does not abandon an acquired owner record.\n fn refuse_new_ownership_during_drain(\n &mut self,\n id: &str,\n cell: &mut Cell,\n prior: Option<&str>,\n effects: &mut Vec,\n ) -> bool {\n if !self.draining || prior == Some(self.node.as_str()) {\n return false;\n }\n cell.waiting_for = None;\n set_phase(&mut self.occupied, cell, Phase::Inactive);\n self.finish_requests(id, cell, Err(RequestError::NodeUnavailable), effects);\n true\n }\n\n fn begin_activation(\n &mut self,\n id: &str,\n cell: &mut Cell,\n activation: Activation,\n effects: &mut Vec,\n ) {\n if let Activation::Claim(claim) = &activation {\n if self.refuse_new_ownership_during_drain(id, cell, claim.prior.as_deref(), effects) {\n return;\n }\n }\n // An ownership-only handoff claim lands `Dormant`: it takes no\n // residency slot, so place_unowned admits it above the cap and\n // the phase carries the exemption.\n let dormant = !cell.requests.is_empty()\n && cell\n .requests\n .iter()\n .all(|request| self.handoff_requests.contains(request));\n debug_assert!(self.has_capacity() || dormant);\n cell.waiting_for = None;\n let op = self.cell_op(id);\n match activation {\n Activation::Claim(claim) => {\n set_phase(\n &mut self.occupied,\n cell,\n Phase::Acquiring {\n op,\n claim: claim.clone(),\n dormant,\n },\n );\n effects.push(Effect::CasOwner {\n op,\n cell: id.to_string(),\n guard: claim.guard,\n epoch: claim.epoch,\n takeover: claim.takeover,\n });\n }\n Activation::Restore(spec) => {\n set_phase(\n &mut self.occupied,\n cell,\n Phase::Restoring {\n op,\n spec: spec.clone(),\n },\n );\n effects.push(Effect::Restore {\n op,\n cell: id.to_string(),\n spec,\n });\n }\n }\n }\n\n fn activate_or_wait(\n &mut self,\n id: &str,\n cell: &mut Cell,\n activation: Activation,\n when_full: WhenFull,\n effects: &mut Vec,\n ) {\n if let Activation::Claim(claim) = &activation {\n if self.refuse_new_ownership_during_drain(id, cell, claim.prior.as_deref(), effects) {\n return;\n }\n }\n if cell.requests.is_empty() && !cell.alarm_wake && !cell.resume_demand && !cell.confirming {\n set_phase(\n &mut self.occupied,\n cell,\n match activation {\n Activation::Claim(_) => Phase::Inactive,\n Activation::Restore(spec) => Phase::Dormant { epoch: spec.epoch },\n },\n );\n } else if self.has_capacity() {\n self.begin_activation(id, cell, activation, effects);\n } else {\n set_phase(&mut self.occupied, cell, Phase::WaitingCapacity);\n cell.waiting_for = Some(activation);\n self.capacity_waiters.push_back(id.to_string());\n self.watch_queued(id, cell, effects);\n if when_full == WhenFull::ShedToAdmit {\n self.shed_one(effects);\n }\n }\n }\n\n fn pump_capacity(&mut self, effects: &mut Vec) {\n while self.has_capacity() {\n let Some(id) = self.capacity_waiters.pop_front() else {\n break;\n };\n let Some(mut cell) = self.cells.remove(&id) else {\n continue;\n };\n if cell.phase != Phase::WaitingCapacity {\n self.cells.insert(id, cell);\n continue;\n }\n let Some(activation) = cell.waiting_for.take() else {\n self.cells.insert(id, cell);\n continue;\n };\n if cell.requests.is_empty() && !cell.alarm_wake && !cell.resume_demand {\n set_phase(\n &mut self.occupied,\n &mut cell,\n match activation {\n Activation::Claim(_) => Phase::Inactive,\n Activation::Restore(spec) => Phase::Dormant { epoch: spec.epoch },\n },\n );\n } else {\n self.begin_activation(&id, &mut cell, activation, effects);\n }\n self.cells.insert(id, cell);\n }\n self.shed_one(effects);\n }\n\n fn start_node_lease(&mut self, now_ms: u64, spec: NodeLeaseSpec, effects: &mut Vec) {\n if !matches!(\n self.node_authority,\n NodeAuthority::Unstarted | NodeAuthority::Failed\n ) {\n return;\n }\n self.begin_node_lease_acquisition(now_ms, spec, effects);\n }\n\n fn begin_node_lease_acquisition(\n &mut self,\n now_ms: u64,\n spec: NodeLeaseSpec,\n effects: &mut Vec,\n ) {\n let desired = NodeLeaseRecord {\n node: self.node.clone(),\n addr: spec.addr.clone(),\n expires_ms: now_ms.saturating_add(spec.ttl_ms),\n peer_protocol: spec.peer_protocol,\n generation: spec.generation.clone(),\n // A fresh session has not opened a log; the predecessor's\n // state was recovered before this install (the engine's\n // startup order), so None is the truth, not a reset.\n log_state: None,\n etag: String::new(),\n };\n let pending = PendingNodeLease {\n spec,\n desired,\n prior: None,\n readback_only: false,\n anchor_mono_ms: self.now_mono_ms,\n resume_local: false,\n };\n let op = self.op();\n self.node_authority = NodeAuthority::Reading { op, pending };\n effects.push(Effect::ReadSelfNodeLease { op });\n }\n\n fn fail_initial_node_lease(&mut self, spec: NodeLeaseSpec, effects: &mut Vec) {\n // Nothing else will ever ask: `StartNodeLease` is a one-shot at\n // process start. Arm the retry here or this node is done serving.\n // The renewal cadence is the right interval -- it is already the\n // rate this fleet has decided its bucket can carry.\n let generation = self.next_timer_generation;\n self.next_timer_generation = self\n .next_timer_generation\n .checked_add(1)\n .expect(\"timer generation exhausted\");\n let retry_after = (spec.ttl_ms / 3).max(1);\n effects.push(Effect::ScheduleTimer {\n timer: Timer::NodeLeaseRenew { generation },\n at_mono_ms: self.now_mono_ms.saturating_add(retry_after),\n });\n self.node_authority = NodeAuthority::Retrying { spec, generation };\n }\n\n /// A lease write landed. Decide whether it actually preserved authority,\n /// and if so install it anchored to the instant its `expires_ms` was\n /// sampled.\n ///\n /// A renewal that lands at or after the PRIOR lease's expiry has not\n /// extended anything: every peer was already entitled to read that record,\n /// find it dead, and seize this node's cells, and one of them may have\n /// begun doing exactly that. Rewriting `nodes/.json` does not\n /// retract a takeover already in flight -- the ownership compare-and-swap\n /// a peer is about to issue is guarded on the ownership record's ETag,\n /// which a resident owner never touches -- so resurrecting the lease here\n /// is precisely how two nodes end up serving one cell. The gap is a fence,\n /// not a hiccup.\n fn complete_node_lease_write(\n &mut self,\n pending: PendingNodeLease,\n record: NodeLeaseRecord,\n now_mono_ms: u64,\n effects: &mut Vec,\n ) {\n // The record just published expires at `anchor + ttl`. A round trip\n // that outlived that window landed a lease which is already dead to\n // every peer reading the bucket, so this process must not act on it.\n let expired = now_mono_ms.saturating_sub(pending.anchor_mono_ms) >= pending.spec.ttl_ms;\n match &pending.prior {\n // A renewal only preserves authority if it lands while the PRIOR\n // lease is still valid. Losing that continuity means peers were\n // already entitled to seize this node's cells.\n Some(prior) => {\n if expired || now_mono_ms.saturating_sub(prior.last_ok_mono_ms) >= prior.spec.ttl_ms\n {\n self.fence_node(HaltReason::NodeLeaseExpired, effects);\n return;\n }\n }\n // An initial acquisition that lands expired never became\n // authoritative: this node owns nothing and is serving nothing, so\n // there is no runtime to fence and no reason to halt. Fail the\n // waiters and let a later request retry from a fresh sample.\n None => {\n if expired {\n self.fail_initial_node_lease(pending.spec, effects);\n return;\n }\n }\n }\n let resume_local = pending.resume_local;\n // The first hold of a process that is not resuming a clean reload\n // starts the inventory confirmation; a renewal never does.\n let confirm_local = pending.prior.is_none() && !resume_local;\n self.hold_node_lease(\n pending.spec,\n record,\n pending.anchor_mono_ms,\n resume_local,\n confirm_local,\n effects,\n );\n }\n\n /// Install a held lease. `anchor_mono_ms` is the monotonic instant the\n /// record's `expires_ms` was computed from -- NOT the instant the write\n /// completed. See [`PendingNodeLease::anchor_mono_ms`].\n fn hold_node_lease(\n &mut self,\n spec: NodeLeaseSpec,\n record: NodeLeaseRecord,\n anchor_mono_ms: u64,\n resume_local: bool,\n confirm_local: bool,\n effects: &mut Vec,\n ) {\n let generation = self.next_timer_generation;\n self.next_timer_generation = self\n .next_timer_generation\n .checked_add(1)\n .expect(\"timer generation exhausted\");\n // A nudge that arrived mid-write could not ride the write that\n // just landed (its body predates the publish): renew again NOW.\n let renew_after = if std::mem::take(&mut self.nudge_pending) {\n 1\n } else {\n (spec.ttl_ms / 3).max(1)\n };\n effects.push(Effect::ScheduleTimer {\n timer: Timer::NodeLeaseRenew { generation },\n at_mono_ms: anchor_mono_ms.saturating_add(renew_after),\n });\n // Exactly the published deadline. A peer treats the record as live\n // while `expires_ms > now_ms`, so this node must have stopped serving\n // by the time `now_ms` reaches `expires_ms` -- not one millisecond\n // after it.\n effects.push(Effect::ScheduleTimer {\n timer: Timer::NodeLeaseFence { generation },\n at_mono_ms: anchor_mono_ms.saturating_add(spec.ttl_ms),\n });\n self.node_authority = NodeAuthority::Held(HeldNodeLease {\n spec,\n record,\n last_ok_mono_ms: anchor_mono_ms,\n last_attempt_mono_ms: anchor_mono_ms,\n timer_generation: generation,\n });\n if resume_local {\n self.resuming = true;\n effects.push(Effect::ReadLocalCells);\n } else if confirm_local {\n self.confirming = true;\n effects.push(Effect::ReadLocalCells);\n }\n }\n\n fn resume_node_lease_after_failure(\n &mut self,\n prior: HeldNodeLease,\n now_mono_ms: u64,\n effects: &mut Vec,\n ) {\n // A publish that arrived mid-write must not wait a full ttl/3\n // behind a transient failure: the\n // OwnLog waiter's deadline is shorter than that.\n let retry_after = if std::mem::take(&mut self.nudge_pending) {\n 1\n } else {\n // A failed CAS and its readback consume authority without moving\n // the published expiry. Waiting another full cadence can put the\n // retry at or beyond that expiry, so a recovered store gets no\n // further attempt and the node fences. Keep the normal cadence\n // only while it leaves at least half of the remaining window\n // for the retry's CAS and readback; the attempt that just\n // failed shows how long an attempt can take. Otherwise retry\n // after a quarter of the window, so a store that answers as\n // slowly as it just did still lands the retry before the\n // expiry. On 2026-09-03 a fleet kept the full cadence with\n // 4.4 s left after a 2.5 s ambiguous write and every node\n // fenced in the same second.\n let authority_deadline = prior.last_ok_mono_ms.saturating_add(prior.spec.ttl_ms);\n let remaining = authority_deadline.saturating_sub(now_mono_ms);\n let cadence = (prior.spec.ttl_ms / 3).max(1);\n if cadence <= remaining / 2 {\n cadence\n } else {\n (remaining / 4).max(1)\n }\n };\n effects.push(Effect::ScheduleTimer {\n timer: Timer::NodeLeaseRenew {\n generation: prior.timer_generation,\n },\n at_mono_ms: now_mono_ms.saturating_add(retry_after),\n });\n self.node_authority = NodeAuthority::Held(prior);\n }\n\n fn begin_node_lease_write(\n &mut self,\n pending: PendingNodeLease,\n guard: CasGuard,\n effects: &mut Vec,\n ) {\n let op = self.op();\n let record = pending.desired.clone();\n let authority_expires_ms = pending.prior.as_ref().map(|prior| prior.record.expires_ms);\n self.node_authority = NodeAuthority::Writing { op, pending };\n effects.push(Effect::CasNodeLease {\n op,\n guard,\n record,\n authority_expires_ms,\n });\n }\n\n fn read_self_node_lease(\n &mut self,\n op: OpId,\n now_ms: u64,\n now_mono_ms: u64,\n result: Result, Failure>,\n effects: &mut Vec,\n ) {\n let authority = std::mem::replace(&mut self.node_authority, NodeAuthority::Failed);\n let NodeAuthority::Reading {\n op: current,\n pending,\n } = authority\n else {\n self.node_authority = authority;\n return;\n };\n if current != op {\n self.node_authority = NodeAuthority::Reading {\n op: current,\n pending,\n };\n return;\n }\n match result {\n Err(_) => {\n if let Some(prior) = pending.prior {\n self.resume_node_lease_after_failure(prior, now_mono_ms, effects);\n } else {\n self.fail_initial_node_lease(pending.spec, effects);\n }\n }\n Ok(Some(record)) if same_node_lease(&record, &pending.desired) => {\n self.complete_node_lease_write(pending, record, now_mono_ms, effects);\n }\n Ok(Some(record))\n if pending\n .prior\n .as_ref()\n .is_some_and(|prior| same_node_lease(&record, &prior.record)) =>\n {\n let mut prior = pending.prior.expect(\"checked above\");\n prior.record.etag = record.etag;\n self.resume_node_lease_after_failure(prior, now_mono_ms, effects);\n }\n Ok(Some(record))\n if pending.prior.as_ref().is_some_and(|prior| {\n record.node == pending.desired.node\n && record.addr == pending.desired.addr\n && record.peer_protocol == pending.desired.peer_protocol\n && record.generation == pending.desired.generation\n && record.log_state == pending.desired.log_state\n && record.expires_ms > prior.record.expires_ms\n && record.expires_ms <= pending.desired.expires_ms\n }) =>\n {\n // Our own zombie: every identity field ours, including the\n // folded log state a peer's recovery stamp would have\n // changed, and only the expiry moved — no further than an\n // attempt of ours could have written. An earlier ambiguous\n // renewal landed after its readback missed it (#482). Adopt\n // the record. Bound the adopted monotonic anchor by both the\n // expiry delta and the expiry observed at this readback. A\n // wall-clock step between the write attempts is present in\n // one bound but not the other, so taking the earlier anchor\n // prevents that step from becoming local lease authority.\n let mut prior = pending.prior.expect(\"checked above\");\n let delta = record.expires_ms.saturating_sub(prior.record.expires_ms);\n let delta_anchor = prior.last_ok_mono_ms.saturating_add(delta);\n let observed_remaining = record.expires_ms.saturating_sub(now_ms);\n if observed_remaining == 0 {\n self.fence_node(HaltReason::NodeLeaseExpired, effects);\n return;\n }\n // The anchor names the deadline as `anchor + ttl_ms`, so it\n // cannot name a translated expiry that falls less than one\n // TTL after the monotonic epoch. A large forward step early\n // in uptime produces exactly that. Saturating the anchor to\n // zero there would read as a full TTL of authority — past\n // the published deadline, and overlapping a successor — so\n // an unrepresentable deadline fences instead.\n let Some(observed_anchor) = now_mono_ms\n .saturating_add(observed_remaining)\n .checked_sub(prior.spec.ttl_ms)\n else {\n self.fence_node(HaltReason::NodeLeaseExpired, effects);\n return;\n };\n prior.last_ok_mono_ms = delta_anchor.min(observed_anchor);\n prior.record = record;\n // An adoption happens only after an ambiguous attempt, its\n // readback, a cadence wait, a rejection, and a second\n // readback have eaten most of the TTL, so the adopted\n // headroom is always thin. The 2026-08-28T07:43Z field\n // fence landed one millisecond past an adopted expiry with\n // the cadence-scheduled renewal 54 ms from completion.\n // Renew NOW, on the nudge lane.\n self.nudge_pending = true;\n effects.push(Effect::ScheduleTimer {\n timer: Timer::NodeLeaseFence {\n generation: prior.timer_generation,\n },\n at_mono_ms: prior.last_ok_mono_ms.saturating_add(prior.spec.ttl_ms),\n });\n self.resume_node_lease_after_failure(prior, now_mono_ms, effects);\n }\n Ok(None) if pending.prior.is_some() => {\n // The record vanished after this process held it. Report the\n // observed loss instead of making the shell call it a TTL\n // expiry while published authority still had time remaining.\n self.fence_node(HaltReason::NodeLeaseMissing, effects);\n }\n Ok(Some(_)) if pending.prior.is_some() => {\n // Every record that this process can safely retain or adopt\n // matched an earlier arm, so authority is lost. The record is\n // NOT proof of a foreign writer: an ambiguous renewal of this\n // process can land late, carrying the folded log stamp it was\n // serialized with, after OwnLog published a newer one. The\n // zombie arm above compares that stamp, so it misses, and the\n // record arrives here self-authored. Report the observation.\n self.fence_node(HaltReason::NodeLeaseMismatch, effects);\n }\n Ok(_) if pending.readback_only => {\n // The ambiguous initial write did not publish the exact\n // desired generation. Fail this activation; a later request\n // may retry from a fresh read, but this one cannot spin.\n self.fail_initial_node_lease(pending.spec, effects);\n }\n Ok(Some(record)) => {\n // The record we are\n // about to replace carries the predecessor's folded log\n // state, and our install writes a fresh one — so a\n // not-sealed predecessor log here means recovery has NOT\n // run, and installing would erase the only evidence it\n // was needed. The engine's boot order makes this\n // unreachable; the guard keeps it structurally so.\n let predecessor_unrecovered = pending.prior.is_none()\n && record.generation != pending.desired.generation\n && !matches!(\n record.log_state,\n None | Some(crate::log_tier::LogState::Sealed)\n );\n if predecessor_unrecovered {\n self.fail_initial_node_lease(pending.spec, effects);\n return;\n }\n // A configured node id is a singleton key: restarting that\n // node replaces its prior process generation immediately. The\n // ETag still serializes competing replacements, and a process\n // which loses that CAS never becomes authoritative.\n let mut pending = pending;\n pending.resume_local = pending.prior.is_none()\n && !pending.readback_only\n && record.expires_ms > now_ms\n && pending.spec.resume_generation.as_deref()\n == Some(record.generation.as_str());\n self.begin_node_lease_write(pending, CasGuard::Match(record.etag), effects);\n }\n Ok(None) => self.begin_node_lease_write(pending, CasGuard::Absent, effects),\n }\n }\n\n fn node_lease_cas_completed(\n &mut self,\n op: OpId,\n now_mono_ms: u64,\n result: Result,\n stamped_log_state: Option,\n effects: &mut Vec,\n ) {\n let authority = std::mem::replace(&mut self.node_authority, NodeAuthority::Failed);\n let NodeAuthority::Writing {\n op: current,\n pending,\n } = authority\n else {\n self.node_authority = authority;\n return;\n };\n if current != op {\n self.node_authority = NodeAuthority::Writing {\n op: current,\n pending,\n };\n return;\n }\n match result {\n Ok(LeaseCasOutcome::Applied { etag }) => {\n let mut record = pending.desired.clone();\n record.etag = etag;\n // The shell stamps the folded log into the body at\n // serialization, after the core chose `desired`: the held\n // record must track the body that actually landed, or the next ambiguous\n // readback compares the bucket's truth against a stale\n // belief and self-fences a healthy node.\n record.log_state = stamped_log_state;\n self.complete_node_lease_write(pending, record, now_mono_ms, effects);\n }\n Ok(LeaseCasOutcome::Rejected) => {\n // A rejected renewal is NOT proof of usurpation: an earlier\n // AMBIGUOUS renewal can land after its own readback missed\n // it, and then the ETag moved under our hand (#482 — R2\n // throttling manufactured exactly this, and fencing here\n // halted a healthy follower with 549ms of real authority\n // left). Read back and decide from the record; the readback\n // compares against what the shell actually serialized, for\n // the same reason the ambiguous arm does.\n let read = self.op();\n let mut pending = pending;\n pending.desired.log_state = stamped_log_state;\n self.node_authority = NodeAuthority::Reading { op: read, pending };\n effects.push(Effect::ReadSelfNodeLease { op: read });\n }\n Err(Failure::Ambiguous) => {\n let read = self.op();\n let mut pending = pending;\n pending.readback_only = true;\n // The readback compares against what the shell actually\n // serialized into this attempt, never the core's\n // pre-attempt belief.\n pending.desired.log_state = stamped_log_state;\n self.node_authority = NodeAuthority::Reading { op: read, pending };\n effects.push(Effect::ReadSelfNodeLease { op: read });\n }\n Err(Failure::Definite) => {\n if let Some(prior) = pending.prior {\n self.resume_node_lease_after_failure(prior, now_mono_ms, effects);\n } else {\n self.fail_initial_node_lease(pending.spec, effects);\n }\n }\n }\n }\n\n /// Arm a deadline for every activation effect this event produced.\n ///\n /// Done once, centrally, rather than at each site that emits one: every\n /// separate site is a chance to forget, and a forgotten deadline is\n /// invisible until something hangs. The effect list already names every\n /// operation the executor is about to start, so it is the natural place to\n /// decide which of them are worth watching.\n fn arm_operation_deadlines(&mut self, effects: &mut Vec) {\n let Some(deadline_ms) = self.config.operation_deadline_ms else {\n return;\n };\n let watched: Vec = effects\n .iter()\n .filter_map(|effect| match effect {\n // Every operation that holds something back. The activation\n // stages hold a caller; a durability proof holds an eviction;\n // a firing alarm holds the cell out of dormancy. All three\n // keep the ownership record claimed while they wait.\n //\n // `StopRuntime` is the deliberate exception. A returned host\n // error can take the failed-stop path, but a pending release\n // can still own the runtime and its database. Expiring that\n // operation would permit an unsafe concurrent restart.\n // A restore is a sequence of individually bounded object-store\n // requests. Its total duration scales with the replica. The\n // shell cannot cancel the task, so retiring the core op would\n // discard late success and let another request duplicate it.\n Effect::Restore { .. } => None,\n Effect::ReadOwner { op, .. }\n | Effect::ReadNodeLease { op, .. }\n | Effect::RecoverNodeLog { op, .. }\n | Effect::ReadCapacityPeers { op, .. }\n | Effect::CasOwner { op, .. }\n | Effect::StartRuntime { op, .. }\n | Effect::Publish { op, .. }\n | Effect::EnsureDurable { op, .. }\n // A stuck owner write must retry while the drain is open.\n // A stuck successor request must release its cohort permit;\n // the owner record is already blank, so later demand can\n // discover whichever peer acquired it.\n | Effect::ReleaseOwner { op, .. }\n | Effect::AdoptReleased { op, .. }\n // A held write response: a swallowed durability proof must not\n // hang a client forever.\n | Effect::AwaitDurable { op, .. }\n | Effect::FireAlarm { op, .. } => Some(*op),\n _ => None,\n })\n .collect();\n let at_mono_ms = self.now_mono_ms.saturating_add(deadline_ms);\n for op in watched {\n effects.push(Effect::ScheduleTimer {\n timer: Timer::OperationDeadline { op },\n at_mono_ms,\n });\n }\n }\n\n /// An activation effect outlived its deadline.\n ///\n /// Expiry deliberately reuses each stage's own failure handling rather\n /// than introducing a second way to abandon work: the core already knows\n /// how to reconcile an ambiguous acquire and how to fail a read, and a\n /// deadline is only a different reason for reaching those paths.\n ///\n /// The classification is the whole substance. A read cannot have committed\n /// anything, so it is definite. Everything past it may have taken effect\n /// on the far side while the answer was lost, so it is ambiguous — the\n /// same distinction that decides whether a retry is safe. Calling a\n /// timed-out compare-and-swap definite would let a second attempt\n /// overwrite an epoch that had in fact been applied.\n fn expire_operation(&mut self, op: OpId, now_ms: u64, effects: &mut Vec) {\n // A gated write is tracked in `barriers`, not the op index.\n // Ambiguous is the only safe class: the write may or may not be\n // durable, so the client must not be told it succeeded.\n if self.barriers.contains_key(&op) {\n // Source is irrelevant on a failed proof; Bucket is the\n // conservative label.\n self.durable_reached(op, Err(Failure::Ambiguous), ProofSource::Bucket, effects);\n return;\n }\n // A peek, not a take: each arm below re-enters the op's own\n // completion handler, and that handler consumes the index entry.\n let Some(id) = self.cell_ops.get(&op).cloned() else {\n // Already answered, superseded, or fenced. A stale deadline has\n // nothing to expire, which is the ordinary case.\n return;\n };\n let Some(cell) = self.cells.get(&id) else {\n return;\n };\n // A release can have committed before its response disappeared. The\n // ordinary failure path is safe in both cases: a retry either blanks\n // our still-owned record or reads the already-blank record and\n // settles as rejected.\n if cell.releasing == Some(op) {\n self.owner_released(op, Err(Failure::Definite), effects);\n return;\n }\n // Ownership is already blank before adoption starts. Retiring a stuck\n // acknowledgement cannot restore this donor's authority. Future\n // demand reads the record and claims or follows the real owner.\n if matches!(cell.phase, Phase::Adopting { op: current, .. } if current == op) {\n self.successor_adopted(op, Err(Failure::Definite), effects);\n return;\n }\n // A firing alarm is tracked on the cell rather than in its phase, so\n // it is looked for first. Expiry re-arms it exactly as a failed\n // handler would, which keeps alarms at-least-once instead of turning\n // a stuck handler into a lost one.\n if matches!(cell.alarm, Some(AlarmState::Firing { op: current, .. }) if current == op) {\n // The cell comes from the index here, unlike the event, which\n // carries its own: an expiry that cannot find the op has nothing\n // to time out, so depending on the index is safe, while the event\n // cannot depend on it because a fold may retire the firing state\n // while the write it reports is still unproven. The epoch is read\n // for completeness and never used — a failure opens no gate, and\n // the gate is the only thing `alarm_finished` reads it for.\n let epoch = match cell.phase {\n Phase::Resident { epoch } => epoch,\n _ => 0,\n };\n let (id, now_mono_ms) = (id.clone(), self.now_mono_ms);\n self.alarm_finished(\n op,\n (id, epoch),\n now_ms,\n now_mono_ms,\n Err(Failure::Ambiguous),\n effects,\n );\n return;\n }\n if phase_op(&cell.phase) != Some(op) {\n return;\n }\n let phase = Some(cell.phase.clone());\n match phase {\n Some(Phase::ReadingOwner { .. }) => {\n self.owner_read(op, 0, Err(Failure::Definite), effects)\n }\n Some(Phase::ReadingNodeLease { .. }) => {\n self.node_lease_read(op, now_ms, Err(Failure::Definite), effects)\n }\n // Recovery can upload data or seal the remote session before its\n // completion is lost, so a timeout cannot prove that nothing\n // changed on the far side.\n Some(Phase::RecoveringOwnerLog { .. }) => {\n self.node_log_recovered(op, Err(Failure::Ambiguous), effects)\n }\n Some(Phase::ReadingCapacity { .. }) => {\n self.capacity_peers_read(op, now_ms, Err(Failure::Definite), effects)\n }\n Some(Phase::Acquiring { .. }) => {\n self.owner_cas_completed(op, Err(Failure::Ambiguous), effects)\n }\n // ReconcilingAcquire waits on the read that follows an ambiguous\n // CAS. The CAS handler accepts only Acquiring, so dispatching the\n // deadline there leaves the read operation and request attached.\n Some(Phase::ReconcilingAcquire { .. }) => {\n self.owner_read(op, now_ms, Err(Failure::Definite), effects)\n }\n // A restore has no aggregate deadline. Each object-store request\n // is bounded in the shell, and the complete task must retain its\n // op so a late success cannot be discarded and duplicated.\n Some(Phase::Restoring { .. }) => {}\n Some(Phase::Starting { .. }) => self.runtime_start_timed_out(op, effects),\n Some(Phase::Publishing { .. }) => self.published(op, Err(Failure::Ambiguous), effects),\n // An unprovable snapshot leaves the cell resident. Evicting on a\n // proof that never arrived is the one outcome that loses data.\n Some(Phase::EnsuringDurability { .. }) => {\n self.durability_checked(op, Err(EvictFailure::DurabilityTimeout), effects)\n }\n _ => {}\n }\n }\n\n fn timer_fired(\n &mut self,\n timer: Timer,\n now_ms: u64,\n now_mono_ms: u64,\n effects: &mut Vec,\n ) {\n let timer = match timer {\n Timer::CellAlarm { cell, generation } => {\n self.cell_alarm_timer(&cell, generation, now_ms, now_mono_ms, effects);\n return;\n }\n // An activation deadline is not conditional on node authority: a\n // request stalled behind a swallowed effect must be released\n // whatever the lease is doing.\n Timer::OperationDeadline { op } => {\n self.expire_operation(op, now_ms, effects);\n return;\n }\n // Also unconditional on node authority: a cell parked behind the\n // gate is waiting on this node's own capacity, not on its lease.\n Timer::QueuedActivation { cell, generation } => {\n self.expire_queued(&cell, generation, effects);\n self.pump_activations(effects);\n self.pump_capacity(effects);\n return;\n }\n // Unconditional as well: the retry re-enters resolution, and\n // resolution itself decides what the lease permits.\n Timer::OwnerLogRecoveryRetry { cell, generation } => {\n self.retry_owner_log_recovery(&cell, generation, effects);\n return;\n }\n timer => timer,\n };\n // A retry has no held lease to validate against, so it is resolved\n // before the authority lookup below discards it.\n if let NodeAuthority::Retrying { spec, generation } = &self.node_authority {\n if matches!(timer, Timer::NodeLeaseRenew { generation: fired } if fired == *generation)\n {\n let spec = spec.clone();\n self.begin_node_lease_acquisition(now_ms, spec, effects);\n }\n return;\n }\n let active = match &self.node_authority {\n NodeAuthority::Held(held) => Some(held.clone()),\n NodeAuthority::Reading { pending, .. } | NodeAuthority::Writing { pending, .. } => {\n pending.prior.clone()\n }\n _ => None,\n };\n let Some(held) = active else {\n return;\n };\n // A VM suspension can advance wall time while its monotonic clock is\n // stopped. The published wall deadline and the local monotonic\n // deadline are independent upper bounds, so reaching either one ends\n // authority. Checking only monotonic elapsed time lets a resumed node\n // retain or renew a record that every peer already considers dead.\n let lease_expired = now_ms >= held.record.expires_ms\n || now_mono_ms.saturating_sub(held.last_ok_mono_ms) >= held.spec.ttl_ms;\n match timer {\n // `>=`, not `>`: a peer stops treating the record as live the\n // instant `now_ms` reaches `expires_ms`, so authority must end\n // there too. Serving through that one millisecond overlaps a\n // takeover that is already entitled to proceed.\n Timer::NodeLeaseFence { generation }\n if generation == held.timer_generation && lease_expired =>\n {\n self.fence_node(HaltReason::NodeLeaseExpired, effects);\n }\n // A renew timer that arrives after the lease it belongs to has\n // already expired is not a renewal opportunity; authority is\n // already gone. Fence rather than issue a write that could\n // resurrect it behind a peer's in-flight takeover.\n Timer::NodeLeaseRenew { generation }\n if generation == held.timer_generation && lease_expired =>\n {\n self.fence_node(HaltReason::NodeLeaseExpired, effects);\n }\n Timer::NodeLeaseRenew { generation }\n if generation == held.timer_generation\n && matches!(self.node_authority, NodeAuthority::Held(_)) =>\n {\n let NodeAuthority::Held(mut prior) =\n std::mem::replace(&mut self.node_authority, NodeAuthority::Failed)\n else {\n unreachable!(\"held checked above\")\n };\n prior.last_attempt_mono_ms = now_mono_ms;\n let spec = prior.spec.clone();\n let desired = NodeLeaseRecord {\n node: self.node.clone(),\n addr: spec.addr.clone(),\n expires_ms: now_ms.saturating_add(spec.ttl_ms),\n peer_protocol: spec.peer_protocol,\n generation: spec.generation.clone(),\n // The write discipline the fold's safety rests on\n // (BrokenLeaseRenewalDropsLog is the spec tooth): a\n // renewal carries the folded log state through\n // unchanged.\n log_state: prior.record.log_state,\n etag: String::new(),\n };\n let guard = CasGuard::Match(prior.record.etag.clone());\n self.begin_node_lease_write(\n PendingNodeLease {\n spec,\n desired,\n prior: Some(prior),\n readback_only: false,\n anchor_mono_ms: now_mono_ms,\n resume_local: false,\n },\n guard,\n effects,\n );\n }\n Timer::CellAlarm { .. } => unreachable!(\"cell timers returned above\"),\n _ => {}\n }\n }\n\n fn schedule_alarm_timer(\n &self,\n cell: &str,\n generation: u64,\n at_ms: i64,\n now_ms: u64,\n now_mono_ms: u64,\n effects: &mut Vec,\n ) {\n let at_ms = u64::try_from(at_ms).unwrap_or(0);\n effects.push(Effect::ScheduleTimer {\n timer: Timer::CellAlarm {\n cell: cell.to_string(),\n generation,\n },\n at_mono_ms: now_mono_ms.saturating_add(at_ms.saturating_sub(now_ms)),\n });\n }\n\n fn alarm_observed(\n &mut self,\n id: &str,\n observed: AlarmObservation,\n now_ms: u64,\n now_mono_ms: u64,\n effects: &mut Vec,\n ) {\n let AlarmObservation {\n alarm,\n covered,\n firing_settled,\n } = observed;\n let at_ms = alarm.at_ms();\n let Some(mut cell) = self.cells.remove(id) else {\n return;\n };\n // Only a cell with a runtime on this node can be observed. A report\n // that lands after the cell left (a handoff coverage refresh that\n // outlived the release, whose snapshot is then `None`) describes\n // nothing, and applying it would delete the wake entry the successor\n // needs.\n if !matches!(\n cell.phase,\n Phase::Starting { .. }\n | Phase::Publishing { .. }\n | Phase::Resident { .. }\n | Phase::EnsuringDurability { .. }\n | Phase::Cleaning { .. }\n ) {\n self.cells.insert(id.to_string(), cell);\n return;\n }\n let source = alarm.source().map(|source| source.id);\n if source.is_some_and(|source| {\n cell.held_epoch != Some(source.epoch)\n || cell.alarm_source.is_some_and(|previous| previous > source)\n }) {\n self.cells.insert(id.to_string(), cell);\n return;\n }\n // A failed output proof has already chosen the durable recovery\n // state and is stopping the unproven runtime. Its alarm cache can\n // still report the locally consumed row while the stop is in flight.\n // That observation belongs to the runtime being discarded, so it\n // cannot replace the re-armed alarm or delete its wake entry.\n if matches!(\n cell.phase,\n Phase::Cleaning {\n cause: StopCause::Reset,\n ..\n }\n ) {\n self.cells.insert(id.to_string(), cell);\n return;\n }\n if let Phase::EnsuringDurability { epoch, .. } = cell.phase {\n // An ordinary alarm change rescues a cell from eviction. A\n // handoff already stopped new routes and deliberately refreshed\n // this observation after nomination. A covered alarm (or no\n // alarm) is therefore confirmation, not new demand. Cancelling\n // here re-emits EnsureDurable forever while the cell has no pin.\n let safe_drain_refresh = self.draining && (at_ms.is_none() || covered);\n if !safe_drain_refresh {\n self.cancel_eviction(id, &mut cell, epoch, EvictCancellation::Alarm, effects);\n }\n }\n // Runtime observations can arrive before `AlarmFinished`. The alarm\n // handler removes its row before its consuming write is durable, so a\n // direct `None` must not replace `Firing` and delete the wake entry.\n // Stage a real move for the completion path. The completion carries\n // the durability verdict that decides whether the observation is safe\n // to apply. An unchanged report only refreshes wake coverage unless\n // it supersedes an earlier staged move.\n if !firing_settled {\n if let Some(AlarmState::Firing {\n at_ms: firing_at_ms,\n covered: firing_covered,\n observed,\n ..\n }) = &mut cell.alarm\n {\n if observed.as_ref().is_some_and(|(staged, _)| {\n staged\n .source()\n .zip(alarm.source())\n .is_some_and(|(staged, next)| staged.id > next.id)\n }) {\n self.cells.insert(id.to_string(), cell);\n return;\n }\n if at_ms == Some(*firing_at_ms) && observed.is_none() && source == cell.alarm_source\n {\n *firing_covered = covered;\n } else {\n *observed = Some((alarm, covered));\n }\n self.cells.insert(id.to_string(), cell);\n return;\n }\n }\n let observed_before = match cell.alarm {\n Some(AlarmState::Armed { at_ms, .. }) | Some(AlarmState::Firing { at_ms, .. }) => at_ms,\n None => -1,\n };\n // The assignment below replaces a firing alarm unconditionally. Its\n // op entry is usually consumed already (`alarm_finished` routes\n // through here), but a direct observation can land mid-fire, and the\n // replaced op will then never match a lookup again.\n if let Some(AlarmState::Firing { op, .. }) = cell.alarm {\n self.cell_ops.remove(&op);\n }\n cell.alarm = at_ms.filter(|at_ms| *at_ms >= 0).map(|at_ms| {\n let generation = self.next_timer_generation;\n self.next_timer_generation = self\n .next_timer_generation\n .checked_add(1)\n .expect(\"timer generation exhausted\");\n self.schedule_alarm_timer(id, generation, at_ms, now_ms, now_mono_ms, effects);\n AlarmState::Armed {\n at_ms,\n generation,\n covered,\n }\n });\n let discovered = cell.alarm_wake;\n if cell.alarm.is_none() {\n cell.alarm_wake = false;\n }\n // Both ways an alarm settles arrive here -- a request that changed it\n // and a firing that consumed it (`alarm_finished` routes through this\n // function). Saying so once, here, is what keeps the bucket entry\n // from depending on the shell noticing each path separately. Only on\n // a change: an activity that left the alarm alone has nothing to\n // mirror, and re-stating it would put a bucket round trip on the end\n // of every request.\n let settled = at_ms.filter(|at_ms| *at_ms >= 0).unwrap_or(-1);\n if settled != observed_before\n || source.is_some_and(|source| {\n // Sequence zero has no alarm history in the restored image.\n // A scan still forces retirement: an old unreplicated PUT\n // can exist outside that image.\n (discovered && at_ms.is_none())\n || (cell.alarm_source != Some(source)\n && (at_ms.is_some() || source.sequence > 0))\n })\n {\n effects.push(Effect::ReconcileWakeEntry {\n cell: id.to_string(),\n alarm,\n });\n }\n if source.is_some() {\n cell.alarm_source = source;\n }\n self.cells.insert(id.to_string(), cell);\n }\n\n fn cell_alarm_timer(\n &mut self,\n id: &str,\n generation: u64,\n now_ms: u64,\n now_mono_ms: u64,\n effects: &mut Vec,\n ) {\n if !self.node_authoritative() {\n return;\n }\n let Some(mut cell) = self.cells.remove(id) else {\n return;\n };\n let Some(AlarmState::Armed {\n at_ms,\n generation: current,\n covered,\n }) = cell.alarm\n else {\n self.cells.insert(id.to_string(), cell);\n return;\n };\n if current != generation {\n self.cells.insert(id.to_string(), cell);\n return;\n }\n if i64::try_from(now_ms).unwrap_or(i64::MAX) < at_ms {\n self.schedule_alarm_timer(id, generation, at_ms, now_ms, now_mono_ms, effects);\n self.cells.insert(id.to_string(), cell);\n return;\n }\n let epoch = match cell.phase {\n // The drain has nominated this cell and is waiting for it to go\n // idle. A handler started now would hold it `Firing` instead\n // (the `@cloudflare/containers` class sleeps for minutes inside\n // its alarm), which `begin_eviction` refuses until the deadline.\n // The alarm stays armed and covered for the successor; the\n // timer below keeps it live should the cell stay after all.\n Phase::Resident { .. } if self.draining && cell.quiescing => {\n effects.push(Effect::ScheduleTimer {\n timer: Timer::CellAlarm {\n cell: id.to_string(),\n generation,\n },\n at_mono_ms: now_mono_ms.saturating_add(100),\n });\n self.cells.insert(id.to_string(), cell);\n return;\n }\n Phase::Resident { epoch } => epoch,\n Phase::EnsuringDurability { epoch, .. } => {\n self.cancel_eviction(id, &mut cell, epoch, EvictCancellation::Alarm, effects);\n epoch\n }\n Phase::Fenced => {\n cell.alarm = None;\n self.cells.insert(id.to_string(), cell);\n return;\n }\n // A draining node has released this cell, or is about to.\n // Re-admitting it here would pull it back for one more\n // ownership CAS and release, and repeat that on every timer\n // fire until the process exits (seen on the fleet as a\n // drain that never settled). The successor reads the alarm\n // from the cell's storage when it activates.\n Phase::Dormant { .. } | Phase::Inactive if self.draining => {\n cell.alarm = None;\n self.cells.insert(id.to_string(), cell);\n return;\n }\n Phase::Dormant { .. } => {\n cell.alarm_wake = true;\n // Every activation takes a fresh epoch. The LTX metadata does\n // not survive an eviction, so a same-epoch wake would restart\n // at TXID 1 inside a populated prefix and mix two writer\n // lineages (CelldWriterGeneration.tla, #158). The CAS also\n // settles the older resume-vs-release race deterministically.\n self.admit_or_queue_activation(id, &mut cell, ColdStart::ReadOwner, effects);\n effects.push(Effect::ScheduleTimer {\n timer: Timer::CellAlarm {\n cell: id.to_string(),\n generation,\n },\n at_mono_ms: now_mono_ms.saturating_add(100),\n });\n self.cells.insert(id.to_string(), cell);\n return;\n }\n Phase::Inactive => {\n cell.alarm_wake = true;\n self.admit_or_queue_activation(id, &mut cell, ColdStart::ReadOwner, effects);\n effects.push(Effect::ScheduleTimer {\n timer: Timer::CellAlarm {\n cell: id.to_string(),\n generation,\n },\n at_mono_ms: now_mono_ms.saturating_add(100),\n });\n self.cells.insert(id.to_string(), cell);\n return;\n }\n _ => {\n effects.push(Effect::ScheduleTimer {\n timer: Timer::CellAlarm {\n cell: id.to_string(),\n generation,\n },\n at_mono_ms: now_mono_ms.saturating_add(100),\n });\n self.cells.insert(id.to_string(), cell);\n return;\n }\n };\n let op = self.cell_op(id);\n cell.alarm = Some(AlarmState::Firing {\n op,\n at_ms,\n generation,\n covered,\n observed: None,\n });\n cell.alarm_wake = false;\n self.cells.insert(id.to_string(), cell);\n effects.push(Effect::FireAlarm {\n op,\n cell: id.to_string(),\n epoch,\n scheduled_ms: at_ms,\n });\n }\n\n /// Settle a firing alarm, once the commit that consumed it is durable.\n ///\n /// The consuming commit is a write like any other, so it takes the same\n /// output gate a request's write takes. Holding the alarm behind that gate\n /// is what lets the core order the wake-entry delete safely: the delete\n /// leaves only from the far side of a proven `DurableReached`, and every\n /// response that can reveal the same commit trails the same gate through\n /// `read_output`. The gate is keyed by the cell and epoch the firing was\n /// dispatched against, because an activity fold can supersede the firing\n /// state while the write is still unproven.\n ///\n /// The gate calls back here with `position: None` once it settles, so the\n /// second pass runs the ordinary settlement below and cannot re-gate.\n ///\n /// `(fired, epoch)` is where the firing was dispatched, carried as a pair\n /// because neither half means anything without the other.\n fn alarm_finished(\n &mut self,\n op: OpId,\n (fired, epoch): (CellId, Epoch),\n now_ms: u64,\n now_mono_ms: u64,\n result: Result<(wake::AlarmSnapshot, bool, Option), Failure>,\n effects: &mut Vec,\n ) {\n if let Ok((snapshot, covered, Some(position))) = result {\n if matches!(\n self.cells.get(&fired).map(|cell| &cell.phase),\n Some(Phase::Resident { epoch: current }) if *current == epoch\n ) {\n let gate = self.op();\n self.barriers.insert(\n gate,\n Barrier {\n owner: GateOwner::Alarm {\n alarm: op,\n snapshot,\n covered,\n },\n cell: fired.clone(),\n epoch,\n position,\n followers: Vec::new(),\n },\n );\n effects.push(Effect::AwaitDurable {\n op: gate,\n cell: fired,\n epoch,\n position,\n });\n return;\n }\n // The cell no longer runs the epoch this write belongs to, so the\n // reset or fence that took it has already refused to acknowledge\n // the write. Settle the alarm below; a superseded op falls out of\n // `take_cell_op`.\n }\n let result = result.map(|(alarm, covered, _)| (alarm, covered));\n let Some(id) = self.take_cell_op(op, |cell| {\n matches!(cell.alarm, Some(AlarmState::Firing { op: current, .. }) if current == op)\n }) else {\n return;\n };\n let Some(mut cell) = self.cells.remove(&id) else {\n return;\n };\n let Some(AlarmState::Firing {\n at_ms,\n generation,\n covered,\n observed,\n ..\n }) = cell.alarm\n else {\n self.cells.insert(id, cell);\n return;\n };\n match result {\n Ok((alarm, covered)) => {\n // Leave the fired alarm in place for `alarm_observed` to\n // replace: it assigns unconditionally, and it compares against\n // what was there to decide whether the bucket entry has to\n // follow. Clearing it here first makes a consumed alarm look\n // like it was never armed, and the entry is never deleted.\n self.cells.insert(id.clone(), cell);\n let (alarm, covered) = observed.unwrap_or((alarm, covered));\n self.alarm_observed(\n &id,\n AlarmObservation {\n alarm,\n covered,\n firing_settled: true,\n },\n now_ms,\n now_mono_ms,\n effects,\n );\n }\n Err(_) => {\n cell.alarm = Some(AlarmState::Armed {\n at_ms,\n generation,\n covered,\n });\n self.cells.insert(id.clone(), cell);\n effects.push(Effect::ScheduleTimer {\n timer: Timer::CellAlarm {\n cell: id,\n generation,\n },\n at_mono_ms: now_mono_ms.saturating_add(500),\n });\n }\n }\n }\n\n fn fence_node(&mut self, reason: HaltReason, effects: &mut Vec) {\n self.node_authority = NodeAuthority::Fenced;\n self.fence(effects);\n effects.push(Effect::Halt { code: 3, reason });\n }\n\n fn request(\n &mut self,\n request: RequestId,\n id: CellId,\n capacity_handoff: bool,\n ownership_handoff: bool,\n effects: &mut Vec,\n ) {\n if self.request_cells.contains_key(&request) {\n return;\n }\n if self.fenced {\n effects.push(Effect::Complete {\n request,\n result: Err(RequestError::NodeFenced),\n });\n return;\n }\n if capacity_handoff {\n self.capacity_requests.insert(request);\n }\n if ownership_handoff {\n self.handoff_requests.insert(request);\n }\n if self.node_authoritative() {\n self.request_authorized(request, id, effects);\n return;\n }\n effects.push(Effect::Complete {\n request,\n result: Err(RequestError::NodeUnavailable),\n });\n }\n\n /// Route one event from an already-registered WebSocket. A regular or\n /// outbound socket keeps its resident runtime alive, so a shutdown drain\n /// must let that socket finish its message and close handlers locally.\n /// Ordinary requests still wait for the successor once the cell quiesces.\n fn websocket_request(\n &mut self,\n request: RequestId,\n id: CellId,\n websocket: WebSocketId,\n effects: &mut Vec,\n ) {\n let held_resident = self.node_authoritative()\n && self.cells.get(&id).is_some_and(|cell| {\n matches!(cell.phase, Phase::Resident { .. })\n && cell.websockets.contains_key(&websocket)\n });\n if held_resident {\n self.complete_request(&id, request, Ok(Route::Local), effects);\n } else {\n // A hibernatable socket can outlive residency. It follows the\n // ordinary route so it can reactivate here or reach a successor.\n self.request(request, id, false, false, effects);\n }\n }\n\n /// Admit an output gate on a resident cell even while it quiesces.\n ///\n /// A swap quiesces a cell so new traffic queues for the restarted runtime,\n /// and it waits for admitted work to finish. A gate is part of that work:\n /// the handler holding the effect cannot finish until the gate answers.\n /// Queued like new traffic, the gate waited for the swap, the swap waited\n /// for the handler, and only the swap's maximum age broke the cycle by\n /// cancelling the handler (denoland/celld#255). The ticket's epoch check\n /// still refuses a gate whose write a restart discarded.\n ///\n /// The cell must have the work that raised the gate: an active request\n /// that is not itself an admitted gate, or a firing alarm. Otherwise the\n /// gate takes the ordinary route. A drain cancels the cell's work instead\n /// of waiting for it, so it has no cycle to break.\n ///\n /// A forced swap admits no further gate. It cancels the cell's activity\n /// once, from a snapshot of the active requests, so a gate admitted after\n /// that snapshot is work nothing cancels, and the swap's maximum age would\n /// no longer bound it. The cancelled handler fails anyway, so its queued\n /// gate costs nothing.\n fn gate_request(&mut self, request: RequestId, id: CellId, effects: &mut Vec) {\n let admitted = self.node_authoritative()\n && !self.draining\n && self.cells.get(&id).is_some_and(|cell| {\n matches!(cell.phase, Phase::Resident { .. })\n && !cell.swap_cancelled\n && (self.runs_gate_source(&id)\n || matches!(cell.alarm, Some(AlarmState::Firing { .. })))\n });\n if admitted {\n self.complete_request(&id, request, Ok(Route::Local), effects);\n self.admitted_gates.insert(request);\n *self.admitted_gate_cells.entry(id).or_insert(0) += 1;\n } else {\n self.request(request, id, false, false, effects);\n }\n }\n\n fn request_authorized(&mut self, request: RequestId, id: CellId, effects: &mut Vec) {\n let mut cell = self.cells.remove(&id).unwrap_or_default();\n match &cell.phase {\n Phase::Resident { .. } if !cell.quiescing => {\n Self::record_local_request(&mut cell, request, self.now_mono_ms);\n self.complete_request(&id, request, Ok(Route::Local), effects)\n }\n Phase::EnsuringDurability { epoch, .. } if !self.draining => {\n // The runtime is still published, so a new request wins the\n // race with voluntary eviction. Retiring the operation makes\n // its eventual durability completion harmless. The permit\n // taken at nomination comes back with the rescue: leaked, it\n // counts against `max_evictions` forever and eventually\n // stands every future eviction down.\n let epoch = *epoch;\n self.cancel_eviction(&id, &mut cell, epoch, EvictCancellation::Activity, effects);\n Self::record_local_request(&mut cell, request, self.now_mono_ms);\n self.complete_request(&id, request, Ok(Route::Local), effects);\n }\n Phase::Remote {\n node,\n addr,\n epoch,\n peer_protocol,\n cache,\n } if cache.reusable(self.now_ms, self.now_mono_ms) => effects.push(Effect::Complete {\n request,\n result: Ok(Route::Remote {\n node: node.clone(),\n addr: addr.clone(),\n epoch: *epoch,\n peer_protocol: *peer_protocol,\n }),\n }),\n Phase::Fenced => effects.push(Effect::Complete {\n request,\n result: Err(RequestError::NodeFenced),\n }),\n Phase::Starting { op, .. } if self.timed_out_runtime_starts.contains(op) => effects\n .push(Effect::Complete {\n request,\n result: Err(RequestError::RuntimeFailed),\n }),\n Phase::Inactive if self.draining => effects.push(Effect::Complete {\n request,\n result: Err(RequestError::NodeUnavailable),\n }),\n Phase::Cleaning {\n op,\n cause: StopCause::Evict { .. },\n ..\n } if !self.draining => {\n // The eviction already left `EnsuringDurability`, so the\n // rescue above cannot reach it, and the runtime is gone. Ask\n // the shell to abandon the stop anyway: it restarts the cell\n // on the same open database at the same epoch, which is what\n // `runtime_stop_failed` does for a stop that ran out of time.\n // Without this the request waits out the whole stop and then\n // a cold route at epoch+1, which is the cost #801 reports.\n //\n // Only the first waiter asks. The signal is one bit, and a\n // burst of requests on one cell must not push one effect\n // each. A draining node has nowhere to put the cell back, so\n // it does not ask at all.\n let op = *op;\n let first = cell.requests.is_empty();\n cell.requests.insert(request);\n self.request_cells.insert(request, id.clone());\n if first {\n effects.push(Effect::AbandonEvictionStop {\n op,\n cell: id.clone(),\n });\n }\n }\n Phase::Dormant { .. } if self.handoff_requests.contains(&request) => {\n // The ownership-only handoff is complete already. Keep the\n // synthetic request pending just long enough for the executor\n // to observe `accepted_epoch`; its cancellation removes it.\n cell.requests.insert(request);\n self.request_cells.insert(request, id.clone());\n }\n phase => {\n cell.requests.insert(request);\n self.request_cells.insert(request, id.clone());\n match phase {\n Phase::Inactive | Phase::Remote { .. } => {\n if matches!(phase, Phase::Remote { cache, .. } if !matches!(cache, RemoteCache::Uncached))\n {\n self.activity.remote_route_refreshes =\n self.activity.remote_route_refreshes.saturating_add(1);\n }\n // A paused owner keeps its connections open, so no\n // failed send invalidates this route. Fresh demand\n // must re-resolve after expiry without replaying\n // requests already sent there. A draining ingress can\n // still resolve a live remote owner during handoff;\n // acquisition admission refuses new local ownership.\n self.admit_or_queue_activation(\n &id,\n &mut cell,\n ColdStart::ReadOwner,\n effects,\n );\n }\n Phase::Dormant { .. } if !self.hands_off(&cell) => {\n // A wake always claims a fresh epoch. The preserved\n // snapshot remains reusable as the previous epoch's\n // baseline after the claim succeeds.\n self.admit_or_queue_activation(\n &id,\n &mut cell,\n ColdStart::ReadOwner,\n effects,\n );\n }\n _ => {}\n }\n }\n }\n self.cells.insert(id, cell);\n }\n\n fn wake_hint(\n &mut self,\n id: CellId,\n entry_ms: i64,\n scope: WakeHintScope,\n effects: &mut Vec,\n ) {\n if self.fenced {\n return;\n }\n if self.node_authoritative() {\n self.wake_hint_authorized(id, entry_ms, scope, effects);\n }\n }\n\n fn wake_hint_authorized(\n &mut self,\n id: CellId,\n _entry_ms: i64,\n scope: WakeHintScope,\n effects: &mut Vec,\n ) {\n if self.draining {\n return;\n }\n // An `Owned` hint is decided before the map is touched.\n // `unwrap_or_default` below creates an `Inactive` entry for a cell\n // this node has never seen, which is what a `Fleet` probe needs and\n // exactly wrong here: a non-elected node sees the whole fleet's due\n // set on every tick, and each entry would leave a cell behind.\n if scope == WakeHintScope::Owned\n && !matches!(\n self.cells.get(&id).map(|cell| &cell.phase),\n Some(Phase::Dormant { .. })\n )\n {\n return;\n }\n let mut cell = self.cells.remove(&id).unwrap_or_default();\n cell.alarm_wake = true;\n match cell.phase {\n Phase::Inactive | Phase::Remote { .. } | Phase::Dormant { .. } => {\n self.admit_or_queue_activation(&id, &mut cell, ColdStart::ReadOwner, effects);\n }\n _ => {}\n }\n self.cells.insert(id, cell);\n }\n\n fn cancel(&mut self, request: RequestId, effects: &mut Vec) {\n // A cancelled route can already be Local-complete: the reply raced\n // the caller's death and was never consumed, so no activity guard\n // exists and no `ActivityFinished` will ever arrive. Release the\n // activity here, or the leaked entry pins its cell forever and a\n // later quiesce can never reach eviction. Every consumed route\n // disarms its cancel guard before this event can be sent, and a\n // guard that did report finished has already left `active_requests`,\n // so this release cannot double-fire.\n if self.active_requests.contains_key(&request) {\n self.activity_finished(request, effects);\n }\n self.capacity_requests.remove(&request);\n self.handoff_requests.remove(&request);\n let Some(id) = self.request_cells.remove(&request) else {\n return;\n };\n let Some(cell) = self.cells.get_mut(&id) else {\n return;\n };\n cell.requests.remove(&request);\n // A clean-generation resume is node demand, not caller demand. If a\n // joined caller cancels, keep that resume queued so readiness can\n // still observe its terminal outcome.\n if cell.requests.is_empty()\n && !cell.alarm_wake\n && !cell.resume_demand\n && cell.phase == Phase::WaitingActivation\n {\n let start = cell.waiting_activation.take();\n set_phase(\n &mut self.occupied,\n cell,\n match start {\n Some(ColdStart::Restore(spec)) => Phase::Dormant { epoch: spec.epoch },\n _ => Phase::Inactive,\n },\n );\n self.activation_waiters.retain(|queued| queued != &id);\n } else if cell.requests.is_empty()\n && !cell.alarm_wake\n && !cell.resume_demand\n && cell.phase == Phase::WaitingCapacity\n {\n let activation = cell.waiting_for.take();\n set_phase(\n &mut self.occupied,\n cell,\n match activation {\n Some(Activation::Restore(spec)) => Phase::Dormant { epoch: spec.epoch },\n _ => Phase::Inactive,\n },\n );\n self.capacity_waiters.retain(|queued| queued != &id);\n }\n }\n\n /// Retire cold work that cannot contribute to a clean local reload.\n ///\n /// Published runtimes remain untouched. The shell drains their active\n /// turns before it snapshots the resident inventory. Cold restore and\n /// recovery phases have no runtime to preserve, and they can legitimately\n /// outlive every caller, so waiting for it would make one cold cell disable\n /// local reuse for the complete node. Retire those phases and fail their\n /// callers. A later completion is ignored because its operation is no\n /// longer current.\n fn begin_preserve(&mut self, effects: &mut Vec) {\n self.activation_waiters.clear();\n self.capacity_waiters.clear();\n\n let ids: Vec = self.cells.keys().cloned().collect();\n for id in ids {\n let mut cell = self.cells.remove(&id).expect(\"id came from map\");\n let fallback = match &cell.phase {\n Phase::WaitingActivation => match cell.waiting_activation.as_ref() {\n Some(ColdStart::Restore(spec)) => Some(Phase::Dormant { epoch: spec.epoch }),\n _ => Some(Phase::Inactive),\n },\n Phase::ReadingOwner { .. }\n | Phase::ReadingNodeLease { .. }\n | Phase::RecoveringOwnerLog { .. }\n // Backoff retains its callers after returning the activation\n // permit. Retire its phase too, or the retry timer restarts\n // cold work while the shell prepares the local reload.\n | Phase::WaitingOwnerLogRecovery { .. }\n | Phase::ReadingCapacity { .. }\n | Phase::Acquiring { .. }\n | Phase::ReconcilingAcquire { .. } => Some(Phase::Inactive),\n Phase::WaitingCapacity => match cell.waiting_for.as_ref() {\n Some(Activation::Restore(spec)) => Some(Phase::Dormant { epoch: spec.epoch }),\n _ => Some(Phase::Inactive),\n },\n Phase::Restoring { spec, .. } => Some(Phase::Dormant { epoch: spec.epoch }),\n _ => None,\n };\n if let Some(fallback) = fallback {\n if let Some(stale) = phase_op(&cell.phase) {\n self.cell_ops.remove(&stale);\n }\n self.finish_requests(&id, &mut cell, Err(RequestError::NodeFenced), effects);\n set_phase(&mut self.occupied, &mut cell, fallback);\n // A later cold route must not inherit failures from the\n // recovery episode that preserve just retired.\n cell.owner_log_recovery_attempts = 0;\n cell.waiting_for = None;\n cell.waiting_activation = None;\n cell.alarm_wake = false;\n cell.resume_demand = false;\n self.resuming_cells.remove(&id);\n }\n self.cells.insert(id, cell);\n }\n if self.resuming_cells.is_empty() {\n self.resuming = false;\n }\n self.activation_permits.retain(|id| {\n self.cells\n .get(id)\n .is_some_and(|cell| phase_holds_activation(&cell.phase))\n });\n self.capacity_requests\n .retain(|request| self.request_cells.contains_key(request));\n self.handoff_requests\n .retain(|request| self.request_cells.contains_key(request));\n }\n\n fn begin_capacity_lookup(\n &mut self,\n id: &str,\n cell: &mut Cell,\n claim: Claim,\n effects: &mut Vec,\n ) {\n let op = self.cell_op(id);\n set_phase(\n &mut self.occupied,\n cell,\n Phase::ReadingCapacity {\n op,\n claim: claim.clone(),\n issued_mono_ms: self.now_mono_ms,\n },\n );\n effects.push(Effect::ReadCapacityPeers {\n op,\n cell: id.to_string(),\n });\n }\n\n /// Place a genuinely unowned cell. Ordinary ingress may look for fleet\n /// capacity before waiting locally; a capacity handoff must either reserve\n /// a real local slot now or explicitly refuse so its caller can traverse.\n fn place_unowned(\n &mut self,\n id: &str,\n cell: &mut Cell,\n claim: Claim,\n effects: &mut Vec,\n ) {\n if self.refuse_new_ownership_during_drain(id, cell, claim.prior.as_deref(), effects) {\n return;\n }\n if self.has_capacity() {\n self.activate_or_wait(\n id,\n cell,\n Activation::Claim(claim),\n WhenFull::ShedToAdmit,\n effects,\n );\n return;\n }\n\n // An ownership-only handoff parks dormant after its claim and\n // holds no residency slot, so the cap is no reason to refuse it.\n // Refusing here starved a whole drain when every peer sat at its\n // cap: each stuck cell held a donor release permit, the donor\n // retried forever, its no-progress deadline fired, and SIGKILL\n // left the fleet's records naming a dead node. The claim's\n // Acquiring phase carries a dormant mark and occupies no\n // residency slot, so the ceiling\n // invariant holds throughout. A cell that live demand also\n // reached still refuses below, because its claim would restore a\n // runtime this node has no slot for.\n if !cell.requests.is_empty()\n && cell\n .requests\n .iter()\n .all(|request| self.handoff_requests.contains(request))\n {\n self.begin_activation(id, cell, Activation::Claim(claim), effects);\n return;\n }\n\n let handoffs: Vec = cell\n .requests\n .iter()\n .copied()\n .filter(|request| self.capacity_requests.contains(request))\n .collect();\n for request in handoffs {\n cell.requests.remove(&request);\n self.request_cells.remove(&request);\n self.complete_request(id, request, Err(RequestError::CapacityExhausted), effects);\n }\n\n if cell.requests.is_empty() || !self.config.require_node_lease {\n // Alarm wakes cannot be proxied as an HTTP capacity handoff. They\n // retain the old local wait semantics until alarm dispatch itself\n // has a fleet transport effect. Lease-disabled mode likewise has\n // no authoritative fleet membership to enumerate.\n self.activate_or_wait(\n id,\n cell,\n Activation::Claim(claim),\n WhenFull::ShedToAdmit,\n effects,\n );\n } else {\n self.begin_capacity_lookup(id, cell, claim, effects);\n }\n }\n\n fn owner_read(\n &mut self,\n op: OpId,\n now_ms: u64,\n result: Result, Failure>,\n effects: &mut Vec,\n ) {\n let Some(id) = self.take_cell_op(op, |cell| {\n matches!(\n cell.phase,\n Phase::ReadingOwner { op: current }\n | Phase::ReconcilingAcquire { op: current, .. }\n if current == op\n )\n }) else {\n return;\n };\n let mut cell = self.cells.remove(&id).expect(\"cell found above\");\n if std::mem::take(&mut cell.confirming) {\n self.settle_confirmation(&id);\n if cell.requests.is_empty() && !cell.alarm_wake {\n // Nobody wants the cell: record what the store says and\n // stop. A record that still names this node leaves the cell\n // dormant at that epoch, exactly as an idle eviction does,\n // and the next request wakes it through the usual claim.\n let phase = match result {\n Ok(Some(record)) if record.node.as_deref() == Some(self.node.as_str()) => {\n Phase::Dormant {\n epoch: record.epoch,\n }\n }\n _ => Phase::Inactive,\n };\n set_phase(&mut self.occupied, &mut cell, phase);\n self.cells.insert(id, cell);\n return;\n }\n }\n let reconciling_claim = match &cell.phase {\n Phase::ReconcilingAcquire { claim, .. } => Some(claim.clone()),\n _ => None,\n };\n // A reconcile re-reads and then acquires again: one attempt\n // continuing, not a new one. The count has to survive the round trip\n // or the bound below can never be reached.\n let reconciles = reconciling_claim\n .as_ref()\n .map_or(0, |claim| claim.reconciles);\n match result {\n Ok(Some(record))\n if reconciling_claim.as_ref().is_some_and(|claim| {\n record.node.as_deref() == Some(self.node.as_str())\n && record.epoch == claim.epoch\n }) =>\n {\n let spec = RestoreSpec {\n epoch: record.epoch,\n fresh: matches!(\n reconciling_claim.as_ref().map(|claim| &claim.guard),\n Some(CasGuard::Absent)\n ),\n took_over: reconciling_claim\n .as_ref()\n .is_some_and(|claim| claim.takeover),\n resume_local: false,\n // The ambiguous acquire applied after all, so the prior\n // owner it displaced is the one the claim carried — the\n // takeover interlock must see it exactly as if the CAS\n // response had confirmed directly.\n prior: reconciling_claim\n .as_ref()\n .and_then(|claim| claim.prior.clone()),\n };\n self.record_acquisition(&spec);\n let ownership_only = !cell.requests.is_empty()\n && cell\n .requests\n .iter()\n .all(|request| self.handoff_requests.contains(request));\n if ownership_only {\n set_phase(\n &mut self.occupied,\n &mut cell,\n Phase::Dormant { epoch: spec.epoch },\n );\n } else {\n self.activate_or_wait(\n &id,\n &mut cell,\n Activation::Restore(spec),\n WhenFull::ShedToAdmit,\n effects,\n );\n }\n }\n Ok(Some(record)) if record.node.as_deref() == Some(self.node.as_str()) => {\n let Some(epoch) = self.next_writer_epoch(&id, &mut cell, record.epoch, effects)\n else {\n self.cells.insert(id, cell);\n return;\n };\n let prior = record.node.clone();\n self.activate_or_wait(\n &id,\n &mut cell,\n Activation::Claim(Claim {\n guard: CasGuard::Match(record.etag),\n epoch,\n takeover: false,\n prior,\n reconciles,\n }),\n WhenFull::ShedToAdmit,\n effects,\n );\n }\n Ok(Some(record)) if record.node.is_none() => {\n let Some(epoch) = self.next_writer_epoch(&id, &mut cell, record.epoch, effects)\n else {\n self.cells.insert(id, cell);\n return;\n };\n self.place_unowned(\n &id,\n &mut cell,\n Claim {\n guard: CasGuard::Match(record.etag),\n epoch,\n takeover: true,\n prior: None,\n reconciles,\n },\n effects,\n );\n }\n Ok(Some(record)) => {\n let owner = record.node.clone().expect(\"foreign owner checked above\");\n let cached = self\n .node_lease_cache\n .get(&owner)\n .filter(|lease| lease.deadline.live(self.now_ms, self.now_mono_ms))\n .cloned();\n if let Some(lease) = cached {\n self.apply_node_lease_result(\n &id,\n &mut cell,\n record,\n now_ms,\n Ok(Some(lease)),\n effects,\n );\n } else {\n self.node_lease_cache.remove(&owner);\n let next = self.cell_op(&id);\n set_phase(\n &mut self.occupied,\n &mut cell,\n Phase::ReadingNodeLease {\n op: next,\n owner: record,\n issued_ms: now_ms,\n issued_mono_ms: self.now_mono_ms,\n },\n );\n effects.push(Effect::ReadNodeLease {\n op: next,\n cell: id.clone(),\n owner,\n });\n }\n }\n Ok(None) => {\n self.place_unowned(\n &id,\n &mut cell,\n Claim {\n guard: CasGuard::Absent,\n epoch: 1,\n takeover: false,\n prior: None,\n reconciles,\n },\n effects,\n );\n }\n Err(_) => {\n set_phase(&mut self.occupied, &mut cell, Phase::Inactive);\n self.finish_requests(&id, &mut cell, Err(RequestError::ResolveFailed), effects);\n }\n }\n self.cells.insert(id, cell);\n if reconciling_claim.is_some() {\n self.pump_capacity(effects);\n }\n }\n\n fn node_lease_read(\n &mut self,\n op: OpId,\n now_ms: u64,\n result: Result, Failure>,\n effects: &mut Vec,\n ) {\n let Some(id) = self.take_cell_op(op, |cell| {\n matches!(cell.phase, Phase::ReadingNodeLease { op: current, .. } if current == op)\n }) else {\n return;\n };\n let mut cell = self.cells.remove(&id).expect(\"cell found above\");\n let Phase::ReadingNodeLease {\n owner: record,\n issued_ms,\n issued_mono_ms,\n ..\n } = &cell.phase\n else {\n unreachable!()\n };\n let record = record.clone();\n let issued_ms = *issued_ms;\n let issued_mono_ms = *issued_mono_ms;\n let result = result.map(|lease| {\n lease.map(|record| CachedNodeLease {\n deadline: RemoteLeaseDeadline::observed(\n record.expires_ms,\n issued_mono_ms,\n now_ms.max(self.now_ms),\n ),\n record,\n })\n });\n if let Ok(Some(lease)) = &result {\n if record.node.as_deref() == Some(lease.record.node.as_str())\n && lease.deadline.live(self.now_ms, self.now_mono_ms)\n && !lease.record.addr.is_empty()\n {\n self.node_lease_cache\n .insert(lease.record.node.clone(), lease.clone());\n }\n }\n self.apply_node_lease_result(&id, &mut cell, record, issued_ms, result, effects);\n self.cells.insert(id, cell);\n }\n\n fn capacity_peers_read(\n &mut self,\n op: OpId,\n now_ms: u64,\n result: Result, Failure>,\n effects: &mut Vec,\n ) {\n let Some(id) = self.take_cell_op(op, |cell| {\n matches!(cell.phase, Phase::ReadingCapacity { op: current, .. } if current == op)\n }) else {\n return;\n };\n let mut cell = self.cells.remove(&id).expect(\"cell found above\");\n let Phase::ReadingCapacity {\n claim,\n issued_mono_ms,\n ..\n } = &cell.phase\n else {\n unreachable!()\n };\n let claim = claim.clone();\n let issued_mono_ms = *issued_mono_ms;\n let peers = match result {\n Ok(peers) => peers,\n Err(_) => {\n // The lookup did not answer, so the fleet's room is unknown.\n // The sample it reads is shared and periodic, so a claim race\n // or a refresh in flight returns an error that carries no\n // load information at all. Shedding on it cut a live resident\n // to admit a placement a peer had room for, and the released\n // record then stayed unowned with its epoch retained until\n // some node read it again.\n self.activate_or_wait(\n &id,\n &mut cell,\n Activation::Claim(claim),\n WhenFull::HoldResidents,\n effects,\n );\n self.cells.insert(id, cell);\n return;\n }\n };\n\n // A newer load sample supersedes both reservations made against and\n // refusals of the prior sample. Equal samples deliberately retain\n // both, which is what makes concurrent read completions compose.\n for peer in &peers {\n let prior = self.capacity_samples.get(&peer.node).copied().unwrap_or(0);\n if peer.sampled_ms > prior {\n self.capacity_samples\n .insert(peer.node.clone(), peer.sampled_ms);\n self.capacity_reservations.remove(&peer.node);\n self.capacity_rejections.remove(&peer.node);\n }\n }\n\n let selected = peers\n .into_iter()\n .filter(|peer| {\n peer.node != self.node\n && peer.expires_ms > now_ms\n && !peer.addr.is_empty()\n && peer.peer_protocol == self.config.peer_protocol\n && peer.sampled_ms != 0\n && !peer.pressured\n && self\n .capacity_rejections\n .get(&peer.node)\n .is_none_or(|sample| peer.sampled_ms > *sample)\n })\n .map(|peer| {\n let projected = peer.resident_cells.saturating_add(\n self.capacity_reservations\n .get(&peer.node)\n .copied()\n .unwrap_or(0),\n );\n (peer, projected)\n })\n .min_by_key(|(peer, projected)| {\n // The third key is the memory the peer holds, which is what\n // the peer decides its own pressure on. A peer from before that\n // field existed reports nothing, and its resident set size is\n // the only number available -- which reads as more loaded than\n // it is, and is the conservative direction for a tiebreak.\n (\n *projected,\n peer.host_websockets,\n peer.in_use_bytes.unwrap_or(peer.rss_bytes),\n peer.node.clone(),\n )\n });\n\n if let Some((peer, _)) = selected {\n *self\n .capacity_reservations\n .entry(peer.node.clone())\n .or_default() += 1;\n set_phase(\n &mut self.occupied,\n &mut cell,\n Phase::Remote {\n node: peer.node.clone(),\n addr: peer.addr.clone(),\n epoch: 0,\n peer_protocol: peer.peer_protocol,\n cache: RemoteCache::CapacitySample {\n sampled_ms: peer.sampled_ms,\n deadline: RemoteLeaseDeadline::observed(\n peer.expires_ms,\n issued_mono_ms,\n now_ms.max(self.now_ms),\n ),\n },\n },\n );\n self.finish_requests(\n &id,\n &mut cell,\n Ok(Route::Remote {\n node: peer.node,\n addr: peer.addr,\n epoch: 0,\n peer_protocol: peer.peer_protocol,\n }),\n effects,\n );\n } else {\n self.activate_or_wait(\n &id,\n &mut cell,\n Activation::Claim(claim),\n WhenFull::ShedToAdmit,\n effects,\n );\n }\n self.cells.insert(id, cell);\n }\n\n fn apply_node_lease_result(\n &mut self,\n id: &str,\n cell: &mut Cell,\n record: OwnerRecord,\n issued_ms: u64,\n result: Result, Failure>,\n effects: &mut Vec,\n ) {\n match result {\n Ok(Some(CachedNodeLease { record: lease, .. }))\n if lease.expires_ms > issued_ms\n && !lease.addr.is_empty()\n && record.node.as_deref() == Some(lease.node.as_str())\n && lease.peer_protocol != self.config.peer_protocol =>\n {\n set_phase(&mut self.occupied, cell, Phase::Inactive);\n self.finish_requests(id, cell, Err(RequestError::PeerIncompatible), effects);\n }\n Ok(Some(CachedNodeLease {\n record: lease,\n deadline,\n })) if lease.expires_ms > issued_ms\n && !lease.addr.is_empty()\n && record.node.as_deref() == Some(lease.node.as_str())\n && lease.peer_protocol == self.config.peer_protocol =>\n {\n let node = lease.node;\n let addr = lease.addr;\n let epoch = record.epoch;\n set_phase(\n &mut self.occupied,\n cell,\n Phase::Remote {\n node: node.clone(),\n addr: addr.clone(),\n epoch,\n peer_protocol: lease.peer_protocol,\n cache: RemoteCache::OwnerLease(deadline),\n },\n );\n self.finish_requests(\n id,\n cell,\n Ok(Route::Remote {\n node,\n addr,\n epoch,\n peer_protocol: lease.peer_protocol,\n }),\n effects,\n );\n }\n Ok(lease) => {\n if self.refuse_new_ownership_during_drain(id, cell, record.node.as_deref(), effects)\n {\n return;\n }\n // The takeover interlock, decided HERE on the lease the\n // core just read (lease-fold): a dead owner whose folded\n // log state is not sealed may hold acked writes only on\n // its followers, and claiming before recovery fixes a\n // bucket state that excludes them. Absence of a lease —\n // or a lease that never opened a log — keeps its meaning\n // as a proof: nothing was acked past the bucket.\n let unrecovered = lease.as_ref().is_some_and(|lease| {\n !matches!(\n lease.record.log_state,\n None | Some(crate::log_tier::LogState::Sealed)\n )\n });\n if unrecovered {\n let owner = record.node.clone().unwrap_or_default();\n let next = self.cell_op(id);\n set_phase(\n &mut self.occupied,\n cell,\n Phase::RecoveringOwnerLog {\n op: next,\n owner: record,\n },\n );\n effects.push(Effect::RecoverNodeLog {\n op: next,\n cell: id.to_string(),\n owner,\n });\n return;\n }\n let Some(epoch) = self.next_writer_epoch(id, cell, record.epoch, effects) else {\n return;\n };\n let prior = record.node.clone();\n self.activate_or_wait(\n id,\n cell,\n Activation::Claim(Claim {\n guard: CasGuard::Match(record.etag),\n epoch,\n takeover: true,\n prior,\n reconciles: 0,\n }),\n WhenFull::ShedToAdmit,\n effects,\n );\n }\n Err(_) => {\n set_phase(&mut self.occupied, cell, Phase::Inactive);\n self.finish_requests(id, cell, Err(RequestError::ResolveFailed), effects);\n }\n }\n }\n\n /// The interlock's completion: recovery sealed (or proved absent)\n /// every session of the dead owner, so the claim the gate deferred\n /// proceeds against the SAME owner record the decision was made on —\n /// the CAS guard still carries its etag, so a record that moved in\n /// the meantime rejects the claim and resolution restarts.\n fn node_log_recovered(\n &mut self,\n op: OpId,\n result: Result<(), Failure>,\n effects: &mut Vec,\n ) {\n let Some(id) = self.take_cell_op(\n op,\n |cell| matches!(cell.phase, Phase::RecoveringOwnerLog { op: p, .. } if p == op),\n ) else {\n return;\n };\n let mut cell = self.cells.remove(&id).expect(\"cell found above\");\n let Phase::RecoveringOwnerLog { owner, .. } = cell.phase.clone() else {\n unreachable!()\n };\n match result {\n Ok(()) => {\n let Some(epoch) = self.next_writer_epoch(&id, &mut cell, owner.epoch, effects)\n else {\n self.cells.insert(id, cell);\n return;\n };\n let prior = owner.node.clone();\n self.activate_or_wait(\n &id,\n &mut cell,\n Activation::Claim(Claim {\n guard: CasGuard::Match(owner.etag),\n epoch,\n takeover: true,\n prior,\n reconciles: 0,\n }),\n WhenFull::ShedToAdmit,\n effects,\n );\n cell.owner_log_recovery_attempts = 0;\n }\n Err(_) => {\n // A recovery fails precisely while the survivors sweep the\n // dead owner's deep log — minutes of work for a large\n // owner. The request waits the sweep out\n // behind a backoff instead of failing, and the permit\n // returns so one dead owner's cells cannot starve the node.\n cell.owner_log_recovery_attempts =\n cell.owner_log_recovery_attempts.saturating_add(1);\n if cell.owner_log_recovery_attempts >= self.config.owner_log_recovery_attempts {\n cell.owner_log_recovery_attempts = 0;\n set_phase(&mut self.occupied, &mut cell, Phase::Inactive);\n self.finish_requests(&id, &mut cell, Err(RequestError::ResolveFailed), effects);\n } else {\n let generation = self.next_timer_generation;\n self.next_timer_generation = self\n .next_timer_generation\n .checked_add(1)\n .expect(\"timer generation exhausted\");\n set_phase(\n &mut self.occupied,\n &mut cell,\n Phase::WaitingOwnerLogRecovery { generation },\n );\n effects.push(Effect::ScheduleTimer {\n timer: Timer::OwnerLogRecoveryRetry {\n cell: id.clone(),\n generation,\n },\n at_mono_ms: self\n .now_mono_ms\n .saturating_add(self.config.owner_log_recovery_backoff_ms),\n });\n }\n }\n }\n self.cells.insert(id, cell);\n self.pump_activations(effects);\n }\n\n /// A backed-off dead-owner recovery re-enters ownership resolution\n /// from the top: the owner record may have moved while the cell\n /// waited, and the admission queue shares permits fairly with every\n /// other cold route.\n fn retry_owner_log_recovery(&mut self, id: &str, generation: u64, effects: &mut Vec) {\n let Some(mut cell) = self.cells.remove(id) else {\n return;\n };\n if cell.phase != (Phase::WaitingOwnerLogRecovery { generation }) {\n self.cells.insert(id.to_string(), cell);\n return;\n }\n if cell.requests.is_empty() && !cell.alarm_wake && !cell.resume_demand {\n // Every waiting request was cancelled during the backoff;\n // nothing demands the cell any more.\n cell.owner_log_recovery_attempts = 0;\n set_phase(&mut self.occupied, &mut cell, Phase::Inactive);\n } else {\n self.admit_or_queue_activation(id, &mut cell, ColdStart::ReadOwner, effects);\n }\n self.cells.insert(id.to_string(), cell);\n }\n\n fn owner_cas_completed(\n &mut self,\n op: OpId,\n result: Result,\n effects: &mut Vec,\n ) {\n let Some(id) = self.take_cell_op(\n op,\n |cell| matches!(cell.phase, Phase::Acquiring { op: current, .. } if current == op),\n ) else {\n return;\n };\n let mut cell = self.cells.remove(&id).expect(\"cell found above\");\n let Phase::Acquiring { claim, dormant, .. } = &cell.phase else {\n unreachable!()\n };\n let dormant = *dormant;\n let claim = claim.clone();\n match result {\n Ok(CasOutcome::Applied) => {\n let spec = RestoreSpec {\n epoch: claim.epoch,\n fresh: matches!(claim.guard, CasGuard::Absent),\n took_over: claim.takeover,\n resume_local: false,\n prior: claim.prior.clone(),\n };\n self.record_acquisition(&spec);\n let ownership_only = !cell.requests.is_empty()\n && cell\n .requests\n .iter()\n .all(|request| self.handoff_requests.contains(request));\n if ownership_only {\n set_phase(\n &mut self.occupied,\n &mut cell,\n Phase::Dormant { epoch: spec.epoch },\n );\n } else if dormant {\n // Live demand joined a dormant-bound claim before the CAS\n // landed. The claim held no residency slot, so the restore\n // must go through the capacity queue rather than start\n // inline on a node that has none.\n self.activate_or_wait(\n &id,\n &mut cell,\n Activation::Restore(spec),\n WhenFull::ShedToAdmit,\n effects,\n );\n } else {\n // An ordinary claim already occupies the slot its restore\n // needs, so the transition stays inline; re-checking\n // capacity here would count the cell against itself.\n let next = self.cell_op(&id);\n set_phase(\n &mut self.occupied,\n &mut cell,\n Phase::Restoring {\n op: next,\n spec: spec.clone(),\n },\n );\n effects.push(Effect::Restore {\n op: next,\n cell: id.clone(),\n spec,\n });\n }\n }\n Ok(CasOutcome::Rejected) => {\n let next = self.cell_op(&id);\n set_phase(\n &mut self.occupied,\n &mut cell,\n Phase::ReadingOwner { op: next },\n );\n effects.push(Effect::ReadOwner {\n op: next,\n cell: id.clone(),\n });\n }\n Err(Failure::Ambiguous) if claim.reconciles < MAX_ACQUIRE_RECONCILES => {\n let next = self.cell_op(&id);\n let claim = Claim {\n reconciles: claim.reconciles + 1,\n ..claim\n };\n set_phase(\n &mut self.occupied,\n &mut cell,\n Phase::ReconcilingAcquire {\n op: next,\n claim,\n dormant,\n },\n );\n effects.push(Effect::ReadOwner {\n op: next,\n cell: id.clone(),\n });\n }\n Err(Failure::Ambiguous) => {\n // Out of reconciles. The claim may or may not have applied, so\n // this cell is left dormant for a later request to resolve\n // from the record rather than guessed at here.\n set_phase(&mut self.occupied, &mut cell, Phase::Inactive);\n self.finish_requests(&id, &mut cell, Err(RequestError::AcquireFailed), effects);\n }\n Err(Failure::Definite) => {\n set_phase(&mut self.occupied, &mut cell, Phase::Inactive);\n self.finish_requests(&id, &mut cell, Err(RequestError::AcquireFailed), effects);\n }\n }\n self.cells.insert(id, cell);\n self.pump_capacity(effects);\n }\n\n fn restore_completed(\n &mut self,\n op: OpId,\n result: Result,\n effects: &mut Vec,\n ) {\n let Some(id) = self.take_cell_op(\n op,\n |cell| matches!(cell.phase, Phase::Restoring { op: current, .. } if current == op),\n ) else {\n return;\n };\n let mut cell = self.cells.remove(&id).expect(\"cell found above\");\n let Phase::Restoring { spec, .. } = &cell.phase else {\n unreachable!()\n };\n let epoch = spec.epoch;\n let mut resume_failed = false;\n match result {\n Ok(outcome) => {\n if outcome.restored {\n self.activity.restored = self.activity.restored.saturating_add(1);\n }\n // Seed the mirror from the durable truth the restore loaded,\n // before the isolate opens the same database and long before\n // it would re-arm anything.\n if let Some(alarm) = outcome.alarm.filter(|alarm| alarm.at_ms >= 0) {\n let generation = self.next_timer_generation;\n self.next_timer_generation = self\n .next_timer_generation\n .checked_add(1)\n .expect(\"timer generation exhausted\");\n let (now_ms, now_mono_ms) = (self.now_ms, self.now_mono_ms);\n self.schedule_alarm_timer(\n &id,\n generation,\n alarm.at_ms,\n now_ms,\n now_mono_ms,\n effects,\n );\n cell.alarm = Some(AlarmState::Armed {\n at_ms: alarm.at_ms,\n generation,\n covered: alarm.covered,\n });\n }\n let next = self.cell_op(&id);\n set_phase(\n &mut self.occupied,\n &mut cell,\n Phase::Starting { op: next, epoch },\n );\n effects.push(Effect::StartRuntime {\n op: next,\n cell: id.clone(),\n epoch,\n });\n }\n Err(_) => {\n set_phase(&mut self.occupied, &mut cell, Phase::Dormant { epoch });\n self.finish_requests(&id, &mut cell, Err(RequestError::RestoreFailed), effects);\n resume_failed = cell.resume_demand;\n }\n }\n self.cells.insert(id.clone(), cell);\n if resume_failed {\n self.settle_local_resume(&id);\n }\n self.pump_capacity(effects);\n }\n\n fn runtime_started(\n &mut self,\n op: OpId,\n isolate: Option,\n generation: u64,\n result: Result<(), Failure>,\n effects: &mut Vec,\n ) {\n let Some(id) = self.take_cell_op(\n op,\n |cell| matches!(cell.phase, Phase::Starting { op: current, .. } if current == op),\n ) else {\n self.compensate_retired_runtime(op, result, effects);\n return;\n };\n self.timed_out_runtime_starts.remove(&op);\n let mut cell = self.cells.remove(&id).expect(\"cell found above\");\n let Phase::Starting { epoch, .. } = cell.phase else {\n unreachable!()\n };\n let mut resume_failed = false;\n match result {\n Ok(()) => {\n cell.isolate = isolate;\n cell.generation = generation;\n let next = self.cell_op(&id);\n set_phase(\n &mut self.occupied,\n &mut cell,\n Phase::Publishing { op: next, epoch },\n );\n effects.push(Effect::Publish {\n op: next,\n cell: id.clone(),\n epoch,\n });\n }\n Err(_) => {\n set_phase(&mut self.occupied, &mut cell, Phase::Dormant { epoch });\n self.finish_requests(&id, &mut cell, Err(RequestError::RuntimeFailed), effects);\n resume_failed = cell.resume_demand;\n }\n }\n self.cells.insert(id.clone(), cell);\n if resume_failed {\n self.settle_local_resume(&id);\n }\n self.pump_capacity(effects);\n }\n\n /// Fail the callers waiting on a runtime start without retiring the start.\n ///\n /// The shell cannot cancel a V8 startup turn. Treating this deadline as a\n /// definite failure lets the next request start a second runtime while the\n /// first task can still insert its handle. The late first completion then\n /// becomes an untracked runtime which every later request reports as an\n /// existing duplicate. Keep the operation indexed instead. Its real\n /// completion publishes the runtime or returns the cell to `Dormant`.\n fn runtime_start_timed_out(&mut self, op: OpId, effects: &mut Vec) {\n let Some(id) = self.cell_ops.get(&op).cloned() else {\n return;\n };\n let Some(mut cell) = self.cells.remove(&id) else {\n return;\n };\n if !matches!(cell.phase, Phase::Starting { op: current, .. } if current == op) {\n self.cells.insert(id, cell);\n return;\n }\n self.timed_out_runtime_starts.insert(op);\n self.finish_requests(&id, &mut cell, Err(RequestError::RuntimeFailed), effects);\n self.cells.insert(id, cell);\n }\n\n fn published(&mut self, op: OpId, result: Result<(), Failure>, effects: &mut Vec) {\n let Some(id) = self.take_cell_op(\n op,\n |cell| matches!(cell.phase, Phase::Publishing { op: current, .. } if current == op),\n ) else {\n self.compensate_retired_runtime(op, result, effects);\n return;\n };\n let mut cell = self.cells.remove(&id).expect(\"cell found above\");\n let Phase::Publishing { epoch, .. } = cell.phase else {\n unreachable!()\n };\n let completing_resume = cell.resume_demand;\n match result {\n Ok(()) => {\n set_phase(&mut self.occupied, &mut cell, Phase::Resident { epoch });\n // A swap ends here: the cell is resident again, on the\n // generation `runtime_started` recorded.\n cell.swapping = false;\n cell.swap_cancelled = false;\n self.finish_requests(&id, &mut cell, Ok(Route::Local), effects);\n }\n Err(_) => {\n let next = self.cell_op(&id);\n set_phase(\n &mut self.occupied,\n &mut cell,\n Phase::Cleaning {\n op: next,\n epoch,\n cause: StopCause::Cleanup,\n },\n );\n effects.push(Effect::StopRuntime {\n op: next,\n cell: id.clone(),\n epoch,\n cause: StopCause::Cleanup,\n bounded: false,\n });\n }\n }\n self.cells.insert(id.clone(), cell);\n if completing_resume {\n self.settle_local_resume(&id);\n }\n }\n\n /// A release either published the cell as unowned or did not. Ownership\n /// is the bucket's answer, not this node's, so a rejected or failed write\n /// simply leaves the record naming this node -- correct, if less useful,\n /// and the next eviction gets another chance.\n fn owner_released(\n &mut self,\n op: OpId,\n result: Result,\n effects: &mut Vec,\n ) {\n let Some(id) = self.take_cell_op(op, |cell| cell.releasing == Some(op)) else {\n return;\n };\n let released_epoch = match self.cells.get(&id).map(|cell| &cell.phase) {\n Some(Phase::Dormant { epoch }) => Some(*epoch),\n _ => None,\n };\n // Only a cell still sitting where the eviction left it may be\n // forgotten. Anything else means it was wanted again while the write\n // was in flight, and that claim outranks a release decided earlier.\n // On a draining node a rejection settles the cell too: the record no\n // longer names this node at this epoch, so nothing here blocks a\n // successor, and the pump must not re-read a record it can never\n // blank. An indefinite failure stays `Dormant`, which is what makes\n // the pump retry it while the drain window is open.\n let mut cell = self.cells.remove(&id).expect(\"cell found above\");\n let hands_off = self.hands_off(&cell);\n let settled = matches!(result, Ok(CasOutcome::Applied))\n || (hands_off && matches!(result, Ok(CasOutcome::Rejected)));\n let adoption =\n (settled && hands_off && released_epoch.is_some()).then(|| self.cell_op(&id));\n self.releasing_cells.remove(&id);\n if adoption.is_some() {\n self.adopting_cells.insert(id.clone());\n }\n cell.releasing = None;\n if let (Some(released_epoch), Some(adopt)) = (released_epoch, adoption) {\n set_phase(\n &mut self.occupied,\n &mut cell,\n Phase::Adopting {\n op: adopt,\n released_epoch,\n },\n );\n effects.push(Effect::AdoptReleased {\n op: adopt,\n cell: id.clone(),\n released_epoch,\n rebalance: cell.handoff,\n });\n } else {\n if settled && matches!(cell.phase, Phase::Dormant { .. }) {\n set_phase(&mut self.occupied, &mut cell, Phase::Inactive);\n }\n // A balancing move that did not reach adoption is over. The\n // requests it parked take the ordinary path: a wake if the\n // record still names this node, an owner read if it does not.\n if std::mem::take(&mut cell.handoff) && !cell.requests.is_empty() {\n self.admit_or_queue_activation(&id, &mut cell, ColdStart::ReadOwner, effects);\n }\n }\n self.cells.insert(id, cell);\n }\n\n /// Finish one shutdown handoff after a successor acquired the cell. The\n /// successor keeps the cell dormant until real traffic needs a runtime,\n /// so a fleet restart does not eagerly materialize every resident cell.\n fn successor_adopted(\n &mut self,\n op: OpId,\n result: Result,\n effects: &mut Vec,\n ) {\n let Some(id) = self.take_cell_op(\n op,\n |cell| matches!(cell.phase, Phase::Adopting { op: current, .. } if current == op),\n ) else {\n return;\n };\n self.adopting_cells.remove(&id);\n let mut cell = self.cells.remove(&id).expect(\"cell found above\");\n let Phase::Adopting { released_epoch, .. } = cell.phase else {\n unreachable!()\n };\n let handoff = std::mem::take(&mut cell.handoff);\n match result {\n Ok(adopted)\n if adopted.node != self.node\n && !adopted.addr.is_empty()\n && adopted.epoch > released_epoch\n && adopted.peer_protocol == self.config.peer_protocol =>\n {\n self.activity.handed_off = self.activity.handed_off.saturating_add(1);\n if handoff {\n self.activity.rebalanced = self.activity.rebalanced.saturating_add(1);\n }\n let route = Route::Remote {\n node: adopted.node.clone(),\n addr: adopted.addr.clone(),\n epoch: adopted.epoch,\n peer_protocol: adopted.peer_protocol,\n };\n set_phase(\n &mut self.occupied,\n &mut cell,\n Phase::Remote {\n node: adopted.node,\n addr: adopted.addr,\n epoch: adopted.epoch,\n peer_protocol: adopted.peer_protocol,\n cache: RemoteCache::Uncached,\n },\n );\n self.finish_requests(&id, &mut cell, Ok(route), effects);\n }\n // A drain has no future here, so its parked requests fail. A\n // balancing move only released the record: the cell is unowned,\n // and its requests re-read ownership like any cold request.\n _ if handoff => {\n self.activity.handoff_failed = self.activity.handoff_failed.saturating_add(1);\n self.activity.rebalance_failed = self.activity.rebalance_failed.saturating_add(1);\n set_phase(&mut self.occupied, &mut cell, Phase::Inactive);\n if !cell.requests.is_empty() {\n self.admit_or_queue_activation(&id, &mut cell, ColdStart::ReadOwner, effects);\n }\n }\n _ => {\n self.activity.handoff_failed = self.activity.handoff_failed.saturating_add(1);\n set_phase(&mut self.occupied, &mut cell, Phase::Inactive);\n self.finish_requests(&id, &mut cell, Err(RequestError::NodeUnavailable), effects);\n }\n }\n self.cells.insert(id, cell);\n }\n\n fn runtime_stopped(&mut self, op: OpId, effects: &mut Vec) {\n let Some(id) = self.take_cell_op(\n op,\n |cell| matches!(cell.phase, Phase::Cleaning { op: current, .. } if current == op),\n ) else {\n return;\n };\n let mut cell = self.cells.remove(&id).expect(\"cell found above\");\n // The realm went with the runtime. Leaving the isolate recorded here\n // would keep counting this cell against that heap, so the walk down\n // would never see an isolate empty and would never aim at one.\n cell.isolate = None;\n let Phase::Cleaning { epoch, cause, .. } = cell.phase else {\n unreachable!()\n };\n if matches!(cause, StopCause::Evict { .. }) {\n // The stop completed even when waiting demand immediately starts\n // a new runtime. Its callers await this operation, not absence\n // after the next activation.\n self.finish_eviction(op, Ok(EvictSuccess::Evicted), effects);\n }\n match cause {\n StopCause::Cleanup => {\n set_phase(&mut self.occupied, &mut cell, Phase::Dormant { epoch });\n self.finish_requests(&id, &mut cell, Err(RequestError::PublishFailed), effects);\n }\n // `Inactive`, not `Dormant`: the next request re-reads the\n // ownership record. A proof that could not be completed is exactly\n // the moment to stop assuming this node still holds the cell, and\n // the re-read costs one GET on a path that was already failing.\n StopCause::Reset => {\n set_phase(&mut self.occupied, &mut cell, Phase::Inactive);\n if let Some(AlarmState::Firing { op, .. }) = cell.alarm {\n self.cell_ops.remove(&op);\n }\n cell.alarm = None;\n cell.alarm_wake = false;\n cell.websockets.clear();\n self.finish_requests(\n &id,\n &mut cell,\n Err(RequestError::DurabilityUnproven),\n effects,\n );\n }\n StopCause::Evict { rebalance }\n if cell.requests.is_empty() || (self.draining && rebalance) =>\n {\n set_phase(&mut self.occupied, &mut cell, Phase::Dormant { epoch });\n self.eviction_permits.remove(&id);\n // The record still names this node, which is the whole cost of\n // stopping here: every later request for the cell routes to a\n // node that has already decided it has no room. Publishing it\n // as unowned is what turns an eviction into shed load.\n if rebalance {\n let op = self.cell_op(&id);\n cell.releasing = Some(op);\n self.releasing_cells.insert(id.clone());\n effects.push(Effect::ReleaseOwner {\n op,\n cell: id.clone(),\n epoch,\n });\n }\n }\n StopCause::Evict { .. } => {\n // A request arrived mid-eviction, so the cell turns straight\n // back around. The eviction is over either way.\n self.eviction_permits.remove(&id);\n // The stop discarded the LTX metadata. Re-read ownership and\n // claim epoch+1 before reusing the preserved SQLite image.\n self.admit_or_queue_activation(&id, &mut cell, ColdStart::ReadOwner, effects);\n }\n // The node began leaving while the cell was between isolates.\n // Started again it would only be one more resident for the\n // drain to evict; dormant, the drain releases its record like\n // any other.\n StopCause::Swap if self.draining => {\n set_phase(&mut self.occupied, &mut cell, Phase::Dormant { epoch });\n self.finish_requests(&id, &mut cell, Err(RequestError::PublishFailed), effects);\n }\n // The swap's second half: the same epoch, the same owner record,\n // the same replica, a new isolate. `Starting` without `Restore`\n // is deliberate — the database is open on disk and the LTX\n // handle never closed — and without `CasOwner`, because nothing\n // about ownership changed. Requests queued while the cell\n // quiesced release when it publishes.\n StopCause::Swap => {\n let op = self.cell_op(&id);\n set_phase(&mut self.occupied, &mut cell, Phase::Starting { op, epoch });\n effects.push(Effect::StartRuntime {\n op,\n cell: id.clone(),\n epoch,\n });\n }\n StopCause::Fence => unreachable!(\"fenced cells do not wait for runtime shutdown\"),\n }\n self.cells.insert(id, cell);\n self.pump_capacity(effects);\n self.shed_toward_floor(effects);\n // The permit this stop returned is spent here, not on the next load\n // sample. Without this re-entry the idle path drains at\n // `max_evictions` cells per sampling period, whatever the proof and\n // stop take: 10,000 idle rooms hibernated at exactly four per second\n // for 42 minutes (GCE, 2026-09-03), and raising the bound only\n // raised the rate. The clock is the last one an event carried, which\n // is at or before the real time, so a cell idle at the sample is\n // still idle here.\n self.evict_idle(self.now_mono_ms, effects);\n }\n\n /// A bounded stop gave up. The host dropped the runtime and kept the\n /// database open, which is the state a generation swap leaves between\n /// its halves, so the cell restarts on the open handle at the same\n /// epoch through the swap's own path. The eviction is recorded as\n /// refused, so idle eviction waits another idle period before it tries\n /// the cell again instead of stopping it on every sample. Without this a\n /// cell whose final proof stalled retried every 10 s without end and\n /// answered nobody (the 2026-09-03 whale runs).\n fn runtime_stop_failed(&mut self, op: OpId, error: EvictError, effects: &mut Vec) {\n let Some(id) = self.take_cell_op(\n op,\n |cell| matches!(cell.phase, Phase::Cleaning { op: current, .. } if current == op),\n ) else {\n return;\n };\n let mut cell = self.cells.remove(&id).expect(\"cell found above\");\n let Phase::Cleaning { epoch, cause, .. } = cell.phase else {\n unreachable!()\n };\n assert!(\n matches!(cause, StopCause::Evict { .. }),\n \"only an eviction stop is bounded\"\n );\n self.finish_eviction(op, Err(error), effects);\n\n cell.isolate = None;\n cell.evict_rebalance = false;\n cell.handoff = false;\n cell.eviction_refused_mono_ms = Some(self.now_mono_ms);\n self.eviction_permits.remove(&id);\n let next = self.cell_op(&id);\n cell.swapping = true;\n set_phase(\n &mut self.occupied,\n &mut cell,\n Phase::Starting { op: next, epoch },\n );\n effects.push(Effect::StartRuntime {\n op: next,\n cell: id.clone(),\n epoch,\n });\n self.cells.insert(id, cell);\n }\n\n /// Announce the generation the node now serves.\n ///\n /// Nothing moves here, and nothing is marked; the pump moves cells after\n /// this and every later event, so the bound, the safe point, and\n /// eagerness are decided by the same code whether a cell became stale\n /// now or lands stale later. A resident cell of an eager class is forced\n /// at once: the engine's reserved cells hold no application state worth\n /// waiting for, and the cron cell must run the new schedule before the\n /// adoption arms it.\n fn generation_changed(&mut self, generation: u64, max_age_ms: u64, eager_classes: &[String]) {\n if generation <= self.current_generation {\n return;\n }\n self.current_generation = generation;\n self.generation_changed_mono_ms = self.now_mono_ms;\n self.swap_max_age_ms = max_age_ms;\n self.swap_eager_classes = eager_classes.iter().cloned().collect();\n }\n\n /// Whether this cell's class was named eager by the last announcement, so\n /// its swap does not wait for the maximum age.\n ///\n /// A scope is `:` and the instance may itself contain a\n /// `:`, so the class is everything before the first one -- the same split\n /// the shell makes when it resolves a cell to its Worker config.\n fn swap_is_eager(&self, id: &str) -> bool {\n let class = id.split_once(':').map_or(id, |(class, _)| class);\n self.swap_eager_classes.contains(class)\n }\n\n /// Whether a swap can stop this cell's runtime now without cutting work\n /// short: no request running, no output withheld by the gate, no alarm\n /// handler in flight, and no regular or outbound socket, which cannot\n /// outlive the isolate. The input gate is not consulted; a\n /// `blockConcurrencyWhile` holder is a running request or alarm, so it is\n /// covered by those.\n fn swap_safe_point(&self, id: &str) -> bool {\n let Some(cell) = self.cells.get(id) else {\n return false;\n };\n !self.active_cells.contains_key(id)\n && !cell\n .websockets\n .values()\n .any(|kind| matches!(kind, WebSocketKind::Regular | WebSocketKind::Outbound))\n && !matches!(cell.alarm, Some(AlarmState::Firing { .. }))\n && !self.barriers.values().any(|barrier| barrier.cell == id)\n }\n\n /// The swap pump, run after every event once a generation change has\n /// been announced: move every resident cell on an older generation to\n /// the current one, at most `max_releases` at a time.\n ///\n /// A nominated cell quiesces first, exactly as a shutdown handoff does:\n /// new requests queue for the restarted runtime instead of routing to the\n /// old one, so continuous traffic cannot keep the cell from ever reaching\n /// a safe point. Work already admitted finishes. At the safe point the\n /// cell stops with `StopCause::Swap`, and `runtime_stopped` starts it\n /// again on the current generation at the same epoch.\n ///\n /// A cell that never reaches a safe point — a regular WebSocket, an\n /// endless request — is forced once the maximum age has passed, or at\n /// once when its class is eager: its activity is cancelled, which is what\n /// a Cloudflare deployment does to every object unconditionally. The\n /// hibernatable sockets survive either way.\n fn pump_swap(&mut self, effects: &mut Vec) {\n if self.current_generation == 0\n || self.draining\n || self.fenced\n || !self.node_authoritative()\n {\n return;\n }\n let current = self.current_generation;\n let forced_by_age = self\n .now_mono_ms\n .saturating_sub(self.generation_changed_mono_ms)\n >= self.swap_max_age_ms;\n let in_flight = self.cells.values().filter(|cell| cell.swapping).count();\n let room = self.config.max_releases.saturating_sub(in_flight);\n let stale: Vec = self\n .cells\n .iter()\n .filter(|(_, cell)| {\n matches!(cell.phase, Phase::Resident { .. })\n && !cell.swapping\n && !cell.quiescing\n && cell.releasing.is_none()\n && cell.generation != current\n })\n .map(|(id, _)| id.clone())\n .collect();\n for id in stale.into_iter().take(room) {\n let cell = self.cells.get_mut(&id).expect(\"stale cell listed above\");\n cell.swapping = true;\n cell.quiescing = true;\n self.quiescing_cells.insert(id);\n }\n let nominated: Vec = self\n .cells\n .iter()\n .filter(|(_, cell)| {\n cell.swapping && cell.quiescing && matches!(cell.phase, Phase::Resident { .. })\n })\n .map(|(id, _)| id.clone())\n .collect();\n for id in nominated {\n if !self.swap_safe_point(&id) {\n let eager = self.swap_is_eager(&id);\n let cell = self.cells.get(&id).expect(\"nominated cell listed above\");\n let alarm = match cell.alarm {\n Some(AlarmState::Firing { op, .. }) => Some(op),\n _ => None,\n };\n if !(forced_by_age || eager) || cell.swap_cancelled {\n continue;\n }\n let requests = self\n .active_requests\n .iter()\n .filter_map(|(request, cell)| (cell == &id).then_some(*request))\n .collect();\n let cell = self\n .cells\n .get_mut(&id)\n .expect(\"nominated cell listed above\");\n cell.swap_cancelled = true;\n let websockets = cell\n .websockets\n .iter()\n .filter(|(_, kind)| !matches!(kind, WebSocketKind::Hibernatable))\n .map(|(websocket, _)| *websocket)\n .collect();\n effects.push(Effect::CancelCellActivity {\n cell: id,\n requests,\n websockets,\n alarm,\n keep_hibernatable: true,\n });\n continue;\n }\n let op = self.cell_op(&id);\n let cell = self\n .cells\n .get_mut(&id)\n .expect(\"nominated cell listed above\");\n let Phase::Resident { epoch } = cell.phase else {\n unreachable!(\"nominated cells are resident\")\n };\n set_phase(\n &mut self.occupied,\n cell,\n Phase::Cleaning {\n op,\n epoch,\n cause: StopCause::Swap,\n },\n );\n self.quiescing_cells.remove(&id);\n effects.push(Effect::StopRuntime {\n op,\n cell: id,\n epoch,\n cause: StopCause::Swap,\n bounded: false,\n });\n }\n }\n\n /// The generation a cell's runtime was started on, if it has one.\n pub fn cell_generation(&self, id: &str) -> Option {\n self.cells.get(id).map(|cell| cell.generation)\n }\n\n /// The generation the node serves, as last announced.\n pub fn current_generation(&self) -> u64 {\n self.current_generation\n }\n\n /// Cells moving to the current generation right now.\n pub fn swapping(&self) -> usize {\n self.cells.values().filter(|cell| cell.swapping).count()\n }\n\n fn durability_checked(\n &mut self,\n op: OpId,\n result: Result<(), EvictFailure>,\n effects: &mut Vec,\n ) {\n let now = self.now_mono_ms;\n let Some(id) = self.take_cell_op(op, |cell| {\n matches!(cell.phase, Phase::EnsuringDurability { op: current, .. } if current == op)\n }) else {\n return;\n };\n let cell = self.cells.get_mut(&id).expect(\"cell found above\");\n let Phase::EnsuringDurability { epoch, .. } = cell.phase else {\n unreachable!()\n };\n match result {\n Ok(()) => {\n // Minted inline: `cell` still borrows the map, and field\n // accesses split where a method call cannot.\n let stop = self.next_op;\n self.next_op = self.next_op.checked_add(1).expect(\"operation id exhausted\");\n if let Some(requests) = self.eviction_notifications.remove(&op) {\n self.eviction_notifications.insert(stop, requests);\n }\n self.cell_ops.insert(stop, id.clone());\n let rebalance = cell.evict_rebalance;\n set_phase(\n &mut self.occupied,\n cell,\n Phase::Cleaning {\n op: stop,\n epoch,\n cause: StopCause::Evict { rebalance },\n },\n );\n effects.push(Effect::StopRuntime {\n op: stop,\n cell: id,\n epoch,\n cause: StopCause::Evict { rebalance },\n bounded: !self.draining,\n });\n }\n Err(reason) => {\n set_phase(&mut self.occupied, cell, Phase::Resident { epoch });\n cell.eviction_refused_mono_ms = Some(now);\n self.eviction_permits.remove(&id);\n self.finish_eviction(op, Err(EvictError::Failed(reason)), effects);\n }\n }\n }\n\n /// Evict on demand. Local, like an idle eviction: the caller asked\n /// this node to drop the cell, not to give it away.\n fn evict(&mut self, id: &str, effects: &mut Vec) {\n let _ = self.begin_eviction(id, false, effects);\n }\n\n fn evict_requested(&mut self, request: RequestId, id: CellId, effects: &mut Vec) {\n let mut admitted_effects = Vec::new();\n let result = self\n .admit_eviction(&id, &mut admitted_effects)\n .map(|op| match op {\n Some(op) => {\n self.eviction_notifications\n .entry(op)\n .or_default()\n .insert(request);\n EvictionAdmission::Pending\n }\n None => EvictionAdmission::Absent,\n });\n effects.push(Effect::EvictionAdmission { request, result });\n effects.extend(admitted_effects);\n }\n\n fn finish_eviction(\n &mut self,\n op: OpId,\n result: Result,\n effects: &mut Vec,\n ) {\n if let Some(requests) = self.eviction_notifications.remove(&op) {\n effects.extend(\n requests\n .into_iter()\n .map(|request| Effect::EvictionFinished { request, result }),\n );\n }\n }\n\n fn admit_eviction(\n &mut self,\n id: &str,\n effects: &mut Vec,\n ) -> Result, EvictError> {\n let refuse = |reason| Err(EvictError::Refused(reason));\n if !self.ready_to_serve() || self.confirming {\n return refuse(EvictRefusal::NodeUnavailable);\n }\n if self.is_active(id) {\n return refuse(EvictRefusal::CellActive);\n }\n let cell = self.cells.get(id);\n let phase = cell.map(|cell| &cell.phase);\n match phase {\n Some(Phase::EnsuringDurability { op, .. }) => return Ok(Some(*op)),\n Some(Phase::Cleaning {\n op,\n cause: StopCause::Evict { .. },\n ..\n }) => return Ok(Some(*op)),\n _ => {}\n }\n // A dormant phase can still owe an ownership release. The operation\n // index is validated against these slots, so no global scan is needed.\n let transitioning = cell.is_some_and(|cell| {\n cell.releasing.is_some()\n || cell.waiting_for.is_some()\n || cell.waiting_activation.is_some()\n || !cell.requests.is_empty()\n || cell.quiescing\n || cell.swapping\n || cell.handoff\n || cell.confirming\n || cell.resume_demand\n });\n if transitioning {\n return refuse(EvictRefusal::CellTransitioning);\n }\n match phase {\n None | Some(Phase::Inactive | Phase::Dormant { .. } | Phase::Remote { .. }) => {\n // A wake hint can outlive a remote lookup or an eviction with\n // a future alarm. Refusing on the hint alone would reject\n // settled absence; only pending lifecycle work prevents it.\n Ok(None)\n }\n Some(Phase::Resident { .. }) => self\n .begin_eviction(id, false, effects)\n .map(Some)\n .map_err(EvictError::Refused),\n _ => refuse(EvictRefusal::CellTransitioning),\n }\n }\n\n /// Shutdown handoff: give every resident cell away by releasing its\n /// owner record, so peers take over at once instead of waiting out the\n /// node lease. `rebalance` is forced true, so this releases even when\n /// `ownership_on_evict` keeps ownership on an ordinary evict.\n ///\n /// Only marks the node draining; `pump_release` does the work, at most\n /// `max_releases` at a time. An unbounded walk here would start a\n /// durability proof per resident cell at once -- a node holding\n /// thousands turns its own shutdown into the restore storm the eviction\n /// bound exists to prevent, and trips the `validate` ceiling besides.\n fn release_all(&mut self, effects: &mut Vec) {\n self.draining = true;\n self.pump_release(effects);\n }\n\n /// Fail every unacknowledged output held on `id` without discarding the\n /// runtime. A shutdown handoff has already quiesced the cell, and its\n /// mandatory final durability proof follows before ownership release. The\n /// caller therefore learns that its write was not acknowledged, while the\n /// handoff can still prove and preserve that write for a later retry.\n fn cancel_cell_outputs(&mut self, id: &str, effects: &mut Vec) {\n let outputs: Vec = self\n .barriers\n .iter()\n .filter(|(_, barrier)| {\n barrier.cell == id && matches!(barrier.owner, GateOwner::Output(_))\n })\n .map(|(op, _)| *op)\n .collect();\n for op in outputs {\n let Some(gate) = self.barriers.remove(&op) else {\n continue;\n };\n let GateOwner::Output(owner) = gate.owner else {\n unreachable!(\"only output gates were selected\")\n };\n self.release_held(owner, Err(RequestError::DurabilityUnproven), effects);\n for follower in gate.followers {\n self.release_held(follower, Err(RequestError::DurabilityUnproven), effects);\n }\n }\n }\n\n /// The drain pump, run after every event once draining: keep\n /// `max_releases` handoffs in flight while any owned cell remains.\n /// The pump first marks a bounded batch as quiescing. A quiescing cell\n /// queues new requests for its successor, so continuous traffic cannot\n /// keep finding a fresh activity between eviction attempts. Work already\n /// admitted on the cell finishes before its durability proof starts.\n /// Releases in flight count against the same ceiling as the proofs: the\n /// bound exists to keep shutdown from flooding the store, and a release\n /// is one more write in that flood.\n fn pump_release(&mut self, effects: &mut Vec) {\n if !self.draining {\n return;\n }\n let mut in_flight = self.handoffs_in_flight();\n\n // A reserved cell can become idle on any event. Promote it into the\n // ordinary eviction state without spending another handoff permit.\n let quiescing: Vec = self.quiescing_cells.iter().cloned().collect();\n for id in quiescing {\n let _ = self.begin_eviction(&id, true, effects);\n }\n\n // A full batch cannot nominate another cell. Stop before rebuilding\n // and sorting the resident inventory: peer traffic still reaches the\n // draining actor, and repeating that complete scan for every queued\n // request can starve the node-lease timer until the donor self-fences.\n if in_flight >= self.config.max_releases {\n return;\n }\n\n // A dormant Queue broker has no runtime to prove or stop, but it still\n // owns delayed deliveries and retry alarms. Move those owner records\n // before resident work can spend the complete bounded batch. The\n // general dormant walk below remains after residents because an idle\n // application cell carries no comparable background obligation.\n #[cfg(celld_internal_tests)]\n self.record_drain_table_visits(self.cells.len());\n let dormant_queues: Vec<(CellId, Epoch)> = self\n .cells\n .iter()\n .filter_map(|(id, cell)| match cell.phase {\n Phase::Dormant { epoch } if cell.releasing.is_none() && is_queue_broker(id) => {\n Some((id.clone(), epoch))\n }\n _ => None,\n })\n .collect();\n for (id, epoch) in dormant_queues {\n if in_flight >= self.config.max_releases {\n return;\n }\n let op = self.cell_op(&id);\n let cell = self.cells.get_mut(&id).expect(\"dormant Queue listed above\");\n cell.releasing = Some(op);\n self.releasing_cells.insert(id.clone());\n effects.push(Effect::ReleaseOwner {\n op,\n cell: id,\n epoch,\n });\n in_flight += 1;\n }\n\n // Reserve the complete batch even when its cells are active. Future\n // requests wait in `cell.requests`; the already-running requests keep\n // the runtime pinned until their activity guards finish.\n #[cfg(celld_internal_tests)]\n self.record_drain_table_visits(self.cells.len());\n let mut residents: Vec<(CellId, bool, u64, RequestId, u64)> = self\n .cells\n .iter()\n .filter(|(_, cell)| !cell.quiescing && matches!(cell.phase, Phase::Resident { .. }))\n .map(|(id, cell)| {\n (\n id.clone(),\n is_queue_broker(id),\n cell.local_request_count,\n cell.last_used_request,\n cell.last_used_mono_ms,\n )\n })\n .collect();\n // A process deadline can cut this walk before it reaches every cell.\n // A Queue broker owns delayed delivery and retry alarms, so its work\n // does not always arrive as a routed request that increments the use\n // count. Protect those persistent brokers first. Then protect cells\n // that current traffic used repeatedly, followed by the newest use.\n // The old BTreeMap order handed off thousands of one-shot cells before\n // shared storage, so complete API surfaces waited for dead-node\n // recovery even though the donor made bounded handoff progress.\n residents.sort_by(\n |(left_id, left_queue, left_count, left_request, left_used),\n (right_id, right_queue, right_count, right_request, right_used)| {\n right_queue\n .cmp(left_queue)\n .then_with(|| right_count.cmp(left_count))\n .then_with(|| right_request.cmp(left_request))\n .then_with(|| right_used.cmp(left_used))\n .then_with(|| left_id.cmp(right_id))\n },\n );\n for (id, _, _, _, _) in residents {\n if in_flight >= self.config.max_releases {\n return;\n }\n let requests = self\n .active_requests\n .iter()\n .filter_map(|(request, cell)| (cell == &id).then_some(*request))\n .collect();\n let cell = self.cells.get_mut(&id).expect(\"resident cell listed above\");\n let alarm = match cell.alarm {\n Some(AlarmState::Firing { op, .. }) => Some(op),\n _ => None,\n };\n cell.quiescing = true;\n self.quiescing_cells.insert(id.clone());\n let websockets = cell.websockets.keys().copied().collect();\n self.cancel_cell_outputs(&id, effects);\n effects.push(Effect::CancelCellActivity {\n cell: id.clone(),\n requests,\n websockets,\n alarm,\n keep_hibernatable: false,\n });\n in_flight += 1;\n let _ = self.begin_eviction(&id, true, effects);\n }\n // A cell evicted before the drain kept its record on this node --\n // a sticky eviction, hibernation, or a release whose write failed\n // -- so a later request could resume locally. On a leaving node\n // that record only makes a successor wait out the node lease. The\n // runtime is already stopped, so the drain releases the record\n // directly.\n #[cfg(celld_internal_tests)]\n self.record_drain_table_visits(self.cells.len());\n let dormant: Vec<(CellId, Epoch)> = self\n .cells\n .iter()\n .filter_map(|(id, cell)| match cell.phase {\n Phase::Dormant { epoch } if cell.releasing.is_none() => Some((id.clone(), epoch)),\n _ => None,\n })\n .collect();\n for (id, epoch) in dormant {\n if in_flight >= self.config.max_releases {\n return;\n }\n self.release_dormant(&id, epoch, effects);\n in_flight += 1;\n }\n }\n\n /// Release one dormant cell's record for a successor. The runtime is\n /// already stopped, so this is the record write alone.\n fn release_dormant(&mut self, id: &str, epoch: Epoch, effects: &mut Vec) {\n let op = self.cell_op(id);\n let handoff = !self.draining;\n let cell = self.cells.get_mut(id).expect(\"dormant cell listed above\");\n cell.releasing = Some(op);\n cell.handoff = handoff;\n self.releasing_cells.insert(id.to_string());\n // A hibernated cell parks its sockets past the runtime, and\n // nothing else on the handoff path visits them: without this\n // close the process exit tears them down as an abnormal\n // 1006, while a resident cell's clients get the clean 1012\n // above.\n if !cell.websockets.is_empty() {\n let websockets = cell.websockets.keys().copied().collect();\n effects.push(Effect::CancelCellActivity {\n cell: id.to_string(),\n requests: Vec::new(),\n websockets,\n alarm: None,\n keep_hibernatable: false,\n });\n }\n effects.push(Effect::ReleaseOwner {\n op,\n cell: id.to_string(),\n epoch,\n });\n }\n\n /// Give up to `count` hibernated cells to the fleet without draining.\n ///\n /// Only a dormant cell moves: its move is one record write and one\n /// signed acquire, it cannot park a request behind a durability proof,\n /// and the receiver restores it when traffic returns. A resident cell\n /// hibernates through the ordinary idle eviction first and moves after.\n /// The drain's eviction pipeline is not used here: a drain cannot\n /// abandon a cell whose final proof stalls, and a balancing move must.\n /// On the 2026-09-03 fleet that pipeline parked whales without end. A\n /// cell with a pending request or an alarm about to fire stays. Every\n /// move shares the drain's release ceiling, and a node that is draining,\n /// fenced, or shedding moves nothing because those paths own its cells.\n fn rebalance(&mut self, count: usize, effects: &mut Vec) {\n if self.draining || self.fenced || self.shedding || !self.node_authoritative() {\n return;\n }\n let budget = count.min(\n self.config\n .max_releases\n .saturating_sub(self.handoffs_in_flight()),\n );\n let mut dormant: Vec<(CellId, Epoch, u64)> = self\n .cells\n .iter()\n .filter_map(|(id, cell)| match cell.phase {\n Phase::Dormant { epoch }\n if cell.releasing.is_none()\n && cell.requests.is_empty()\n && !self.alarm_is_imminent(cell) =>\n {\n Some((id.clone(), epoch, cell.last_used_mono_ms))\n }\n _ => None,\n })\n .collect();\n dormant.sort_by(|left, right| (left.2, &left.0).cmp(&(right.2, &right.0)));\n for (id, epoch, _) in dormant.into_iter().take(budget) {\n self.release_dormant(&id, epoch, effects);\n }\n }\n\n /// Does an eviction made for room hand the cell away?\n fn rebalances(&self) -> bool {\n self.config.ownership_on_evict == OwnershipOnEvict::Release\n }\n\n /// A drain may hold as many evictions as releases, because every\n /// resident cell leaves through one.\n fn eviction_ceiling(&self) -> usize {\n if self.draining {\n self.config.max_releases.max(self.config.max_evictions)\n } else {\n self.config.max_evictions\n }\n }\n\n /// An alarm about to fire is not worth an eviction: the wake costs\n /// more than the residency it would save, so the cell is held even\n /// though its entry is perfectly durable. Coverage says the alarm can\n /// survive the eviction; this says it is not worth surviving it.\n fn alarm_is_imminent(&self, cell: &Cell) -> bool {\n match cell.alarm {\n Some(AlarmState::Armed { at_ms, .. }) | Some(AlarmState::Firing { at_ms, .. }) => {\n at_ms >= 0\n && (at_ms as u64).saturating_sub(self.now_ms) <= self.config.alarm_resident_ms\n }\n None => false,\n }\n }\n\n fn begin_eviction(\n &mut self,\n id: &str,\n rebalance: bool,\n effects: &mut Vec,\n ) -> Result {\n let Some(cell) = self.cells.get(id) else {\n return Err(EvictRefusal::CellTransitioning);\n };\n // Only while there is room to spare. Holding a cell to save a wake is\n // worth it on an idle node and indefensible on a full one: the window\n // defaults to an hour, so on an alarm-driven workload this would hold\n // most of the node and pin it at its ceiling -- trading a real\n // admission failure for a saved activation. Under pressure the node\n // takes the wake.\n if self.alarm_is_imminent(cell) && !self.shedding && !self.draining {\n return Err(EvictRefusal::AlarmImminent);\n }\n if self.is_active(id) {\n return Err(EvictRefusal::CellActive);\n }\n let Phase::Resident { epoch } = cell.phase else {\n return Err(EvictRefusal::CellTransitioning);\n };\n if !cell\n .alarm\n .as_ref()\n .is_none_or(|alarm| matches!(alarm, AlarmState::Armed { covered: true, .. }))\n {\n return Err(EvictRefusal::AlarmUncovered);\n }\n // The permit is taken here, so the ceiling holds here. A balancing\n // batch reserves more cells than permits, and the pump promotes\n // them as permits return.\n if self.eviction_permits.len() >= self.eviction_ceiling() {\n return Err(EvictRefusal::ConcurrencyLimit);\n }\n let op = self.cell_op(id);\n let cell = self.cells.get_mut(id).expect(\"resident cell checked above\");\n cell.evict_rebalance = rebalance;\n set_phase(\n &mut self.occupied,\n cell,\n Phase::EnsuringDurability { op, epoch },\n );\n // Retire the quiescing reservation with the resident phase. Demand\n // shedding can enter here before the drain pump, so caller-only\n // cleanup double-counts a handoff and breaks the phase index.\n self.quiescing_cells.remove(id);\n self.eviction_permits.insert(id.to_string());\n effects.push(Effect::EnsureDurable {\n op,\n cell: id.to_string(),\n epoch,\n // The same condition `request_authorized` applies when it cancels\n // an eviction for an arriving request. `worker_request` cancels\n // one without testing it, which this field is what makes safe;\n // `Effect::EnsureDurable::revocable` records the argument.\n revocable: !self.draining,\n });\n Ok(op)\n }\n\n fn shed_one(&mut self, effects: &mut Vec) {\n // Only when the blocker is headroom. A latched node is already walking\n // down on its own schedule, and admission stays closed until it\n // reaches the low watermark -- so shedding to make room for a waiter\n // that cannot be admitted has no stopping condition, and every\n // completed eviction re-enters here through `pump_capacity` and starts\n // another. That empties the node.\n //\n // Count the evictions already in flight against the waiters: this is\n // reachable from every activity finish and websocket close, and a\n // waiter whose eviction is already under way must not turn each of\n // those triggers into another victim. Spend the cut on commit, never\n // on nomination.\n if self.shedding\n || self.capacity_waiters.len() <= self.eviction_permits.len()\n || self.eviction_permits.len() >= self.config.max_evictions\n {\n return;\n }\n if let Some(victim) = self.shed_candidate() {\n let _ = self.begin_eviction(&victim, self.rebalances(), effects);\n }\n }\n\n /// The cell to shed right now: resident, idle, not holding an alarm the\n /// node still owes, and the least recently used of those. Demand shedding\n /// and pressure shedding both come through here, so they cannot disagree\n /// about what is safe to take or which one to take first.\n /// Bound the time a cell can sit behind the admission gate.\n ///\n /// Every other stall in a cold route is bounded by `operation_deadline_ms`\n /// through `watch_operations`, which arms per emitted effect. A parked\n /// cell has emitted none, so it is the one stall nothing expires, and a\n /// gate that closes on memory the queue itself holds can never reopen.\n fn watch_queued(&mut self, id: &str, cell: &mut Cell, effects: &mut Vec) {\n let Some(deadline_ms) = self.config.operation_deadline_ms else {\n return;\n };\n let generation = self.next_timer_generation;\n self.next_timer_generation = self\n .next_timer_generation\n .checked_add(1)\n .expect(\"timer generation exhausted\");\n cell.queued_generation = Some(generation);\n effects.push(Effect::ScheduleTimer {\n timer: Timer::QueuedActivation {\n cell: id.to_string(),\n generation,\n },\n at_mono_ms: self.now_mono_ms.saturating_add(deadline_ms),\n });\n }\n\n /// A cell that has waited out its deadline behind the gate.\n ///\n /// Answering `CapacityExhausted` is the existing refusal for \"not here,\n /// not now\": it already carries the stale-route header, so the caller\n /// re-resolves instead of retrying a node that cannot take it.\n fn expire_queued(&mut self, cell: &str, generation: u64, effects: &mut Vec) {\n let Some(queued) = self.cells.get(cell) else {\n return;\n };\n if queued.queued_generation != Some(generation) {\n return;\n }\n if !matches!(\n queued.phase,\n Phase::WaitingActivation | Phase::WaitingCapacity\n ) {\n return;\n }\n for request in queued.requests.iter().copied().collect::>() {\n effects.push(Effect::Complete {\n request,\n result: Err(RequestError::CapacityExhausted),\n });\n self.request_cells.remove(&request);\n self.capacity_requests.remove(&request);\n if let Some(queued) = self.cells.get_mut(cell) {\n queued.requests.remove(&request);\n }\n }\n let Some(queued) = self.cells.get_mut(cell) else {\n return;\n };\n queued.queued_generation = None;\n // An alarm wake or a record confirmation is not a client waiting, so\n // it keeps the cell queued rather than being refused with the\n // requests. A clean-reload resume is retired here instead, so a slow\n // resume does not hold readiness.\n if queued.alarm_wake || queued.confirming {\n return;\n }\n self.activation_waiters.retain(|queued| queued != cell);\n self.capacity_waiters.retain(|queued| queued != cell);\n let Some(queued) = self.cells.get_mut(cell) else {\n return;\n };\n let start = queued.waiting_activation.take();\n let activation = queued.waiting_for.take();\n let dormant = match (start, activation) {\n (Some(ColdStart::Restore(spec)), _) => Some(spec.epoch),\n (_, Some(Activation::Restore(spec))) => Some(spec.epoch),\n _ => None,\n };\n let settles_resume = queued.resume_demand;\n set_phase(\n &mut self.occupied,\n queued,\n match dormant {\n Some(epoch) => Phase::Dormant { epoch },\n None => Phase::Inactive,\n },\n );\n // A saved restore has no caller that can retire its resume demand.\n // Leaving that demand indexed after this terminal timeout keeps the\n // complete node unready with no operation or timer that can clear it.\n if settles_resume {\n self.settle_local_resume(cell);\n }\n }\n\n fn shed_candidate(&self) -> Option {\n self.shed_candidate_where(|_| true)\n }\n\n /// The shed victim among the cells `keep` accepts. The idle path\n /// passes a coldness filter here: without it, one warm cell in the\n /// emptiest isolate is the single candidate every sample, and idle\n /// eviction stalls while cold cells sit in fuller isolates.\n fn shed_candidate_where(&self, keep: impl Fn(&Cell) -> bool) -> Option {\n let evictable = |id: &CellId, cell: &Cell| {\n self.is_hibernatable(id)\n && keep(cell)\n && cell\n .alarm\n .as_ref()\n .is_none_or(|alarm| matches!(alarm, AlarmState::Armed { covered: true, .. }))\n };\n // How many cells each isolate still holds. The victim comes from the\n // one closest to empty, because only the cut that takes an isolate's\n // last cell gives the heap back -- and the heap is where the memory\n // is. Evicting by recency alone spreads the cuts over every isolate,\n // so the node gives up its working set and frees nothing.\n let mut occupancy: BTreeMap = BTreeMap::new();\n for cell in self.cells.values() {\n if let Some(isolate) = cell.isolate {\n *occupancy.entry(isolate).or_default() += 1;\n }\n }\n self.cells\n .iter()\n .filter(|(id, cell)| evictable(id, cell))\n // Within an isolate the order is unchanged: never-refused first,\n // then by how long ago the refusal was, then least recently used,\n // with the id as a tiebreak so the choice is a function of the\n // state and not of map iteration order.\n //\n // A cell the shell never placed sorts last rather than first. It\n // has no heap to give back, so preferring it would reintroduce the\n // scatter this ordering exists to remove.\n .min_by_key(|(id, cell)| {\n let emptiest = cell\n .isolate\n .and_then(|isolate| occupancy.get(&isolate).copied())\n .unwrap_or(usize::MAX);\n (\n emptiest,\n cell.eviction_refused_mono_ms,\n cell.last_used_mono_ms,\n (*id).clone(),\n )\n })\n .map(|(id, _)| id.clone())\n }\n\n /// Release one cell that has gone cold, with nothing asking for the room.\n ///\n /// Shedding answers \"this node is in trouble\"; this answers the ordinary\n /// question of a cell nobody has touched in a long time. Without it a node\n /// only ever gives a cell back under pressure or when another cell wants\n /// the slot, so a quiet node holds every cell it ever served until it\n /// reaches a watermark -- paying to keep runtimes alive for traffic that\n /// stopped hours ago.\n fn evict_idle(&mut self, now_mono_ms: u64, effects: &mut Vec) {\n let Some(idle_ms) = self.config.idle_evict_ms else {\n return;\n };\n // A latched node is already walking down on its own schedule, and\n // `shed_toward_floor` prefers the same cold cells. This is reached\n // from every eviction stop as well as from the sample, so the guard\n // lives here rather than at each caller.\n if self.shedding {\n return;\n }\n // Fill the permits, exactly as `shed_toward_floor` does. One\n // candidate per sample drains an idle backlog at the sampling\n // period: a few thousand parked hibernatable cells — the shape\n // of a game or chat fleet after activity stops — then keep\n // their isolates pinned for the better part of an hour, and an\n // autoscaler reading occupancy never sees the node go quiet.\n while self.eviction_permits.len() < self.config.max_evictions {\n // A refused eviction waits out another idle period. Retrying it\n // on every sample stops and restarts a cell whose proof cannot\n // complete for as long as that lasts.\n let Some(candidate) = self.shed_candidate_where(|cell| {\n now_mono_ms.saturating_sub(cell.last_used_mono_ms) >= idle_ms\n && cell\n .eviction_refused_mono_ms\n .is_none_or(|refused| now_mono_ms.saturating_sub(refused) >= idle_ms)\n }) else {\n return;\n };\n // Idle eviction is a local residency decision, not a handoff.\n if self.begin_eviction(&candidate, false, effects).is_err() {\n return;\n }\n }\n }\n\n /// Fold a resource sample into the shedding latch, then act on it.\n ///\n /// The latch is the whole point. Shedding on the instantaneous crossing\n /// alone flaps: the eviction relieves the pressure, admission resumes, the\n /// ceiling is crossed again. `PressureConfig` holds the node in shedding\n /// until every configured low watermark clears, so the node walks down to\n /// its target instead of oscillating around its ceiling.\n fn load_sampled(&mut self, load: pressure::Load, now_mono_ms: u64, effects: &mut Vec) {\n self.memory_headroom = self.config.pressure.has_headroom(load);\n // Both latches are folded, and the reason names the more serious of the\n // two. The latches are state of their own: derived from the reason\n // instead, one crossing holds the node against the other's watermark.\n let (latches, reason) = self.config.pressure.classify(load, self.latches);\n self.latches = latches;\n self.shed_reason = reason;\n self.shedding = reason.is_some();\n if !self.shedding {\n // Relieved. Whatever was queued for capacity may proceed.\n self.shed_cut = None;\n self.pump_capacity(effects);\n self.evict_idle(now_mono_ms, effects);\n return;\n }\n // Evicting only helps if it can finish the job. Without a stopping\n // condition, a ceiling below the process's memory floor cuts a\n // proportion of whatever remains on every sample and walks the node to\n // zero. So the latch holds -- admission stays closed, the node is over\n // its ceiling -- while the walk down stops spending the working set.\n // `walk_metric` documents which measurement it reads and why. Changing\n // that measurement discards the baseline: the two numbers are not\n // comparable, and they sit close enough to find each other flat by\n // coincidence.\n let metric = pressure::PressureConfig::walk_metric(latches);\n let sample_bytes = load.metric_bytes(metric);\n if self.shed_cut.is_some_and(|cut| cut.metric != metric) {\n self.shed_cut = None;\n }\n if let Some(cut) = self.shed_cut {\n let cut_landed = self.occupied() <= self.shed_floor && self.eviction_permits.is_empty();\n // A rising sample re-arms the walk down: the projection was built\n // from a measurement that no longer describes the node, so take a\n // fresh baseline. The band absorbs the jitter of a live sample and\n // decides nothing about what a cut must achieve.\n let rose = sample_bytes > cut.bytes.saturating_add(cut.bytes / 100);\n if cut_landed && !rose && !self.shedding_can_reach_resume_line(cut, sample_bytes) {\n return;\n }\n }\n // How far this resource sample asks the node to come down: a proportion\n // of what was just measured, because the effect of an eviction on the\n // sample is not visible until the next one.\n self.shed_floor = pressure::PressureConfig::release_target(load.resident_cells);\n self.shed_cut = Some(ShedCut {\n metric,\n bytes: sample_bytes,\n cells: load.resident_cells,\n });\n self.shed_toward_floor(effects);\n }\n\n /// Is there a heap this walk down can still take back whole?\n ///\n /// True when some isolate holds only evictable cells and it is not the\n /// last one. The last one is excluded deliberately: emptying it means\n /// walking the node to zero, which is the case the stopping condition\n /// exists to prevent, and the node keeps one heap's worth of working set\n /// instead.\n ///\n /// A cell the shell never placed counts against nothing here, so a node\n /// with no placement reported falls through to the projection unchanged.\n fn can_empty_an_isolate(&self) -> bool {\n let mut evictable: BTreeMap = BTreeMap::new();\n for (id, cell) in &self.cells {\n let Some(isolate) = cell.isolate else {\n continue;\n };\n let can_go = self.is_hibernatable(id)\n && cell\n .alarm\n .as_ref()\n .is_none_or(|alarm| matches!(alarm, AlarmState::Armed { covered: true, .. }));\n *evictable.entry(isolate).or_insert(true) &= can_go;\n }\n evictable.len() > 1 && evictable.values().any(|all_evictable| *all_evictable)\n }\n\n /// Can shedding still get the sample down to the resume line?\n ///\n /// Do not replace this with a band on the last two samples, at any width.\n /// A cut is a proportion of the cells while the sample also holds memory no\n /// cell owns, so once that fixed part dominates, a cut that helps moves the\n /// sample by less than the band and the walk down stops above a line it\n /// could reach. That is #50, and 5% and 1% both had it.\n ///\n /// The yield is wanted honest, not corrected: a node that keeps part of a\n /// stopped cell really does return less per eviction, and holding the\n /// working set is the right answer there. A lagging sample under-reports\n /// the yield for a sample or two, and the walk down survives that because\n /// a stop takes no fresh baseline: the sample keeps falling against the\n /// bytes of the last cut until the memory that came back late has landed\n /// against it.\n fn shedding_can_reach_resume_line(&self, cut: ShedCut, sample_bytes: u64) -> bool {\n let Some(resume_line) = self.config.pressure.resume_line(cut.metric) else {\n // Unreachable: `walk_metric` only names a measurement whose ceiling\n // is configured. With no line, the safe answer is that shedding\n // buys nothing, so the caller keeps the working set.\n return false;\n };\n // The projection below prices the next cut from the last one, which\n // only holds while every cut is worth the same. It is not: a cut that\n // takes an isolate's last cell returns that heap, and the heap is\n // most of what a cell costs. So while a heap can still be emptied,\n // the last cut does not describe the next one and the projection has\n // nothing to say.\n if self.can_empty_an_isolate() {\n return true;\n }\n let removed = cut.cells.saturating_sub(self.occupied());\n if removed == 0 {\n // Only a node holding nothing arrives here. Reaching this call\n // means `occupied` is at or below `release_target(cut.cells)`,\n // which sits strictly below `cut.cells` for any residency but\n // zero, so zero is the one residency where a cut can vacate\n // nothing and still be measured. It vacated nothing because there\n // was nothing to vacate, so there is no yield to project and the\n // division below would be by zero.\n //\n // Such a node is over its ceiling on memory no cell owns, and it\n // holds no cell to spend on the crossing. The two answers here\n // therefore differ only in which sample the baseline remembers,\n // and neither changes what the node does. This one lets the walk\n // down carry on, the way the isolate arm above does.\n return true;\n }\n let returned = cut.bytes.saturating_sub(sample_bytes);\n // One division, at the end: a per-cell yield rounded to a whole number\n // first is zero for any cell worth less than a byte more than the\n // rounding, which reports every walk down on a large node as futile.\n let projected = u128::from(returned).saturating_mul(self.occupied() as u128)\n / u128::from(removed as u64);\n let floor = u128::from(sample_bytes).saturating_sub(projected);\n // Inclusive: `classify` holds the latch only *above* the low watermark,\n // so a projection landing exactly on the line reaches a node that\n // releases. A strict `<` here wedges that node.\n floor <= u128::from(resume_line)\n }\n\n /// Continue a latched walk down as each eviction lands.\n ///\n /// `shed_one` stands down while the latch is hot, because its stopping\n /// condition -- the waiter got in -- cannot be met until the node is\n /// relieved. This is the other half: a stopping condition that can be met,\n /// so the node reaches its floor at the speed evictions complete rather\n /// than one per sampling period. Serialized like every other eviction\n /// path, so a walk down never puts the whole working set in flight.\n fn shed_toward_floor(&mut self, effects: &mut Vec) {\n if !self.shedding || self.occupied() <= self.shed_floor {\n return;\n }\n // Fill the permits rather than starting one and waiting for it. Each\n // proof is a round trip, so a serialized drain costs the number of\n // cells times that latency; running the bound's worth at once is the\n // difference between a walk down measured in seconds and one measured\n // in minutes.\n // Count what is already leaving against the target. `occupied` still\n // includes a cell whose proof is in flight, so comparing it directly\n // nominates cells the evictions already under way will account for,\n // and the node settles below its floor by up to the whole bound --\n // the mistake celld's eviction budget documents: spend the cut on\n // commit, never on nomination.\n while self.eviction_permits.len() < self.config.max_evictions\n && self.occupied().saturating_sub(self.eviction_permits.len()) > self.shed_floor\n {\n let Some(victim) = self.shed_candidate() else {\n return;\n };\n if self\n .begin_eviction(&victim, self.rebalances(), effects)\n .is_err()\n {\n return;\n }\n }\n }\n\n fn activity_finished(&mut self, request: RequestId, effects: &mut Vec) {\n // A write still on the output gate keeps its request pinned, so the\n // cell cannot be evicted before the write is proven durable. The unpin\n // moves to whichever path drains the last gate for this request.\n if self.barriers.values().any(|barrier| {\n matches!(&barrier.owner, GateOwner::Output(held) if held.request == request)\n || barrier.followers.iter().any(|held| held.request == request)\n }) {\n self.gate_pinned.insert(request);\n return;\n }\n if let Some(id) = self.deactivate_request(request) {\n let now = self.now_mono_ms;\n if let Some(cell) = self.cells.get_mut(&id) {\n cell.last_used_mono_ms = now;\n cell.last_used_request = cell.last_used_request.max(request);\n }\n }\n self.shed_one(effects);\n }\n\n /// Discard a runtime whose durability could not be proved.\n ///\n /// The caller has been told its write failed. If the cell stays resident,\n /// the very next read is served from the same unreplicated memory and\n /// returns that write anyway -- the node disagreeing with what it just\n /// told a client. Stopping with [`StopCause::Reset`] keeps no local\n /// snapshot, so the next activation restores from the bucket.\n ///\n /// Note this is a stop, not a truncation. The restored database is not\n /// invalidated by a failed proof; only this node's unreplicated additions\n /// to it are in doubt, and re-reading durable state is how they are\n /// discarded.\n fn reset_cell(&mut self, id: &str, epoch: Epoch, effects: &mut Vec) {\n if !matches!(\n self.cells.get(id).map(|cell| &cell.phase),\n Some(Phase::Resident { epoch: current }) if *current == epoch\n ) {\n return;\n }\n // Every other write still on the gate for this cell loses its runtime\n // here, so none of them may be acknowledged either.\n let doomed: Vec = self\n .barriers\n .iter()\n .filter(|(_, barrier)| barrier.cell == id && barrier.epoch == epoch)\n .map(|(op, _)| *op)\n .collect();\n for op in doomed {\n let Some(gate) = self.barriers.remove(&op) else {\n continue;\n };\n match gate.owner {\n GateOwner::Output(held) => {\n self.gate_pinned.remove(&held.request);\n effects.push(Effect::Release {\n request: held.request,\n channel: held.channel,\n result: Err(RequestError::DurabilityUnproven),\n });\n }\n // This proof can no longer succeed against a runtime being\n // discarded, so the alarm settles as a failed handler's does:\n // re-armed, with the wake entry left in place for the next\n // activation to find.\n GateOwner::Alarm { alarm, .. } => {\n let (now_ms, now_mono_ms) = (self.now_ms, self.now_mono_ms);\n self.alarm_finished(\n alarm,\n (gate.cell.clone(), gate.epoch),\n now_ms,\n now_mono_ms,\n Err(Failure::Ambiguous),\n effects,\n );\n }\n }\n for held in gate.followers {\n effects.push(Effect::Release {\n request: held.request,\n channel: held.channel,\n result: Err(RequestError::DurabilityUnproven),\n });\n }\n }\n // Requests already running against this runtime lose it here, exactly\n // as a fence does. Leaving them in `active_requests` would leave the\n // core believing a stopped cell is still serving, which its own\n // invariant catches.\n let running: Vec = self\n .active_requests\n .iter()\n .filter(|(_, cell)| cell.as_str() == id)\n .map(|(request, _)| *request)\n .collect();\n for request in running {\n self.deactivate_request(request);\n self.gate_pinned.remove(&request);\n }\n let op = self.cell_op(id);\n if let Some(cell) = self.cells.get_mut(id) {\n // The phase being replaced may hold an op that will now never\n // complete against it; its index entry must not outlive it.\n if let Some(stale) = phase_op(&cell.phase) {\n self.cell_ops.remove(&stale);\n }\n // A late output can fail after drain or swap nomination. Retire\n // its reservation with residency, or the stale index consumes\n // a drain handoff slot even after the runtime is gone.\n self.quiescing_cells.remove(id);\n set_phase(\n &mut self.occupied,\n cell,\n Phase::Cleaning {\n op,\n epoch,\n cause: StopCause::Reset,\n },\n );\n }\n effects.push(Effect::StopRuntime {\n op,\n cell: id.to_string(),\n epoch,\n cause: StopCause::Reset,\n bounded: false,\n });\n }\n\n /// How many cells are currently held resident by a non-hibernatable\n /// transport.\n fn outbound_pinned(&self) -> usize {\n self.cells\n .values()\n .filter(|cell| {\n cell.websockets\n .values()\n .any(|kind| *kind == WebSocketKind::Outbound)\n })\n .count()\n }\n\n fn websocket_opened(\n &mut self,\n id: &str,\n websocket: WebSocketId,\n kind: WebSocketKind,\n effects: &mut Vec,\n ) {\n if self.fenced {\n return;\n }\n // A quiescing cell is leaving this node, and a socket accepted now\n // would keep it active for the rest of the drain: the Sandbox SDK\n // reconnects its RPC socket the moment the drain closes it, and a\n // cell pinned that way waited out the whole handoff window on the\n // fleet, until the successor took it after the lease expired.\n let leaving = self.cells.get(id).is_some_and(|cell| cell.quiescing);\n if kind == WebSocketKind::Outbound && leaving {\n effects.push(Effect::CloseWebSocket {\n cell: id.to_string(),\n websocket,\n });\n return;\n }\n // A non-hibernatable transport holds its cell resident for as long as\n // it is open, so every one of them is a cell the node can never shed.\n // Pin the whole ceiling and there is nothing left to nominate:\n // residency cannot fall, admission waits on capacity that will never\n // be freed, and the node is wedged by its own applications. Cells that\n // already hold one are not counted again -- the budget is on how much\n // of the node is held, not on how many sockets exist.\n let already_pinned = self.cells.get(id).is_some_and(|cell| {\n cell.websockets\n .values()\n .any(|kind| *kind == WebSocketKind::Outbound)\n });\n let cell_outbound = self.cells.get(id).map_or(0, |cell| {\n cell.websockets\n .values()\n .filter(|kind| **kind == WebSocketKind::Outbound)\n .count()\n });\n if kind == WebSocketKind::Outbound\n && (cell_outbound >= self.config.max_outbound_websockets\n || (!already_pinned\n && !pressure::may_pin_outbound(\n self.outbound_pinned(),\n Some(self.config.max_resident),\n )))\n {\n effects.push(Effect::CloseWebSocket {\n cell: id.to_string(),\n websocket,\n });\n return;\n }\n let Some(cell) = self.cells.get_mut(id) else {\n return;\n };\n cell.websockets.insert(websocket, kind);\n }\n\n fn websocket_closed(&mut self, id: &str, websocket: WebSocketId, effects: &mut Vec) {\n if let Some(cell) = self.cells.get_mut(id) {\n cell.websockets.remove(&websocket);\n }\n self.shed_one(effects);\n }\n\n fn invalidate_remote(&mut self, id: &str, node: &str, epoch: Epoch) {\n self.node_lease_cache.remove(node);\n let rejected_sample = self.cells.get(id).and_then(|cell| match &cell.phase {\n Phase::Remote {\n node: current,\n epoch: current_epoch,\n cache: RemoteCache::CapacitySample { sampled_ms, .. },\n ..\n } if current == node && *current_epoch == epoch => Some(*sampled_ms),\n _ => None,\n });\n if let Some(sample) = rejected_sample {\n self.capacity_rejections.insert(node.to_string(), sample);\n self.capacity_reservations.remove(node);\n }\n let Some(cell) = self.cells.get_mut(id) else {\n return;\n };\n if matches!(\n &cell.phase,\n Phase::Remote {\n node: current,\n epoch: current_epoch,\n ..\n } if current == node && *current_epoch == epoch\n ) {\n set_phase(&mut self.occupied, cell, Phase::Inactive);\n }\n }\n\n fn fence(&mut self, effects: &mut Vec) {\n if self.fenced {\n return;\n }\n self.fenced = true;\n self.activation_waiters.clear();\n self.activation_permits.clear();\n self.capacity_waiters.clear();\n self.quiescing_cells.clear();\n self.adopting_cells.clear();\n self.active_requests.clear();\n self.active_cells.clear();\n self.admitted_gates.clear();\n self.admitted_gate_cells.clear();\n // Any write still waiting on the output gate loses its cell here, so it\n // must fail rather than be acknowledged — the fence and the fail are\n // atomic. A late DurableReached for a drained op is ignored.\n self.gate_pinned.clear();\n for (_, gate) in std::mem::take(&mut self.barriers) {\n // An alarm has no output to fail, and unlike `reset_cell` it is not\n // replayed either. The per-cell teardown below retires its firing\n // op and clears the alarm, and the wake entry is left where it is,\n // so the node that takes the cell next discovers it and fires\n // again.\n if let GateOwner::Output(held) = gate.owner {\n effects.push(Effect::Release {\n request: held.request,\n channel: held.channel,\n result: Err(RequestError::NodeFenced),\n });\n }\n for held in gate.followers {\n effects.push(Effect::Release {\n request: held.request,\n channel: held.channel,\n result: Err(RequestError::NodeFenced),\n });\n }\n }\n let ids: Vec = self.cells.keys().cloned().collect();\n for id in ids {\n let mut cell = self.cells.remove(&id).expect(\"id came from map\");\n match &cell.phase {\n Phase::Starting { op, epoch } | Phase::Publishing { op, epoch } => {\n self.timed_out_runtime_starts.remove(op);\n self.retired_runtime_ops.insert(*op, (id.clone(), *epoch));\n }\n _ => {}\n }\n if let Some(epoch) = runtime_epoch(&cell.phase) {\n let op = self.op();\n effects.push(Effect::StopRuntime {\n op,\n cell: id.clone(),\n epoch,\n cause: StopCause::Fence,\n bounded: false,\n });\n }\n self.finish_requests(&id, &mut cell, Err(RequestError::NodeFenced), effects);\n // The phase and alarm die here without completing; their ops can\n // never match a lookup again. The release op survives on purpose:\n // `owner_released` still consumes it after a fence, as it did\n // when resolution was a scan.\n if let Phase::EnsuringDurability { op, .. }\n | Phase::Cleaning {\n op,\n cause: StopCause::Evict { .. },\n ..\n } = cell.phase\n {\n self.finish_eviction(\n op,\n Err(EvictError::Cancelled(EvictCancellation::NodeFenced)),\n effects,\n );\n }\n if let Some(stale) = phase_op(&cell.phase) {\n self.cell_ops.remove(&stale);\n }\n if let Some(AlarmState::Firing { op, .. }) = cell.alarm {\n self.cell_ops.remove(&op);\n }\n set_phase(&mut self.occupied, &mut cell, Phase::Fenced);\n cell.waiting_for = None;\n cell.waiting_activation = None;\n cell.alarm = None;\n cell.alarm_wake = false;\n cell.websockets.clear();\n self.cells.insert(id, cell);\n }\n }\n\n fn compensate_retired_runtime(\n &mut self,\n op: OpId,\n result: Result<(), Failure>,\n effects: &mut Vec,\n ) {\n let Some((cell, epoch)) = self.retired_runtime_ops.remove(&op) else {\n return;\n };\n // Definite failure created nothing. Success or ambiguity may have\n // created/published a runtime after authority was revoked, so cleanup\n // is mandatory and idempotent.\n if result != Err(Failure::Definite) {\n let cleanup = self.op();\n effects.push(Effect::StopRuntime {\n op: cleanup,\n cell,\n epoch,\n // This completion lost node authority before it became\n // visible. It can close the database, but it cannot run the\n // durability pass that an ordinary live cleanup uses to make\n // the image a successor-epoch base.\n cause: StopCause::Fence,\n bounded: false,\n });\n }\n }\n}\n\n/// The sole state transition entry point.\n/// The monotonic instant an event carries, if it carries one. Events without\n/// a timestamp leave the remembered instant alone rather than resetting it.\n/// The wall-clock reading an event carried, if it carried one.\nfn event_now_ms(event: &Event) -> Option {\n match event {\n Event::StartNodeLease { now_ms, .. }\n | Event::SelfNodeLeaseRead { now_ms, .. }\n | Event::OwnerRead { now_ms, .. }\n | Event::NodeLeaseRead { now_ms, .. }\n | Event::TimerFired { now_ms, .. }\n | Event::AlarmObserved { now_ms, .. }\n | Event::AlarmFinished { now_ms, .. } => Some(*now_ms),\n _ => None,\n }\n}\n\nfn event_mono_ms(event: &Event) -> Option {\n match event {\n Event::StartNodeLease { now_mono_ms, .. }\n | Event::SelfNodeLeaseRead { now_mono_ms, .. }\n | Event::NodeLeaseCasCompleted { now_mono_ms, .. }\n | Event::RequestAt { now_mono_ms, .. }\n | Event::EvictRequested { now_mono_ms, .. }\n | Event::CapacityRequestAt { now_mono_ms, .. }\n | Event::HandoffRequestAt { now_mono_ms, .. }\n | Event::WebSocketRequestAt { now_mono_ms, .. }\n | Event::GateRequestAt { now_mono_ms, .. }\n | Event::OutputAt { now_mono_ms, .. }\n | Event::WakeHintAt { now_mono_ms, .. }\n | Event::TimerFired { now_mono_ms, .. }\n | Event::AlarmObserved { now_mono_ms, .. }\n | Event::AlarmFinished { now_mono_ms, .. }\n | Event::LoadSampled { now_mono_ms, .. }\n | Event::GenerationChanged { now_mono_ms, .. } => Some(*now_mono_ms),\n _ => None,\n }\n}\n\npub fn on_event(state: &mut State, event: Event) -> Vec {\n let mut effects = Vec::new();\n if let Some(now_mono_ms) = event_mono_ms(&event) {\n state.now_mono_ms = state.now_mono_ms.max(now_mono_ms);\n }\n if let Some(now_ms) = event_now_ms(&event) {\n state.now_ms = state.now_ms.max(now_ms);\n }\n match event {\n Event::StartNodeLease { now_ms, spec, .. } => {\n state.start_node_lease(now_ms, spec, &mut effects)\n }\n Event::NodeLogRecovered { op, result } => {\n state.node_log_recovered(op, result, &mut effects)\n }\n Event::NudgeNodeLease {\n now_ms,\n now_mono_ms,\n } => match &state.node_authority {\n NodeAuthority::Held(held) => {\n let generation = held.timer_generation;\n state.timer_fired(\n Timer::NodeLeaseRenew { generation },\n now_ms,\n now_mono_ms,\n &mut effects,\n );\n }\n // A write is in flight (or a read preceding one): its body was\n // serialized before this nudge's publish, so it cannot confirm\n // it. Remember, and hold_node_lease renews again on arrival.\n NodeAuthority::Writing { .. } | NodeAuthority::Reading { .. } => {\n state.nudge_pending = true;\n }\n _ => {}\n },\n Event::SelfNodeLeaseRead {\n op,\n now_ms,\n now_mono_ms,\n result,\n } => state.read_self_node_lease(op, now_ms, now_mono_ms, result, &mut effects),\n Event::NodeLeaseCasCompleted {\n op,\n now_mono_ms,\n result,\n stamped_log_state,\n } => {\n state.node_lease_cas_completed(op, now_mono_ms, result, stamped_log_state, &mut effects)\n }\n Event::LocalCellsRead { result } => state.local_cells_read(result, &mut effects),\n Event::TimerFired {\n timer,\n now_ms,\n now_mono_ms,\n } => state.timer_fired(timer, now_ms, now_mono_ms, &mut effects),\n Event::Request { request, cell } => {\n state.request(request, cell, false, false, &mut effects)\n }\n Event::RequestAt { request, cell, .. } => {\n state.request(request, cell, false, false, &mut effects)\n }\n Event::CapacityRequestAt { request, cell, .. } => {\n state.request(request, cell, true, false, &mut effects)\n }\n Event::HandoffRequestAt { request, cell, .. } => {\n state.request(request, cell, true, true, &mut effects)\n }\n Event::WebSocketRequestAt {\n request,\n cell,\n websocket,\n ..\n } => state.websocket_request(request, cell, websocket, &mut effects),\n Event::GateRequestAt { request, cell, .. } => {\n state.gate_request(request, cell, &mut effects)\n }\n Event::WorkerRequest { request } => state.worker_request(request, &mut effects),\n Event::BeginPreserve => state.begin_preserve(&mut effects),\n Event::Cancel { request } => state.cancel(request, &mut effects),\n Event::ActivityFinished { request } => state.activity_finished(request, &mut effects),\n Event::Output {\n request,\n channel,\n position,\n observed,\n epoch,\n } => state.output(request, channel, position, observed, epoch, &mut effects),\n Event::OutputAt {\n request,\n channel,\n position,\n observed,\n epoch,\n ..\n } => state.output(request, channel, position, observed, epoch, &mut effects),\n Event::DurableReached { op, result, source } => {\n state.durable_reached(op, result, source, &mut effects)\n }\n Event::OwnershipVerified { op, result } => {\n state.ownership_verified(op, result, &mut effects)\n }\n Event::WebSocketOpened {\n cell,\n websocket,\n kind,\n } => state.websocket_opened(&cell, websocket, kind, &mut effects),\n Event::WebSocketClosed { cell, websocket } => {\n state.websocket_closed(&cell, websocket, &mut effects)\n }\n Event::AlarmObserved {\n cell,\n alarm,\n covered,\n now_ms,\n now_mono_ms,\n } => state.alarm_observed(\n &cell,\n AlarmObservation {\n alarm,\n covered,\n firing_settled: false,\n },\n now_ms,\n now_mono_ms,\n &mut effects,\n ),\n Event::AlarmFinished {\n op,\n cell,\n epoch,\n now_ms,\n now_mono_ms,\n result,\n } => state.alarm_finished(op, (cell, epoch), now_ms, now_mono_ms, result, &mut effects),\n Event::WakeHint {\n cell,\n entry_ms,\n scope,\n } => state.wake_hint(cell, entry_ms, scope, &mut effects),\n Event::WakeHintAt {\n cell,\n entry_ms,\n scope,\n ..\n } => state.wake_hint(cell, entry_ms, scope, &mut effects),\n Event::OwnerRead { op, now_ms, result } => {\n state.owner_read(op, now_ms, result, &mut effects)\n }\n Event::NodeLeaseRead { op, now_ms, result } => {\n state.node_lease_read(op, now_ms, result, &mut effects)\n }\n Event::CapacityPeersRead { op, now_ms, result } => {\n state.capacity_peers_read(op, now_ms, result, &mut effects)\n }\n Event::OwnerCasCompleted { op, result } => {\n state.owner_cas_completed(op, result, &mut effects)\n }\n Event::OwnerReleased { op, result } => state.owner_released(op, result, &mut effects),\n Event::SuccessorAdopted { op, result } => state.successor_adopted(op, result, &mut effects),\n Event::RestoreCompleted { op, result } => state.restore_completed(op, result, &mut effects),\n Event::RuntimeStarted {\n op,\n isolate,\n generation,\n result,\n } => state.runtime_started(op, isolate, generation, result, &mut effects),\n Event::GenerationChanged {\n generation,\n max_age_ms,\n eager_classes,\n ..\n } => state.generation_changed(generation, max_age_ms, &eager_classes),\n Event::Published { op, result } => state.published(op, result, &mut effects),\n Event::DurabilityChecked { op, result } => state.durability_checked(\n op,\n result.map_err(|_| EvictFailure::Durability),\n &mut effects,\n ),\n Event::RuntimeStopped { op } => state.runtime_stopped(op, &mut effects),\n Event::RuntimeStopFailed { op } => state.runtime_stop_failed(\n op,\n EvictError::Failed(EvictFailure::RuntimeStop),\n &mut effects,\n ),\n Event::RuntimeStopCancelled { op } => state.runtime_stop_failed(\n op,\n EvictError::Cancelled(EvictCancellation::Activity),\n &mut effects,\n ),\n Event::Evict { cell } => state.evict(&cell, &mut effects),\n Event::EvictRequested { request, cell, .. } => {\n state.evict_requested(request, cell, &mut effects)\n }\n Event::LoadSampled { load, now_mono_ms } => {\n state.load_sampled(load, now_mono_ms, &mut effects)\n }\n Event::InvalidateRemote { cell, node, epoch } => {\n state.invalidate_remote(&cell, &node, epoch)\n }\n Event::NodeFenced => state.fence_node(HaltReason::NodeLeaseExpired, &mut effects),\n Event::ReleaseAll => state.release_all(&mut effects),\n Event::Rebalance { cells } => state.rebalance(cells, &mut effects),\n }\n state.pump_activations(&mut effects);\n state.pump_release(&mut effects);\n state.pump_swap(&mut effects);\n if cfg!(debug_assertions) {\n state.validate().expect(\"state invariant\");\n }\n state.arm_operation_deadlines(&mut effects);\n effects\n}\n\nfn is_queue_broker(id: &str) -> bool {\n id.split_once(':')\n .is_some_and(|(class, _)| class == queue::RESERVED_CLASS)\n}\n\nfn same_node_lease(left: &NodeLeaseRecord, right: &NodeLeaseRecord) -> bool {\n left.node == right.node\n && left.addr == right.addr\n && left.expires_ms == right.expires_ms\n && left.peer_protocol == right.peer_protocol\n && left.generation == right.generation\n // The folded log state distinguishes \"our record, untouched\" from\n // \"our record, fenced to recovering by a peer\":\n // ignoring it let an ambiguous renewal's readback adopt recovery's\n // etag and the NEXT renewal un-fenced the recovery. Any difference\n // here is authority loss.\n && left.log_state == right.log_state\n}\n\nfn phase_occupies_capacity(phase: &Phase) -> bool {\n matches!(\n phase,\n Phase::Acquiring { dormant: false, .. }\n | Phase::ReconcilingAcquire { dormant: false, .. }\n | Phase::Restoring { .. }\n | Phase::Starting { .. }\n | Phase::Publishing { .. }\n | Phase::EnsuringDurability { .. }\n | Phase::Cleaning { .. }\n | Phase::Resident { .. }\n )\n}\n\nfn phase_holds_activation(phase: &Phase) -> bool {\n matches!(\n phase,\n Phase::ReadingOwner { .. }\n | Phase::ReadingNodeLease { .. }\n | Phase::RecoveringOwnerLog { .. }\n | Phase::ReadingCapacity { .. }\n | Phase::WaitingCapacity\n | Phase::Acquiring { .. }\n | Phase::ReconcilingAcquire { .. }\n | Phase::Restoring { .. }\n | Phase::Starting { .. }\n | Phase::Publishing { .. }\n | Phase::Cleaning { .. }\n )\n}\n\n/// Identify phases that cannot exist outside one admitted cold route.\n///\n/// `Starting` and `Publishing` also serve a live generation swap, so the swap\n/// flag disambiguates them. `Cleaning` has cold and resident entry points and\n/// is therefore not a safe converse even though it retains an existing permit.\nfn phase_requires_activation_permit(phase: &Phase, swapping: bool) -> bool {\n matches!(\n phase,\n Phase::ReadingOwner { .. }\n | Phase::ReadingNodeLease { .. }\n | Phase::RecoveringOwnerLog { .. }\n | Phase::ReadingCapacity { .. }\n | Phase::WaitingCapacity\n | Phase::Acquiring { .. }\n | Phase::ReconcilingAcquire { .. }\n | Phase::Restoring { .. }\n ) || (!swapping && matches!(phase, Phase::Starting { .. } | Phase::Publishing { .. }))\n}\n\n/// The one gate every phase change passes through: it keeps `occupied`\n/// equal to the number of cells in a capacity-occupying phase, so nothing\n/// ever has to count them. `validate` asserts the equality by walking.\nfn set_phase(occupied: &mut usize, cell: &mut Cell, phase: Phase) {\n if !matches!(&phase, Phase::Resident { .. }) {\n cell.quiescing = false;\n }\n // Ownership follows the confirmed phases and survives the transient\n // wake phases between them; a phase that names another owner, no owner,\n // or a released record ends it.\n match (&phase, phase_owned_epoch(&phase)) {\n (_, Some(epoch)) => cell.held_epoch = Some(epoch),\n (Phase::Inactive | Phase::Remote { .. } | Phase::Fenced | Phase::Adopting { .. }, None) => {\n cell.held_epoch = None\n }\n _ => {}\n }\n // A proof covers positions of one runtime: the next runtime opens a new\n // connection whose positions start over, at the same epoch after a\n // generation swap. Every residency starts with no proof on record, so a\n // stale one cannot cover a new runtime's commit; a residency that kept\n // its runtime, as one a failed eviction returns to, proves once more.\n if matches!(&phase, Phase::Resident { .. }) {\n cell.proven_position = None;\n }\n // A swap lives on exactly these phases. Any other change — an eviction\n // that won the race, a fence, a failed start — ends it, and the cell\n // picks the current generation on its next activation instead.\n if !matches!(\n &phase,\n Phase::Resident { .. }\n | Phase::Cleaning {\n cause: StopCause::Swap,\n ..\n }\n | Phase::Starting { .. }\n | Phase::Publishing { .. }\n ) {\n cell.swapping = false;\n cell.swap_cancelled = false;\n }\n match (\n phase_occupies_capacity(&cell.phase),\n phase_occupies_capacity(&phase),\n ) {\n (false, true) => *occupied += 1,\n (true, false) => *occupied -= 1,\n _ => {}\n }\n cell.phase = phase;\n}\n\n/// The in-flight operation a phase is waiting on, if it is waiting on one.\nfn phase_op(phase: &Phase) -> Option {\n match phase {\n Phase::ReadingOwner { op }\n | Phase::ReadingNodeLease { op, .. }\n | Phase::RecoveringOwnerLog { op, .. }\n | Phase::ReadingCapacity { op, .. }\n | Phase::Acquiring { op, .. }\n | Phase::ReconcilingAcquire { op, .. }\n | Phase::Restoring { op, .. }\n | Phase::Starting { op, .. }\n | Phase::Publishing { op, .. }\n | Phase::EnsuringDurability { op, .. }\n | Phase::Cleaning { op, .. }\n | Phase::Adopting { op, .. } => Some(*op),\n _ => None,\n }\n}\n\n/// Return the epoch only after this node has confirmed its ownership CAS.\n/// Keep the census and handoff acknowledgement on this one classification so\n/// a new owned phase cannot silently affect routing without affecting load.\nfn phase_owned_epoch(phase: &Phase) -> Option {\n match phase {\n Phase::Restoring { spec, .. } => Some(spec.epoch),\n Phase::Starting { epoch, .. }\n | Phase::Publishing { epoch, .. }\n | Phase::EnsuringDurability { epoch, .. }\n | Phase::Cleaning { epoch, .. }\n | Phase::Dormant { epoch }\n | Phase::Resident { epoch } => Some(*epoch),\n _ => None,\n }\n}\n\nfn runtime_epoch(phase: &Phase) -> Option {\n match phase {\n Phase::Publishing { epoch, .. }\n | Phase::EnsuringDurability { epoch, .. }\n | Phase::Cleaning { epoch, .. }\n | Phase::Resident { epoch } => Some(*epoch),\n _ => None,\n }\n}\n","editable",false,"branch","documentNavigation","repo","denoland/celld","repoId","a23b8a10-92cf-4dc4-a7ab-a04e58298dde","ownerId","org:42048915","site","nexthub","private","admin","settings","maintainer","triage","starred","stars",1,"watchers","watching","pinned","notify","views",[134,135,136,137,138,139,140,141,142,143,144,145,146],"navigation",{"_55":56,"_91":92,"_93":94,"_95":96,"_97":98,"_99":100,"_101":102},"sources",[],"viewer","meta",{"_83":84,"_85":86,"_87":88,"_43":32,"_89":14,"_90":32,"_49":50,"_51":50},"branches",[14],"tags",[70,71,72,73,74,75,76,77,78,79,80,81,82],"cloneUrl","https://git.nexthub.sh/denoland/celld.git","v0.0.1","v0.0.2","v0.1.0","v0.2.0","v0.2.1","v0.3.0","v0.4.0","v0.4.1","v0.5.0","v0.5.1","v0.6.0","v0.6.1","v0.6.2","description","self-hosted, distributed Durable Objects","owner","denoland","section","","head","empty","pins",[116,117,118],"documents",{"_111":112,"_113":114},"code",{"_108":109,"_110":104},"automations",{"_106":50,"_107":-5},"delivery",{"_103":104,"_105":-5},"tag","7gTA52Z9a1vO6FP6eCQPix9zO4O","releases",0,"applications","workflows","deployments","commits",13,"changes","counts",{"_115":114},"total",38,"issue",{"_119":110,"_121":132,"_15":110,"_123":133,"_53":125,"_126":32,"_127":125,"_128":104},{"_119":129,"_121":130,"_15":129,"_123":131,"_53":125,"_126":32,"_127":125,"_128":104},{"_119":120,"_121":122,"_15":120,"_123":124,"_53":125,"_126":32,"_127":125,"_128":114},"id","issues","name","Issues","filters",{},true,"saved","pinnable","count","tasks","Tasks",{},"Changes",{},{"_119":110,"_121":132,"_15":110,"_123":177,"_53":125,"_126":32,"_127":125},{"_119":129,"_121":130,"_15":129,"_123":176,"_53":125,"_126":32,"_127":125},{"_119":120,"_121":122,"_15":120,"_123":175,"_53":125,"_126":32,"_127":125},{"_119":172,"_121":173,"_15":172,"_123":174,"_53":32,"_126":32,"_127":125},{"_119":169,"_121":170,"_15":169,"_123":171,"_53":32,"_126":32,"_127":125},{"_119":106,"_121":167,"_15":97,"_123":168,"_53":32,"_126":32,"_127":125},{"_119":107,"_121":164,"_15":165,"_123":166,"_53":32,"_126":32,"_127":125},{"_119":103,"_121":162,"_15":103,"_123":163,"_53":32,"_126":32,"_127":125},{"_119":158,"_121":159,"_15":160,"_123":161,"_53":32,"_126":32,"_127":125},{"_119":93,"_121":156,"_15":93,"_123":157,"_53":32,"_126":32,"_127":32},{"_119":153,"_121":154,"_15":153,"_123":155,"_53":32,"_126":32,"_127":32},{"_119":150,"_121":151,"_15":150,"_123":152,"_53":32,"_126":32,"_127":32},{"_119":147,"_121":148,"_15":147,"_123":149,"_53":32,"_126":32,"_127":32},"decisions","Decisions",{},"changelog","Changelog",{},"roadmap","Roadmap",{},"Documents",{},"activity","Activity","insights",{},"Releases",{},"Deployments","automations/deployments",{},"Workflows",{},"discussions","Discussions",{},"goals","Goals",{},{},{},{},"groups","user","agents",[],"now",1794435252904,"sidebar","expanded"]