[{"_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",{"_40":41,"_178":-5,"_179":-5,"_180":181},"routes/repo",{"_33":-7,"_34":35,"_36":37,"_38":39,"_40":41,"_42":31,"_43":31,"_44":31,"_45":31,"_46":31,"_47":31,"_48":49,"_50":49,"_51":31,"_52":-7,"_53":-7,"_54":55,"_56":57,"_58":59,"_60":-5,"_61":62,"_63":64,"_65":66,"_67":68},"routes/repo.blob",{"_13":14,"_15":16,"_17":18,"_19":20,"_21":14,"_22":23,"_24":-5,"_25":26,"_27":28,"_28":29,"_30":31,"_32":14},"ref","90b43017241f81189453d326d05948f388b34652","path","crates/celld/runtime.rs","oid","74941d5e7fa94e2a3509327e6801f1103ca6b80f","size",139873,"commit","mounts",[],"source","plain",{"_13":14,"_15":16},"kind","text","// Copyright 2026 Deno Land Inc. Apache-2.0 license.\n\n// RuntimeManager is the V8 cell-host arm, so its executor and observability\n// clocks remain ambient.\n#![allow(clippy::disallowed_methods)]\n\n//! V8 runtime materialization behind core-authorized lifecycle effects.\n//!\n//! The manager owns handles and filesystem paths, never lifecycle policy.\n//! StartRuntime, Publish, and StopRuntime decide when a cell handle moves from\n//! starting to externally dispatchable to closed.\n\nuse crate::asyncrt;\npub(crate) use crate::engine_api::StopMode;\nuse crate::generation::{Generation, GenerationId};\nuse crate::js::{self, CellJob, CellStorage, HttpResponse, Worker, WorkerConfig};\n\npub use crate::clean_reload::take_clean_reload_generation;\npub use crate::clean_reload::write_clean_reload_marker;\npub use crate::engine_api::admission_wait;\npub use crate::engine_api::AlarmObserver;\npub use crate::engine_api::RuntimeFetch;\npub use crate::engine_api::RuntimeOptions;\npub use crate::engine_api::ServiceFetch;\npub use crate::ltx_replication::Replication;\nuse anyhow::{anyhow, Context};\nuse futures_util::StreamExt as _;\nuse std::collections::{BTreeMap, BTreeSet, HashMap, VecDeque};\nuse std::path::Path;\nuse std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};\nuse std::sync::{Arc, Mutex};\nuse std::time::{Duration, Instant};\n\nconst REMOTE_ABORT_TTL: Duration = Duration::from_secs(600);\nconst REMOTE_COMPLETION_TTL: Duration = Duration::from_secs(60);\nconst MAX_REMOTE_PENDING_ABORTS: usize = 65_536;\n#[doc(hidden)]\npub const MAX_REMOTE_COMPLETIONS: usize = 65_536;\nconst MAX_ALARM_COMPLETIONS: usize = 65_536;\n\n/// How often the isolate pool gives back what it no longer needs.\nconst REAP_INTERVAL: Duration = Duration::from_secs(30);\n\n/// How long a reclaim waits before it retries when admission holds the pool.\nconst RECLAIM_RETRY: Duration = Duration::from_millis(10);\n\n/// How often a suspended request re-reads its cancellation flag. Matches the\n/// blocking run loop's own cap, which exists for the same reason: a client\n/// disconnect is raised on another thread and has nothing to wake this one.\nconst CANCELLATION_TICK: Duration = Duration::from_millis(10);\n/// Cell fetches that one target can hold before celld refuses excess work.\n///\n/// The last nonsaturated Queue step in the AWS partition staircase held\n/// approximately 40 requests by Little's Law. A limit of 64 keeps that step\n/// below the gate and prevents one target from consuming hundreds of client\n/// slots after its throughput stops increasing.\npub const DEFAULT_MAX_CELL_REQUESTS: usize = 64;\n\n/// Serialize each publisher that owns this lifetime with the target request\n/// driver's final cancellation cleanup. Without this shared lifetime, the\n/// driver can clear and exit between a publisher observing a pending reply and\n/// publishing its abort. This race leaves a tombstone with no remaining driver.\n///\n/// This hidden public type is the embedder boundary for a stateless request.\n/// An embedder can give clones to abort publishers, but it must give the same\n/// lifetime to the driver through [`Self::drive_stateless_fetch`].\n#[doc(hidden)]\npub struct RequestCancellationLifetime {\n request_id: js::RequestId,\n finished: Mutex,\n}\n\nimpl RequestCancellationLifetime {\n #[doc(hidden)]\n pub fn stateless() -> Arc {\n Self::from_request_id(js::next_request_id())\n }\n\n fn from_request_id(request_id: js::RequestId) -> Arc {\n Arc::new(Self {\n request_id,\n finished: Mutex::new(false),\n })\n }\n\n #[doc(hidden)]\n pub fn request_id(&self) -> js::RequestId {\n self.request_id\n }\n\n /// Publish an abort from an embedder that owns this lifetime.\n #[doc(hidden)]\n pub fn publish_abort(&self) {\n #[cfg(all(test, celld_internal_tests))]\n // Pause before taking `finished`: the test must let retirement win\n // this ordering without deadlocking on the very lock under test.\n js::pause_abort_request_if_armed_for_test(self.request_id);\n let finished = self\n .finished\n .lock()\n .expect(\"request cancellation lifetime poisoned\");\n if !*finished {\n js::abort_request(self.request_id);\n }\n }\n\n pub(crate) fn finish(&self) {\n let mut finished = self\n .finished\n .lock()\n .expect(\"request cancellation lifetime poisoned\");\n js::clear_request_cancellation(self.request_id);\n *finished = true;\n }\n\n /// Drive one cancellable stateless fetch with this same lifetime.\n ///\n /// This function constructs the driver future and its retirement guard\n /// before it returns. Thus, dropping the future before its first poll also\n /// retires the request. The request ID is derived here, so a publisher\n /// cannot name one request while the driver retires another request.\n #[doc(hidden)]\n pub fn drive_stateless_fetch(\n self: Arc,\n slot: Arc,\n url: String,\n method: String,\n body: js::RequestBody,\n headers: Vec<(String, String)>,\n reply: tokio::sync::oneshot::Sender>,\n ) -> impl std::future::Future + Send + 'static {\n let job = stateless_fetch_job_factory(None, url, method, body, headers, Some(self))(reply);\n drive_affiliated(slot.affiliate(), job, None)\n }\n\n /// Drive a fully built fetch job, such as a loaded Worker's, with this\n /// lifetime. The job's request id is overwritten from the lifetime, so the\n /// caller cannot pair an abort publisher with a different request.\n pub(crate) fn drive_fetch_job(\n self: Arc,\n slot: Arc,\n mut job: crate::WorkerJob,\n ) -> impl std::future::Future + Send + 'static {\n if let crate::WorkerJob::Fetch { request_id, .. } = &mut job {\n *request_id = Some(self.request_id);\n }\n let job = StatelessWorkerJob {\n job,\n cancellation: RequestCancellationGuard::shared(self),\n };\n drive_affiliated(slot.affiliate(), job, None)\n }\n}\n\n/// A stateless job and the only cancellation lifetime that can name it.\n/// Cancellable constructors derive the copied JavaScript request id from the\n/// lifetime, so a caller cannot pair one id with another lifetime.\nstruct StatelessWorkerJob {\n job: crate::WorkerJob,\n cancellation: RequestCancellationGuard,\n}\n\nimpl StatelessWorkerJob {\n fn fetch(\n entrypoint: Option,\n url: String,\n method: String,\n body: js::RequestBody,\n headers: Vec<(String, String)>,\n cancellation: RequestCancellationGuard,\n reply: tokio::sync::oneshot::Sender>,\n ) -> Self {\n let request_id = cancellation\n .lifetime()\n .map(|lifetime| lifetime.request_id());\n Self {\n job: crate::WorkerJob::Fetch {\n queued_at: Instant::now(),\n entrypoint,\n invocation_limits: None,\n url,\n method,\n body,\n headers,\n request_id,\n tail_report: None,\n reply,\n },\n cancellation,\n }\n }\n\n fn driver_owned(job: crate::WorkerJob) -> Self {\n // This compatibility constructor is for a job with no publisher. A\n // publisher must give its lifetime to the fetch factory because a\n // copied request id cannot prove that the publisher and driver match.\n let lifetime = match &job {\n crate::WorkerJob::Fetch { request_id, .. } => {\n request_id.map(RequestCancellationLifetime::from_request_id)\n }\n crate::WorkerJob::Rpc { .. } | crate::WorkerJob::Queue { .. } => None,\n };\n let cancellation = RequestCancellationGuard::new(lifetime);\n Self { job, cancellation }\n }\n}\n\nfn stateless_fetch_job_factory(\n entrypoint: Option,\n url: String,\n method: String,\n body: js::RequestBody,\n headers: Vec<(String, String)>,\n cancellation: Option>,\n) -> impl FnOnce(tokio::sync::oneshot::Sender>) -> StatelessWorkerJob {\n // The factory crosses the admission await, so it owns retirement until it\n // transfers the same guard into the spawned driver's job.\n let cancellation = RequestCancellationGuard::new(cancellation);\n move |reply| {\n StatelessWorkerJob::fetch(entrypoint, url, method, body, headers, cancellation, reply)\n }\n}\n\n/// Remove a request's cancellation state when its driver leaves. Each\n/// publisher for this request must share this lifetime, so publication\n/// serializes with cleanup.\nstruct RequestCancellationGuard(Option>);\n\nimpl RequestCancellationGuard {\n fn new(lifetime: Option>) -> Self {\n Self(lifetime)\n }\n\n fn shared(lifetime: Arc) -> Self {\n Self::new(Some(lifetime))\n }\n\n fn lifetime(&self) -> Option<&Arc> {\n self.0.as_ref()\n }\n}\n\nimpl Drop for RequestCancellationGuard {\n fn drop(&mut self) {\n if let Some(lifetime) = &self.0 {\n lifetime.finish();\n }\n }\n}\n\nasync fn drive_with_request_cancellation(\n driving: impl std::future::Future + Send + 'static,\n cancellation: RequestCancellationGuard,\n) {\n // The caller constructs the guard before it creates this future. The\n // future therefore owns retirement before its first poll, and this local\n // keeps that ownership until the complete driver future returns.\n let _request_cancellation = cancellation;\n driving.await;\n}\n\n#[derive(Clone)]\n#[doc(hidden)]\npub enum RemoteRequestState {\n PendingAbort(Instant),\n Active(Arc),\n Completed(Instant),\n}\n\n#[derive(Default)]\n#[doc(hidden)]\npub struct RemoteRequestRegistry {\n pub states: HashMap,\n pending: VecDeque<(Instant, js::RequestId)>,\n completed: VecDeque<(Instant, js::RequestId)>,\n}\n\nimpl RemoteRequestRegistry {\n fn prune(&mut self) {\n while self.pending.len() > MAX_REMOTE_PENDING_ABORTS\n || self\n .pending\n .front()\n .is_some_and(|(created, _)| created.elapsed() >= REMOTE_ABORT_TTL)\n {\n let (created, request) = self.pending.pop_front().expect(\"checked pending abort\");\n if matches!(\n self.states.get(&request),\n Some(RemoteRequestState::PendingAbort(current)) if *current == created\n ) {\n self.states.remove(&request);\n }\n }\n while self.completed.len() > MAX_REMOTE_COMPLETIONS\n || self\n .completed\n .front()\n .is_some_and(|(created, _)| created.elapsed() >= REMOTE_COMPLETION_TTL)\n {\n let (created, request) = self.completed.pop_front().expect(\"checked completion\");\n if matches!(\n self.states.get(&request),\n Some(RemoteRequestState::Completed(current)) if *current == created\n ) {\n self.states.remove(&request);\n }\n }\n }\n\n fn pending_abort(&mut self, request: js::RequestId) {\n let created = Instant::now();\n self.states\n .insert(request, RemoteRequestState::PendingAbort(created));\n self.pending.push_back((created, request));\n }\n\n #[doc(hidden)]\n pub fn completed(&mut self, request: js::RequestId) {\n let created = Instant::now();\n self.states\n .insert(request, RemoteRequestState::Completed(created));\n self.completed.push_back((created, request));\n self.prune();\n }\n\n /// Record that `request` has been handed to a cell isolate.\n #[doc(hidden)]\n pub fn active(&mut self, request: js::RequestId) -> Arc {\n let lifetime = RequestCancellationLifetime::from_request_id(request);\n self.states\n .insert(request, RemoteRequestState::Active(lifetime.clone()));\n lifetime\n }\n\n /// Record and publish a hang-up for `request`, and report whether it\n /// reached an active fetch.\n ///\n /// An `Active` fetch is tombstoned here instead of being left in place.\n /// `fetch_cell` is what normally retires an id, but a caller that\n /// disconnects mid-fetch has that future dropped underneath it, so the\n /// retiring `completed()` never runs. `prune` reclaims only the ids it can\n /// reach through `pending` or `completed`, so without this tombstone the\n /// entry stays in `states` for the life of the process, and one routine\n /// disconnect leaks one entry. A `fetch_cell` that does outlive the abort\n /// enqueues a fresher tombstone, and `prune`'s generation guard discards\n /// the stale enqueue without disturbing the live entry.\n #[doc(hidden)]\n pub fn abort(&mut self, request: js::RequestId) -> bool {\n match self.states.get(&request).cloned() {\n Some(RemoteRequestState::Active(lifetime)) => {\n self.completed(request);\n // Publish while the registry still owns the state transition,\n // so every direct abort reaches the same lifetime handshake.\n lifetime.publish_abort();\n true\n }\n Some(RemoteRequestState::Completed(_) | RemoteRequestState::PendingAbort(_)) => false,\n None => {\n self.pending_abort(request);\n false\n }\n }\n }\n}\n\n#[derive(Default)]\nstruct AlarmRequestRegistry {\n states: BTreeMap<(String, celld_logic::OpId), AlarmRequestState>,\n completed: VecDeque<(String, celld_logic::OpId)>,\n}\n\n#[derive(Clone, Copy)]\nenum AlarmRequestState {\n PendingAbort,\n Active(js::RequestId),\n Completed,\n}\n\nimpl AlarmRequestRegistry {\n fn begin(&mut self, cell: String, op: celld_logic::OpId, request: js::RequestId) -> bool {\n let key = (cell, op);\n let cancel = matches!(\n self.states.remove(&key),\n Some(AlarmRequestState::PendingAbort)\n );\n assert!(\n self.states\n .insert(key, AlarmRequestState::Active(request))\n .is_none(),\n \"one shell alarm task per core operation\"\n );\n cancel\n }\n\n fn finish(&mut self, cell: &str, op: celld_logic::OpId, request: js::RequestId) {\n let key = (cell.to_string(), op);\n if matches!(\n self.states.get(&key),\n Some(AlarmRequestState::Active(active)) if *active == request\n ) {\n self.states\n .insert(key.clone(), AlarmRequestState::Completed);\n self.completed.push_back(key);\n }\n while self.completed.len() > MAX_ALARM_COMPLETIONS {\n let completed = self.completed.pop_front().expect(\"checked completion\");\n if matches!(\n self.states.get(&completed),\n Some(AlarmRequestState::Completed)\n ) {\n self.states.remove(&completed);\n }\n }\n }\n\n fn aborting(&mut self, cell: &str, op: celld_logic::OpId) -> Option {\n let key = (cell.to_string(), op);\n match self.states.get(&key).copied() {\n Some(AlarmRequestState::Active(request)) => Some(request),\n Some(AlarmRequestState::PendingAbort | AlarmRequestState::Completed) => None,\n None => {\n self.states.insert(key, AlarmRequestState::PendingAbort);\n None\n }\n }\n }\n}\n\nstruct AlarmRequestGuard {\n registry: Arc>,\n cell: String,\n op: celld_logic::OpId,\n request: js::RequestId,\n}\n\nimpl Drop for AlarmRequestGuard {\n fn drop(&mut self) {\n self.registry\n .lock()\n .expect(\"alarm registry poisoned\")\n .finish(&self.cell, self.op, self.request);\n }\n}\n\n/// The isolate pool's limits, built from the environment here because the\n/// decision core never reads it.\n///\n/// `max_requests` is the node's only bound on stateless memory, and it is\n/// live: `Slot::affiliate` counts an affiliation for a request's whole life,\n/// `observe` reports it, and `isolate::admit` refuses against it. Unset\n/// means unbounded, not unwired — `engine/load-under-pressure.md` measures\n/// `CELLD_MAX_REQUESTS=32` admitting 641 rps against a theoretical 640.\npub fn pool_limits() -> celld_logic::isolate::PoolLimits {\n const GROW_AT: usize = 2;\n const SHRINK_UNDER: usize = 1;\n const MAX_CELLS_PER_ISOLATE: usize = 32;\n let cores = std::thread::available_parallelism()\n .map(|n| n.get())\n .unwrap_or(4);\n celld_logic::isolate::PoolLimits {\n // These thresholds form one hysteresis policy, so they are constants\n // rather than two independently configurable values.\n grow_at: GROW_AT,\n shrink_under: SHRINK_UNDER,\n max_stateless: env_usize(\"CELLD_MAX_STATELESS_ISOLATES\").unwrap_or(cores),\n max_requests: env_usize(\"CELLD_MAX_REQUESTS\"),\n // This is an engine blast-radius policy. The resident-cell and RSS\n // limits are the operator controls for node memory.\n max_cells: MAX_CELLS_PER_ISOLATE,\n }\n}\n\nfn env_usize(name: &str) -> Option {\n crate::env_vars::positive(name).expect(\"validated positive runtime limit\")\n}\n\n#[derive(Clone)]\n#[doc(hidden)]\npub struct StatelessRuntime {\n pub node: Arc,\n pub region: Arc,\n /// The isolates fetch runs on, entered one turn at a time from whichever\n /// tokio worker is driving the request.\n pub isolates: Arc,\n}\n\n#[derive(Clone, Copy)]\n#[doc(hidden)]\npub enum StatelessVerb {\n Fetch,\n Rpc,\n Queue,\n}\n\nimpl StatelessVerb {\n fn task_died(self, error: tokio::task::JoinError) -> anyhow::Error {\n match self {\n Self::Fetch => anyhow!(\"stateless request task died: {error}\"),\n Self::Rpc => anyhow!(\"stateless RPC task died: {error}\"),\n Self::Queue => anyhow!(\"stateless queue task died: {error}\"),\n }\n }\n\n fn dropped_result(self) -> anyhow::Error {\n match self {\n Self::Fetch => anyhow!(\"stateless Worker dropped response\"),\n Self::Rpc => anyhow!(\"stateless Worker dropped RPC result\"),\n Self::Queue => anyhow!(\"stateless Worker dropped queue result\"),\n }\n }\n}\n\nstruct CellHandle {\n epoch: u64,\n /// The application generation whose isolate holds this cell, reported\n /// when the cell leaves for a swap.\n generation: GenerationId,\n startup_us: u64,\n /// The cell's claim on the isolate holding its realm. An event knows\n /// where to run from it, and dropping it gives the placement back.\n residency: crate::pool::Residency,\n /// The source value and wake revision stay together in the registry.\n /// Reading a newer flusher revision beside an older cached value would\n /// authorize that old value to delete a later arm's entry.\n alarm: celld_logic::wake::AlarmSnapshot,\n requests: Arc,\n}\n\nstruct CellRequestAdmission {\n in_flight: AtomicUsize,\n saturated: AtomicBool,\n limit: usize,\n}\n\nimpl CellRequestAdmission {\n fn acquire(self: &Arc) -> Option {\n let admitted = self\n .in_flight\n .fetch_update(Ordering::AcqRel, Ordering::Acquire, |held| {\n (held < self.limit).then_some(held + 1)\n })\n .is_ok();\n if admitted {\n // A successful admission proves that the target left its prior\n // saturated interval. Reset here as well as on release because a\n // refusal and a release can race between the failed count check\n // and the saturation transition.\n self.saturated.store(false, Ordering::Release);\n }\n admitted.then(|| CellRequestPermit(self.clone()))\n }\n}\n\nstruct CellRequestPermit(Arc);\n\nimpl Drop for CellRequestPermit {\n fn drop(&mut self) {\n let held = self.0.in_flight.fetch_sub(1, Ordering::AcqRel);\n if held == self.0.limit {\n self.0.saturated.store(false, Ordering::Release);\n }\n }\n}\n\nstruct AdmittedCellRequest {\n affiliation: crate::pool::Affiliation,\n permit: CellRequestPermit,\n}\n\n/// How a turn's alarm move reaches the host. `drive_cell` calls it with\n/// what `take_alarm_moves` drained; it caches the value on the cell's\n/// handle and forwards a real change to the observer.\n#[doc(hidden)]\npub type AlarmReporter = Arc;\n\n/// Each committed installation reaches the host, including same-time re-arms.\n/// Duplicate observations of that installation are deduplicated. A\n/// scope without a handle was stopped mid-flight; its ActivityFinished\n/// report is gone with it, so a move for it says nothing and is dropped.\n///\n/// The observer is called under the registry lock, and `with_alarm` reads\n/// and reports under the same lock, so what reaches the core is monotone\n/// with the cache: a stale end-of-request read cannot land *after* a\n/// fresher report and unarm an alarm the core just learned about — the\n/// core overwrites on observation and deletes the wake entry on `None`,\n/// so that ordering loses the alarm outright. The observer only sends on\n/// an unbounded channel, so holding the lock across it cannot block.\nfn alarm_reporter(cells: &Arc>, observe: &AlarmObserver) -> AlarmReporter {\n let cells_ = cells.clone();\n let observe_ = observe.clone();\n Arc::new(\n move |scope: String, alarm: celld_logic::wake::AlarmSnapshot| {\n let mut registry = cells_.lock().expect(\"cell registry poisoned\");\n let handle = if let Some(handle) = registry.published.get_mut(&scope) {\n Some(handle)\n } else {\n registry.starting.get_mut(&scope)\n };\n if let Some(handle) = handle {\n let changed = handle.alarm.at_ms() != alarm.at_ms();\n // Refresh the revision even when a same-time rearm needs no timer\n // change. An ActivityFinished must use that source's snapshot.\n handle.alarm = alarm;\n if changed {\n observe_(scope, alarm);\n }\n }\n },\n )\n}\n\n#[derive(Default)]\nstruct CellRegistry {\n starting: HashMap,\n published: HashMap,\n}\n\n#[derive(Clone)]\npub struct RuntimeManager {\n /// What the node's cell runtime holds whichever engine runs the cells.\n core: crate::cell_runtime::CellRuntime,\n cells: Arc>,\n alarm_reporter: AlarmReporter,\n /// A peer abort can arrive before the forwarded fetch. The tombstone and\n /// cell enqueue share this lock so neither ordering can lose cancellation.\n remote_requests: Arc>,\n /// Shell alarm tasks keyed by the core firing operation. A shutdown\n /// cancellation must target the firing it observed, not a later retry.\n alarm_requests: Arc>,\n max_cell_requests: usize,\n}\n\n/// The engine-neutral half, so a method of either reads as the manager's own.\nimpl std::ops::Deref for RuntimeManager {\n type Target = crate::cell_runtime::CellRuntime;\n fn deref(&self) -> &crate::cell_runtime::CellRuntime {\n &self.core\n }\n}\n\n/// The engine's part of `Generation::build`: V8 starts once per process.\npub(crate) fn init_engine() {\n init_v8();\n}\n\n/// The isolates a script's cells live in, built from the script's config.\npub(crate) fn cell_pool(config: Arc) -> Arc {\n Arc::new(crate::pool::Pool::new(\n pool_limits(),\n admission_wait(),\n Box::new(move || load_cell_isolate(config.clone())),\n ))\n}\n\n/// Wait for one cell fetch reply without retaining the internal receiver after\n/// its external caller disconnects. Dropping that receiver is the direct wake\n/// for a reply that already belongs to a detached wake-entry gate task.\nasync fn receive_cell_fetch_reply(\n mut receive: tokio::sync::oneshot::Receiver>,\n cancellation: Option>,\n cancel: Option>,\n) -> anyhow::Result {\n let received = match (cancellation, cancel) {\n (Some(cancellation), Some(mut cancel)) => asyncrt::select_biased! {\n \"a completed cell reply wins a tie with an external disconnect\";\n result = &mut receive => result,\n cancelled = &mut cancel => match cancelled {\n Ok(()) => {\n // The driver reads this between turns. Dropping the\n // receiver also wakes a reply that has already left V8\n // and is waiting only on its event's durability gates.\n cancellation.publish_abort();\n drop(receive);\n return Err(anyhow!(\"The client has disconnected\"));\n }\n Err(_) => receive.await,\n }\n },\n _ => receive.await,\n };\n received.context(\"cell isolate dropped response\")?\n}\n\npub(crate) async fn receive_service_fetch_response(\n response: impl std::future::Future>,\n cancellation: Arc,\n cancel: Option>,\n) -> anyhow::Result {\n match cancel {\n Some(mut cancel) => crate::asyncrt::select_biased! {\n \"a completed service response wins a tie with an external disconnect\";\n response = response => response,\n _ = &mut cancel => {\n cancellation.publish_abort();\n Err(anyhow!(\"service-binding caller disconnected\"))\n }\n },\n None => response.await,\n }\n}\n\n#[cfg(celld_internal_tests)]\n#[doc(hidden)]\npub async fn receive_cell_fetch_reply_for_test(\n receive: tokio::sync::oneshot::Receiver>,\n request_id: Option,\n cancel: Option>,\n) -> anyhow::Result {\n let cancellation = request_id.map(RequestCancellationLifetime::from_request_id);\n receive_cell_fetch_reply(receive, cancellation, cancel).await\n}\n\n#[cfg(celld_internal_tests)]\n#[derive(Clone)]\n#[doc(hidden)]\npub struct RequestCancellationLifetimeForTest(Arc);\n\n#[cfg(celld_internal_tests)]\n#[doc(hidden)]\npub fn request_cancellation_lifetime_for_test(\n request_id: js::RequestId,\n) -> RequestCancellationLifetimeForTest {\n RequestCancellationLifetimeForTest(RequestCancellationLifetime::from_request_id(request_id))\n}\n\n#[cfg(celld_internal_tests)]\n#[doc(hidden)]\npub fn finish_request_cancellation_for_test(lifetime: &RequestCancellationLifetimeForTest) {\n lifetime.0.finish();\n}\n\n#[cfg(celld_internal_tests)]\n#[doc(hidden)]\npub fn finish_active_request_cancellation_for_test(lifetime: &Arc) {\n lifetime.finish();\n}\n\n#[cfg(celld_internal_tests)]\n#[doc(hidden)]\npub async fn receive_cell_fetch_reply_with_lifetime_for_test(\n receive: tokio::sync::oneshot::Receiver>,\n lifetime: RequestCancellationLifetimeForTest,\n cancel: tokio::sync::oneshot::Receiver<()>,\n entered: tokio::sync::oneshot::Sender<()>,\n) -> anyhow::Result {\n let _ = entered.send(());\n receive_cell_fetch_reply(receive, Some(lifetime.0), Some(cancel)).await\n}\n\n#[cfg(all(test, celld_internal_tests))]\npub(crate) async fn receive_service_fetch_response_for_test(\n response: impl std::future::Future>,\n lifetime: Arc,\n cancel: tokio::sync::oneshot::Receiver<()>,\n) -> anyhow::Result {\n receive_service_fetch_response(response, lifetime, Some(cancel)).await\n}\n\nimpl RuntimeManager {\n /// A deployment with no Durable Object classes can never land a Worker fetch\n /// on a cell, so the core's round-robin routing always returns `None`. Lets\n /// the request path skip the core round-trip entirely for stateless workers.\n pub fn has_cell_classes(&self) -> bool {\n self.generation().has_cell_classes()\n }\n\n /// The reserved cell carrying the current deployment's cron schedule, or\n /// `None` when it declares no `triggers.crons`. Asked after every\n /// adoption so the schedule is armed without a request.\n pub fn cron_cell(&self) -> Option {\n self.generation().cron_cell()\n }\n\n /// Start the node's cell runtime around its first generation.\n ///\n /// The generation is built before this is called, by `Generation::build`,\n /// which is also what a reload calls: a deployment reaches a running\n /// node through exactly one function whether the node is booting or\n /// adopting. What starts here is the node-level state that outlives any\n /// generation — the cell registry, replication, the wake flusher.\n pub fn start(generation: Generation, options: RuntimeOptions) -> anyhow::Result {\n let RuntimeOptions {\n data_dir,\n replication,\n wake,\n alarm_observer,\n node,\n region,\n bucket,\n } = options;\n let max_cell_requests =\n crate::env_vars::positive_or(\"CELLD_MAX_CELL_REQUESTS\", DEFAULT_MAX_CELL_REQUESTS)?;\n let generation = Arc::new(generation);\n let core = crate::cell_runtime::CellRuntime::new(\n generation.clone(),\n data_dir,\n replication,\n wake,\n alarm_observer,\n node,\n region,\n )?;\n let cells = Arc::new(Mutex::new(CellRegistry::default()));\n let alarm_reporter = alarm_reporter(&cells, &core.alarm_observer);\n let manager = Self {\n core,\n cells,\n alarm_reporter,\n remote_requests: Arc::new(Mutex::new(RemoteRequestRegistry::default())),\n alarm_requests: Arc::new(Mutex::new(AlarmRequestRegistry::default())),\n max_cell_requests,\n };\n manager.watch_generation(&generation);\n crate::container::configure(\n manager.node.to_string(),\n bucket,\n manager.data_dir.as_ref().clone(),\n );\n install_container_specs(&generation);\n Ok(manager)\n }\n\n /// Make `generation` current: the flip.\n ///\n /// From the moment this returns, new stateless requests, new cell\n /// activations, ingress asset lookups, service-binding resolution, and\n /// queue dispatch use the new generation. The previous one stops taking\n /// new work and drains; it is dropped once every isolate it built has\n /// been freed. Requests and cells already on it finish there.\n pub fn adopt(&self, generation: Generation) -> Arc {\n let next = Arc::new(generation);\n self.watch_generation(&next);\n install_container_specs(&next);\n let previous = {\n let mut generations = self.generations.write().expect(\"generation lock poisoned\");\n let previous = std::mem::replace(&mut generations.current, next.clone());\n // Installed in the same instant the new one becomes current. A\n // call from an isolate of the previous generation must never\n // find neither, because `generation_by_id` answers \"neither\"\n // with the current generation and the call would cross.\n generations.draining.push(previous.clone());\n previous\n };\n tracing::info!(\n event = \"deployment_generation_adopted\",\n generation = next.id,\n version = %next.version,\n prefix = %next.prefix,\n previous_generation = previous.id,\n previous_version = %previous.version,\n \"application generation adopted\"\n );\n // After the install, never before: retiring walks every pool of the\n // previous generation and this must not hold the generation lock,\n // which every request reads.\n previous.retire();\n next\n }\n\n /// Spawn the maintenance loop for one generation's cell pools.\n ///\n /// The loop holds a weak reference. A strong one would keep a superseded\n /// generation — its compiled scripts and every pool — alive for the life\n /// of the process, so the loop ends when the generation is dropped. The\n /// draining list is what keeps a superseded generation alive until then,\n /// and this loop is what removes it from that list once every isolate it\n /// built has been freed.\n fn watch_generation(&self, generation: &Arc) {\n let Ok(handle) = tokio::runtime::Handle::try_current() else {\n return;\n };\n let reaping = Arc::downgrade(generation);\n let generations = self.generations.clone();\n handle.spawn(async move {\n let mut tick = tokio::time::interval_at(\n tokio::time::Instant::now() + REAP_INTERVAL,\n REAP_INTERVAL,\n );\n tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);\n loop {\n tick.tick().await;\n let Some(generation) = reaping.upgrade() else {\n return;\n };\n generation.reap_cell_pools();\n if !generation.is_drained() {\n continue;\n }\n let mut generations = generations.write().expect(\"generation lock poisoned\");\n if let Some(index) = generations\n .draining\n .iter()\n .position(|candidate| Arc::ptr_eq(candidate, &generation))\n {\n generations.draining.remove(index);\n tracing::info!(\n event = \"deployment_generation_freed\",\n generation = generation.id,\n version = %generation.version,\n \"application generation freed\"\n );\n }\n }\n });\n }\n\n pub async fn fetch_worker(\n &self,\n url: String,\n method: String,\n body: Vec,\n headers: Vec<(String, String)>,\n ) -> anyhow::Result {\n // The snapshot is held for the whole request, so the generation it\n // started on outlives it even after the node adopts another.\n let generation = self.generation();\n generation\n .stateless\n .fetch(url, method, body.into(), headers, None)\n .await\n }\n\n /// Dispatch a cancellable top-level Worker request to the stateless pool.\n pub fn fetch_worker_pool(\n &self,\n url: String,\n method: String,\n body: js::RequestBody,\n headers: Vec<(String, String)>,\n cancellation: Arc,\n ) -> impl std::future::Future> + Send + 'static {\n let generation = self.generation();\n let fetching = generation\n .stateless\n .fetch(url, method, body, headers, Some(cancellation));\n async move {\n let _generation = generation;\n fetching.await\n }\n }\n\n /// Dispatch a top-level Worker request on the exact resident runtime the\n /// decision core reserved. The activity token pins that lifecycle choice\n /// until the queued event has completely left the isolate loop.\n pub async fn fetch_worker_on_cell(\n &self,\n cell: String,\n epoch: u64,\n request: RuntimeFetch,\n inline_activity: crate::CellActivityGuard,\n ) -> anyhow::Result {\n let RuntimeFetch {\n url,\n method,\n body,\n headers,\n request_id,\n // A resident Worker fetch is not a cell event; its inbound\n // traceparent is honored by the drive, from the headers.\n order: _,\n parent: _,\n } = request;\n let request_id = request_id.context(\"resident Worker fetch requires a request id\")?;\n let isolate = self\n .cells\n .lock()\n .expect(\"cell registry poisoned\")\n .published\n .get(&cell)\n .filter(|handle| handle.epoch == epoch)\n // Affiliated under the registry lock for the driver's whole\n // lifetime, exactly like cell_isolate (denoland/celld#147).\n .map(|handle| handle.residency.slot().affiliate())\n .ok_or_else(|| anyhow!(\"cell runtime is not published at epoch {epoch}: {cell}\"))?;\n // The Worker entry, run in the isolate that hosts the cell it will\n // route to. The call still goes out through the host -- every cell\n // dispatch does -- but it comes back to an isolate that is already\n // warm for this cell, with its storage open and its instance live.\n //\n // It needs no rescheduling any more. A Worker fetch could not be\n // nested inside an actor event, and delivery used to nest, so a job\n // that arrived mid-event had to be handed back to the stateless\n // pool. Events no longer nest: this is an entry like any other, and\n // it waits for its turn rather than for the isolate to go idle.\n let (reply, receive) = tokio::sync::oneshot::channel();\n let job = crate::WorkerJob::Fetch {\n queued_at: Instant::now(),\n entrypoint: None,\n invocation_limits: None,\n url,\n method,\n body,\n headers,\n request_id: Some(request_id),\n tail_report: None,\n reply,\n };\n tokio::spawn(async move {\n let _inline_activity = inline_activity;\n drive_worker_on_cell(isolate, job).await;\n });\n receive\n .await\n .context(\"cell isolate dropped Worker response\")?\n }\n\n /// Call a service binding. `generation` is the caller's, so the target\n /// comes from the deployment graph the caller was built with.\n pub async fn fetch_service(&self, call: ServiceFetch) -> anyhow::Result {\n let ServiceFetch {\n generation,\n script,\n entrypoint,\n url,\n method,\n body,\n headers,\n cancel,\n } = call;\n let generation = self.generation_by_id(generation);\n let pool = generation\n .service(&script)\n .ok_or_else(|| anyhow!(\"no service Worker for script {script}\"))?;\n let cancellation = RequestCancellationLifetime::stateless();\n let response = pool.fetch_target(\n entrypoint,\n url,\n method,\n body,\n headers,\n Some(cancellation.clone()),\n );\n receive_service_fetch_response(response, cancellation, cancel).await\n }\n\n pub async fn rpc_service(\n &self,\n generation: GenerationId,\n script: &str,\n entrypoint: String,\n props: Vec,\n operation: crate::WorkerRpcOperation,\n ) -> anyhow::Result> {\n let generation = self.generation_by_id(generation);\n generation\n .service(script)\n .ok_or_else(|| anyhow!(\"no service Worker for script {script}\"))?\n .rpc(entrypoint, props, operation)\n .await\n }\n\n /// Dispatch one broker-leased batch to its attached consumer script.\n pub async fn queue_service(\n &self,\n generation: GenerationId,\n script: &str,\n batch: js::QueueBatch,\n ) -> anyhow::Result {\n let generation = self.generation_by_id(generation);\n generation\n .service(script)\n .ok_or_else(|| anyhow!(\"no Queue consumer Worker for script {script}\"))?\n .queue(batch)\n .await\n }\n\n pub async fn restore_cell(\n &self,\n cell: &str,\n spec: &celld_logic::RestoreSpec,\n ) -> anyhow::Result {\n let path = self.db_path(cell, spec.epoch);\n self.facets.register(cell, spec);\n if let Some(replication) = &self.replication {\n let (restored_path, restored, vfs) = replication.restore(cell, spec, true).await?;\n if restored_path != path {\n return Err(anyhow!(\n \"replication restored {} instead of {}\",\n restored_path.display(),\n path.display()\n ));\n }\n return Ok(celld_logic::RestoreOutcome {\n restored,\n alarm: self.restored_alarm(cell, &path, vfs.as_deref()).await,\n });\n }\n let parent = path.parent().context(\"cell database has no parent\")?;\n let parent = parent.to_path_buf();\n let parent_display = parent.display().to_string();\n let filesystem = asyncrt::fs();\n asyncrt::blocking(move || filesystem.create_dir_all(&parent))\n .await?\n .with_context(|| format!(\"create cell data directory {parent_display}\"))?;\n Ok(celld_logic::RestoreOutcome {\n restored: false,\n alarm: self.restored_alarm(cell, &path, None).await,\n })\n }\n\n /// The alarm the restored database already had armed, read directly by\n /// path. Read-only, and the connection is dropped here -- the isolate\n /// opens the same file moments later through `spawn_cell`.\n async fn restored_alarm(\n &self,\n cell: &str,\n path: &std::path::Path,\n vfs: Option<&str>,\n ) -> Option {\n // A paged cell's alarm read faults pages in on the reading thread, so\n // it runs on a blocking thread rather than on this runtime worker.\n let path_ = path.to_string_lossy().into_owned();\n let cell_ = cell.to_string();\n let vfs_ = vfs.map(str::to_string);\n let persisted = asyncrt::blocking(move || {\n crate::storage::persisted_alarm(&path_, &cell_, vfs_.as_deref())\n })\n .await\n .ok()\n .flatten();\n restored_alarm_from_persisted(cell, persisted, |at_ms| {\n self.alarm_covered(\n cell,\n celld_logic::wake::AlarmSnapshot::without_wake(Some(at_ms)),\n )\n })\n }\n\n /// Detach every resident runtime, prove and retain its exact database path,\n /// and remove stale live-named epochs. The caller has already stopped\n /// admission and drained request effects.\n pub async fn prepare_clean_reload(\n &self,\n cells: &[celld_logic::PresenceCell],\n ) -> anyhow::Result {\n let replication = self\n .replication\n .as_ref()\n .context(\"clean reload requires replication\")?;\n let keep: BTreeSet<_> = cells\n .iter()\n .map(|cell| (cell.id.clone(), cell.epoch))\n .collect();\n anyhow::ensure!(\n keep.len() == cells.len(),\n \"clean reload resident inventory contains duplicates\"\n );\n let mut closes = futures_util::stream::iter(cells.iter().cloned())\n .map(|cell| {\n let runtime = self.clone();\n let replication = replication.clone();\n async move {\n // A normal stop releases the replica before returning.\n // Keep it registered until the final capture proves the\n // stopped database position present in the bucket.\n runtime.swap_out_cell(&cell.id, cell.epoch).await?;\n for facet in runtime.facets.stopping(&cell.id, cell.epoch) {\n replication.close_for_reload(&facet, cell.epoch).await?;\n runtime.facets.stopped(&cell.id, cell.epoch, &facet);\n }\n replication.close_for_reload(&cell.id, cell.epoch).await?;\n runtime.facets.forget(&cell.id, cell.epoch);\n anyhow::Ok(())\n }\n })\n .buffer_unordered(128);\n while let Some(result) = closes.next().await {\n result?;\n }\n let pruned = replication.prune_stale_live(&keep)?;\n Ok(pruned)\n }\n\n /// Take a cell out of its isolate for a generation swap.\n ///\n /// The first half of `stop_cell` and nothing more: close the application\n /// database in the isolate that holds it, and drop the residency so the\n /// isolate can be reclaimed once its last cell leaves. The replica keeps\n /// its handle and its file, the owner record keeps its epoch, and the\n /// core starts the cell again on the current generation as soon as it\n /// hears the stop. A cell already gone is not an error: the swap and an\n /// eviction can race, and the eviction's stop did this work.\n pub async fn swap_out_cell(&self, cell: &str, epoch: u64) -> anyhow::Result<()> {\n // The isolates to release, taken by reference rather than by handle.\n // A release that fails can leave this cell's database open in that\n // realm, and the handle owns both the residency that keeps the realm\n // from being reclaimed and the registry entry a retry finds it\n // through -- so only a release that succeeded gives them up.\n let slots: Vec<(Arc, GenerationId)> = {\n let cells = self.cells.lock().expect(\"cell registry poisoned\");\n [&cells.starting, &cells.published]\n .into_iter()\n .filter_map(|residents| residents.get(cell))\n .filter(|handle| handle.epoch == epoch)\n .map(|handle| (handle.residency.slot().clone(), handle.generation))\n .collect()\n };\n for (slot, generation) in slots {\n // Taking the isolate for this turn is the barrier: an event of\n // this cell either finished its turn before it or has not\n // started one, so closing its SQLite cannot land mid-handler.\n slot.turn(|worker| {\n #[cfg(debug_assertions)]\n if let Some(error) = crate::engine_api::injected_swap_release_failure() {\n return Err(error);\n }\n worker.own_cell(cell, None)\n })\n .await\n .with_context(|| format!(\"release {cell} from its isolate for a generation swap\"))?;\n crate::js::release_root_facets(cell, epoch, crate::js::RootRelease::Swap).await;\n tracing::info!(\n event = \"generation_swap_out\",\n scope = %cell,\n epoch,\n generation,\n \"cell left its isolate for a generation swap\"\n );\n }\n // Released, so the placements can go back. Dropping each handle drops\n // its residency, which is what lets the isolate be reclaimed once its\n // last cell has left.\n let mut cells = self.cells.lock().expect(\"cell registry poisoned\");\n let CellRegistry {\n starting,\n published,\n } = &mut *cells;\n for residents in [starting, published] {\n if residents\n .get(cell)\n .is_some_and(|handle| handle.epoch == epoch)\n {\n residents.remove(cell);\n }\n }\n Ok(())\n }\n\n /// Materialize an isolate and retain it as non-routable until publication.\n /// Start a cell's runtime and report which isolate took its realm and the\n /// generation it belongs to. The core groups eviction on the isolate and\n /// the swap pump keys on the generation, so both come back from here.\n pub async fn start_cell(\n &self,\n cell: String,\n epoch: u64,\n fresh: bool,\n ) -> anyhow::Result<(celld_logic::isolate::HeapId, GenerationId)> {\n let db_path = self.db_path(&cell, epoch);\n let class = cell\n .split_once(':')\n .map(|(class, _)| class)\n .ok_or_else(|| anyhow!(\"cell scope has no class: {cell}\"))?;\n let generation = self.generation();\n let config = generation\n .cell_config(class)\n .ok_or_else(|| anyhow!(\"no Worker exports Durable Object class {class}\"))?;\n // The object's `running` is read synchronously, so the engine\n // answers it here, before the object exists. A container a previous\n // activation on this node left running is adopted by name.\n if config.container_class(class) {\n let engine = crate::container::engine().await?;\n engine.attach(&cell, class).await?;\n }\n let startup_timing = CellIsolateStartupTiming {\n started: Instant::now(),\n scope: cell.clone(),\n node: self.node.clone(),\n region: self.region.clone(),\n epoch,\n fresh,\n };\n\n let isolates = generation\n .cell_isolates(&config.script_name)\n .ok_or_else(|| anyhow!(\"no cell isolates for script {}\", config.script_name))?;\n // Building an isolate compiles the script, so it runs on a blocking\n // thread: the pool builds before taking its lock, but the caller is a\n // tokio worker either way.\n let placed = {\n let isolates = isolates.clone();\n tokio::task::spawn_blocking(move || isolates.place_cell())\n .await\n .context(\"cell placement panicked\")?\n };\n let residency = match placed {\n Ok(residency) => residency,\n Err(error) => {\n startup_timing.emit(\"error\", \"worker_load\");\n return Err(error);\n }\n };\n let placed_in = residency.slot().heap_id();\n\n // Everything the cell needs that the isolate must do: open its\n // SQLite — which the isolate owns, not the caller — and restore its\n // persisted id name. A paged restore leaves the file sparse behind\n // the activation's VFS, so the actor's connection must open through\n // it too.\n //\n // A direct call rather than a job: adoption is not an event, it runs\n // no handler, and it needs one turn.\n // The activation's own answer, not a guess from an absent handle: a\n // cell removed between restore and adoption must not be opened\n // plainly over a sparse or missing file.\n let paged_vfs = match self.replication.as_ref() {\n Some(replication) => match replication.ltx().activation_vfs(&cell, epoch) {\n Ok(vfs) => vfs,\n Err(error) => {\n startup_timing.emit(\"error\", \"storage_open\");\n return Err(error);\n }\n },\n None => None,\n };\n let adopted = residency\n .adopt(\n &cell,\n CellStorage {\n path: path_text(&db_path),\n epoch,\n replicated_wake: self.wake.is_some(),\n vfs: paged_vfs.as_deref(),\n },\n )\n .await;\n let (residency, alarm) = match adopted {\n Ok(adopted) => adopted,\n Err(error) => {\n startup_timing.emit(\"error\", \"storage_open\");\n return Err(error);\n }\n };\n (self.alarm_observer)(cell.clone(), alarm);\n let startup_us = startup_timing.emit(\"ready\", \"\");\n\n {\n let mut cells = self.cells.lock().expect(\"cell registry poisoned\");\n if cells.starting.contains_key(&cell) || cells.published.contains_key(&cell) {\n return Err(anyhow!(\"cell runtime already exists: {cell}\"));\n }\n cells.starting.insert(\n cell.clone(),\n CellHandle {\n epoch,\n generation: generation.id,\n startup_us,\n residency,\n alarm,\n requests: Arc::new(CellRequestAdmission {\n in_flight: AtomicUsize::new(0),\n saturated: AtomicBool::new(false),\n limit: self.max_cell_requests,\n }),\n },\n );\n Ok((placed_in, generation.id))\n }\n }\n\n /// Make the exact started generation visible to request dispatch.\n pub fn publish_cell(&self, cell: &str, epoch: u64) -> anyhow::Result<()> {\n let mut cells = self.cells.lock().expect(\"cell registry poisoned\");\n if cells\n .starting\n .get(cell)\n .is_none_or(|handle| handle.epoch != epoch)\n {\n return Err(anyhow!(\"no started cell runtime for {cell} epoch {epoch}\"));\n }\n let handle = cells\n .starting\n .remove(cell)\n .expect(\"checked started runtime\");\n let startup_us = handle.startup_us;\n if let Some(replaced) = cells.published.insert(cell.to_string(), handle) {\n // Nothing to shut down: the isolate serves other cells, and\n // dropping the handle drops the residency that held its place.\n drop(replaced);\n return Err(anyhow!(\"replaced published cell runtime for {cell}\"));\n }\n drop(cells);\n tracing::info!(\n event = \"cell_runtime_publication\",\n outcome = \"published\",\n scope = %cell,\n node = %self.node,\n region = %self.region,\n runtime_version = env!(\"CARGO_PKG_VERSION\"),\n epoch,\n isolate_startup_us = startup_us,\n \"cell runtime published\"\n );\n Ok(())\n }\n\n pub(crate) async fn stop_cell(\n &self,\n cell: &str,\n epoch: u64,\n mode: StopMode,\n ) -> anyhow::Result<()> {\n // No facet opens once the stop begins; the facet loop below stops\n // the streams this answers again, with any that opened before.\n self.facets.stopping(cell, epoch);\n if let Some(engine) = crate::container::engine_if_ready() {\n // Only an idle eviction keeps the cell on this node, so only it\n // keeps the container: the next activation here reconnects.\n let release = match &mode {\n StopMode::Evict {\n preserve_local: true,\n ..\n } => crate::container::Release::Keep,\n _ => crate::container::Release::Destroy,\n };\n engine.release(cell, release).await;\n }\n let mut stopped = Vec::new();\n {\n let mut cells = self.cells.lock().expect(\"cell registry poisoned\");\n if cells\n .starting\n .get(cell)\n .is_some_and(|handle| handle.epoch == epoch)\n {\n if let Some(handle) = cells.starting.remove(cell) {\n stopped.push(handle);\n }\n }\n if cells\n .published\n .get(cell)\n .is_some_and(|handle| handle.epoch == epoch)\n {\n if let Some(handle) = cells.published.remove(cell) {\n stopped.push(handle);\n }\n }\n }\n for handle in stopped {\n // Give the cell back rather than shutting the isolate down: it\n // serves other cells. Taking the isolate for this turn is the\n // barrier — an event of this cell either finished its turn\n // before it, or has not started one — so closing its SQLite\n // cannot land under a handler that is mid-turn.\n // The cell's own lane, not the shared stateless one: this stop\n // follows the turns already admitted for this cell, and it does\n // not wait behind every stateless turn queued on the isolate. On\n // a loaded node that queue held a handoff batch for minutes.\n let _ = handle\n .residency\n .slot()\n .turn_cell(cell, |worker| worker.own_cell(cell, None))\n .await;\n // Dropping the handle drops its residency, which is what gives\n // the isolate its place back — and what lets `retire` reclaim\n // the isolate once no cell is left in it.\n drop(handle);\n }\n // After the give-back, so the root cannot start another facet call,\n // and before the facet streams stop, so no write reaches a file that\n // the facet's eviction unlinked.\n crate::js::release_root_facets(cell, epoch, crate::js::RootRelease::Stop).await;\n let Some(replication) = &self.replication else {\n self.facets.forget(cell, epoch);\n return Ok(());\n };\n // The root and its facets stop in the order `FacetStreams::stop_root`\n // owns, which a test drives through its failures.\n let stop_root = || async {\n // Every stop releases the handle, and this does not consult\n // `stopped_runtime`. The replication entry is created by\n // `Effect::Restore` and the registry entry by\n // `Effect::StartRuntime`, so the entry outlives a start that fails\n // between the two and there is nothing else that would ever remove\n // it. Its lifetime is the activation's, not the registry's.\n //\n // The mode pairs the durability authority with the file outcome.\n // Passing these as independent booleans let cleanup close a proved\n // remote restore in place, where no later activation could use it.\n match &mode {\n // A failed final sync leaves the handle and files in place.\n // This call is therefore intentionally retryable after the\n // runtime itself has already stopped. An abandoned one is not:\n // it returns `EvictionAbandoned`, and the caller ends the stop\n // instead of trying again.\n StopMode::Evict {\n preserve_local,\n abandon,\n } => replication\n .evict(cell, epoch, *preserve_local, abandon.as_deref())\n .await\n .map(|_| ()),\n StopMode::Rebase => replication.release(cell, epoch).await,\n StopMode::CloseInPlace => replication.close_in_place(cell, epoch).await,\n StopMode::Discard => {\n replication.ltx().discard(cell, epoch);\n Ok(())\n }\n }\n };\n let stop_facet = |facet: String| {\n let mode = &mode;\n async move {\n // A retry after the facet's stream stopped, or after shutdown\n // took the handle, has nothing left to stop.\n if !replication.ltx().is_resident(&facet, epoch) {\n return Ok(());\n }\n match mode {\n // A facet keeps no local file past an eviction: its handler\n // writes through a connection of its own, which can outlive\n // the root's stop, and a preserved file would become the\n // next activation's baseline. The root is gone by now, so\n // the eviction is not abandoned.\n StopMode::Evict { .. } => replication\n .evict(&facet, epoch, false, None)\n .await\n .map(|_| ()),\n StopMode::Rebase => replication.release(&facet, epoch).await,\n StopMode::CloseInPlace => replication.close_in_place(&facet, epoch).await,\n StopMode::Discard => {\n replication.ltx().discard(&facet, epoch);\n Ok(())\n }\n }\n }\n };\n self.facets\n .stop_root(\n cell,\n epoch,\n replication.ltx().is_resident(cell, epoch),\n stop_root,\n stop_facet,\n )\n .await\n }\n\n pub async fn fetch_cell(\n &self,\n cell: String,\n name: Option,\n request: RuntimeFetch,\n cancel: Option>,\n ) -> anyhow::Result {\n let RuntimeFetch {\n url,\n method,\n body,\n headers,\n request_id,\n order,\n parent,\n } = request;\n let admitted = match self.admit_cell_request(&cell)? {\n Some(admitted) => admitted,\n None => {\n return Ok(HttpResponse {\n status: 503,\n headers: vec![\n (\"retry-after\".to_string(), \"1\".to_string()),\n (\"x-celld-overload\".to_string(), \"cell\".to_string()),\n ],\n body: b\"cell request limit reached\".to_vec(),\n websocket: None,\n stream: None,\n write_position: None,\n observed_position: None,\n });\n }\n };\n let AdmittedCellRequest {\n affiliation,\n permit,\n } = admitted;\n let (reply, receive) = tokio::sync::oneshot::channel();\n let job = CellJob::Fetch {\n request_id,\n scope: cell,\n name,\n url,\n method,\n body,\n headers,\n reply,\n order,\n };\n let cancellation = if let Some(request_id) = request_id {\n let mut requests = self\n .remote_requests\n .lock()\n .expect(\"request registry poisoned\");\n requests.prune();\n if matches!(\n requests.states.remove(&request_id),\n Some(RemoteRequestState::PendingAbort(_))\n ) {\n requests.completed(request_id);\n return Err(anyhow!(\"the client disconnected before dispatch\"));\n }\n Some(requests.active(request_id))\n } else {\n None\n };\n let alarm_reporter = self.alarm_reporter.clone();\n let drive_cancellation = cancellation.clone();\n tokio::spawn(async move {\n let _permit = permit;\n drive_cell_with_request_cancellation(\n affiliation,\n job,\n Some(alarm_reporter),\n parent,\n drive_cancellation,\n )\n .await;\n });\n let result = receive_cell_fetch_reply(receive, cancellation, cancel).await;\n if let Some(request_id) = request_id {\n let mut requests = self\n .remote_requests\n .lock()\n .expect(\"request registry poisoned\");\n requests.completed(request_id);\n }\n result\n }\n\n /// Tell a cell to abandon a fetch, by name.\n ///\n /// `fetch_cell` drops its reply receiver and returns when its explicit\n /// cancellation signal arrives. A caller that learns about a hang-up only\n /// in a destructor has no future left to receive that signal, so it uses\n /// this direct path instead.\n pub fn abort_fetch(&self, cell: &str, request_id: js::RequestId) {\n let mut requests = self\n .remote_requests\n .lock()\n .expect(\"request registry poisoned\");\n requests.prune();\n requests.abort(request_id);\n drop(requests);\n let _ = cell;\n }\n\n pub fn published_epoch(&self, cell: &str) -> Option {\n self.cells\n .lock()\n .expect(\"cell registry poisoned\")\n .published\n .get(cell)\n .map(|handle| handle.epoch)\n }\n\n pub fn alarm(&self, cell: &str) -> Option {\n self.with_alarm(cell, |at_ms| at_ms)\n }\n\n /// Read the cell's alarm cache and call `f` before the registry lock is\n /// released. The reporter sends under the same lock, so whatever `f`\n /// sends is ordered with the reporter's sends: a read taken here cannot\n /// reach the core after a fresher report (see `alarm_reporter`).\n pub fn with_alarm(&self, cell: &str, f: impl FnOnce(Option) -> T) -> T {\n self.with_alarm_snapshot(cell, |alarm| f(alarm.at_ms()))\n }\n\n pub(crate) fn with_alarm_snapshot(\n &self,\n cell: &str,\n f: impl FnOnce(celld_logic::wake::AlarmSnapshot) -> T,\n ) -> T {\n let cells = self.cells.lock().expect(\"cell registry poisoned\");\n let alarm = cells\n .published\n .get(cell)\n .or_else(|| cells.starting.get(cell))\n .map(|handle| handle.alarm)\n .unwrap_or_else(|| celld_logic::wake::AlarmSnapshot::without_wake(None));\n f(alarm)\n }\n\n pub async fn fire_alarm(\n &self,\n op: celld_logic::OpId,\n cell: String,\n scheduled_ms: i64,\n ) -> anyhow::Result<(celld_logic::wake::AlarmSnapshot, bool, Option)> {\n let request_id = js::next_request_id();\n let cancel = self\n .alarm_requests\n .lock()\n .expect(\"alarm registry poisoned\")\n .begin(cell.clone(), op, request_id);\n let (reply, receive) = tokio::sync::oneshot::channel();\n let job = CellJob::Alarm {\n request_id: Some(request_id),\n scope: cell.clone(),\n scheduled_ms,\n claim: js::AlarmDispatch::Due,\n reply,\n };\n let guard = AlarmRequestGuard {\n registry: self.alarm_requests.clone(),\n cell: cell.clone(),\n op,\n request: request_id,\n };\n let isolate = self.cell_isolate(&cell)?;\n let alarm_reporter = self.alarm_reporter.clone();\n let drive = crate::asyncrt::spawn(async move {\n let _guard = guard;\n drive_alarm(isolate, job, Some(alarm_reporter)).await\n });\n if cancel {\n js::abort_request_for_shutdown(request_id);\n }\n let result = receive.await.context(\"cell isolate dropped alarm result\")?;\n match result {\n // A handler that returned settled its claim inside its completion\n // turn (`Answer::Alarm`), so the snapshot and the delta it replied\n // with are already final and nothing below reads the drive's\n // position. The drive itself runs on, because a completed cell\n // event keeps its pending I/O (`completed_cell_event`), exactly as\n // it does after a fetch answers. Waiting for it here held the\n // core in `AlarmState::Firing` until the leftover I/O ended or the\n // operation deadline expired the firing, so an `alarm()` that left\n // one pending timer — what `AbortSignal.timeout` leaves behind\n // after its fetch finished — delayed every re-arm by up to\n // `CELLD_OPERATION_DEADLINE_MS`, and the core booked a handler\n // that succeeded as a stuck one (denoland/celld#228).\n Ok((alarm, wrote)) => {\n watch_detached_alarm_drive(drive, &cell, op);\n Ok((alarm, self.alarm_covered(&cell, alarm), wrote))\n }\n Err(error) => {\n // A cancelled or failed alarm settles its claim in the drive's\n // final isolate turn. The event reply is itself held behind\n // every wake-entry arm, so waiting for the drive makes that\n // final cache authoritative before the core sees completion.\n let final_write = drive.await.expect(\"cell alarm drive task panicked\")?;\n // The drive completed its final isolate turn before this\n // branch. Its alarm cache is therefore authoritative even\n // when the handler failed: `Some` is the automatic retry or\n // an explicit re-arm, and `None` is an explicit change which\n // wins over retry. Returning the old firing as an error would\n // resurrect an alarm which storage no longer contains and\n // leave a draining cell permanently uncovered.\n //\n // The position travels as it does for a success. What the\n // handler committed before it failed, and the retry record\n // the final turn wrote, are unproven writes the core must\n // prove before a reader can reveal them; without a position\n // the alarm settled at once and opened no barrier (#715).\n // A handler that rejected settled its claim before its error\n // left, so the error carries the whole delta; one that\n // failed before or between turns had its record written by\n // the final turn, whose sample is the later one.\n let alarm = self.with_alarm_snapshot(&cell, |alarm| alarm);\n Ok((\n alarm,\n self.alarm_covered(&cell, alarm),\n final_write.max(js::failed_write_position(&error)),\n ))\n }\n }\n }\n\n pub fn abort_alarm(&self, cell: &str, op: celld_logic::OpId) {\n let request = self\n .alarm_requests\n .lock()\n .expect(\"alarm registry poisoned\")\n .aborting(cell, op);\n if let Some(request) = request {\n js::abort_request_for_shutdown(request);\n }\n }\n\n /// Run `webSocketOpen`. The answer is the position the handler's writes\n /// reached, or the error carries it, so the caller can open their barrier.\n pub async fn ws_open(\n &self,\n cell: String,\n ws_id: u64,\n protocol: String,\n ) -> anyhow::Result> {\n let (reply, receive) = tokio::sync::oneshot::channel();\n let job = CellJob::WsOpen {\n scope: cell.clone(),\n ws_id,\n facet_opening: None,\n protocol,\n reply,\n };\n self.cell_event(&cell, job, receive, \"cell isolate dropped WebSocket open\")\n .await\n }\n\n /// Run `webSocketOpen` in the facet `scope` that dialed `ws_id`. A write\n /// of the handler is proved by the facet's next output, as for\n /// `facet_ws_closed`.\n pub async fn facet_ws_open(\n &self,\n scope: String,\n facet: &js::FacetSocket,\n ws_id: u64,\n protocol: String,\n ) -> anyhow::Result<()> {\n let (reply, receive) = tokio::sync::oneshot::channel();\n let job = CellJob::WsOpen {\n scope: scope.clone(),\n ws_id,\n facet_opening: Some(facet.opening),\n protocol,\n reply,\n };\n self.facet_event(\n &scope,\n facet,\n job,\n receive,\n \"facet isolate dropped WebSocket open\",\n )\n .await\n .map(|_| ())\n }\n\n pub async fn rpc(\n &self,\n cell: String,\n name: Option,\n method: String,\n args: js::RpcData,\n request_id: Option,\n ) -> anyhow::Result {\n let (reply, receive) = tokio::sync::oneshot::channel();\n let job = CellJob::Rpc {\n request_id,\n scope: cell.clone(),\n name,\n method,\n args,\n reply,\n };\n self.cell_event(&cell, job, receive, \"cell isolate dropped RPC result\")\n .await\n }\n\n pub async fn ws_message(\n &self,\n cell: String,\n ws_id: u64,\n data: js::WsIn,\n started: tokio::sync::oneshot::Sender<()>,\n ) -> anyhow::Result {\n let (reply, receive) = tokio::sync::oneshot::channel();\n let job = CellJob::WsMessage {\n scope: cell.clone(),\n ws_id,\n facet_opening: None,\n data,\n started: Some(started),\n reply,\n };\n self.cell_event(\n &cell,\n job,\n receive,\n \"cell isolate dropped WebSocket message\",\n )\n .await\n }\n\n /// Run `webSocketClose`. The answer is the position the handler's writes\n /// reached, or the error carries it, so the caller can open their barrier.\n pub async fn ws_closed(\n &self,\n cell: String,\n ws_id: u64,\n code: u16,\n reason: String,\n was_clean: bool,\n ) -> anyhow::Result> {\n let (reply, receive) = tokio::sync::oneshot::channel();\n let job = CellJob::WsClosed {\n scope: cell.clone(),\n ws_id,\n facet_opening: None,\n code,\n reason,\n was_clean,\n reply,\n };\n self.cell_event(&cell, job, receive, \"cell isolate dropped WebSocket close\")\n .await\n }\n\n /// Run `webSocketMessage` in the facet `scope` that accepted `ws_id`.\n ///\n /// The frames leave only after the facet's stream proves every write the\n /// facet committed, as a facet's reply waits in `op_facet_fetch`. The\n /// answer carries no position: a facet's write positions count in the\n /// facet's stream, and the root cell's gate would read them as its own.\n pub async fn facet_ws_message(\n &self,\n scope: String,\n facet: &js::FacetSocket,\n ws_id: u64,\n data: js::WsIn,\n started: tokio::sync::oneshot::Sender<()>,\n ) -> anyhow::Result {\n let (reply, receive) = tokio::sync::oneshot::channel();\n let job = CellJob::WsMessage {\n scope: scope.clone(),\n ws_id,\n facet_opening: Some(facet.opening),\n data,\n started: Some(started),\n reply,\n };\n let dispatch = self\n .facet_event(\n &scope,\n facet,\n job,\n receive,\n \"facet isolate dropped WebSocket message\",\n )\n .await?;\n js::prove_facet_stream(facet.stream.clone(), facet.epoch).await?;\n Ok(js::WsDispatch {\n frames: dispatch.frames,\n write_position: None,\n observed_position: None,\n })\n }\n\n /// Run `webSocketClose` in the facet `scope` that accepted `ws_id`. A\n /// write of the handler is proved by the facet's next output, which\n /// waits for the facet's whole stream.\n pub async fn facet_ws_closed(\n &self,\n scope: String,\n facet: &js::FacetSocket,\n ws_id: u64,\n code: u16,\n reason: String,\n was_clean: bool,\n ) -> anyhow::Result<()> {\n let (reply, receive) = tokio::sync::oneshot::channel();\n let job = CellJob::WsClosed {\n scope: scope.clone(),\n ws_id,\n facet_opening: Some(facet.opening),\n code,\n reason,\n was_clean,\n reply,\n };\n self.facet_event(\n &scope,\n facet,\n job,\n receive,\n \"facet isolate dropped WebSocket close\",\n )\n .await\n .map(|_| ())\n }\n\n /// Start one event of a facet's socket and wait for its answer.\n ///\n /// A loaded Worker's isolate stays alive while its root's record holds\n /// it. Any other facet runs in the root's own isolate, which the root's\n /// residency pins. A facet that closed in either place is refused by the\n /// event's own first turn (`facet_socket_refusal`).\n async fn facet_event(\n &self,\n scope: &str,\n facet: &js::FacetSocket,\n job: CellJob,\n receive: tokio::sync::oneshot::Receiver>,\n dropped: &'static str,\n ) -> anyhow::Result {\n let isolate = match js::recorded_facet_slot(&facet.root, facet.epoch, scope) {\n Some(slot) => slot.affiliate(),\n None => self.cell_isolate(&facet.root)?,\n };\n // No alarm reporter: a facet cannot set an alarm.\n run_cell_event(isolate, job, None, receive, dropped).await\n }\n\n /// The isolate a published cell's events run in.\n fn cell_isolate(&self, cell: &str) -> anyhow::Result {\n // The affiliation is taken UNDER the registry lock, while the\n // cell's Residency provably pins the slot, and it is held by the\n // driver for the event's entire async lifetime — including while\n // the event is suspended awaiting host I/O. Without it, a\n // suspended event holds only a bare Arc: stop_cell() drops\n // the Residency, the pool reaps the \"drained\" isolate, and the\n // resumed event enters a freed worker and aborts the process\n // (denoland/celld#147).\n self.cells\n .lock()\n .expect(\"cell registry poisoned\")\n .published\n .get(cell)\n .map(|handle| handle.residency.slot().affiliate())\n .ok_or_else(|| anyhow!(\"cell runtime is not published: {cell}\"))\n }\n\n /// Reserve one fetch against a published cell and return its isolate\n /// affiliation with the reservation. Returning either value alone would\n /// let a caller run work that the target did not admit or release the\n /// admission before the event's asynchronous lifetime ended.\n fn admit_cell_request(&self, cell: &str) -> anyhow::Result> {\n let cells = self.cells.lock().expect(\"cell registry poisoned\");\n let handle = cells\n .published\n .get(cell)\n .ok_or_else(|| anyhow!(\"cell runtime is not published: {cell}\"))?;\n let Some(permit) = handle.requests.acquire() else {\n let in_flight = handle.requests.in_flight.load(Ordering::Acquire);\n if !handle.requests.saturated.swap(true, Ordering::AcqRel) {\n tracing::warn!(\n event = \"cell_overload_refused\",\n scope = %cell,\n node = %self.node,\n region = %self.region,\n in_flight,\n limit = handle.requests.limit,\n \"refused work for a saturated cell\"\n );\n }\n return Ok(None);\n };\n Ok(Some(AdmittedCellRequest {\n affiliation: handle.residency.slot().affiliate(),\n permit,\n }))\n }\n\n /// Start one cell event and wait for its answer.\n ///\n /// The event is driven by its own task, so this future holds nothing\n /// while it waits: the isolate is taken and given back one turn at a\n /// time by `drive_cell`.\n async fn cell_event(\n &self,\n cell: &str,\n job: CellJob,\n receive: tokio::sync::oneshot::Receiver>,\n dropped: &'static str,\n ) -> anyhow::Result {\n let isolate = self.cell_isolate(cell)?;\n run_cell_event(\n isolate,\n job,\n Some(self.alarm_reporter.clone()),\n receive,\n dropped,\n )\n .await\n }\n}\n\n/// Drive one cell event in `isolate` on a task of its own, and wait for its\n/// answer.\nasync fn run_cell_event(\n isolate: crate::pool::Affiliation,\n job: CellJob,\n report: Option,\n receive: tokio::sync::oneshot::Receiver>,\n dropped: &'static str,\n) -> anyhow::Result {\n tokio::spawn(drive_cell(isolate, job, report, None));\n receive.await.context(dropped)?\n}\n\nfn restored_alarm_from_persisted(\n _cell: &str,\n persisted: Option<(i64, i64, u32, u32)>,\n covered: impl FnOnce(i64) -> bool,\n) -> Option {\n let at_ms = match persisted {\n Some((at_ms, ..)) => at_ms,\n None => -1,\n };\n if at_ms < 0 {\n return None;\n }\n // The restore hint grants no publication or retirement authority. The\n // opened SQLite connection installs the new writer identity first.\n Some(celld_logic::RestoredAlarm {\n at_ms,\n covered: covered(at_ms),\n })\n}\n\n#[cfg(all(test, celld_internal_tests))]\npub(crate) fn restored_alarm_for_test(\n cell: &str,\n path: &std::path::Path,\n covered: bool,\n) -> Option {\n let persisted = crate::storage::persisted_alarm(&path.to_string_lossy(), cell, None);\n restored_alarm_from_persisted(cell, persisted, |_| covered)\n}\n\n/// The caller's trace context, when the ingress request carried one.\n/// Malformed headers are ignored; nothing here is trusted for anything\n/// but correlation and (under parentbased samplers, deliberately) the\n/// sampling decision.\nfn inbound_parent(job: &crate::WorkerJob) -> Option {\n if !crate::telemetry::active() {\n return None;\n }\n let crate::WorkerJob::Fetch { headers, .. } = job else {\n return None;\n };\n headers\n .iter()\n .find(|(name, _)| name.eq_ignore_ascii_case(\"traceparent\"))\n .and_then(|(_, value)| crate::telemetry::parse_traceparent(value))\n}\n\n/// An op in flight, carrying the id of the promise it will resolve.\ntype PendingOp = std::pin::Pin<\n Box)> + Send>,\n>;\n\n/// The ops one request is waiting on. Per request, not per isolate: that is\n/// what deleting the pump means, and it is why no attribution table exists.\ntype Ops = futures_util::stream::FuturesUnordered;\n\nfn adopt(ops: &mut Ops, started: Vec) {\n for (id, future) in started {\n ops.push(Box::pin(async move { (id, future.await) }));\n }\n}\n\n/// Cancel native futures and remove their JavaScript resolvers together.\n///\n/// A failed handler can still own a gated reply, but none of its handler ops\n/// can re-enter the isolate while that detached gate waiter finishes.\nfn abort_ops(ops: &mut Ops, entry: &mut js::InFlight) {\n ops.clear();\n entry.abandon();\n}\n\n/// Drive one stateless request to completion, one turn at a time.\n///\n/// The loop the pump used to run, owned by the request instead. Between turns\n/// it holds no isolate — only its affiliation, which is memory rather than\n/// CPU — so a handler awaiting I/O stops nothing else in that isolate.\n/// This compatibility entry owns any cancellation lifetime that it creates.\n/// A caller with an abort publisher must use\n/// `RequestCancellationLifetime::drive_stateless_fetch` instead.\n#[doc(hidden)]\npub fn drive(\n slot: Arc,\n job: crate::WorkerJob,\n telemetry: Option<(Arc, Arc)>,\n) -> impl std::future::Future + Send + 'static {\n drive_affiliated(\n slot.affiliate(),\n StatelessWorkerJob::driver_owned(job),\n telemetry,\n )\n}\n\nfn drive_affiliated(\n affiliation: crate::pool::Affiliation,\n job: StatelessWorkerJob,\n telemetry: Option<(Arc, Arc)>,\n) -> impl std::future::Future + Send + 'static {\n drive_affiliated_with_budget(affiliation, job, telemetry, js::handler_budget())\n}\n\nfn drive_affiliated_with_budget(\n affiliation: crate::pool::Affiliation,\n job: StatelessWorkerJob,\n telemetry: Option<(Arc, Arc)>,\n budget: Duration,\n) -> impl std::future::Future + Send + 'static {\n let StatelessWorkerJob { job, cancellation } = job;\n let driving = drive_affiliated_inner(affiliation, job, telemetry, budget);\n drive_with_request_cancellation(driving, cancellation)\n}\n\n#[cfg(celld_internal_tests)]\n#[doc(hidden)]\npub fn drive_affiliated_with_budget_for_test(\n affiliation: crate::pool::Affiliation,\n job: crate::WorkerJob,\n telemetry: Option<(Arc, Arc)>,\n budget: Duration,\n) -> impl std::future::Future + Send + 'static {\n drive_affiliated_with_budget(\n affiliation,\n StatelessWorkerJob::driver_owned(job),\n telemetry,\n budget,\n )\n}\n\nasync fn drive_affiliated_inner(\n affiliation: crate::pool::Affiliation,\n job: crate::WorkerJob,\n telemetry: Option<(Arc, Arc)>,\n budget: Duration,\n) {\n let slot = affiliation.slot().clone();\n // One sampling decision per request, shared by the SERVER span and\n // the turn context the handler's console/fetch children inherit. A\n // caller's traceparent is honored per the spec: ids adopted either\n // way, its sampled flag deciding only under a parentbased sampler.\n let remote = telemetry.as_ref().and_then(|_| inbound_parent(&job));\n let trace = telemetry\n .as_ref()\n .and_then(|_| crate::telemetry::start_trace_with_parent(remote.as_ref()));\n let recording = trace.and_then(crate::telemetry::TraceContext::recording_ids);\n let mut timing = telemetry.map(|(node, region)| {\n StatelessTiming::start(&job, slot.id, node, region, recording, remote)\n });\n // Admission created `affiliation` before returning the isolate. It stays\n // here for the request's whole life, so maintenance cannot free the heap\n // between placement and this first turn or while a promise is suspended.\n let _affiliation = affiliation;\n drive_worker(\n &slot,\n job,\n trace,\n budget,\n |entry| {\n if let Some(timing) = &mut timing {\n timing.answered(entry);\n }\n },\n |_| {},\n )\n .await;\n}\n\n// Both worker placements use the same execution loop. Keep observation at\n// each turn separate from completion: response timing excludes waitUntil,\n// while the worker-on-cell span includes it. Neither observer owns cleanup.\nasync fn drive_worker(\n slot: &crate::pool::Slot,\n job: crate::WorkerJob,\n trace: Option,\n budget: Duration,\n mut after_turn: impl FnMut(&js::InFlight),\n on_finish: impl FnOnce(&js::InFlight),\n) {\n let mut ops = Ops::new();\n\n let (begun, started) = slot.turn(|worker| worker.turn_begin(job, trace)).await;\n // Nothing is in flight; the reply already carries the error.\n let Some(mut entry) = begun else {\n drop(started);\n return;\n };\n if entry.keeps_native_ops() {\n adopt(&mut ops, started);\n } else {\n drop(started);\n abort_ops(&mut ops, &mut entry);\n }\n after_turn(&entry);\n\n while !entry.finished() {\n let started = match wake_with_cross_entry_gate(&mut ops, &mut entry, budget).await {\n Wake::Op(op, result) => {\n slot.turn(|worker| worker.turn_deliver(&mut entry, op, result))\n .await\n }\n Wake::GatedReply(completion) => {\n entry.finish_gated_reply(completion);\n Vec::new()\n }\n Wake::CancelGatedReply => {\n entry.cancel_gated_reply();\n Vec::new()\n }\n Wake::CrossEntryGateChanged => {\n entry.finish_cross_entry_gates();\n Vec::new()\n }\n Wake::Cancelled { shutdown } => {\n let started = slot\n .turn(|worker| {\n if shutdown {\n worker.turn_cancel_for_shutdown(&mut entry)\n } else {\n worker.turn_cancel(&mut entry)\n }\n })\n .await;\n entry.cancel_gated_reply();\n started\n }\n Wake::Expired => {\n entry.time_out(budget);\n Vec::new()\n }\n Wake::Idle => {\n entry.stuck();\n Vec::new()\n }\n Wake::PendingEventIdle => {\n slot.turn(|worker| worker.turn_cancel_pending_events(&mut entry))\n .await\n }\n Wake::Poll => slot.turn(|worker| worker.turn_poll(&mut entry)).await,\n };\n if entry.keeps_native_ops() {\n adopt(&mut ops, started);\n } else {\n drop(started);\n abort_ops(&mut ops, &mut entry);\n }\n after_turn(&entry);\n }\n\n on_finish(&entry);\n // Dropping `ops` aborts whatever is still pending, which is what a region\n // does on every exit path; their resolvers have to go with them.\n entry.finish_tail_report();\n entry.abandon();\n}\n\n/// What next moves a suspended request.\nenum Wake {\n /// One of its own ops finished.\n Op(u64, Result),\n /// The detached output-gate waiter sent, or abandoned, the final reply.\n GatedReply(Result),\n /// The handler has answered, so cancel its detached reply gate without\n /// entering JavaScript again.\n CancelGatedReply,\n /// A cross-entry claim changed, or subscribing closed a retirement gap.\n CrossEntryGateChanged,\n /// Its client hung up, or shutdown forced the complete event to retire.\n Cancelled { shutdown: bool },\n /// It ran past the handler budget without answering.\n Expired,\n /// Nothing outstanding could ever move it.\n Idle,\n /// A nested Worker entrypoint is pending without native work. Reject that\n /// call inside JavaScript so the enclosing event can catch the failure.\n PendingEventIdle,\n /// Nothing of its own is outstanding, but another event of the same cell\n /// still could settle it. Look in and see.\n Poll,\n}\n\nfn take_cancellation_wake(request_id: Option) -> Option {\n js::take_request_cancellation(request_id).then(|| Wake::Cancelled {\n shutdown: js::take_shutdown_cancellation(request_id),\n })\n}\n\n/// Wait for whichever comes first, holding no isolate.\n///\n/// This is the whole of what a request does between turns, and it is\n/// deliberately the only place that waits: everything else in `drive` either\n/// holds the isolate or is arithmetic.\nasync fn wake_with_cross_entry_gate(\n ops: &mut Ops,\n entry: &mut js::InFlight,\n budget: Duration,\n) -> Wake {\n let wait = entry.prepare_cross_entry_gate_wait();\n match wait\n .wait(wake(ops, entry, budget), |wake| matches!(wake, Wake::Idle))\n .await\n {\n js::input_gate_lifecycle::WaitOutcome::StateChanged => Wake::CrossEntryGateChanged,\n js::input_gate_lifecycle::WaitOutcome::Ordinary(wake) => wake,\n }\n}\n\nasync fn wake(ops: &mut Ops, entry: &mut js::InFlight, budget: Duration) -> Wake {\n loop {\n // An op this event enqueued from inside another event's turn reaches\n // this driver here rather than through `adopt`, because the turn that\n // took it belongs to another entry. See `js::adopt`.\n adopt(ops, entry.take_handed_ops());\n // A WorkerEntrypoint call is its own PendingEvent. When no referenced\n // native operation can move it, reject that call inside JavaScript.\n // An unreferenced signal listener cannot make progress and must not\n // hide a hung call. This check precedes the answered branch because\n // the enclosing handler can keep the call alive through waitUntil\n // after its response.\n if !entry.has_refed_ops() && entry.has_pending_events() {\n return Wake::PendingEventIdle;\n }\n let Some(left) = entry.remaining(budget) else {\n // The handler settled, so neither its reply gate nor waitUntil\n // work is charged to the handler budget. Poll both because the\n // background can progress while the gate owns the reply.\n let request_id = entry.request_id();\n let (gated_reply, context) = entry.gated_reply_and_io_context();\n if let Some(gated_reply) = gated_reply {\n let Some(request_id) = request_id else {\n let next = if ops.is_empty() {\n asyncrt::select_biased! {\n \"a completed gated reply wins a tie with an op another turn hands over\";\n completion = gated_reply => Some(Wake::GatedReply(completion)),\n _ = context.handed_ready() => None,\n }\n } else {\n asyncrt::select_biased! {\n \"a completed gated reply wins a tie with another operation wake\";\n completion = gated_reply => Some(Wake::GatedReply(completion)),\n result = ops.next() => result.map(|(op, result)| Wake::Op(op, result)),\n _ = context.handed_ready() => None,\n }\n };\n match next {\n Some(wake) => return wake,\n // A hand-off, or an exhausted stream after its final\n // completion; take the mailbox before waiting again.\n None => continue,\n }\n };\n // The domain select is declaration-order biased. A reply that\n // completed wins over an op, a mailbox hand-off wins over a\n // cancellation tick, and cancellation is sampled immediately\n // after a mailbox or timer wake so continuous hand-offs cannot\n // starve it.\n let next = if ops.is_empty() {\n asyncrt::select_biased! {\n \"a completed gated reply wins a tie with another request wake\";\n completion = gated_reply => Some(Wake::GatedReply(completion)),\n _ = context.handed_ready() => None,\n _ = asyncrt::sleep(CANCELLATION_TICK) => None,\n }\n } else {\n asyncrt::select_biased! {\n \"a completed gated reply wins a tie with another request wake\";\n completion = gated_reply => Some(Wake::GatedReply(completion)),\n next = async {\n asyncrt::select_biased! {\n \"an operation result wins a tie with another request wake\";\n result = ops.next() => Some(match result {\n Some((op, result)) => Wake::Op(op, result),\n None => Wake::Idle,\n }),\n _ = context.handed_ready() => None,\n _ = asyncrt::sleep(CANCELLATION_TICK) => None,\n }\n } => next,\n }\n };\n if let Some(next) = next {\n return next;\n }\n if let Some(cancelled) = take_cancellation_wake(Some(request_id)) {\n return if matches!(cancelled, Wake::Cancelled { shutdown: true }) {\n cancelled\n } else {\n Wake::CancelGatedReply\n };\n }\n continue;\n }\n // The reply arrived, so only waitUntil work remains. A client\n // disconnect no longer matters, but a lifecycle cancellation\n // must still retire the background work before the runtime stops.\n let Some(request_id) = request_id else {\n let context = entry.io_context();\n let next = asyncrt::select_biased! {\n \"an operation result wins a tie with an op another turn hands over\";\n result = ops.next() => Some(result),\n _ = context.handed_ready() => None,\n };\n match next {\n Some(Some((op, result))) => return Wake::Op(op, result),\n Some(None) => return Wake::Idle,\n // Handed over; the next pass takes it.\n None => continue,\n }\n };\n let next = if ops.is_empty() {\n asyncrt::sleep(CANCELLATION_TICK).await;\n None\n } else {\n asyncrt::select_biased! {\n \"completed waitUntil work wins a tie with periodic lifecycle cancellation sampling\";\n result = ops.next() => Some(match result {\n Some((op, result)) => Wake::Op(op, result),\n None => Wake::Idle,\n }),\n _ = asyncrt::sleep(CANCELLATION_TICK) => None,\n }\n };\n if let Some(next) = next {\n return next;\n }\n if let Some(cancelled) = take_cancellation_wake(Some(request_id)) {\n if matches!(cancelled, Wake::Cancelled { shutdown: true }) {\n return cancelled;\n }\n }\n continue;\n };\n // A disconnect is raised on another thread with nothing to wake this\n // one, so the wait is capped and the flag re-read — as the blocking\n // run loop capped its own. The difference is that reading it costs no\n // isolate, so a request enters V8 only once the client has really gone.\n // An unrefed op can wake a handler that awaits it, but it cannot be\n // the reason this driver stops polling the isolate. A retained signal\n // listener otherwise hides the empty-op path that lets a loopback\n // entry make progress, and both events wait on each other forever.\n let only_unrefed = !ops.is_empty() && !entry.has_refed_ops();\n let capped = if entry.cancellable() || only_unrefed {\n left.min(CANCELLATION_TICK)\n } else {\n left\n };\n // Nothing of this entry's own is outstanding, so there is no future\n // to wait on — only the chance that some *other* entry in this\n // isolate has settled it since the last look. That is an ordinary\n // thing rather than a stall: a cell awaits the alarm it armed, and a\n // Worker awaits a Durable Object it dispatched to. Both used to\n // resolve inside the caller's own run loop, so there was nothing to\n // wait for; both are separate entries now.\n //\n // So \"waiting on nothing\" is a verdict the budget reaches, not one\n // an empty op set proves.\n let context = entry.io_context();\n if ops.is_empty() {\n if left.is_zero() {\n return Wake::Idle;\n }\n // The handed wake belongs on this path too. This entry holds no op\n // of its own, so without it a handed op waits out the whole\n // cancellation tick before the next pass can take it — the delay\n // this branch has no reason to add.\n asyncrt::select_biased! {\n \"an op another turn hands over wins a tie with the cancellation tick\";\n _ = context.handed_ready() => {},\n _ = tokio::time::sleep(capped.min(CANCELLATION_TICK)) => {},\n }\n // Re-read the flag on this path too. A request with nothing\n // outstanding can still have its client hang up, and only the\n // branch below used to look.\n if let Some(cancelled) = take_cancellation_wake(entry.request_id()) {\n return cancelled;\n }\n return Wake::Poll;\n }\n let next = tokio::time::timeout(capped, async {\n asyncrt::select_biased! {\n \"an operation result wins a tie with an op another turn hands over\";\n result = ops.next() => Some(result),\n _ = context.handed_ready() => None,\n }\n })\n .await;\n match next {\n Ok(Some(Some((op, result)))) => return Wake::Op(op, result),\n Ok(Some(None)) => return Wake::Idle,\n // Another turn handed this event an op. The next pass takes it.\n Ok(None) => continue,\n Err(_) => {\n if let Some(cancelled) = take_cancellation_wake(entry.request_id()) {\n return cancelled;\n }\n if capped == left {\n return Wake::Expired;\n }\n if only_unrefed {\n return Wake::Poll;\n }\n continue;\n }\n }\n }\n}\n\n/// One stateless request's canonical timing event.\n///\n/// The phases still mean what they always did, but what they measure has\n/// moved: `queue_wait_us` was the wait for a free worker thread and is now\n/// the wait for admission and the isolate's async gate, and `execution_us`\n/// spans every turn the request took rather than one uninterrupted run.\nstruct StatelessTiming {\n queued_at: Instant,\n request_id: Option,\n node: Arc,\n region: Arc,\n isolate: usize,\n admitted: Instant,\n emitted: bool,\n /// Sampled at creation: `None` is off or unsampled, and nothing more\n /// is ever built for this request.\n trace: Option,\n /// The caller's span, when the request arrived with a traceparent.\n remote_parent: Option,\n /// Why the handler failed, read from the entry when it was answered.\n failure: Option,\n span_name: &'static str,\n span_kind: u8,\n}\n\nimpl StatelessTiming {\n fn start(\n job: &crate::WorkerJob,\n isolate: usize,\n node: Arc,\n region: Arc,\n trace: Option,\n remote_parent: Option,\n ) -> Self {\n let (queued_at, request_id, span_name, span_kind) = match job {\n crate::WorkerJob::Fetch {\n queued_at,\n request_id,\n ..\n } => (\n *queued_at,\n *request_id,\n \"celld.fetch\",\n crate::telemetry::KIND_SERVER,\n ),\n crate::WorkerJob::Queue { queued_at, .. } => (\n *queued_at,\n None,\n \"celld.queue\",\n crate::telemetry::KIND_CONSUMER,\n ),\n crate::WorkerJob::Rpc { .. } => (\n Instant::now(),\n None,\n \"celld.fetch\",\n crate::telemetry::KIND_SERVER,\n ),\n };\n StatelessTiming {\n queued_at,\n request_id,\n node,\n region,\n isolate,\n admitted: Instant::now(),\n emitted: false,\n trace,\n remote_parent,\n failure: None,\n span_name,\n span_kind,\n }\n }\n\n /// Emit once, the first time the client has been answered. `waitUntil`\n /// work continues afterwards and is not part of the response's timing.\n fn answered(&mut self, entry: &js::InFlight) {\n if self.emitted || !entry.answered() {\n return;\n }\n self.emitted = true;\n self.failure = entry.failure().map(str::to_string);\n self.emit();\n }\n\n fn emit(&self) {\n if let Some(ids) = self.trace {\n let total_us = self.queued_at.elapsed().as_micros() as i64;\n let mut span = crate::telemetry::Span::new(ids, self.span_name, self.span_kind);\n span.start_unix_us = crate::telemetry::now_unix_us() - total_us;\n span.duration_us = total_us;\n span.ok = self.failure.is_none();\n span.error = self.failure.clone();\n span.request_id = self.request_id.map(js::request_id_string);\n span.isolate = Some(self.isolate as u64);\n span.parent_span_id = self.remote_parent.map(|parent| parent.span_id);\n span.parent_remote = self.remote_parent.map(|_| true);\n span.queue_wait_us =\n Some(self.admitted.duration_since(self.queued_at).as_micros() as i64);\n crate::telemetry::record(span);\n }\n // An info!-per-request costs real throughput on the hot path, so the\n // `enabled!` guard skips the elapsed math and the formatting when the\n // target is off. The lab turns it on with RUST_LOG=info,timing=debug.\n let Some(request_id) = self\n .request_id\n .filter(|_| tracing::enabled!(target: \"timing\", tracing::Level::DEBUG))\n else {\n return;\n };\n tracing::debug!(\n target: \"timing\",\n event = \"worker_fetch_timing\",\n outcome = \"completed\",\n request_id = %js::request_id_string(request_id),\n node = %self.node,\n region = %self.region,\n runtime_version = env!(\"CARGO_PKG_VERSION\"),\n total_us = self.queued_at.elapsed().as_micros() as u64,\n queue_wait_us = self.admitted.duration_since(self.queued_at).as_micros() as u64,\n execution_us = self.admitted.elapsed().as_micros() as u64,\n isolate = self.isolate,\n \"stateless Worker fetch completed\"\n );\n }\n}\n\n/// Initialise V8, at most once.\n///\n/// **Must run before any thread that will enter an isolate is created.**\n/// V8 protects its pointer tables with a memory protection key, and the\n/// `PKRU` register granting access to that key is per-thread and inherited\n/// at thread creation. A thread created before this runs never receives\n/// access, so its first read of a dispatch table traps with `SEGV_PKUERR`\n/// on any CPU that supports protection keys.\npub fn init_v8() {\n js::Engine::init();\n}\n\nimpl StatelessRuntime {\n #[doc(hidden)]\n pub fn start(\n config: Arc,\n node: Arc,\n region: Arc,\n ) -> anyhow::Result {\n let build = {\n let config = config.clone();\n move || Worker::load_config(config.clone())\n };\n let isolates = Arc::new(crate::pool::Pool::new(\n pool_limits(),\n admission_wait(),\n Box::new(build),\n ));\n // Eagerly, so a script that does not load fails here rather than on\n // every request, and so the first request does not pay for compiling\n // it. Growth past this one stays lazy.\n isolates.warm().context(\"stateless Worker failed to load\")?;\n // Give isolates back when the burst that grew them is over. Without\n // this the pool only grows, and every heap a burst created is held\n // for the life of the process. Long relative to a request, because\n // retiring is not urgent and a short period would thrash a pool that\n // is about to be busy again.\n if let Ok(handle) = tokio::runtime::Handle::try_current() {\n // Weak, so this loop cannot keep a superseded generation's pool\n // alive: it ends when the pool is dropped.\n let reaping = Arc::downgrade(&isolates);\n let reclaim = isolates.reclaim_bell();\n handle.spawn(async move {\n // `interval` fires its first tick immediately, which\n // reaped the isolate `warm` had just built and handed the\n // first request the compile cost warming exists to avoid.\n let mut tick = tokio::time::interval_at(\n tokio::time::Instant::now() + REAP_INTERVAL,\n REAP_INTERVAL,\n );\n tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);\n loop {\n let reap = crate::asyncrt::select_biased! {\n \"the paced reap also frees, so it wins a tie with the reclaim bell\";\n _ = tick.tick() => true,\n _ = reclaim.notified() => false,\n };\n let Some(mut pool) = reaping.upgrade() else {\n return;\n };\n if reap {\n pool.reap();\n continue;\n }\n // A condemned isolate is freed as soon as it drains, not\n // at the next tick. Admission holds the guard only while\n // it places one request, so a short retry gets it.\n while !pool.free_drained() {\n drop(pool);\n crate::asyncrt::sleep(RECLAIM_RETRY).await;\n let Some(again) = reaping.upgrade() else {\n return;\n };\n pool = again;\n }\n }\n });\n }\n Ok(Self {\n isolates,\n node,\n region,\n })\n }\n\n /// Admit and drive one stateless event. A verb supplies only its job and\n /// result type, so admission, shedding, pressure, and task failure cannot\n /// drift between fetch, RPC, and queue dispatch.\n async fn dispatch(\n &self,\n verb: StatelessVerb,\n make_job: impl FnOnce(tokio::sync::oneshot::Sender>) -> crate::WorkerJob,\n ) -> anyhow::Result {\n let shedding = crate::ownership_store::node_is_shedding();\n self.dispatch_stateless(verb, shedding, move |reply| {\n StatelessWorkerJob::driver_owned(make_job(reply))\n })\n .await\n }\n\n #[doc(hidden)]\n pub async fn dispatch_with_shedding(\n &self,\n verb: StatelessVerb,\n shedding: bool,\n make_job: impl FnOnce(tokio::sync::oneshot::Sender>) -> crate::WorkerJob,\n ) -> anyhow::Result {\n self.dispatch_stateless(verb, shedding, move |reply| {\n StatelessWorkerJob::driver_owned(make_job(reply))\n })\n .await\n }\n\n async fn dispatch_stateless(\n &self,\n verb: StatelessVerb,\n shedding: bool,\n make_job: impl FnOnce(tokio::sync::oneshot::Sender>) -> StatelessWorkerJob,\n ) -> anyhow::Result {\n let affiliation = self.isolates.admit_or_wait(shedding).await?;\n let (reply, receive) = tokio::sync::oneshot::channel();\n let job = make_job(reply);\n // Spawned rather than awaited inline, because the response and the\n // event are not the same lifetime: `waitUntil` work outlives the\n // answer, and the driver keeps turning until it settles.\n let driving = tokio::spawn(drive_affiliated(\n affiliation,\n job,\n Some((self.node.clone(), self.region.clone())),\n ));\n match receive.await {\n Ok(result) => result,\n // The driver dropped the reply without sending. Joining it here —\n // and only here — turns a bare \"channel closed\" into the panic\n // that actually caused it, at no cost on the path that works.\n Err(_) => match driving.await {\n Err(error) => Err(verb.task_died(error)),\n Ok(()) => Err(verb.dropped_result()),\n },\n }\n }\n\n /// Serve one stateless request, entering an isolate once per turn.\n ///\n /// The request drives itself: it is admitted and placed, runs its first\n /// turn, then awaits its *own* ops with no isolate held, re-entering for\n /// each completion. Nothing multiplexes and nothing demultiplexes, which\n /// is what deleting the pump buys.\n #[doc(hidden)]\n pub fn fetch(\n &self,\n url: String,\n method: String,\n body: js::RequestBody,\n headers: Vec<(String, String)>,\n cancellation: Option>,\n ) -> impl std::future::Future> + Send + 'static {\n self.fetch_target(None, url, method, body, headers, cancellation)\n }\n\n /// Serve one default or named stateless Worker entrypoint.\n fn fetch_target(\n &self,\n entrypoint: Option,\n url: String,\n method: String,\n body: js::RequestBody,\n headers: Vec<(String, String)>,\n cancellation: Option>,\n ) -> impl std::future::Future> + Send + 'static {\n let runtime = self.clone();\n let make_job =\n stateless_fetch_job_factory(entrypoint, url, method, body, headers, cancellation);\n async move {\n let shedding = crate::ownership_store::node_is_shedding();\n runtime\n .dispatch_stateless(StatelessVerb::Fetch, shedding, make_job)\n .await\n }\n }\n\n /// An entrypoint RPC, on the isolate pool like fetch.\n ///\n /// It used to go to the worker threads, because the old RPC dispatcher\n /// blocked while the handler awaited — and a\n /// blocking call cannot hold a pool slot without parking a tokio worker\n /// on V8. Turning it into `begin`/`drive` removes the blocking, and with\n /// it the last reason `WorkerPool` existed.\n async fn rpc(\n &self,\n entrypoint: String,\n props: Vec,\n operation: crate::WorkerRpcOperation,\n ) -> anyhow::Result> {\n self.dispatch(StatelessVerb::Rpc, move |reply| crate::WorkerJob::Rpc {\n entrypoint,\n operation,\n props,\n invocation_limits: None,\n reply,\n })\n .await\n }\n\n /// Dispatch one leased queue batch through the stateless isolate pool.\n #[doc(hidden)]\n pub async fn queue(&self, batch: js::QueueBatch) -> anyhow::Result {\n let queued_at = Instant::now();\n self.dispatch(StatelessVerb::Queue, move |reply| crate::WorkerJob::Queue {\n queued_at,\n batch,\n reply,\n })\n .await\n }\n}\n\nstruct CellIsolateStartupTiming {\n started: Instant,\n scope: String,\n node: Arc,\n region: Arc,\n epoch: u64,\n fresh: bool,\n}\n\nimpl CellIsolateStartupTiming {\n fn emit(&self, outcome: &str, failure_phase: &str) -> u64 {\n let total_us = self.started.elapsed().as_micros() as u64;\n if let Some(ids) =\n crate::telemetry::start_trace().and_then(crate::telemetry::TraceContext::recording_ids)\n {\n let mut span = crate::telemetry::Span::new(\n ids,\n \"celld.cell_startup\",\n crate::telemetry::KIND_INTERNAL,\n );\n span.start_unix_us = crate::telemetry::now_unix_us() - total_us as i64;\n span.duration_us = total_us as i64;\n span.ok = outcome == \"ready\";\n span.error = (!span.ok).then(|| failure_phase.to_string());\n span.cell = Some(self.scope.clone());\n span.epoch = Some(self.epoch);\n crate::telemetry::record(span);\n }\n tracing::info!(\n event = \"cell_isolate_startup_timing\",\n outcome,\n failure_phase,\n scope = %self.scope,\n node = %self.node,\n region = %self.region,\n runtime_version = env!(\"CARGO_PKG_VERSION\"),\n epoch = self.epoch,\n fresh = self.fresh,\n total_us,\n \"cell isolate startup completed\"\n );\n total_us\n }\n}\n\n/// Build one isolate for a Worker script's cells.\n///\n/// No cells yet: `Worker::own_cell` opens each cell's storage as it is\n/// placed here, because this isolate outlives any of them.\nfn load_cell_isolate(config: Arc) -> anyhow::Result {\n #[cfg(debug_assertions)]\n if let Ok(barrier) = std::env::var(\"CELLD_TEST_CELL_STARTUP_BARRIER\") {\n while !Path::new(&barrier).exists() {\n std::thread::sleep(Duration::from_millis(10));\n }\n }\n #[cfg(debug_assertions)]\n if std::env::var(\"CELLD_TEST_CELL_STARTUP_FAILURE\").as_deref() == Ok(\"1\") {\n return Err(anyhow!(\"injected cell isolate startup failure\"));\n }\n Worker::load_config(config).map_err(|error| error.context(\"cell isolate load failed\"))\n}\n\n/// Drive the Worker entry in a cell's isolate.\n///\n/// The stateless `drive` loop, with the isolate named rather than admitted:\n/// this request must run *here*, because the cell it will route to lives here\n/// and arriving in a warm isolate is the point.\nasync fn drive_worker_on_cell(affiliation: crate::pool::Affiliation, job: crate::WorkerJob) {\n // Held to the end of this function: the request's claim on the isolate\n // outlives every suspension, so the pool cannot free the worker under\n // a parked event (denoland/celld#147).\n let slot = affiliation.slot().clone();\n let remote = inbound_parent(&job);\n let trace = crate::telemetry::start_trace_with_parent(remote.as_ref());\n let recording = trace.and_then(crate::telemetry::TraceContext::recording_ids);\n let span_started = recording.map(|_| (Instant::now(), crate::telemetry::now_unix_us()));\n drive_worker(\n &slot,\n job,\n trace,\n js::handler_budget(),\n |_| {},\n |entry| {\n if let (Some(ids), Some((started, start_unix))) = (recording, span_started) {\n let mut span =\n crate::telemetry::Span::new(ids, \"celld.fetch\", crate::telemetry::KIND_SERVER);\n span.start_unix_us = start_unix;\n span.duration_us = started.elapsed().as_micros() as i64;\n span.ok = entry.finished() && entry.failure().is_none();\n span.error = entry.failure().map(str::to_string);\n span.parent_span_id = remote.map(|parent| parent.span_id);\n span.parent_remote = remote.map(|_| true);\n crate::telemetry::record(span);\n }\n },\n )\n .await;\n}\n\n/// Report a turn's alarm moves to the host.\n///\n/// The host otherwise hears about a cell's alarm only when the request\n/// finishes, because `ActivityGuard` reports it on drop. That is too late\n/// for a handler that arms an alarm and then *awaits* it: the timer would\n/// not be scheduled until the request ended, and the request cannot end\n/// until the alarm fires. The blocking run loop hid this by polling\n/// `get_alarm` between turns and firing a due alarm inline; with events as\n/// entries, the host has to be told as soon as the arming turn returns.\nfn report_alarm_moves(\n report: &Option,\n moves: Vec<(String, celld_logic::wake::AlarmSnapshot)>,\n) {\n if let Some(report) = report {\n for (scope, at_ms) in moves {\n report(scope, at_ms);\n }\n }\n}\n\n/// Drive one cell event to completion, one turn at a time.\n///\n/// The same loop as `drive`, and deliberately so: what makes a cell event\n/// different is which realm its turns enter and that it waits for the input\n/// gate first — not how it is pumped. Between turns it holds no isolate, so\n/// a handler awaiting I/O stops neither its own cell's next event nor any\n/// other cell sharing the isolate.\n#[doc(hidden)]\npub async fn drive_cell(\n affiliation: crate::pool::Affiliation,\n job: CellJob,\n report: Option,\n parent: Option,\n) {\n let cancellation = job\n .request_id()\n .map(RequestCancellationLifetime::from_request_id);\n #[cfg(celld_internal_tests)]\n let _ = drive_cell_inner(\n affiliation,\n job,\n report,\n parent,\n js::handler_budget(),\n cancellation,\n DriveCellTestObservers::default(),\n )\n .await;\n #[cfg(not(celld_internal_tests))]\n let _ = drive_cell_inner(\n affiliation,\n job,\n report,\n parent,\n js::handler_budget(),\n cancellation,\n )\n .await;\n}\n\n/// Let an answered alarm's leftover I/O finish on its own, and report how it\n/// ended.\n///\n/// `fire_alarm` stops waiting for the drive as soon as the handler returns, so\n/// nothing joins the task after that. Without this watcher a panic in the\n/// leftover work, which the former `drive.await` turned into a loud process\n/// panic, would end the task in silence.\nfn watch_detached_alarm_drive(\n drive: crate::asyncrt::TaskHandle>>,\n cell: &str,\n op: celld_logic::OpId,\n) {\n let cell = cell.to_string();\n crate::asyncrt::spawn(async move {\n let failure = match drive.await {\n Ok(Ok(_)) => return,\n Ok(Err(error)) => format!(\"{error:#}\"),\n Err(panic) => panic.to_string(),\n };\n tracing::warn!(\n event = \"alarm_leftover_work_failed\",\n scope = %cell,\n op,\n failure = %failure,\n \"work an alarm handler left pending ended in a failure\"\n );\n })\n .detach();\n}\n\n/// Drive an alarm event and answer the position its final turn reached.\n///\n/// A handler that failed between turns, or threw before it suspended, has\n/// its retry record written by the final alarm turn, after its error left.\n/// That commit is as unproven as the handler's own, so `fire_alarm` gates on\n/// the position from here. `None` when the event settled its claim itself;\n/// its error then carries the position.\npub(crate) async fn drive_alarm(\n affiliation: crate::pool::Affiliation,\n job: CellJob,\n report: Option,\n) -> anyhow::Result> {\n let cancellation = job\n .request_id()\n .map(RequestCancellationLifetime::from_request_id);\n drive_cell_inner(\n affiliation,\n job,\n report,\n None,\n js::handler_budget(),\n cancellation,\n #[cfg(celld_internal_tests)]\n DriveCellTestObservers::default(),\n )\n .await\n}\n\n#[cfg(celld_internal_tests)]\n#[doc(hidden)]\npub async fn drive_cell_with_budget_for_test(\n affiliation: crate::pool::Affiliation,\n job: CellJob,\n report: Option,\n parent: Option,\n budget: Duration,\n) {\n let cancellation = job\n .request_id()\n .map(RequestCancellationLifetime::from_request_id);\n let _ = drive_cell_inner(\n affiliation,\n job,\n report,\n parent,\n budget,\n cancellation,\n DriveCellTestObservers::default(),\n )\n .await;\n}\n\nasync fn drive_cell_with_request_cancellation(\n affiliation: crate::pool::Affiliation,\n job: CellJob,\n report: Option,\n parent: Option,\n cancellation: Option>,\n) {\n assert_eq!(\n job.request_id(),\n cancellation.as_ref().map(|lifetime| lifetime.request_id),\n \"a cell fetch and its cancellation lifetime must have the same request id\"\n );\n #[cfg(celld_internal_tests)]\n let _ = drive_cell_inner(\n affiliation,\n job,\n report,\n parent,\n js::handler_budget(),\n cancellation,\n DriveCellTestObservers::default(),\n )\n .await;\n #[cfg(not(celld_internal_tests))]\n let _ = drive_cell_inner(\n affiliation,\n job,\n report,\n parent,\n js::handler_budget(),\n cancellation,\n )\n .await;\n}\n\n#[cfg(celld_internal_tests)]\n#[doc(hidden)]\npub async fn drive_cell_observing_gated_failure_for_test(\n affiliation: crate::pool::Affiliation,\n job: CellJob,\n gated_failure: tokio::sync::oneshot::Sender,\n) {\n let cancellation = job\n .request_id()\n .map(RequestCancellationLifetime::from_request_id);\n let _ = drive_cell_inner(\n affiliation,\n job,\n None,\n None,\n js::handler_budget(),\n cancellation,\n DriveCellTestObservers {\n gated_failure: Some(gated_failure),\n ..DriveCellTestObservers::default()\n },\n )\n .await;\n}\n\n#[cfg(celld_internal_tests)]\n#[doc(hidden)]\npub async fn drive_cell_observing_gated_op_drop_for_test(\n affiliation: crate::pool::Affiliation,\n job: CellJob,\n gated_failure: tokio::sync::oneshot::Sender,\n native_op_dropped: tokio::sync::oneshot::Sender<()>,\n) {\n let cancellation = job\n .request_id()\n .map(RequestCancellationLifetime::from_request_id);\n let _ = drive_cell_inner(\n affiliation,\n job,\n None,\n None,\n js::handler_budget(),\n cancellation,\n DriveCellTestObservers {\n gated_failure: Some(gated_failure),\n native_op_dropped: Some(native_op_dropped),\n },\n )\n .await;\n}\n\n#[cfg(celld_internal_tests)]\n#[derive(Default)]\nstruct DriveCellTestObservers {\n gated_failure: Option>,\n native_op_dropped: Option>,\n}\n\n#[cfg(celld_internal_tests)]\nstruct NativeOpDropProbe(Option>);\n\n#[cfg(celld_internal_tests)]\nimpl Drop for NativeOpDropProbe {\n fn drop(&mut self) {\n if let Some(dropped) = self.0.take() {\n let _ = dropped.send(());\n }\n }\n}\n\n#[cfg(celld_internal_tests)]\nfn adopt_cell_ops_for_test(\n ops: &mut Ops,\n started: Vec,\n observer: &mut Option>,\n) {\n for (id, future) in started {\n let probe = observer\n .take()\n .map(|observer| NativeOpDropProbe(Some(observer)));\n ops.push(Box::pin(async move {\n let _probe = probe;\n (id, future.await)\n }));\n }\n}\n\n#[cfg(celld_internal_tests)]\nfn notify_gated_failure_for_test(\n entry: &mut js::InFlight,\n observer: &mut Option>,\n) {\n let gated = entry.gated_reply_and_io_context().0.is_some();\n if let Some(observer) = observer.take() {\n let _ = observer.send(gated);\n }\n}\n\nasync fn drive_cell_inner(\n affiliation: crate::pool::Affiliation,\n mut job: CellJob,\n report: Option,\n parent: Option,\n budget: Duration,\n request_cancellation: Option>,\n #[cfg(celld_internal_tests)] mut test_observers: DriveCellTestObservers,\n) -> anyhow::Result> {\n let _request_cancellation = request_cancellation.map(RequestCancellationGuard::shared);\n // Held to the end of this function: the event's claim on the isolate\n // outlives every suspension, so the pool cannot free the worker under\n // a parked event (denoland/celld#147).\n let slot = affiliation.slot().clone();\n let scope = job.scope().to_string();\n let _queue_producer = if job.is_queue_producer() {\n match slot.queue_producer(&scope) {\n Some(permit) => Some(permit),\n None => {\n job.fail(anyhow!(js::CellOverloaded));\n return Ok(None);\n }\n }\n } else {\n None\n };\n // Two calls a caller made back-to-back reach the cell in that order.\n // This is the only place that can hold it: everything upstream is a\n // race, and everything downstream has already been delivered. Held\n // until the event *begins*, not until it finishes — a handler that\n // waits must not stop the next call arriving, or cell events would\n // stop interleaving and the DO contract with them.\n let mut order = job.take_order();\n if let Some(order) = order.as_mut() {\n order.wait().await;\n }\n // Join the dispatching Worker's trace when there is one — the root\n // already made the sampling decision — else decide a fresh root.\n // The seed is captured before the job moves, recorded once the entry\n // settles.\n let trace = match parent.as_ref() {\n Some(parent) => crate::telemetry::child_of(parent),\n None => crate::telemetry::start_trace(),\n };\n let recording = trace.and_then(crate::telemetry::TraceContext::recording_ids);\n let span_seed = recording.map(|_| {\n let name = match &job {\n CellJob::Fetch { .. } => \"celld.cell_fetch\",\n CellJob::Alarm { .. } => \"celld.alarm\",\n CellJob::Rpc { .. } => \"celld.rpc\",\n CellJob::WsOpen { .. } => \"celld.ws_open\",\n CellJob::WsMessage { .. } => \"celld.ws_message\",\n CellJob::WsClosed { .. } => \"celld.ws_close\",\n #[cfg(celld_internal_tests)]\n CellJob::SyncErrorForTest { .. } => \"celld.sync_error_test\",\n };\n (\n name,\n job.scope().to_string(),\n Instant::now(),\n crate::telemetry::now_unix_us(),\n )\n });\n let mut ops = Ops::new();\n // `blockConcurrencyWhile` shuts the cell's gate, and a shut gate means no\n // event reaches that cell until it opens. The blocking loop left a\n // refused job on the channel; there is no channel now, so the event waits\n // here.\n //\n // Asked *inside* the turn, which is the whole of it. A handler shuts the\n // gate while holding the isolate, so a check made before taking the\n // isolate can pass and then queue behind the very block it should have\n // waited for. On an idle machine the blocking event always won that race\n // and the bug was invisible; under load it is not.\n //\n // `cell_gate_wait` checks and enqueues under the gate's own lock, so the\n // ticket cannot be missed by a release landing between the two. Only the\n // waiting happens out here, because a turn may not await.\n let mut pending = Some(job);\n let started_event = match pending.as_mut() {\n Some(CellJob::WsMessage { started, .. }) => started.take(),\n _ => None,\n };\n let (begun, started, moves) = loop {\n let mut waiting = None;\n let taken = slot\n .turn_cell(&scope, |worker| {\n let job = pending.take().expect(\"one job per attempt\");\n if let Some(open) = js::cell_gate_wait(job.scope()) {\n waiting = Some(open);\n pending = Some(job);\n return None;\n }\n let (begun, started) = worker.turn_begin_cell(job, trace);\n Some((begun, started, worker.take_alarm_moves()))\n })\n .await;\n match taken {\n Some(taken) => break taken,\n None => match waiting {\n None => {}\n Some(open) => match open.await {\n // The gate opened normally; try for the isolate again.\n Ok(Ok(())) => {}\n // The critical section this event queued behind failed,\n // which reset the cell. Delivering now would run against\n // state that no longer exists, so refuse instead and say\n // why the caller is being refused.\n Ok(Err(failure)) => {\n if let Some(job) = pending.take() {\n job.fail(anyhow!(failure));\n }\n return Ok(None);\n }\n // The cell stopped while this event waited.\n Err(_) => return Ok(None),\n },\n },\n }\n };\n // Delivered. Whatever the caller sent next may go.\n // A socket waits for this turn, including the input gate, rather than\n // handler completion. Acknowledging a queued task instead can reorder\n // messages; awaiting completion prevents a later message cancelling it.\n if let Some(started) = started_event {\n let _ = started.send(());\n }\n if let Some(order) = order.as_mut() {\n order.delivered();\n }\n drop(order);\n report_alarm_moves(&report, moves);\n // Nothing is in flight; the reply already carries the error.\n let Some(mut entry) = begun else {\n drop(started);\n return Ok(None);\n };\n if entry.keeps_native_ops() {\n #[cfg(celld_internal_tests)]\n adopt_cell_ops_for_test(&mut ops, started, &mut test_observers.native_op_dropped);\n #[cfg(not(celld_internal_tests))]\n adopt(&mut ops, started);\n } else {\n #[cfg(celld_internal_tests)]\n adopt_cell_ops_for_test(&mut ops, started, &mut test_observers.native_op_dropped);\n #[cfg(not(celld_internal_tests))]\n drop(started);\n abort_ops(&mut ops, &mut entry);\n }\n #[cfg(celld_internal_tests)]\n if entry.gated_reply_and_io_context().0.is_some() {\n notify_gated_failure_for_test(&mut entry, &mut test_observers.gated_failure);\n }\n\n while !entry.finished() {\n let (started, moves) = match wake_with_cross_entry_gate(&mut ops, &mut entry, budget).await\n {\n Wake::Op(op, result) => {\n slot.turn(|worker| {\n let started = worker.turn_deliver(&mut entry, op, result);\n (started, worker.take_alarm_moves())\n })\n .await\n }\n Wake::GatedReply(completion) => {\n entry.finish_gated_reply(completion);\n (Vec::new(), Vec::new())\n }\n Wake::CancelGatedReply => {\n entry.cancel_gated_reply();\n (Vec::new(), Vec::new())\n }\n Wake::CrossEntryGateChanged => {\n entry.finish_cross_entry_gates();\n (Vec::new(), Vec::new())\n }\n Wake::Cancelled { shutdown } => {\n let cancelled = slot\n .turn(|worker| {\n let started = if shutdown {\n worker.turn_cancel_for_shutdown(&mut entry)\n } else {\n worker.turn_cancel(&mut entry)\n };\n (started, worker.take_alarm_moves())\n })\n .await;\n entry.cancel_gated_reply();\n #[cfg(celld_internal_tests)]\n notify_gated_failure_for_test(&mut entry, &mut test_observers.gated_failure);\n cancelled\n }\n Wake::Expired => {\n entry.time_out(budget);\n slot.turn(|worker| worker.turn_retire_input_gates(&entry))\n .await;\n #[cfg(celld_internal_tests)]\n notify_gated_failure_for_test(&mut entry, &mut test_observers.gated_failure);\n (Vec::new(), Vec::new())\n }\n Wake::Idle => {\n entry.stuck();\n slot.turn(|worker| worker.turn_retire_input_gates(&entry))\n .await;\n #[cfg(celld_internal_tests)]\n notify_gated_failure_for_test(&mut entry, &mut test_observers.gated_failure);\n (Vec::new(), Vec::new())\n }\n Wake::PendingEventIdle => {\n slot.turn(|worker| {\n let started = worker.turn_cancel_pending_events(&mut entry);\n (started, worker.take_alarm_moves())\n })\n .await\n }\n Wake::Poll => {\n slot.turn(|worker| {\n let started = worker.turn_poll(&mut entry);\n (started, worker.take_alarm_moves())\n })\n .await\n }\n };\n #[cfg(celld_internal_tests)]\n if entry.gated_reply_and_io_context().0.is_some() {\n notify_gated_failure_for_test(&mut entry, &mut test_observers.gated_failure);\n }\n if entry.keeps_native_ops() {\n #[cfg(celld_internal_tests)]\n adopt_cell_ops_for_test(&mut ops, started, &mut test_observers.native_op_dropped);\n #[cfg(not(celld_internal_tests))]\n adopt(&mut ops, started);\n } else {\n drop(started);\n abort_ops(&mut ops, &mut entry);\n }\n // A turn that ran JS may have armed an alarm this very request is\n // waiting on, so the host hears about it now rather than when the\n // request ends.\n report_alarm_moves(&report, moves);\n }\n\n // An alarm that ended without running JS again still owes its outcome —\n // `fail` deliberately leaves the claim, because it runs between turns\n // where cell storage is unreachable (denoland/celld#170) — and\n // recording it is storage only the isolate can reach.\n let mut alarm_write = Ok(None);\n if entry.owes_alarm() {\n let (moves, write) = slot\n .turn(|worker| {\n let write = worker.turn_finish_alarm(&mut entry);\n (worker.take_alarm_moves(), write)\n })\n .await;\n report_alarm_moves(&report, moves);\n alarm_write = write;\n }\n if let (Some(ids), Some((name, cell, started, start_unix))) = (recording, span_seed) {\n let kind = if name == \"celld.cell_fetch\" {\n crate::telemetry::KIND_SERVER\n } else {\n crate::telemetry::KIND_INTERNAL\n };\n let mut span = crate::telemetry::Span::new(ids, name, kind);\n span.start_unix_us = start_unix;\n span.duration_us = started.elapsed().as_micros() as i64;\n span.ok = entry.finished() && entry.failure().is_none();\n span.error = entry.failure().map(str::to_string);\n span.cell = Some(cell);\n span.parent_span_id = parent.map(|parent| parent.span_id);\n span.parent_remote = parent.map(|_| false);\n crate::telemetry::record(span);\n }\n slot.turn(|worker| worker.turn_abandon_input_gates(&entry))\n .await;\n entry.abandon();\n alarm_write\n}\n\nfn path_text(path: &Path) -> &str {\n path.to_str().expect(\"celld data path must be UTF-8\")\n}\n\n/// Tell the container module which classes the current generation\n/// attaches a container to, and warm their images off the request path.\n/// A generation with no container class installs an empty list, so a\n/// class dropped by a redeploy cannot start a container.\nfn install_container_specs(generation: &Generation) {\n crate::container::install_specs(\n generation.containers.clone(),\n generation.fence_image.clone(),\n );\n if generation.containers.is_empty() {\n return;\n }\n if let Ok(handle) = tokio::runtime::Handle::try_current() {\n handle.spawn(crate::container::prewarm());\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",{"_54":55,"_91":92,"_93":94,"_95":96,"_97":98,"_99":100,"_101":102},"sources",[],"viewer","meta",{"_83":84,"_85":86,"_87":88,"_42":31,"_89":82,"_90":31,"_48":49,"_50":49},"branches",[82],"tags",[69,70,71,72,73,74,75,76,77,78,79,80,81],"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","main","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":49,"_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,"_52":125,"_126":31,"_127":125,"_128":104},{"_119":129,"_121":130,"_15":129,"_123":131,"_52":125,"_126":31,"_127":125,"_128":104},{"_119":120,"_121":122,"_15":120,"_123":124,"_52":125,"_126":31,"_127":125,"_128":114},"id","issues","name","Issues","filters",{},true,"saved","pinnable","count","tasks","Tasks",{},"Changes",{},{"_119":110,"_121":132,"_15":110,"_123":177,"_52":125,"_126":31,"_127":125},{"_119":129,"_121":130,"_15":129,"_123":176,"_52":125,"_126":31,"_127":125},{"_119":120,"_121":122,"_15":120,"_123":175,"_52":125,"_126":31,"_127":125},{"_119":172,"_121":173,"_15":172,"_123":174,"_52":31,"_126":31,"_127":125},{"_119":169,"_121":170,"_15":169,"_123":171,"_52":31,"_126":31,"_127":125},{"_119":106,"_121":167,"_15":97,"_123":168,"_52":31,"_126":31,"_127":125},{"_119":107,"_121":164,"_15":165,"_123":166,"_52":31,"_126":31,"_127":125},{"_119":103,"_121":162,"_15":103,"_123":163,"_52":31,"_126":31,"_127":125},{"_119":158,"_121":159,"_15":160,"_123":161,"_52":31,"_126":31,"_127":125},{"_119":93,"_121":156,"_15":93,"_123":157,"_52":31,"_126":31,"_127":31},{"_119":153,"_121":154,"_15":153,"_123":155,"_52":31,"_126":31,"_127":31},{"_119":150,"_121":151,"_15":150,"_123":152,"_52":31,"_126":31,"_127":31},{"_119":147,"_121":148,"_15":147,"_123":149,"_52":31,"_126":31,"_127":31},"decisions","Decisions",{},"changelog","Changelog",{},"roadmap","Roadmap",{},"Documents",{},"activity","Activity","insights",{},"Releases",{},"Deployments","automations/deployments",{},"Workflows",{},"discussions","Discussions",{},"goals","Goals",{},{},{},{},"groups","user","agents",[],"now",1794434629344,"sidebar","expanded"]