Ack scheduled task

Author: max-lt

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "openworkers-runner"
-version = "0.1.8-multi"
+version = "0.1.9"
 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.8"}
+openworkers-runtime = { git = "https://github.com/openworkers/openworkers-runtime", tag = "v0.1.8"}
 # openworkers-runtime =  { path = "../openworkers-runtime" }
 nats = "0.25.0"
 serde_json = "1.0.135"

Cargo.toml

@@ -19,7 +19,7 @@ pub struct ScheduledData {
     pub worker_id: String,
 }
 
-fn run_scheduled(data: ScheduledData, script: Script) {
+fn run_scheduled(data: ScheduledData, script: Script, nc: nats::Connection) {
     let (res_tx, res_rx) = tokio::sync::oneshot::channel::<()>();
 
     let task = Task::Scheduled(Some(ScheduledInit::new(res_tx, data.scheduled_time)));
@@ -47,12 +47,17 @@ 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}"),
         }
 
         log::debug!("scheduled task listener stopped");
+
+        match nc.publish(&format!("scheduled.ack.{}", data.id), b"") {
+            Ok(()) => {}
+            Err(err) => log::error!("failed to ack: {err}"),
+        }
     });
 }
 
@@ -108,11 +113,11 @@ 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,
                     },
                 };
 
-                run_scheduled(data, script);
+                run_scheduled(data, script, nc.clone());
             }
 
             log::debug!("scheduled task listener stopped");