feat: cooperative interleaving, keyed routing, and warm isolate fixes

Worker pool:
- Replace flume with tokio::sync::mpsc for proper LocalSet integration
- Persistent LocalSet + spawn_local for cooperative task interleaving
- try_recv() drain to batch-dispatch bursts
- Add consistent-hash routing (spawn_keyed, spawn_await_keyed)
- Add interleaving and spawn_local concurrency tests with pitfall docs

Task executor:
- Use keyed routing (hash of worker_id) for warm cache affinity
- Fix event loop crash on warm reuse (spawn_local → tokio::spawn)

Ops:
- Interior mutability with Mutex<RequestState> for warm context reuse
- update_request() to swap per-request state (log_tx, span)
- Implement as_any() for downcast in warm hit callback

Author: max-lt

Cargo.lock

@@ -35,7 +35,6 @@ bytes = "1.11.0"
 log = "0.4.29"
 tokio = "1.49.0"
 tokio-util = { version = "0.7", features = ["rt"] }
-flume = "0.12.0"
 once_cell = "1.19"
 # filetime = "0.2"  # was used by disk-based snapshot cache
 env_logger = "0.11.6"
@@ -56,14 +55,14 @@ sqlx = { version = "0.8.6", features = [ "runtime-tokio", "postgres", "uuid", "b
 reqwest = { version = "0.13.1", default-features = false, features = ["rustls", "json", "stream"] }
 
 # Core types
-openworkers-core = { git = "https://github.com/openworkers/openworkers-core", tag = "v0.13.0", features = ["hyper"] }
+openworkers-core = { path = "../openworkers-core", features = ["hyper"] }
 
 # Transform (TS strip + export default → globalThis.default)
 # openworkers-transform = { path = "../openworkers-transform" }
 openworkers-transform = { git = "https://github.com/openworkers/openworkers-transform", tag = "v0.1.0" }
 
 # Runtime backend (v8 only for now, others require older version of core)
-openworkers-runtime-v8 = { git = "https://github.com/openworkers/openworkers-runtime-v8", tag = "v0.13.3", optional = true, features = ["ptrcomp"] }
+openworkers-runtime-v8 = { path = "../openworkers-runtime-v8", optional = true, features = ["ptrcomp"] }
 
 # WASM runtime (optional)
 # openworkers-runtime-wasm = { path = "../openworkers-runtime-wasm", optional = true }

Cargo.toml

@@ -35,6 +35,7 @@ use openworkers_core::{
 };
 use std::collections::HashMap;
 use std::sync::Arc;
+use std::sync::Mutex;
 use std::sync::atomic::{AtomicU64, Ordering};
 
 use tracing::Instrument;
@@ -115,26 +116,36 @@ impl BindingConfigs {
 /// Database pool type alias
 pub type DbPool = sqlx::Pool<sqlx::Postgres>;
 
+/// Per-request state that changes between requests for context reuse.
+///
+/// When warm isolates are enabled, the same `RunnerOperations` handle persists
+/// across requests. Only `log_tx` and `span` change per-request.
+pub struct RequestState {
+    pub log_tx: Option<LogTx>,
+    pub span: tracing::Span,
+}
+
 /// Runner's implementation of OperationsHandler
 ///
 /// Implements fetch via reqwest, bindings via config lookup, and logs via channel.
+///
+/// For warm isolates: `request_state` is swappable via interior mutability so that
+/// the same ops handle (captured in the event loop) can serve multiple requests.
 pub struct RunnerOperations {
     /// Stats for this worker
     pub stats: Arc<OperationsStats>,
     /// User ID for logging/quotas
     pub user_id: Option<String>,
     /// Worker ID for logging/quotas
     pub worker_id: Option<String>,
-    /// Log sender (optional - if None, uses default stderr)
-    pub log_tx: Option<LogTx>,
+    /// Per-request state (log_tx + span), swappable between requests
+    request_state: Mutex<RequestState>,
     /// Binding configs for this worker
     pub bindings: BindingConfigs,
     /// Database pool for KV operations
     pub db_pool: Option<DbPool>,
     /// Binding limiters (rate limiting for fetch, KV, database, storage)
     pub limiters: BindingLimiters,
-    /// Tracing span for async operations (propagates trace context)
-    pub span: tracing::Span,
 }
 
 impl RunnerOperations {
@@ -143,11 +154,13 @@ impl RunnerOperations {
             stats: Arc::new(OperationsStats::new()),
             user_id: None,
             worker_id: None,
-            log_tx: None,
+            request_state: Mutex::new(RequestState {
+                log_tx: None,
+                span: tracing::Span::none(),
+            }),
             bindings: BindingConfigs::new(),
             db_pool: None,
             limiters: BindingLimiters::default(),
-            span: tracing::Span::none(),
         }
     }
 
@@ -158,8 +171,8 @@ impl RunnerOperations {
     }
 
     /// Attach tracing span for context propagation
-    pub fn with_span(mut self, span: tracing::Span) -> Self {
-        self.span = span;
+    pub fn with_span(self, span: tracing::Span) -> Self {
+        self.request_state.lock().unwrap().span = span;
         self
     }
 
@@ -173,8 +186,8 @@ impl RunnerOperations {
         self
     }
 
-    pub fn with_log_tx(mut self, log_tx: LogTx) -> Self {
-        self.log_tx = Some(log_tx);
+    pub fn with_log_tx(self, log_tx: LogTx) -> Self {
+        self.request_state.lock().unwrap().log_tx = Some(log_tx);
         self
     }
 
@@ -188,6 +201,20 @@ impl RunnerOperations {
         self
     }
 
+    /// Swap per-request state (log handler + tracing span) for warm context reuse.
+    ///
+    /// Called when an existing ops handle is reused for a new request.
+    pub fn update_request(&self, log_tx: LogTx, span: tracing::Span) {
+        let mut state = self.request_state.lock().unwrap();
+        state.log_tx = Some(log_tx);
+        state.span = span;
+    }
+
+    /// Get a clone of the current span
+    fn span(&self) -> tracing::Span {
+        self.request_state.lock().unwrap().span.clone()
+    }
+
     /// Execute a binding fetch with the given config (shared by assets and storage).
     fn do_binding_fetch(
         &self,
@@ -196,7 +223,7 @@ impl RunnerOperations {
         request: HttpRequest,
     ) -> OpFuture<'_, Result<HttpResponse, String>> {
         let binding_name = binding.to_string();
-        let span = self.span.clone();
+        let span = self.span();
 
         Box::pin(
             async move {
@@ -261,6 +288,10 @@ impl Default for RunnerOperations {
 }
 
 impl OperationsHandler for RunnerOperations {
+    fn as_any(&self) -> &dyn std::any::Any {
+        self
+    }
+
     /// Handle direct fetch: `fetch("https://example.com")`
     ///
     /// This is a pass-through fetch with no auth modification.
@@ -269,7 +300,7 @@ impl OperationsHandler for RunnerOperations {
     /// Special case: URLs matching `*.workers.rocks` or `*.workers.dev.localhost`
     /// are routed internally to avoid DNS lookup and external network hop.
     fn handle_fetch(&self, request: HttpRequest) -> OpFuture<'_, Result<HttpResponse, String>> {
-        let span = self.span.clone();
+        let span = self.span();
         Box::pin(
             async move {
                 // Acquire fetch limiter permit (blocks if at concurrent limit, errors if total exceeded)
@@ -366,7 +397,7 @@ impl OperationsHandler for RunnerOperations {
             }
         };
 
-        let span = self.span.clone();
+        let span = self.span();
         Box::pin(
             async move {
                 // Acquire storage limiter permit
@@ -436,7 +467,7 @@ impl OperationsHandler for RunnerOperations {
         };
 
         let namespace_id = config.id.clone();
-        let span = self.span.clone();
+        let span = self.span();
 
         Box::pin(
             async move {
@@ -489,7 +520,7 @@ impl OperationsHandler for RunnerOperations {
 
         // Get shared pool for schema mode
         let shared_pool = self.db_pool.clone();
-        let span = self.span.clone();
+        let span = self.span();
 
         Box::pin(
             async move {
@@ -590,7 +621,7 @@ impl OperationsHandler for RunnerOperations {
             }
         };
 
-        let span = self.span.clone();
+        let span = self.span();
 
         Box::pin(
             async move {
@@ -641,11 +672,15 @@ impl OperationsHandler for RunnerOperations {
         self.stats.log_count.fetch_add(1, Ordering::Relaxed);
 
         // Send to log channel if available (for log collection/storage)
-        if let Some(ref tx) = self.log_tx {
-            let _ = tx.send(LogEvent {
-                level,
-                message: message.clone(),
-            });
+        {
+            let state = self.request_state.lock().unwrap();
+
+            if let Some(ref tx) = state.log_tx {
+                let _ = tx.send(LogEvent {
+                    level,
+                    message: message.clone(),
+                });
+            }
         }
 
         // Also log via the log crate for debugging
@@ -670,6 +705,14 @@ mod tests {
         assert!(ops.bindings.kv.is_empty());
     }
 
+    #[test]
+    fn test_runner_operations_update_request() {
+        let ops = RunnerOperations::new();
+        let (tx, _rx) = std::sync::mpsc::channel();
+        ops.update_request(tx, tracing::Span::none());
+        assert!(ops.request_state.lock().unwrap().log_tx.is_some());
+    }
+
     #[test]
     fn test_runner_operations_with_user() {
         let ops = RunnerOperations::new().with_user_id("user-123".to_string());

src/ops.rs

@@ -4,7 +4,7 @@ use tokio::sync::OwnedSemaphorePermit;
 use tracing::Instrument;
 
 use crate::log::WorkerLogHandler;
-use crate::ops::{DbPool, RunnerOperations};
+use crate::ops::{DbPool, LogTx, RunnerOperations};
 use crate::store::{CodeType, WorkerWithBindings};
 use crate::worker::{Worker, create_worker, prepare_script};
 use crate::worker_pool::{TaskPermit, WORKER_POOL};
@@ -81,6 +81,10 @@ struct TaskComponents {
     ops: Arc<RunnerOperations>,
     code_type: CodeType,
     log_handler: WorkerLogHandler,
+    /// Raw log sender for warm hit callback (to update cached ops)
+    log_tx: LogTx,
+    /// Tracing span for warm hit callback
+    span: tracing::Span,
 }
 
 /// Prepare task components: parse script, setup logging, and create operations handle.
@@ -101,7 +105,7 @@ fn prepare_task_components(config: &TaskExecutionConfig) -> Option<TaskComponent
     let ops = Arc::new(
         RunnerOperations::new()
             .with_worker_id(config.worker_data.id.clone())
-            .with_log_tx(log_tx)
+            .with_log_tx(log_tx.clone())
             .with_bindings(config.worker_data.bindings.clone())
             .with_db_pool(config.db_pool.clone())
             .with_span(config.span.clone()),
@@ -112,6 +116,8 @@ fn prepare_task_components(config: &TaskExecutionConfig) -> Option<TaskComponent
         ops,
         code_type: config.worker_data.code_type.clone(),
         log_handler,
+        log_tx,
+        span: config.span.clone(),
     })
 }
 
@@ -184,20 +190,45 @@ pub async fn execute_task_await_v8_pooled(
     let execute_mode = V8ExecuteMode::get();
     let span = config.span.clone();
 
+    // Route to a consistent thread based on worker_id for warm cache affinity
+    let routing_key = worker_id_for_snapshot.clone();
+
     WORKER_POOL
-        .spawn_await(move || {
+        .spawn_await_keyed(&routing_key, move || {
             async move {
                 // Wrap permit to automatically notify drain monitor on drop
                 let _permit = TaskPermit::new(permit);
 
                 let result = match execute_mode {
                     V8ExecuteMode::Pinned => {
                         // Thread-pinned pool (default, best performance)
+                        // Pass worker_id + version for warm context caching.
+                        // The warm hit callback updates the cached ops' per-request state
+                        // so the existing event loop sends logs to the new handler.
+                        let warm_log_tx = components.log_tx.clone();
+                        let warm_span = components.span.clone();
+                        let on_warm_hit: openworkers_runtime_v8::WarmHitCallback =
+                            Box::new(move |cached_ops| {
+                                // Downcast to RunnerOperations to call update_request.
+                                // This updates the cached ops' per-request state (log_tx, span)
+                                // so the existing event loop sends logs to the new handler.
+                                if let Some(runner_ops) =
+                                    cached_ops.as_any().downcast_ref::<RunnerOperations>()
+                                {
+                                    runner_ops.update_request(warm_log_tx, warm_span);
+                                }
+                            });
+
                         openworkers_runtime_v8::execute_pinned(
-                            &owner_id,
-                            components.script,
-                            components.ops,
-                            task,
+                            openworkers_runtime_v8::PinnedExecuteRequest {
+                                owner_id,
+                                worker_id: worker_id_for_snapshot.clone(),
+                                version: version_for_snapshot,
+                                script: components.script,
+                                ops: components.ops,
+                                task,
+                                on_warm_hit: Some(on_warm_hit),
+                            },
                         )
                         .await
                     }