fix: harden Node SDK workflows, worker, dashboard

crates/taskito-core/src/storage/diesel_common/logs.rs

@@ -43,6 +43,27 @@ macro_rules! impl_diesel_log_ops {
                 Ok(rows)
             }
 
+            /// Logs for a job with id strictly after `after_id` (cursor scan).
+            pub fn get_task_logs_after(
+                &self,
+                job_id: &str,
+                after_id: Option<&str>,
+            ) -> Result<Vec<TaskLogRow>> {
+                let mut conn = self.conn()?;
+                let mut query = task_logs::table
+                    .filter(task_logs::job_id.eq(job_id))
+                    .into_boxed();
+                if let Some(cursor) = after_id {
+                    query = query.filter(task_logs::id.gt(cursor.to_string()));
+                }
+                // UUIDv7 ids sort by creation time, so id order == time order.
+                let rows = query
+                    .order(task_logs::id.asc())
+                    .select(TaskLogRow::as_select())
+                    .load(&mut conn)?;
+                Ok(rows)
+            }
+
             /// Query logs by task name, level, etc.
             pub fn query_task_logs(
                 &self,

crates/taskito-core/src/storage/mod.rs

@@ -399,6 +399,13 @@ macro_rules! impl_storage {
             ) -> $crate::error::Result<Vec<$crate::storage::models::TaskLogRow>> {
                 self.get_task_logs(job_id)
             }
+            fn get_task_logs_after(
+                &self,
+                job_id: &str,
+                after_id: Option<&str>,
+            ) -> $crate::error::Result<Vec<$crate::storage::models::TaskLogRow>> {
+                self.get_task_logs_after(job_id, after_id)
+            }
             fn query_task_logs(
                 &self,
                 task_name: Option<&str>,
@@ -931,6 +938,13 @@ impl Storage for StorageBackend {
     fn get_task_logs(&self, job_id: &str) -> Result<Vec<models::TaskLogRow>> {
         delegate!(self, get_task_logs, job_id)
     }
+    fn get_task_logs_after(
+        &self,
+        job_id: &str,
+        after_id: Option<&str>,
+    ) -> Result<Vec<models::TaskLogRow>> {
+        delegate!(self, get_task_logs_after, job_id, after_id)
+    }
     fn query_task_logs(
         &self,
         task_name: Option<&str>,

crates/taskito-core/src/storage/redis_backend/logs.rs

@@ -93,6 +93,40 @@ impl RedisStorage {
         Ok(rows)
     }
 
+    /// Logs for a job with id strictly after `after_id` (cursor scan).
+    pub fn get_task_logs_after(
+        &self,
+        job_id: &str,
+        after_id: Option<&str>,
+    ) -> Result<Vec<TaskLogRow>> {
+        let mut conn = self.conn()?;
+        let by_job_key = self.key(&["logs", "by_job", job_id]);
+
+        // Filter ids before fetching values so old entries cost one cheap
+        // ZRANGEBYSCORE, not a GET per historical log.
+        let mut ids: Vec<String> = conn
+            .zrangebyscore(&by_job_key, "-inf", "+inf")
+            .map_err(map_err)?;
+        if let Some(cursor) = after_id {
+            ids.retain(|id| id.as_str() > cursor);
+        }
+
+        let mut rows = Vec::new();
+        for id in ids {
+            let log_key = self.key(&["log", &id]);
+            let data: Option<String> = conn.get(&log_key).map_err(map_err)?;
+            if let Some(d) = data {
+                let entry: LogEntry =
+                    serde_json::from_str(&d).map_err(|e| QueueError::Other(e.to_string()))?;
+                rows.push(TaskLogRow::from(entry));
+            }
+        }
+
+        // UUIDv7 ids sort by creation time, so id order == time order.
+        rows.sort_by(|a, b| a.id.cmp(&b.id));
+        Ok(rows)
+    }
+
     pub fn query_task_logs(
         &self,
         task_name: Option<&str>,

crates/taskito-core/src/storage/traits.rs

@@ -153,6 +153,9 @@ pub trait Storage: Send + Sync + Clone {
         extra: Option<&str>,
     ) -> Result<()>;
     fn get_task_logs(&self, job_id: &str) -> Result<Vec<TaskLogRow>>;
+    /// Logs for a job with id strictly after `after_id` (UUIDv7 ids are
+    /// time-ordered, so the id doubles as a stream cursor). `None` = all.
+    fn get_task_logs_after(&self, job_id: &str, after_id: Option<&str>) -> Result<Vec<TaskLogRow>>;
     fn query_task_logs(
         &self,
         task_name: Option<&str>,

crates/taskito-core/tests/rust/storage_tests.rs

@@ -885,6 +885,32 @@ fn run_storage_tests(s: &impl Storage) {
     test_archived_job_payload_resolves(s);
     test_concurrent_dequeue_no_double_claim(s);
     test_rate_limit_token_exhaustion(s);
+    test_task_logs_after_cursor(s);
+}
+
+fn test_task_logs_after_cursor(s: &impl Storage) {
+    let job = s.enqueue(make_job("q-logs", "log_task")).unwrap();
+    for i in 0..3 {
+        s.write_task_log(&job.id, "log_task", "result", &format!("m{i}"), None)
+            .unwrap();
+    }
+
+    // No cursor → everything, in id (time) order, matching get_task_logs.
+    let all = s.get_task_logs_after(&job.id, None).unwrap();
+    assert_eq!(all.len(), 3);
+    assert!(all.windows(2).all(|w| w[0].id < w[1].id));
+
+    // A cursor at entry N yields only the entries written after it.
+    let after_first = s.get_task_logs_after(&job.id, Some(&all[0].id)).unwrap();
+    assert_eq!(
+        after_first
+            .iter()
+            .map(|r| r.id.as_str())
+            .collect::<Vec<_>>(),
+        all[1..].iter().map(|r| r.id.as_str()).collect::<Vec<_>>()
+    );
+    let after_last = s.get_task_logs_after(&job.id, Some(&all[2].id)).unwrap();
+    assert!(after_last.is_empty());
 }
 
 fn test_rate_limit_token_exhaustion(s: &impl Storage) {

crates/taskito-node/src/dispatcher.rs

@@ -8,7 +8,7 @@
 use std::time::{Duration, Instant};
 
 use crossbeam_channel::Sender;
-use napi::bindgen_prelude::{spawn, Buffer, Promise};
+use napi::bindgen_prelude::{spawn, spawn_blocking, Buffer, Promise};
 use napi::threadsafe_function::{ErrorStrategy, ThreadsafeFunction, ThreadsafeFunctionCallMode};
 use taskito_core::job::Job;
 use taskito_core::scheduler::JobResult;
@@ -48,8 +48,19 @@ impl WorkerDispatcher for NodeDispatcher {
             let storage = self.storage.clone();
             let result_tx = result_tx.clone();
             spawn(async move {
+                let job_id = job.id.clone();
                 let result = run_one(&callback, &storage, job).await;
-                let _ = result_tx.send(result);
+                // A full bounded channel parks the sender — do it on the
+                // blocking pool, never on the shared async runtime.
+                match spawn_blocking(move || result_tx.send(result)).await {
+                    Ok(Ok(())) => {}
+                    Ok(Err(_)) | Err(_) => {
+                        // A closed channel means the drain loop died — surface it.
+                        log::error!(
+                            "[taskito-node] result channel closed; dropping outcome for {job_id}"
+                        );
+                    }
+                }
             });
         }
     }
@@ -58,12 +69,14 @@ impl WorkerDispatcher for NodeDispatcher {
 }
 
 /// Invoke the JS task for one job and translate the outcome into a [`JobResult`].
-async fn run_one(callback: &TaskCallback, storage: &StorageBackend, job: Job) -> JobResult {
+async fn run_one(callback: &TaskCallback, storage: &StorageBackend, mut job: Job) -> JobResult {
     let started = Instant::now();
     let invocation = JsTaskInvocation {
         id: job.id.clone(),
         task_name: job.task_name.clone(),
-        payload: Buffer::from(job.payload.clone()),
+        // Never read from `job` again — take, don't clone (payloads can be
+        // large); `job` stays whole for the later `failure(job, ...)` moves.
+        payload: Buffer::from(std::mem::take(&mut job.payload)),
     };
 
     // The callback runs on the JS thread and returns a Promise; bridge its

crates/taskito-node/src/error.rs

@@ -8,6 +8,14 @@ pub fn to_napi_err(err: QueueError) -> Error {
     Error::new(Status::GenericFailure, err.to_string())
 }
 
+/// Map a blocking-task join failure onto a napi [`Error`].
+pub fn join_to_napi_err(err: tokio::task::JoinError) -> Error {
+    Error::new(
+        Status::GenericFailure,
+        format!("blocking storage task failed: {err}"),
+    )
+}
+
 /// Build an `InvalidArg` error for caller-supplied input that fails validation
 /// at the N-API boundary (negative pagination, zero pool size, bad rate limit…).
 pub fn invalid_arg(message: impl Into<String>) -> Error {

crates/taskito-node/src/queue/admin.rs

@@ -1,52 +1,72 @@
-//! Management (mutating) methods on `JsQueue`.
+//! Management (mutating) methods on `JsQueue`. List/purge methods scan whole
+//! tables, so they are async and offload to the blocking pool; single-row
+//! operations stay sync.
 
 use std::collections::HashMap;
 
-use napi::bindgen_prelude::Result;
+use napi::bindgen_prelude::{spawn_blocking, Result};
 use napi_derive::napi;
 use taskito_core::Storage;
 
 use super::JsQueue;
 use crate::convert::{dead_job_to_js, JsDeadJob};
-use crate::error::{non_negative, to_napi_err};
+use crate::error::{join_to_napi_err, non_negative, to_napi_err};
 
 const DEFAULT_LIMIT: i64 = 50;
 
 #[napi]
 impl JsQueue {
     /// List dead-letter entries (paginated).
     #[napi]
-    pub fn dead_letters(&self, limit: Option<i64>, offset: Option<i64>) -> Result<Vec<JsDeadJob>> {
+    pub async fn dead_letters(
+        &self,
+        limit: Option<i64>,
+        offset: Option<i64>,
+    ) -> Result<Vec<JsDeadJob>> {
         let limit = non_negative(limit.unwrap_or(DEFAULT_LIMIT), "limit")?;
         let offset = non_negative(offset.unwrap_or(0), "offset")?;
-        let dead = self.storage.list_dead(limit, offset).map_err(to_napi_err)?;
-        Ok(dead.into_iter().map(dead_job_to_js).collect())
+        let storage = self.storage.clone();
+        spawn_blocking(move || {
+            let dead = storage.list_dead(limit, offset).map_err(to_napi_err)?;
+            Ok(dead.into_iter().map(dead_job_to_js).collect())
+        })
+        .await
+        .map_err(join_to_napi_err)?
     }
 
     /// List dead-letter entries for a single task (paginated, newest first).
     #[napi]
-    pub fn dead_letters_by_task(
+    pub async fn dead_letters_by_task(
         &self,
         task_name: String,
         limit: Option<i64>,
         offset: Option<i64>,
     ) -> Result<Vec<JsDeadJob>> {
         let limit = non_negative(limit.unwrap_or(DEFAULT_LIMIT), "limit")?;
         let offset = non_negative(offset.unwrap_or(0), "offset")?;
-        let dead = self
-            .storage
-            .list_dead_by_task(&task_name, limit, offset)
-            .map_err(to_napi_err)?;
-        Ok(dead.into_iter().map(dead_job_to_js).collect())
+        let storage = self.storage.clone();
+        spawn_blocking(move || {
+            let dead = storage
+                .list_dead_by_task(&task_name, limit, offset)
+                .map_err(to_napi_err)?;
+            Ok(dead.into_iter().map(dead_job_to_js).collect())
+        })
+        .await
+        .map_err(join_to_napi_err)?
     }
 
     /// Delete every dead-letter entry for a task. Returns the count removed.
     #[napi]
-    pub fn purge_dead_by_task(&self, task_name: String) -> Result<i64> {
-        self.storage
-            .purge_dead_by_task(&task_name)
-            .map(|n| n as i64)
-            .map_err(to_napi_err)
+    pub async fn purge_dead_by_task(&self, task_name: String) -> Result<i64> {
+        let storage = self.storage.clone();
+        spawn_blocking(move || {
+            storage
+                .purge_dead_by_task(&task_name)
+                .map(|n| n as i64)
+                .map_err(to_napi_err)
+        })
+        .await
+        .map_err(join_to_napi_err)?
     }
 
     /// Re-enqueue a dead-letter entry. Returns the new job id.
@@ -63,22 +83,32 @@ impl JsQueue {
 
     /// Purge dead-letter entries older than `older_than_ms`. Returns the count removed.
     #[napi]
-    pub fn purge_dead(&self, older_than_ms: i64) -> Result<i64> {
+    pub async fn purge_dead(&self, older_than_ms: i64) -> Result<i64> {
         let older_than_ms = non_negative(older_than_ms, "olderThanMs")?;
-        self.storage
-            .purge_dead(older_than_ms)
-            .map(|n| n as i64)
-            .map_err(to_napi_err)
+        let storage = self.storage.clone();
+        spawn_blocking(move || {
+            storage
+                .purge_dead(older_than_ms)
+                .map(|n| n as i64)
+                .map_err(to_napi_err)
+        })
+        .await
+        .map_err(join_to_napi_err)?
     }
 
     /// Purge completed jobs older than `older_than_ms`. Returns the count removed.
     #[napi]
-    pub fn purge_completed(&self, older_than_ms: i64) -> Result<i64> {
+    pub async fn purge_completed(&self, older_than_ms: i64) -> Result<i64> {
         let older_than_ms = non_negative(older_than_ms, "olderThanMs")?;
-        self.storage
-            .purge_completed(older_than_ms)
-            .map(|n| n as i64)
-            .map_err(to_napi_err)
+        let storage = self.storage.clone();
+        spawn_blocking(move || {
+            storage
+                .purge_completed(older_than_ms)
+                .map(|n| n as i64)
+                .map_err(to_napi_err)
+        })
+        .await
+        .map_err(join_to_napi_err)?
     }
 
     /// Pause a queue — workers stop dispatching its jobs until resumed.