Add resource limits

Author: max-lt

.github/workflows/docker.yml

@@ -1,4 +1,3 @@
-name: Build docker image
 
 on:
   push:
@@ -8,12 +7,14 @@ on:
     tags:
       - 'v*.*.*'
   pull_request:
-    branches: [ master ]
+    branches:
+      - master
+      - dev
 
 jobs:
-  build_types:
+  build:
     name: Build
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-24.04
     steps:
       - name: Check Out Repo
         uses: actions/checkout@v4

.vscode/settings.json

@@ -1,3 +1,12 @@
 {
-  "cSpell.words": ["actix", "dotenv", "nats", "oneshot", "sqlx"]
+  "cSpell.words": [
+    "actix",
+    "codegen",
+    "dotenv",
+    "nats",
+    "oneshot",
+    "openworkers",
+    "serde",
+    "sqlx"
+  ]
 }

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "openworkers-runner"
-version = "0.1.9"
+version = "0.1.10"
 edition = "2021"
 default-run = "openworkers-runner"
 
@@ -14,7 +14,7 @@ tokio = "1.43.0"
 env_logger = "0.11.6"
 http_v02 = { package = "http", version = "0.2.12" }
 sqlx = { version = "0.8.3", features = [ "runtime-tokio", "postgres", "uuid", "bigdecimal", "rust_decimal" ] }
-openworkers-runtime ={ git = "https://github.com/openworkers/openworkers-runtime", tag = "v0.1.9"}
+openworkers-runtime ={ git = "https://github.com/openworkers/openworkers-runtime", tag = "v0.1.10"}
 # openworkers-runtime =  { path = "../openworkers-runtime" }
 nats = "0.25.0"
 serde_json = "1.0.137"

Cargo.toml

@@ -1,8 +1,10 @@
 use std::ops::Deref;
 use std::thread::JoinHandle;
+use std::time::Duration;
 
 use bytes::Bytes;
 use openworkers_runtime::FetchInit;
+use openworkers_runtime::RuntimeLimits;
 use openworkers_runtime::Script;
 use openworkers_runtime::Task;
 use openworkers_runtime::Worker;
@@ -11,6 +13,9 @@ use crate::store::WorkerData;
 
 type ResTx = tokio::sync::oneshot::Sender<http_v02::Response<Bytes>>;
 
+// Default timeout for fetch events
+const FETCH_TIMEOUT_MS: u64 = 64_000; // 64 seconds
+
 pub fn run_fetch(
     worker: WorkerData,
     req: http_v02::Request<Bytes>,
@@ -31,7 +36,14 @@ pub fn run_fetch(
 
         let tasks = local.spawn_local(async move {
             log::debug!("create worker");
-            let mut worker = match Worker::new(script, Some(log_tx)).await {
+
+            let limits = RuntimeLimits {
+                max_cpu_time_ms: 100,           // 100ms CPU time for fetch tasks
+                max_wall_clock_time_ms: 60_000, // 60s total time for fetch tasks
+                ..Default::default()
+            };
+
+            let mut worker = match Worker::new(script, Some(log_tx), Some(limits)).await {
                 Ok(worker) => worker,
                 Err(err) => {
                     log::error!("failed to create worker: {err}");
@@ -50,10 +62,18 @@ pub fn run_fetch(
 
             let task = Task::Fetch(Some(FetchInit::new(req, res_tx)));
 
-            log::debug!("exec fetch task");
-            match worker.exec(task).await {
-                Ok(()) => log::debug!("exec completed"),
-                Err(err) => log::error!("exec did not complete: {err}"),
+            log::debug!("exec fetch task with {}ms timeout", FETCH_TIMEOUT_MS);
+
+            // Wrap execution with timeout
+            let timeout_duration = Duration::from_millis(FETCH_TIMEOUT_MS);
+            match tokio::time::timeout(timeout_duration, worker.exec(task)).await {
+                Ok(Ok(())) => log::debug!("exec completed"),
+                Ok(Err(err)) => log::error!("exec did not complete: {err}"),
+                Err(_) => {
+                    log::error!("exec timeout after {}ms", FETCH_TIMEOUT_MS);
+                    // Note: Worker may have already sent a response via FetchInit
+                    // If no response was sent, res_tx will be dropped and client gets an error
+                }
             }
         });
 

src/event_fetch.rs

@@ -1,5 +1,6 @@
 use std::ops::Deref;
 
+use openworkers_runtime::RuntimeLimits;
 use openworkers_runtime::ScheduledInit;
 use openworkers_runtime::Script;
 use openworkers_runtime::Task;
@@ -36,7 +37,16 @@ fn run_scheduled(data: ScheduledData, script: Script) {
 
         local.spawn_local(async move {
             log::debug!("create worker");
-            let mut worker = Worker::new(script, Some(log_tx)).await.unwrap();
+
+            let limits = RuntimeLimits {
+                max_cpu_time_ms: 100,           // 100ms CPU time for scheduled tasks
+                max_wall_clock_time_ms: 60_000, // 60s total time for scheduled tasks
+                ..Default::default()
+            };
+
+            let mut worker = Worker::new(script, Some(log_tx), Some(limits))
+                .await
+                .unwrap();
 
             log::debug!("exec scheduled task");
             match worker.exec(task).await {
@@ -47,7 +57,7 @@ fn run_scheduled(data: ScheduledData, script: Script) {
 
         log::debug!("scheduled task listener started");
 
-        match local.block_on(&rt, async { res_rx.await } ) {
+        match local.block_on(&rt, async { res_rx.await }) {
             Ok(()) => {}
             Err(err) => log::error!("failed to wait for end: {err}"),
         }
@@ -108,7 +118,7 @@ pub fn handle_scheduled(db: sqlx::Pool<sqlx::Postgres>) {
                     code: crate::transform::parse_worker_code(&worker),
                     env: match worker.env {
                         Some(env) => Some(env.deref().to_owned()),
-                        None => None
+                        None => None,
                     },
                 };
 

src/event_scheduled.rs

@@ -23,7 +23,10 @@ pub struct WorkerData {
     pub language: WorkerLanguage,
 }
 
-pub async fn get_worker(conn: &mut sqlx::PgConnection, identifier: WorkerIdentifier) -> Option<WorkerData> {
+pub async fn get_worker(
+    conn: &mut sqlx::PgConnection,
+    identifier: WorkerIdentifier,
+) -> Option<WorkerData> {
     log::debug!("get_worker: {:?}", identifier);
 
     let query = format!(
@@ -58,7 +61,12 @@ pub async fn get_worker(conn: &mut sqlx::PgConnection, identifier: WorkerIdentif
         .await
     {
         Ok(worker) => {
-            log::debug!("worker found: id: {}, checksum: {}, language: {:?}", worker.id, worker.checksum, worker.language);
+            log::debug!(
+                "worker found: id: {}, checksum: {}, language: {:?}",
+                worker.id,
+                worker.checksum,
+                worker.language
+            );
             Some(worker)
         }
         Err(err) => {
@@ -68,7 +76,10 @@ pub async fn get_worker(conn: &mut sqlx::PgConnection, identifier: WorkerIdentif
     }
 }
 
-pub async fn get_worker_id_from_domain(conn: &mut sqlx::PgConnection, domain: String) -> Option<String> {
+pub async fn get_worker_id_from_domain(
+    conn: &mut sqlx::PgConnection,
+    domain: String,
+) -> Option<String> {
     let query = sqlx::query_scalar!(
         "SELECT worker_id::text FROM domains WHERE name = $1 LIMIT 1",
         domain