fix(base): make user worker reuse code-aware

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<usize>,
   notify_pair: (flume::Sender<Option<Uuid>>, flume::Receiver<Option<Uuid>>),
   sem: Arc<Semaphore>,
+
+  // 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<WorkerCodeIdentity>,
 }
 
 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::<Vec<_>>();
+    for key in stale {
+      self.retire(&key);
+    }
+  }
+
   fn maybe_active_worker(
     &mut self,
     service_path: &String,
     force_create: bool,
+    code_identity: WorkerCodeIdentity,
   ) -> Option<Uuid> {
     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)
       }
     }
   }

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"));

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"));

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<u8> {
+    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<u8>) -> 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() {

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