Support wasm runtime

Author: max-lt

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "openworkers-runner"
-version = "0.8.0"
+version = "0.8.1"
 edition = "2024"
 license = "MIT"
 default-run = "openworkers-runner"
@@ -10,6 +10,7 @@ default-run = "openworkers-runner"
 [features]
 default = ["database"]
 v8 = ["dep:openworkers-runtime-v8"]
+wasm = ["dep:openworkers-runtime-wasm", "openworkers-core/wasm"]
 database = ["dep:postgate"]
 # Disabled runtimes (require core 0.5, we're on 0.7) - kept as empty features to avoid cfg warnings
 deno = []
@@ -34,11 +35,15 @@ reqwest = { version = "0.13.1", default-features = false, features = ["rustls",
 
 # Core types
 # openworkers-core = { path = "../openworkers-core", features = ["hyper"] }
-openworkers-core = { git = "https://github.com/openworkers/openworkers-core", tag = "v0.8.0", features = ["hyper"] }
+openworkers-core = { git = "https://github.com/openworkers/openworkers-core", tag = "v0.8.1", features = ["hyper"] }
 
 # Runtime backend (v8 only for now, others require core 0.5)
 # openworkers-runtime-v8 = { path = "../openworkers-runtime-v8", optional = true }
-openworkers-runtime-v8 = { git = "https://github.com/openworkers/openworkers-runtime-v8", tag = "v0.8.0", optional = true }
+openworkers-runtime-v8 = { git = "https://github.com/openworkers/openworkers-runtime-v8", tag = "v0.8.1", optional = true }
+
+# WASM runtime (optional)
+# openworkers-runtime-wasm = { path = "../openworkers-runtime-wasm", optional = true }
+openworkers-runtime-wasm = {  git = "https://github.com/openworkers/openworkers-runtime-wasm", tag = "v0.1.0", optional = true }
 
 # Database bindings (optional)
 # postgate = { path = "../postgate", default-features = false, optional = true }
@@ -87,6 +92,7 @@ lto = true # Enable link-time optimization.
 
 [dev-dependencies]
 criterion = { version = "0.8.1", features = ["async_tokio"] }
+wat = "1.0"
 
 [[bin]]
 name = "openworkers-runner"

Cargo.toml

@@ -10,7 +10,7 @@ use std::time::Duration;
 use tokio::net::TcpSocket;
 use tokio::sync::oneshot::channel;
 
-use openworkers_core::{HttpRequest, HyperBody};
+use openworkers_core::{HttpRequest, HttpResponse, HyperBody};
 use openworkers_runner::store::WorkerIdentifier;
 
 use sqlx::postgres::PgPoolOptions;
@@ -181,7 +181,7 @@ async fn handle_request(
         }
     };
 
-    let (res_tx, res_rx) = channel::<openworkers_runner::runtime::HttpResponse>();
+    let (res_tx, res_rx) = channel::<HttpResponse>();
     let (termination_tx, termination_rx) =
         channel::<Result<(), openworkers_core::TerminationReason>>();
 

bin/main.rs

@@ -3,13 +3,13 @@ use std::time::Duration;
 use tokio::sync::OwnedSemaphorePermit;
 
 use crate::ops::{DbPool, RunnerOperations};
-use crate::runtime::Worker;
-use crate::store::{WorkerWithBindings, bindings_to_infos};
+use crate::store::WorkerWithBindings;
+use crate::worker::{create_worker, prepare_script};
 use crate::worker_pool::WORKER_POOL;
 
 use openworkers_core::{
-    FetchInit, HttpRequest, HttpResponse, ResponseBody, ResponseSender, RuntimeLimits, Script,
-    Task, TerminationReason, WorkerCode,
+    FetchInit, HttpRequest, HttpResponse, ResponseBody, ResponseSender, RuntimeLimits, Task,
+    TerminationReason,
 };
 
 type TerminationTx = tokio::sync::oneshot::Sender<Result<(), TerminationReason>>;
@@ -18,53 +18,36 @@ type TerminationTx = tokio::sync::oneshot::Sender<Result<(), TerminationReason>>
 const FETCH_TIMEOUT_MS: u64 = 64_000; // 64 seconds
 
 pub fn run_fetch(
-    worker: WorkerWithBindings,
+    worker_data: WorkerWithBindings,
     req: HttpRequest,
     res_tx: ResponseSender,
     termination_tx: TerminationTx,
     global_log_tx: std::sync::mpsc::Sender<crate::log::LogMessage>,
     permit: OwnedSemaphorePermit,
     db_pool: DbPool,
 ) {
-    let worker_id = worker.id.clone();
-    let (log_tx, log_handler) = crate::log::create_log_handler(worker_id.clone(), global_log_tx);
-
-    let code = match crate::transform::parse_worker_code_str(&worker.script, &worker.language) {
-        Ok(code) => WorkerCode::js(code),
-        Err(e) => {
-            log::error!("Failed to parse worker code: {}", e);
+    let worker_id = worker_data.id.clone();
+    let bindings = worker_data.bindings.clone();
+    let code_type = worker_data.code_type.clone();
+
+    // Parse script before spawning (fail fast)
+    let script = match prepare_script(&worker_data) {
+        Ok(s) => s,
+        Err(err) => {
+            log::error!("Failed to prepare script: {err:?}");
             res_tx
                 .send(HttpResponse {
                     status: 500,
                     headers: vec![],
-                    body: ResponseBody::Bytes(format!("Failed to parse worker code: {}", e).into()),
+                    body: ResponseBody::Bytes(format!("Failed to prepare script: {err:?}").into()),
                 })
                 .ok();
-            termination_tx
-                .send(Err(TerminationReason::InitializationError(format!(
-                    "Failed to parse worker code: {}",
-                    e
-                ))))
-                .ok();
+            termination_tx.send(Err(err)).ok();
             return;
         }
     };
 
-    // Convert bindings to BindingInfo for Script (names + types only)
-    let binding_infos = bindings_to_infos(&worker.bindings);
-
-    // Clone bindings for RunnerOperations (full configs)
-    let bindings_for_ops = worker.bindings.clone();
-
-    let script = Script {
-        code,
-        env: if worker.env.is_empty() {
-            None
-        } else {
-            Some(worker.env.clone())
-        },
-        bindings: binding_infos,
-    };
+    let (log_tx, log_handler) = crate::log::create_log_handler(worker_id.clone(), global_log_tx);
 
     // Use the sequential worker pool - ensures ONE V8 isolate per thread at a time
     WORKER_POOL.spawn(move || async move {
@@ -85,28 +68,31 @@ pub fn run_fetch(
             RunnerOperations::new()
                 .with_worker_id(worker_id)
                 .with_log_tx(log_tx)
-                .with_bindings(bindings_for_ops)
+                .with_bindings(bindings)
                 .with_db_pool(db_pool),
         );
 
-        let mut worker = match Worker::new_with_ops(script, Some(limits), ops).await {
-            Ok(worker) => worker,
+        // Create worker
+        let mut worker = match create_worker(script, limits, ops, &code_type).await {
+            Ok(w) => w,
             Err(err) => {
-                log::error!("failed to create worker: {err}");
+                log::error!("failed to create worker: {err:?}");
                 res_tx
                     .send(HttpResponse {
                         status: 500,
                         headers: vec![],
-                        body: ResponseBody::Bytes(format!("failed to create worker: {err}").into()),
+                        body: ResponseBody::Bytes(
+                            format!("failed to create worker: {err:?}").into(),
+                        ),
                     })
-                    .unwrap();
-
+                    .ok();
+                termination_tx.send(Err(err)).ok();
+                log_handler.flush();
                 return;
             }
         };
 
         let task = Task::Fetch(Some(FetchInit::new(req, res_tx)));
-
         log::debug!("exec fetch task with {}ms timeout", FETCH_TIMEOUT_MS);
 
         // Wrap execution with timeout

src/event_fetch.rs

@@ -1,11 +1,11 @@
 use std::sync::Arc;
 
 use crate::ops::{DbPool, RunnerOperations};
-use crate::runtime::Worker;
-use crate::store::{self, Binding, bindings_to_infos};
+use crate::store::{self, WorkerWithBindings};
+use crate::worker::{create_worker, prepare_script};
 use crate::worker_pool::WORKER_POOL;
 
-use openworkers_core::{RuntimeLimits, ScheduledInit, Script, Task, WorkerCode};
+use openworkers_core::{RuntimeLimits, ScheduledInit, Task, TerminationReason};
 
 use serde::Deserialize;
 use serde::Serialize;
@@ -21,11 +21,19 @@ pub struct ScheduledData {
 
 fn run_scheduled(
     data: ScheduledData,
-    script: Script,
-    bindings: Vec<Binding>,
+    worker_data: WorkerWithBindings,
     db_pool: DbPool,
     global_log_tx: std::sync::mpsc::Sender<crate::log::LogMessage>,
 ) {
+    // Parse script before spawning (fail fast)
+    let script = match prepare_script(&worker_data) {
+        Ok(s) => s,
+        Err(err) => {
+            log::error!("Failed to prepare script for scheduled task: {err:?}");
+            return;
+        }
+    };
+
     // Try to acquire a worker slot
     let permit = match crate::worker_pool::WORKER_SEMAPHORE
         .clone()
@@ -41,7 +49,9 @@ fn run_scheduled(
         }
     };
 
-    let worker_id = data.worker_id.clone();
+    let worker_id = worker_data.id.clone();
+    let bindings = worker_data.bindings.clone();
+    let code_type = worker_data.code_type.clone();
     let (log_tx, log_handler) = crate::log::create_log_handler(worker_id.clone(), global_log_tx);
 
     // Use the sequential worker pool - ensures ONE V8 isolate per thread at a time
@@ -65,10 +75,11 @@ fn run_scheduled(
                 .with_db_pool(db_pool),
         );
 
-        let mut worker = match Worker::new_with_ops(script, Some(limits), ops).await {
-            Ok(worker) => worker,
+        // Create worker
+        let mut worker = match create_worker(script, limits, ops, &code_type).await {
+            Ok(w) => w,
             Err(err) => {
-                log::error!("failed to create scheduled worker: {err}");
+                log::error!("failed to create scheduled worker: {err:?}");
                 log_handler.flush();
                 return;
             }
@@ -79,6 +90,7 @@ fn run_scheduled(
         let task = Task::Scheduled(Some(ScheduledInit::new(res_tx, data.scheduled_time)));
 
         log::debug!("exec scheduled task");
+
         match worker.exec(task).await {
             Ok(()) => {
                 log::debug!("scheduled task completed successfully");
@@ -88,27 +100,23 @@ fn run_scheduled(
                     Err(err) => log::error!("scheduled task response error: {err}"),
                 }
             }
-            Err(reason) => {
-                use openworkers_core::TerminationReason;
-
-                match reason {
-                    TerminationReason::CpuTimeLimit => {
-                        log::warn!("scheduled task terminated: CPU time limit exceeded");
-                    }
-                    TerminationReason::WallClockTimeout => {
-                        log::warn!("scheduled task terminated: wall-clock timeout");
-                    }
-                    TerminationReason::MemoryLimit => {
-                        log::warn!("scheduled task terminated: memory limit exceeded");
-                    }
-                    TerminationReason::Exception(msg) => {
-                        log::error!("scheduled task terminated: uncaught exception: {}", msg);
-                    }
-                    _ => {
-                        log::error!("scheduled task terminated: {:?}", reason);
-                    }
+            Err(reason) => match reason {
+                TerminationReason::CpuTimeLimit => {
+                    log::warn!("scheduled task terminated: CPU time limit exceeded");
                 }
-            }
+                TerminationReason::WallClockTimeout => {
+                    log::warn!("scheduled task terminated: wall-clock timeout");
+                }
+                TerminationReason::MemoryLimit => {
+                    log::warn!("scheduled task terminated: memory limit exceeded");
+                }
+                TerminationReason::Exception(msg) => {
+                    log::error!("scheduled task terminated: uncaught exception: {}", msg);
+                }
+                _ => {
+                    log::error!("scheduled task terminated: {:?}", reason);
+                }
+            },
         }
 
         // CRITICAL: Flush logs before worker is dropped to prevent log loss
@@ -165,44 +173,16 @@ pub fn handle_scheduled(
                 log::debug!("scheduled task parsed: {:?}", data);
 
                 let worker_id = store::WorkerIdentifier::Id(data.worker_id.clone());
-                let worker = match store::get_worker_with_bindings(&mut conn, worker_id).await {
-                    Some(worker) => worker,
+                let worker_data = match store::get_worker_with_bindings(&mut conn, worker_id).await
+                {
+                    Some(w) => w,
                     None => {
                         log::error!("worker not found: {:?}", data.worker_id);
                         continue;
                     }
                 };
 
-                let code =
-                    match crate::transform::parse_worker_code_str(&worker.script, &worker.language)
-                    {
-                        Ok(code) => WorkerCode::js(code),
-                        Err(e) => {
-                            log::error!("Failed to parse worker code for scheduled task: {}", e);
-                            continue;
-                        }
-                    };
-
-                // Convert bindings to BindingInfo for Script (names + types only)
-                let binding_infos = bindings_to_infos(&worker.bindings);
-
-                let script = Script {
-                    code,
-                    env: if worker.env.is_empty() {
-                        None
-                    } else {
-                        Some(worker.env.clone())
-                    },
-                    bindings: binding_infos,
-                };
-
-                run_scheduled(
-                    data,
-                    script,
-                    worker.bindings,
-                    db.clone(),
-                    global_log_tx.clone(),
-                );
+                run_scheduled(data, worker_data, db.clone(), global_log_tx.clone());
             }
 
             log::debug!("scheduled task listener stopped");

src/event_scheduled.rs

@@ -6,7 +6,9 @@ pub mod nats;
 pub mod ops;
 pub mod runtime;
 pub mod store;
+#[cfg(feature = "v8")]
 mod transform;
+pub mod worker;
 pub mod worker_pool;
 
 // Re-export TerminationReason for use in bin/main.rs

src/lib.rs

@@ -25,12 +25,29 @@ pub use openworkers_runtime_boa::*;
 #[cfg(feature = "jsc")]
 pub use openworkers_runtime_jsc::*;
 
-// Compile-time check: ensure exactly one runtime is selected
+// Snapshot stub for WASM runtime (WASM doesn't need/support snapshots)
+#[cfg(feature = "wasm")]
+pub mod snapshot {
+    /// Snapshot output structure (stub for compatibility)
+    pub struct SnapshotOutput {
+        pub output: Vec<u8>,
+    }
+
+    /// Create a runtime snapshot (stub - WASM doesn't need/support snapshots)
+    pub fn create_runtime_snapshot() -> Result<SnapshotOutput, String> {
+        Err("WASM runtime does not support snapshots".to_string())
+    }
+}
+
+// Compile-time check: ensure at least one runtime is selected
 #[cfg(not(any(
     feature = "deno",
     feature = "quickjs",
     feature = "v8",
     feature = "boa",
-    feature = "jsc"
+    feature = "jsc",
+    feature = "wasm"
 )))]
-compile_error!("At least one runtime feature must be enabled: deno, quickjs, v8, boa, or jsc");
+compile_error!(
+    "At least one runtime feature must be enabled: deno, quickjs, v8, boa, jsc, or wasm"
+);