diff --git a/Cargo.lock b/Cargo.lock index 22f734e5e..fbd373eec 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3373,6 +3373,7 @@ dependencies = [ "once_cell", "scopeguard", "serde", + "sha2", "thiserror 2.0.8", "tokio", "tokio-util", diff --git a/crates/base/src/worker/pool.rs b/crates/base/src/worker/pool.rs index e47de2bdd..48c5439f7 100644 --- a/crates/base/src/worker/pool.rs +++ b/crates/base/src/worker/pool.rs @@ -26,6 +26,7 @@ use ext_workers::context::Timing; use ext_workers::context::TimingStatus; use ext_workers::context::UserWorkerMsgs; use ext_workers::context::UserWorkerProfile; +use ext_workers::context::WorkerCodeIdentity; use ext_workers::context::WorkerContextInitOpts; use ext_workers::context::WorkerRuntimeOpts; use ext_workers::errors::WorkerError; @@ -155,6 +156,13 @@ pub struct ActiveWorkerRegistry { next: Option, notify_pair: (flume::Sender>, flume::Receiver>), sem: Arc, + + // Executable identity shared by every worker currently registered here. + // Reuse for this service path is only handed out when the incoming request + // resolves to the same identity; a mismatch retires these workers so the + // request falls through to normal creation (see supabase/edge-runtime#721). + // `None` until the first worker for the service path is registered. + code_identity: Option, } impl ActiveWorkerRegistry { @@ -164,6 +172,7 @@ impl ActiveWorkerRegistry { next: Option::default(), notify_pair: flume::unbounded(), sem: Arc::new(Semaphore::const_new(max_parallelism)), + code_identity: None, } } @@ -293,8 +302,12 @@ impl WorkerPool { .as_user_worker() .is_some_and(|it| !is_oneshot_policy && it.force_create); + // Identity of the executable artifact this request carries. A warm worker + // is only eligible for reuse when it was built from the same artifact. + let code_identity = worker_options.code_identity(); + if let Some(ref active_worker_uuid) = - self.maybe_active_worker(&service_path, force_create) + self.maybe_active_worker(&service_path, force_create, code_identity) { if tx .send(Ok(CreateUserWorkerResult { @@ -507,6 +520,7 @@ impl WorkerPool { early_drop_tx, timing_tx_pair: (req_start_timing_tx, req_end_timing_tx), service_path, + code_identity, permit: permit.map(Arc::new), status: status.clone(), exit: surface.exit, @@ -541,16 +555,31 @@ impl WorkerPool { } pub fn add_user_worker(&mut self, key: Uuid, profile: UserWorkerProfile) { - let registry = self + let service_path = profile.service_path.clone(); + let code_identity = profile.code_identity; + + // A concurrent create for the same service path may have finished building + // a worker from a different artifact while this one was in flight. The + // most recently built worker reflects the newest request, so supersede the + // now-stale workers instead of pooling mismatched code under one key. + let supersedes_active = self .active_workers - .entry(profile.service_path.clone()) - .or_insert_with(|| { + .get(&service_path) + .and_then(|it| it.code_identity) + .is_some_and(|current| !current.can_serve(&code_identity)); + + if supersedes_active { + self.retire_active_workers(&service_path); + } + + let is_per_worker = self.policy.supervisor_policy.is_per_worker(); + let registry = + self.active_workers.entry(service_path).or_insert_with(|| { ActiveWorkerRegistry::new(self.policy.max_parallelism) }); - registry - .workers - .insert(WorkerId(key, self.policy.supervisor_policy.is_per_worker())); + registry.workers.insert(WorkerId(key, is_per_worker)); + registry.code_identity = Some(code_identity); self.user_workers.insert(key, profile); self.metric_src.incl_active_user_workers(); @@ -735,15 +764,50 @@ impl WorkerPool { } } + /// Retire every worker currently registered as active for `service_path`. + /// + /// Used when an incoming code artifact supersedes what the pool is holding: + /// the stale workers are removed from the active registry (and stop being + /// handed out for reuse) while any in-flight requests already dispatched to + /// them by uuid keep running until they finish. + fn retire_active_workers(&mut self, service_path: &str) { + let Some(registry) = self.active_workers.get(service_path) else { + return; + }; + + let stale = registry.workers.iter().map(|it| it.0).collect::>(); + for key in stale { + self.retire(&key); + } + } + fn maybe_active_worker( &mut self, service_path: &String, force_create: bool, + code_identity: WorkerCodeIdentity, ) -> Option { if force_create { return None; } + // Reject workers that were built from a different executable artifact. If + // we handed one back here it would serve code that was never deployed for + // this request (supabase/edge-runtime#721); retire it instead so the + // caller proceeds through normal creation. + if self + .active_workers + .get(service_path) + .and_then(|it| it.code_identity) + .is_some_and(|current| !current.can_serve(&code_identity)) + { + self.retire_active_workers(service_path); + if let Some(registry) = self.active_workers.get_mut(service_path) { + registry.code_identity = Some(code_identity); + } + return None; + } + let registry = self.active_workers.get_mut(service_path)?; let policy = self.policy.supervisor_policy; @@ -760,7 +824,7 @@ impl WorkerPool { _ => { self.retire(&worker_uuid); - self.maybe_active_worker(service_path, force_create) + self.maybe_active_worker(service_path, force_create, code_identity) } } } diff --git a/crates/base/test_cases/code-aware-worker-reuse/marker_a.ts b/crates/base/test_cases/code-aware-worker-reuse/marker_a.ts new file mode 100644 index 000000000..d7774c5c3 --- /dev/null +++ b/crates/base/test_cases/code-aware-worker-reuse/marker_a.ts @@ -0,0 +1,3 @@ +// Regression fixture for supabase/edge-runtime#721. +// Deployed as artifact "A" for a given service path. +Deno.serve(() => new Response("MARKER_A")); diff --git a/crates/base/test_cases/code-aware-worker-reuse/marker_b.ts b/crates/base/test_cases/code-aware-worker-reuse/marker_b.ts new file mode 100644 index 000000000..de0b05494 --- /dev/null +++ b/crates/base/test_cases/code-aware-worker-reuse/marker_b.ts @@ -0,0 +1,4 @@ +// Regression fixture for supabase/edge-runtime#721. +// Deployed as artifact "B" for the SAME service path as marker_a.ts; the pool +// must not serve this request from a warm worker that is still running "A". +Deno.serve(() => new Response("MARKER_B")); diff --git a/crates/base/tests/integration_tests.rs b/crates/base/tests/integration_tests.rs index 0cc559a6c..622402273 100644 --- a/crates/base/tests/integration_tests.rs +++ b/crates/base/tests/integration_tests.rs @@ -2982,6 +2982,108 @@ async fn test_should_be_able_to_bundle_against_various_exts() { test_serve_simple_fn("tsx", REACT_RESULT.as_bytes()).await; } +// Regression for supabase/edge-runtime#721: the worker pool reuses a warm +// worker purely by `servicePath`. When a second `EdgeRuntime.userWorkers.create` +// call for the same path carries a *different* code artifact (a redeploy, or a +// host funnelling multiple functions through one path), the pool used to hand +// back the worker still running the old bundle, silently serving code that was +// never deployed for that request. +// +// This drives the real `EdgeRuntime.userWorkers.create()` path through the +// `main_eszip` main worker, which derives `servicePath` from the request path +// and takes the eszip bundle from the request body. Every request below hits +// the same `servicePath`; only the bundle changes. +#[tokio::test] +#[serial] +async fn test_user_worker_reuse_is_code_aware() { + async fn eszip_bundle_for(entrypoint: &str) -> Vec { + let mut emitter_factory = EmitterFactory::new(); + + emitter_factory.set_permissions_options(Some(get_default_permissions( + WorkerKind::UserWorker, + ))); + emitter_factory.set_deno_options( + DenoOptionsBuilder::new() + .entrypoint(PathBuf::from(entrypoint)) + .build() + .unwrap(), + ); + + let mut metadata = Metadata::default(); + let eszip = generate_binary_eszip( + &mut metadata, + Arc::new(emitter_factory), + None, + None, + None, + ) + .await + .unwrap(); + + eszip.into_bytes() + } + + let eszip_a = + eszip_bundle_for("./test_cases/code-aware-worker-reuse/marker_a.ts").await; + let eszip_b = + eszip_bundle_for("./test_cases/code-aware-worker-reuse/marker_b.ts").await; + + async fn serve(tb: &TestBed, bundle: Vec) -> String { + let mut resp = tb + .request(move |b| { + b.uri("/code-aware-worker-reuse") + .method("POST") + .body(Body::from(bundle)) + .context("can't make request") + }) + .await + .unwrap(); + + assert_eq!(resp.status().as_u16(), 200); + + let body = to_bytes(resp.body_mut()).await.unwrap(); + String::from_utf8_lossy(&body).to_string() + } + + let (boot_tx, mut boot_rx) = mpsc::unbounded_channel(); + let tb = TestBedBuilder::new("./test_cases/main_eszip") + .with_per_worker_policy(None) + .with_worker_event_sender(Some(boot_tx)) + .build() + .await; + + // 1. Deploy artifact A, confirm the worker serves "A". + assert_eq!(serve(&tb, eszip_a.clone()).await, "MARKER_A"); + + // 2. Same servicePath, same artifact: reuse is still allowed. + assert_eq!(serve(&tb, eszip_a.clone()).await, "MARKER_A"); + + // 3. Same servicePath, DIFFERENT artifact: the warm "A" worker must not be + // reused; the new bundle has to execute. + assert_eq!(serve(&tb, eszip_b.clone()).await, "MARKER_B"); + + // 4. Back to artifact A: its worker was superseded in step 3, so this runs a + // fresh worker built from A again. + assert_eq!(serve(&tb, eszip_a.clone()).await, "MARKER_A"); + + tb.exit(Duration::from_secs(TESTBED_DEADLINE_SEC)).await; + + // Dropping the test bed closes every worker event sender, so this drains. + // One boot for step 1 (A), none for step 2 (reused), one for step 3 (B), + // one for step 4 (A rebuilt) => three fresh workers total. + let mut boots = 0; + while let Some(ev) = boot_rx.recv().await { + if matches!(ev.event, WorkerEvents::Boot(_)) { + boots += 1; + } + } + + assert_eq!( + boots, 3, + "expected fresh boots for A, then B, then A again (step 2 must reuse)" + ); +} + #[tokio::test] #[serial] async fn test_private_npm_package_import() { diff --git a/ext/workers/Cargo.toml b/ext/workers/Cargo.toml index 25b8ae680..2fb5e0dc3 100644 --- a/ext/workers/Cargo.toml +++ b/ext/workers/Cargo.toml @@ -31,6 +31,7 @@ log.workspace = true once_cell.workspace = true scopeguard.workspace = true serde.workspace = true +sha2.workspace = true thiserror.workspace = true tokio.workspace = true tokio-util.workspace = true diff --git a/ext/workers/context.rs b/ext/workers/context.rs index 20d2b5c95..df448a7b9 100644 --- a/ext/workers/context.rs +++ b/ext/workers/context.rs @@ -25,6 +25,8 @@ use fs::tmp_fs::TmpFsConfig; use hyper_v014::Body; use hyper_v014::Request; use hyper_v014::Response; +use sha2::Digest as _; +use sha2::Sha256; use tokio::sync::mpsc; use tokio::sync::mpsc::unbounded_channel; use tokio::sync::oneshot; @@ -138,6 +140,45 @@ impl Default for UserWorkerRuntimeOpts { } } +/// Identity of the executable artifact a user worker was created from. +/// +/// The pool keys warm-worker reuse by `service_path`, but the same +/// `service_path` can be handed completely different executable code across +/// `EdgeRuntime.userWorkers.create()` calls — a redeployed bundle, or a host +/// that funnels several functions through one path. Reusing an already-active +/// worker in that situation silently runs code that was never deployed for the +/// incoming request (supabase/edge-runtime#721). The pool therefore also +/// compares this identity and only reuses a worker whose artifact matches. +/// +/// The digest is deterministic (content-derived, no pointer or process-random +/// input) so the same bytes always map to the same identity, and it is a full +/// SHA-256 (`[u8; 32]`, never truncated) so two different executable artifacts +/// cannot compare equal through a hash collision — worker identity is a +/// correctness boundary, not a cache hint. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum WorkerCodeIdentity { + /// A SHA-256 digest over the executable artifact (inline eszip bundle or + /// inline module code) together with the service path, any explicit + /// entrypoint override, and the effective import map path. + Digest([u8; 32]), + /// The artifact could not be reduced to a stable digest (e.g. a pre-parsed + /// eszip handed straight to the pool). Such a request is never considered + /// compatible with an existing worker, so it always gets a fresh one. + Opaque, +} + +impl WorkerCodeIdentity { + /// Whether a worker created with `self` may serve a request carrying + /// `incoming`. Reuse is only sound when both sides resolve to the same + /// deterministic digest. + pub fn can_serve(&self, incoming: &WorkerCodeIdentity) -> bool { + matches!( + (self, incoming), + (Self::Digest(a), Self::Digest(b)) if a == b + ) + } +} + #[derive(Debug, Clone)] pub struct UserWorkerProfile { pub worker_request_msg_tx: mpsc::UnboundedSender, @@ -147,6 +188,7 @@ pub struct UserWorkerProfile { mpsc::UnboundedSender<()>, ), pub service_path: String, + pub code_identity: WorkerCodeIdentity, pub permit: Option>, pub cancel: CancellationToken, pub status: TimingStatus, @@ -273,6 +315,252 @@ pub struct WorkerContextInitOpts { pub maybe_otel_config: Option, } +impl WorkerContextInitOpts { + /// Derive the [`WorkerCodeIdentity`] for this creation request. + /// + /// This folds in every input that determines *which code* the worker will + /// run: the inline eszip bundle bytes, the inline module code, the service + /// path, any explicit entrypoint override, and the effective import map path + /// (which steers module resolution both when the runtime builds the eszip and + /// when it loads a pre-built one). It intentionally does not depend on runtime + /// knobs (memory/CPU limits, env vars, timing) — those may legitimately differ + /// between two requests that should still share a warm worker. + /// + /// Filesystem source read from `service_path` is *not* hashed: a plain + /// file-backed worker keeps the pre-existing reuse semantics (a caller that + /// wants an on-disk edit picked up must pass `force_create`, exactly as + /// before this change). What the digest adds for that case is that a changed + /// entrypoint or import map path now correctly forces a fresh worker. + /// + /// The digest is SHA-256 and is never truncated, so two different artifacts + /// cannot be treated as compatible through a hash collision. + pub fn code_identity(&self) -> WorkerCodeIdentity { + let mut hasher = Sha256::new(); + + // The service path is the pool's reuse key already, but include it so a + // path-backed worker can never collide with an inline-artifact worker that + // happens to resolve to the same key. + hasher.update(self.service_path.to_string_lossy().as_bytes()); + + if let Some(entrypoint) = self.maybe_entrypoint.as_deref() { + hasher.update(b"\0entrypoint\0"); + hasher.update(entrypoint.as_bytes()); + } + + // The import map path is read out of the worker creation context + // (`context.importMapPath`) in `crates/base/src/runtime`. It changes how + // bare specifiers resolve, so the same source can produce a different + // executable under a different import map — it must be part of identity. + if let Some(import_map_path) = self + .conf + .context() + .and_then(|it| it.get("importMapPath")) + .and_then(|it| it.as_str()) + { + hasher.update(b"\0import_map\0"); + hasher.update(import_map_path.as_bytes()); + } + + match self.maybe_eszip.as_ref() { + Some(EszipPayloadKind::JsBufferKind(buf)) => { + hasher.update(b"\0eszip\0"); + hasher.update(&buf[..]); + } + Some(EszipPayloadKind::VecKind(buf)) => { + hasher.update(b"\0eszip\0"); + hasher.update(&buf[..]); + } + // A pre-parsed eszip does not expose its original bytes cheaply. This + // shape is not produced for pooled user workers, but stay conservative + // rather than risk treating two different bundles as equal. + Some(EszipPayloadKind::Eszip(_)) => return WorkerCodeIdentity::Opaque, + None => {} + } + + if let Some(code) = self.maybe_module_code.as_ref() { + hasher.update(b"\0module\0"); + hasher.update(code.as_str().as_bytes()); + } + + WorkerCodeIdentity::Digest(hasher.finalize().into()) + } +} + +#[cfg(test)] +mod code_identity_tests { + use super::*; + + #[derive(Default)] + struct Artifact<'a> { + eszip: Option>, + module_code: Option<&'a str>, + entrypoint: Option<&'a str>, + import_map_path: Option<&'a str>, + } + + fn opts(service_path: &str, artifact: Artifact<'_>) -> WorkerContextInitOpts { + let Artifact { + eszip, + module_code, + entrypoint, + import_map_path, + } = artifact; + + let context = import_map_path.map(|path| { + let mut map = crate::JsonMap::new(); + map.insert("importMapPath".to_string(), path.into()); + map + }); + + WorkerContextInitOpts { + service_path: std::path::PathBuf::from(service_path), + no_module_cache: false, + no_npm: None, + env_vars: HashMap::new(), + conf: WorkerRuntimeOpts::UserWorker(UserWorkerRuntimeOpts { + context, + ..Default::default() + }), + static_patterns: vec![], + timing: None, + maybe_eszip: eszip.map(EszipPayloadKind::VecKind), + maybe_module_code: module_code.map(|it| it.to_string().into()), + maybe_entrypoint: entrypoint.map(str::to_string), + maybe_s3_fs_config: None, + maybe_tmp_fs_config: None, + maybe_otel_config: None, + } + } + + fn eszip(service_path: &str, bytes: &[u8]) -> WorkerContextInitOpts { + opts( + service_path, + Artifact { + eszip: Some(bytes.to_vec()), + ..Default::default() + }, + ) + } + + #[test] + fn identity_is_deterministic_and_content_addressed() { + // Inline eszip: identical bytes may reuse, different bytes may not. + let a = eszip("svc", b"bundle-A"); + let a_again = eszip("svc", b"bundle-A"); + let b = eszip("svc", b"bundle-B"); + assert!(a.code_identity().can_serve(&a_again.code_identity())); + assert!(!a.code_identity().can_serve(&b.code_identity())); + + // Inline module code behaves the same way. + let m = opts( + "svc", + Artifact { + module_code: Some("export default 1"), + ..Default::default() + }, + ); + let m_again = opts( + "svc", + Artifact { + module_code: Some("export default 1"), + ..Default::default() + }, + ); + let m_changed = opts( + "svc", + Artifact { + module_code: Some("export default 2"), + ..Default::default() + }, + ); + assert!(m.code_identity().can_serve(&m_again.code_identity())); + assert!(!m.code_identity().can_serve(&m_changed.code_identity())); + + // Different artifact kinds for one service path never look equivalent. + assert!(!a.code_identity().can_serve(&m.code_identity())); + + // An explicit entrypoint override is part of the executable identity. + let e1 = opts( + "svc", + Artifact { + entrypoint: Some("a.ts"), + ..Default::default() + }, + ); + let e2 = opts( + "svc", + Artifact { + entrypoint: Some("b.ts"), + ..Default::default() + }, + ); + assert!(!e1.code_identity().can_serve(&e2.code_identity())); + + // A pre-parsed eszip cannot be digested, so it is never reusable. + assert!(!WorkerCodeIdentity::Opaque.can_serve(&WorkerCodeIdentity::Opaque)); + } + + #[test] + fn import_map_path_is_part_of_identity() { + // `context.importMapPath` steers module resolution, so the same source + // under a different import map is a different executable and must not + // silently reuse a warm worker. + let base = eszip("svc", b"bundle-A"); + + let map_a = opts( + "svc", + Artifact { + eszip: Some(b"bundle-A".to_vec()), + import_map_path: Some("/etc/import_map_a.json"), + ..Default::default() + }, + ); + let map_a_again = opts( + "svc", + Artifact { + eszip: Some(b"bundle-A".to_vec()), + import_map_path: Some("/etc/import_map_a.json"), + ..Default::default() + }, + ); + let map_b = opts( + "svc", + Artifact { + eszip: Some(b"bundle-A".to_vec()), + import_map_path: Some("/etc/import_map_b.json"), + ..Default::default() + }, + ); + + assert!(map_a + .code_identity() + .can_serve(&map_a_again.code_identity())); + assert!(!map_a.code_identity().can_serve(&map_b.code_identity())); + // Adding an import map to an otherwise identical request also changes it. + assert!(!base.code_identity().can_serve(&map_a.code_identity())); + } + + #[test] + fn digest_uses_full_sha256() { + // The strong identity is the full 32-byte SHA-256, not a truncated hash. + let WorkerCodeIdentity::Digest(bytes) = + eszip("svc", b"bundle-A").code_identity() + else { + panic!("inline eszip must produce a digest"); + }; + assert_eq!(bytes.len(), 32); + + // Known-answer: SHA-256 of the exact byte stream the hasher folds in for + // this request (service path, then the framed eszip bytes). + let mut expected = Sha256::new(); + expected.update(b"svc"); + expected.update(b"\0eszip\0"); + expected.update(b"bundle-A"); + let expected: [u8; 32] = expected.finalize().into(); + assert_eq!(bytes, expected); + } +} + #[derive(Debug)] #[allow(clippy::large_enum_variant)] // This is a low-frequency control channel; avoid API churn. pub enum UserWorkerMsgs {