[{"_1":2,"_3":-5,"_4":-5},"loaderData",{"_5":6,"_7":8,"_9":10,"_11":12},"actionData","errors","root",{"_180":-5,"_183":184,"_185":186},"routes/shell",{"_43":44,"_179":-5,"_180":-5,"_181":182},"routes/repo",{"_34":-7,"_35":36,"_37":38,"_39":40,"_41":42,"_43":44,"_45":32,"_46":32,"_47":32,"_48":32,"_49":32,"_50":32,"_51":52,"_53":52,"_54":32,"_55":-7,"_56":-7,"_57":58,"_59":60,"_61":62,"_63":-5,"_64":65,"_66":67,"_68":69,"_70":71},"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/output_gate.rs","oid","22b4766971b1b63875443bb30e688142fb13982a","size",17219,"commit","90b43017241f81189453d326d05948f388b34652","mounts",[],"source","plain",{"_13":14,"_15":16},"kind","text","// Copyright 2026 Deno Land Inc. Apache-2.0 license.\n\n//! The output gate: one choke point every egress passes through.\n//!\n//! # The rule\n//!\n//! Nothing that reveals a cell's state may leave this process while that cell\n//! has a write it cannot prove durable. A client told about a write that a\n//! crash then discards has been told something false, and no later message can\n//! retract a side effect that has already left.\n//!\n//! # The implementation contract\n//!\n//! [`Channel`] enumerates every route that can reveal a cell's state. The shell\n//! holds each effect and sends an [`Event::OutputAt`] with its route and the\n//! release-time monotonic instant. This module\n//! applies one durability rule to every route and returns the same route in an\n//! [`Effect::Release`].\n//!\n//! Each part of that contract has one implementation counterpart:\n//!\n//! | contract | implementation |\n//! | --- | --- |\n//! | enumerate every output route | the [`Channel`] enum |\n//! | apply one rule to every route | the `channel` argument to [`State::output`] |\n//! | release only revealable state | the output's barrier, or the barrier it trails, has settled |\n//! | let the held effect leave | [`Effect::Release`] with `Ok` |\n//!\n//! **The core does not read the channel.** It stores what the shell gave it and\n//! hands the same value back. A `match` on [`Channel`] inside this module would\n//! split the invariant into one independent rule per channel. Per-channel code\n//! belongs in the shell and is limited to how an effect is held and how it is\n//! released.\n//!\n//! # Why one choke point\n//!\n//! Earlier implementations split this rule between a core reader gate, a shell\n//! WebSocket queue, and an in-handler position check. Each mechanism had a\n//! different view of the outstanding writes, so one channel could miss a write\n//! that another mechanism knew about. One core barrier map gives every channel\n//! the same view.\n//!\n//! # The second way a barrier opens\n//!\n//! Every channel passes through [`State::output`]. Not every barrier is opened\n//! by one: `alarm_finished` opens its own when a firing's consuming commit is\n//! unproven, owned by [`GateOwner::Alarm`] rather than by an output. An alarm\n//! is not a channel -- it settles by replay into the core, not by\n//! release to the shell, so a [`Channel`] variant could not carry it -- but its\n//! barrier lives in the same per-cell map, which is what matters here: a\n//! read-only output trails the newest barrier on its cell without asking who\n//! opened it, so it trails an alarm's unproven write exactly as it trails a\n//! request's.\n//!\n//! An earlier alarm path proved its write inside [`Effect::FireAlarm`] without\n//! registering a core barrier. A read-only output could therefore miss that\n//! unproven write. The alarm now opens a barrier in the same map, so every\n//! reader observes it.\n//!\n//! # A streamed body is more than one output\n//!\n//! The model treats a response as atomic, and for a complete response it is:\n//! one release covers every byte. A response whose body streams is not atomic.\n//! The head is released while nothing is pending, and the producer then keeps\n//! running; a chunk it makes later can read a commit another event made after\n//! that release. So the shell takes one ticket per chunk on\n//! [`Channel::Response`], not one ticket for the response. The producer is a\n//! single pump that awaits each chunk's ticket before it asks for the next\n//! chunk, which is what keeps a held chunk from overtaking a released one.\n\nuse crate::*;\n\nimpl State {\n /// **The output gate. Every channel passes through here.**\n ///\n /// The implementation contract is meant to be checked line by line:\n ///\n /// - `channel` names the route. This function stores it but does not branch\n /// on it, so every channel uses one rule.\n /// - no barrier means that no unproven write remains for this output to\n /// reveal, so the core emits [`Effect::Release`] with `Ok`.\n ///\n /// A write opens its own barrier and waits for a proof that covers its\n /// position. A read-only output that observed a committed write no\n /// verified proof covers yet waits for that proof itself: the handler\n /// that wrote takes its ticket only when it answers, so a reader that\n /// answers before that finds no barrier for what it read. Otherwise a\n /// read-only output joins the newest barrier open on its cell, because a\n /// reader can start after a write commits and before its proof lands.\n /// Comparing the reader's own start and end positions therefore decides\n /// nothing.\n pub(crate) fn output(\n &mut self,\n request: RequestId,\n channel: Channel,\n position: Option,\n observed: Option,\n epoch: Option,\n effects: &mut Vec,\n ) {\n let held = Held { request, channel };\n // Authority is checked here, not only at request start: a handler can\n // start under a live node lease and reach its output after that lease\n // expired. Releasing then would reveal work a successor owner can\n // already have replaced.\n if !self.node_authoritative() {\n effects.push(Effect::Release {\n request,\n channel,\n result: Err(RequestError::NodeFenced),\n });\n return;\n }\n // A ticket sampled before its request pinned the cell names the epoch\n // it sampled at. A reset between the sample and the pin discarded that\n // epoch's unproven writes, and the pin then activated the next epoch:\n // a proof there covers nothing the ticket asks about, so the ticket\n // fails rather than acknowledge a discarded write. A request whose\n // cell is not resident at all fails here too, rather than by the\n // branch below: the write branch refuses that case itself, but the\n // read-only branch releases an active request whatever its cell's\n // phase, and a sampled epoch is a claim this check must settle\n // instead of leaving it to the reset's request cleanup.\n if let Some(sampled) = epoch {\n if self\n .resident_cell(request)\n .is_none_or(|(_, current)| current != sampled)\n {\n effects.push(Effect::Release {\n request,\n channel,\n result: Err(RequestError::DurabilityUnproven),\n });\n return;\n }\n }\n match (position, observed) {\n (Some(position), _) => self.open_write_barrier(held, position, effects),\n (None, Some(observed)) => match self.resident_cell(request) {\n Some((cell, epoch)) if self.proven_covers(&cell, epoch, observed) => {\n self.trail_open_barrier(held, effects)\n }\n Some((cell, epoch)) => self.hold_observed(held, cell, epoch, observed, effects),\n // The ticket names a commit no proof covers, on a cell that is\n // no longer resident to prove it. Fail closed, as a write in\n // that state does, rather than release what a read-only\n // ticket on a cell with nothing to prove would release.\n None => effects.push(Effect::Release {\n request,\n channel,\n result: Err(RequestError::DurabilityUnproven),\n }),\n },\n (None, None) => self.trail_open_barrier(held, effects),\n }\n }\n\n /// Whether a verified proof of this residency, at `epoch`, already covers\n /// `observed`.\n fn proven_covers(&self, cell: &CellId, epoch: Epoch, observed: u64) -> bool {\n self.cells\n .get(cell)\n .and_then(|record| record.proven_position)\n .is_some_and(|(proven_at, proven)| proven_at == epoch && proven >= observed)\n }\n\n /// Hold a read-only output that observed a committed write no verified\n /// proof covers yet. It joins a barrier on its cell whose position covers\n /// what it observed, and opens one at that position otherwise: the proof\n /// is then the reader's own, and the writer's later ticket opens a barrier\n /// a proof already satisfies.\n fn hold_observed(\n &mut self,\n held: Held,\n cell: CellId,\n epoch: Epoch,\n observed: u64,\n effects: &mut Vec,\n ) {\n if let Some((_, barrier)) = self.barriers.iter_mut().rev().find(|(_, barrier)| {\n barrier.cell == cell && barrier.epoch == epoch && barrier.position >= observed\n }) {\n barrier.followers.push(held);\n return;\n }\n self.open_write_barrier(held, observed, effects);\n }\n\n /// Record a verified proof, so a later read-only output that observed no\n /// more than `position` at this epoch releases without a proof of its own.\n fn note_proven(&mut self, cell: &CellId, epoch: Epoch, position: u64) {\n let Some(record) = self.cells.get_mut(cell) else {\n return;\n };\n record.proven_position = Some(match record.proven_position {\n Some((proven_at, proven)) if proven_at == epoch => (epoch, proven.max(position)),\n _ => (epoch, position),\n });\n }\n\n /// Open the output gate for a local write: hold its output until the\n /// cell's committed `position` is proven replicated. The request must still\n /// be a live local activity on its cell, resident at the epoch that\n /// committed the write; otherwise durability cannot be proven for it and\n /// the output fails rather than falsely acknowledging the write.\n fn open_write_barrier(&mut self, held: Held, position: u64, effects: &mut Vec) {\n let request = held.request;\n let Some((cell, epoch)) = self.resident_cell(request) else {\n effects.push(Effect::Release {\n request,\n channel: held.channel,\n result: Err(RequestError::DurabilityUnproven),\n });\n return;\n };\n let op = self.op();\n self.barriers.insert(\n op,\n Barrier {\n owner: GateOwner::Output(held),\n cell: cell.clone(),\n epoch,\n position,\n followers: Vec::new(),\n },\n );\n effects.push(Effect::AwaitDurable {\n op,\n cell,\n epoch,\n position,\n });\n }\n\n /// The cell a live local request runs against, and the epoch it is\n /// resident at. `None` when the request is not active or its cell is not\n /// resident.\n fn resident_cell(&self, request: RequestId) -> Option<(CellId, Epoch)> {\n self.active_requests.get(&request).and_then(|id| {\n match self.cells.get(id).map(|cell| &cell.phase) {\n Some(Phase::Resident { epoch }) => Some((id.clone(), *epoch)),\n _ => None,\n }\n })\n }\n\n /// Hold a read-only output behind the newest barrier open on its cell.\n ///\n /// Residency is not required here, unlike a write: a read on a cell that\n /// has left `Resident` has no barrier to find and no write of its own to\n /// prove, so it releases rather than fails.\n fn trail_open_barrier(&mut self, held: Held, effects: &mut Vec) {\n let Some(cell) = self.active_requests.get(&held.request).cloned() else {\n effects.push(Effect::Release {\n request: held.request,\n channel: held.channel,\n result: Err(RequestError::DurabilityUnproven),\n });\n return;\n };\n if let Some((_, barrier)) = self\n .barriers\n .iter_mut()\n .rev()\n .find(|(_, barrier)| barrier.cell == cell)\n {\n barrier.followers.push(held);\n } else {\n effects.push(Effect::Release {\n request: held.request,\n channel: held.channel,\n result: Ok(()),\n });\n }\n }\n\n /// A gated write's durability proof completed. Acknowledge the write only\n /// when the replica proved a position that *covers* it — a shorter proof\n /// (a lagging or lying replicator) fails it rather than acknowledging a\n /// write the node cannot actually restore. Any error fails it. A completion\n /// for a gate already drained (fence or deadline) is ignored — the\n /// versioned-op discipline used throughout the core.\n pub(crate) fn durable_reached(\n &mut self,\n op: OpId,\n result: Result,\n source: ProofSource,\n effects: &mut Vec,\n ) {\n let Some(gate) = self.barriers.get(&op) else {\n return;\n };\n let proven = matches!(result, Ok(durable) if durable >= gate.position);\n // A bucket proof reveals nothing until an ownership read confirms the\n // record still names this node at this epoch: \"durable in `e/`\" is not\n // durable if the prefix was orphaned, and the bucket cannot refuse a\n // stale writer. A fleet proof needs no read — the ensemble\n // arbitrated it: a takeover seals a member before restoring, so a\n // stale owner's ack-all fails closed.\n if proven && source == ProofSource::Bucket {\n let (cell, epoch) = (gate.cell.clone(), gate.epoch);\n effects.push(Effect::VerifyOwnership { op, cell, epoch });\n return;\n }\n self.settle_gate(op, proven, effects);\n }\n\n /// Ownership verification for a bucket-proof gate. A record that no longer\n /// names this node at this epoch fails the write exactly like an unproven\n /// durability result: refuse and reset, never acknowledge into an\n /// orphaned lineage.\n pub(crate) fn ownership_verified(\n &mut self,\n op: OpId,\n result: Result<(), Failure>,\n effects: &mut Vec,\n ) {\n if !self.barriers.contains_key(&op) {\n return;\n }\n self.settle_gate(op, result.is_ok(), effects);\n }\n\n /// Release one withheld output, unpinning it first so the cleanup that\n /// ends -- shedding, eviction -- is queued ahead of the release.\n pub(crate) fn release_held(\n &mut self,\n held: Held,\n result: Result<(), RequestError>,\n effects: &mut Vec,\n ) {\n if self.gate_pinned.remove(&held.request) {\n self.activity_finished(held.request, effects);\n }\n effects.push(Effect::Release {\n request: held.request,\n channel: held.channel,\n result,\n });\n }\n\n fn settle_gate(&mut self, op: OpId, proven: bool, effects: &mut Vec) {\n let Some(gate) = self.barriers.remove(&op) else {\n return;\n };\n let result = if proven {\n self.note_proven(&gate.cell, gate.epoch, gate.position);\n Ok(())\n } else {\n Err(RequestError::DurabilityUnproven)\n };\n match gate.owner {\n // Unpin before releasing: the cleanup this ends -- shedding,\n // eviction -- is queued ahead of the response, so a caller that\n // sees its write acknowledged sees the residency it released too.\n // `activity_finished` re-checks the barrier map, so a request\n // holding a second gated write simply re-pins itself here.\n GateOwner::Output(held) => self.release_held(held, result, effects),\n // The alarm settles only now. A proven commit replays the\n // observation the handler made, which routes through\n // `alarm_observed` and orders the consume-side wake-entry delete\n // -- after the proof, by construction. An unproven one takes the\n // re-arm branch a failed handler takes, so the entry stays\n // discoverable and at-least-once holds. The replay carries no\n // position, so it settles rather than opening a second barrier.\n GateOwner::Alarm {\n alarm,\n snapshot,\n covered,\n } => {\n let outcome = if proven {\n Ok((snapshot, covered, None))\n } else {\n Err(Failure::Ambiguous)\n };\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 outcome,\n effects,\n );\n }\n }\n // A follower is always an output: an alarm opens a barrier, it never\n // trails one.\n for held in gate.followers {\n self.release_held(held, result, effects);\n }\n if !proven {\n let (id, epoch) = (gate.cell.clone(), gate.epoch);\n self.reset_cell(&id, epoch, effects);\n }\n }\n}\n","editable",false,"branch","documentNavigation","repo","denoland/celld","repoId","a23b8a10-92cf-4dc4-a7ab-a04e58298dde","ownerId","org:42048915","organization",true,"site","nexthub","private","admin","settings","maintainer","triage","starred","stars",1,"watchers","watching","pinned","notify","views",[135,136,137,138,139,140,141,142,143,144,145,146,147],"navigation",{"_57":58,"_93":94,"_95":96,"_97":98,"_99":100,"_101":102,"_103":104},"sources",[],"viewer","meta",{"_85":86,"_87":88,"_89":90,"_45":32,"_91":14,"_92":32,"_51":52,"_53":52},"branches",[14],"tags",[72,73,74,75,76,77,78,79,80,81,82,83,84],"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",[118,119,120],"documents",{"_113":114,"_115":116},"code",{"_110":111,"_112":106},"automations",{"_108":52,"_109":-5},"delivery",{"_105":106,"_107":-5},"tag","7gTA52Z9a1vO6FP6eCQPix9zO4O","releases",0,"applications","workflows","deployments","commits",13,"changes","counts",{"_117":116},"total",38,"issue",{"_121":112,"_123":133,"_15":112,"_125":134,"_55":42,"_127":32,"_128":42,"_129":106},{"_121":130,"_123":131,"_15":130,"_125":132,"_55":42,"_127":32,"_128":42,"_129":106},{"_121":122,"_123":124,"_15":122,"_125":126,"_55":42,"_127":32,"_128":42,"_129":116},"id","issues","name","Issues","filters",{},"saved","pinnable","count","tasks","Tasks",{},"Changes",{},{"_121":112,"_123":133,"_15":112,"_125":178,"_55":42,"_127":32,"_128":42},{"_121":130,"_123":131,"_15":130,"_125":177,"_55":42,"_127":32,"_128":42},{"_121":122,"_123":124,"_15":122,"_125":176,"_55":42,"_127":32,"_128":42},{"_121":173,"_123":174,"_15":173,"_125":175,"_55":32,"_127":32,"_128":42},{"_121":170,"_123":171,"_15":170,"_125":172,"_55":32,"_127":32,"_128":42},{"_121":108,"_123":168,"_15":99,"_125":169,"_55":32,"_127":32,"_128":42},{"_121":109,"_123":165,"_15":166,"_125":167,"_55":32,"_127":32,"_128":42},{"_121":105,"_123":163,"_15":105,"_125":164,"_55":32,"_127":32,"_128":42},{"_121":159,"_123":160,"_15":161,"_125":162,"_55":32,"_127":32,"_128":42},{"_121":95,"_123":157,"_15":95,"_125":158,"_55":32,"_127":32,"_128":32},{"_121":154,"_123":155,"_15":154,"_125":156,"_55":32,"_127":32,"_128":32},{"_121":151,"_123":152,"_15":151,"_125":153,"_55":32,"_127":32,"_128":32},{"_121":148,"_123":149,"_15":148,"_125":150,"_55":32,"_127":32,"_128":32},"decisions","Decisions",{},"changelog","Changelog",{},"roadmap","Roadmap",{},"Documents",{},"activity","Activity","insights",{},"Releases",{},"Deployments","automations/deployments",{},"Workflows",{},"discussions","Discussions",{},"goals","Goals",{},{},{},{},"groups","user","agents",[],"now",1794712329248,"sidebar","expanded"]