feat(node): resource health-check daemon

crates/taskito-node/src/config.rs

@@ -140,6 +140,9 @@ pub struct WorkerOptions {
     pub batch_size: Option<u32>,
     pub task_configs: Option<Vec<TaskConfigInput>>,
     pub queue_configs: Option<Vec<QueueConfigInput>>,
+    /// Names of the injectable resources this worker serves, advertised on
+    /// registration so inspection surfaces them alongside heartbeat health.
+    pub resources: Option<Vec<String>>,
     /// Opt-in decentralized mesh overlay (requires the `mesh` build feature).
     pub mesh: Option<MeshWorkerConfig>,
 }

crates/taskito-node/src/convert/stats.rs

@@ -109,6 +109,11 @@ pub struct JsWorkerRow {
     pub pool_type: Option<String>,
     pub threads: i32,
     pub tags: Option<String>,
+    /// JSON array of resource names the worker advertised at registration.
+    pub resources: Option<String>,
+    /// JSON object of per-resource health (`"healthy"`/`"unhealthy"`), written
+    /// by the worker's heartbeat.
+    pub resource_health: Option<String>,
 }
 
 pub fn worker_to_js(worker: WorkerRow) -> JsWorkerRow {
@@ -123,6 +128,8 @@ pub fn worker_to_js(worker: WorkerRow) -> JsWorkerRow {
         pool_type: worker.pool_type,
         threads: worker.threads,
         tags: worker.tags,
+        resources: worker.resources,
+        resource_health: worker.resource_health,
     }
 }
 

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

@@ -126,4 +126,25 @@ impl JsQueue {
         .await
         .map_err(join_to_napi_err)?
     }
+
+    /// Record a heartbeat for a running worker, carrying its current resource
+    /// health as a JSON object (`null` clears it). Called from the JS shell
+    /// every 5s. Returns the ids of dead workers reaped as a side effect.
+    #[napi]
+    pub async fn worker_heartbeat(
+        &self,
+        worker_id: String,
+        resource_health: Option<String>,
+    ) -> Result<Vec<String>> {
+        let storage = self.storage.clone();
+        spawn_blocking(move || {
+            storage
+                .heartbeat(&worker_id, resource_health.as_deref())
+                .map_err(to_napi_err)?;
+            // Reaping is opportunistic — a failure must not fail the heartbeat.
+            Ok(storage.reap_dead_workers().unwrap_or_default())
+        })
+        .await
+        .map_err(join_to_napi_err)?
+    }
 }

crates/taskito-node/src/worker.rs

@@ -1,8 +1,9 @@
 //! Worker wiring — scheduler, dispatcher, result-drain, and worker lifecycle
-//! (registration + heartbeat) so the worker shows up in the dashboard.
+//! (registration + unregistration) so the worker shows up in the dashboard.
+//! Heartbeats are driven from JS via [`crate::queue::JsQueue::worker_heartbeat`]
+//! so they can carry resource health without a second writer flapping the column.
 
 use std::sync::Arc;
-use std::time::Duration;
 
 use napi::bindgen_prelude::{spawn, spawn_blocking, Result};
 use napi::threadsafe_function::{ErrorStrategy, ThreadsafeFunction, ThreadsafeFunctionCallMode};
@@ -19,29 +20,36 @@ use crate::error::invalid_arg;
 
 const DEFAULT_QUEUE: &str = "default";
 const DEFAULT_CHANNEL_CAPACITY: usize = 128;
-const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(5);
 
 /// Handle to a running worker. Hold it for the worker's lifetime; call
 /// [`JsWorker::stop`] to shut it down.
 #[napi]
 pub struct JsWorker {
+    worker_id: String,
     shutdown: Arc<Notify>,
-    heartbeat_stop: Arc<Notify>,
+    lifecycle_stop: Arc<Notify>,
     #[cfg(feature = "mesh")]
     mesh_shutdown: Arc<Notify>,
 }
 
 #[napi]
 impl JsWorker {
-    /// Stop the worker: the scheduler stops dispatching, the heartbeat loop ends
-    /// and unregisters, mesh gossip/steal tasks shut down, and the background
+    /// The id this worker registered (and claims executions) under. The JS
+    /// shell heartbeats with it via [`crate::queue::JsQueue::worker_heartbeat`].
+    #[napi(getter)]
+    pub fn id(&self) -> String {
+        self.worker_id.clone()
+    }
+
+    /// Stop the worker: the scheduler stops dispatching, the lifecycle task
+    /// unregisters, mesh gossip/steal tasks shut down, and the background
     /// tasks exit once in-flight results drain.
     #[napi]
     pub fn stop(&self) {
         // `notify_one` stores a permit if no waiter is parked yet, so the signal
         // is never lost between loop iterations.
         self.shutdown.notify_one();
-        self.heartbeat_stop.notify_one();
+        self.lifecycle_stop.notify_one();
         #[cfg(feature = "mesh")]
         self.mesh_shutdown.notify_one();
     }
@@ -112,14 +120,16 @@ pub fn start_worker(
     let (job_tx, job_rx) = tokio::sync::mpsc::channel(capacity);
     let (result_tx, result_rx) = crossbeam_channel::bounded(capacity);
 
-    // Worker lifecycle: register, heartbeat every 5s, unregister on stop.
-    let heartbeat_stop = Arc::new(Notify::new());
+    // Worker lifecycle: register (advertising resource names), unregister on
+    // stop. Heartbeats come from JS so they carry current resource health.
+    let lifecycle_stop = Arc::new(Notify::new());
     spawn_worker_lifecycle(
         lifecycle_storage,
         worker_id.clone(),
         queues_csv,
+        resources_json(options.resources.as_deref()),
         capacity,
-        heartbeat_stop.clone(),
+        lifecycle_stop.clone(),
     );
 
     // Scheduler loop: poll storage, dispatch ready jobs onto `job_tx`. With the
@@ -200,18 +210,29 @@ pub fn start_worker(
     });
 
     Ok(JsWorker {
+        worker_id,
         shutdown,
-        heartbeat_stop,
+        lifecycle_stop,
         #[cfg(feature = "mesh")]
         mesh_shutdown,
     })
 }
 
-/// Register this worker and heartbeat until `stop` is signalled, then unregister.
+/// Serialize advertised resource names to the cross-SDK wire shape: a sorted
+/// JSON array, or `None` when the worker registered no resources.
+fn resources_json(resources: Option<&[String]>) -> Option<String> {
+    let names = resources.filter(|names| !names.is_empty())?;
+    let mut sorted = names.to_vec();
+    sorted.sort();
+    serde_json::to_string(&sorted).ok()
+}
+
+/// Register this worker, then unregister once `stop` is signalled.
 fn spawn_worker_lifecycle(
     storage: StorageBackend,
     worker_id: String,
     queues_csv: String,
+    resources: Option<String>,
     capacity: usize,
     stop: Arc<Notify>,
 ) {
@@ -228,7 +249,7 @@ fn spawn_worker_lifecycle(
                 &reg_id,
                 &queues_csv,
                 None,
-                None,
+                resources.as_deref(),
                 None,
                 capacity as i32,
                 Some(&hostname),
@@ -243,20 +264,7 @@ fn spawn_worker_lifecycle(
             Ok(Ok(_)) => {}
         }
 
-        loop {
-            tokio::select! {
-                _ = stop.notified() => break,
-                _ = tokio::time::sleep(HEARTBEAT_INTERVAL) => {
-                    let hb_storage = storage.clone();
-                    let hb_id = worker_id.clone();
-                    match spawn_blocking(move || hb_storage.heartbeat(&hb_id, None)).await {
-                        Ok(Err(err)) => log::warn!("[taskito-node] worker heartbeat failed: {err}"),
-                        Err(err) => log::warn!("[taskito-node] worker heartbeat task failed: {err}"),
-                        Ok(Ok(_)) => {}
-                    }
-                }
-            }
-        }
+        stop.notified().await;
 
         match spawn_blocking(move || storage.unregister_worker(&worker_id)).await {
             Ok(Err(err)) => log::warn!("[taskito-node] worker unregister failed: {err}"),

sdks/node/src/queue.ts

@@ -262,6 +262,13 @@ export class Queue<TTasks extends TaskMap = TaskMap> {
       scope?: ResourceScope;
       dispose?: (value: T) => void | Promise<void>;
       pool?: PoolOptions;
+      /** Returns truthy while healthy; failures trigger recreation. Worker scope only. */
+      healthCheck?: (value: T) => boolean | Promise<boolean>;
+      /** Milliseconds between health checks. 0 or absent disables checking. */
+      healthCheckIntervalMs?: number;
+      /** Failed checks tolerated (while recreation also fails) before the
+       * resource is marked permanently unhealthy. Default 3. */
+      maxRecreationAttempts?: number;
     },
   ): this {
     const scope = options?.scope ?? "worker";
@@ -270,11 +277,20 @@ export class Queue<TTasks extends TaskMap = TaskMap> {
         `Resource "${name}": pool options require scope "pooled" (got "${scope}")`,
       );
     }
+    if (options?.healthCheck && scope !== "worker") {
+      throw new ResourceError(
+        `Resource "${name}": health checks require scope "worker" (got "${scope}") — ` +
+          "task and pooled instances are already rebuilt or recycled per job",
+      );
+    }
     this.resources.register<T>(name, {
       factory,
       scope,
       dispose: options?.dispose,
       pool: options?.pool,
+      healthCheck: options?.healthCheck,
+      healthCheckIntervalMs: options?.healthCheckIntervalMs,
+      maxRecreationAttempts: options?.maxRecreationAttempts,
     });
     return this;
   }

sdks/node/src/resources/health.ts

@@ -0,0 +1,133 @@
+import { createLogger } from "../utils";
+import type { ResourceRuntime } from "./runtime";
+import type { ResourceDefinition } from "./types";
+
+const log = createLogger("resources");
+
+/** How often the daemon wakes to see whether any resource's check is due. */
+const TICK_MS = 500;
+
+/** Fallback for {@link ResourceDefinition.maxRecreationAttempts}. */
+const DEFAULT_MAX_RECREATION_ATTEMPTS = 3;
+
+/**
+ * Periodically checks worker-resource health and recreates failing instances.
+ *
+ * One unref'd timer wakes every {@link TICK_MS}; each resource with a positive
+ * `healthCheckIntervalMs` is checked independently on its own schedule. A
+ * failed (or throwing) check triggers recreation; when recreation also fails
+ * and the failure budget is spent, the resource is marked permanently
+ * unhealthy and never rechecked.
+ */
+export class HealthChecker {
+  private timer: ReturnType<typeof setInterval> | undefined;
+  /** The tick currently running, if any — doubles as the overlap guard. */
+  private inFlight: Promise<void> | undefined;
+  /** Set by {@link HealthChecker.stop} — a stopped checker never restarts. */
+  private stopped = false;
+  private readonly nextDue = new Map<string, number>();
+  private readonly failCount = new Map<string, number>();
+
+  constructor(private readonly runtime: ResourceRuntime) {}
+
+  /** Start the daemon. No-op unless some resource asked for health checks. */
+  start(): void {
+    if (this.timer || this.stopped) {
+      return;
+    }
+    const checked = this.runtime.healthChecked();
+    if (checked.size === 0) {
+      return;
+    }
+    const now = Date.now();
+    for (const [name, def] of checked) {
+      this.nextDue.set(name, now + (def.healthCheckIntervalMs ?? 0));
+      this.failCount.set(name, 0);
+    }
+    // The tick never rejects (see below), so the timer can't leak an
+    // unhandled rejection; unref keeps it from pinning the process open.
+    this.timer = setInterval(() => this.onTick(checked), TICK_MS);
+    this.timer.unref();
+  }
+
+  /**
+   * Stop the daemon and wait for any in-flight tick to drain, so no
+   * check/recreate can mutate runtime state once this resolves — safe to run
+   * worker teardown right after. Terminal and idempotent: a stopped checker
+   * never restarts.
+   */
+  async stop(): Promise<void> {
+    this.stopped = true;
+    if (this.timer) {
+      clearInterval(this.timer);
+      this.timer = undefined;
+    }
+    await this.inFlight;
+  }
+
+  /** One timer wake-up: skip when the previous tick is still in flight. */
+  private onTick(checked: ReadonlyMap<string, ResourceDefinition>): void {
+    if (this.inFlight || this.stopped) {
+      return;
+    }
+    this.inFlight = this.tick(checked).finally(() => {
+      this.inFlight = undefined;
+    });
+  }
+
+  /** Run every due check. Catches everything — timers must survive. */
+  private async tick(checked: ReadonlyMap<string, ResourceDefinition>): Promise<void> {
+    try {
+      for (const [name, def] of checked) {
+        if (this.stopped) {
+          return; // shutdown began mid-tick — leave the runtime alone
+        }
+        if (Date.now() < (this.nextDue.get(name) ?? 0) || this.runtime.isUnhealthy(name)) {
+          continue;
+        }
+        await this.checkAndRepair(name, def);
+        this.nextDue.set(name, Date.now() + (def.healthCheckIntervalMs ?? 0));
+      }
+    } catch (error) {
+      log.warn(() => "health-check tick failed", error);
+    }
+  }
+
+  /** Check one resource; on failure recreate it, exhausting into unhealthy. */
+  private async checkAndRepair(name: string, def: ResourceDefinition): Promise<void> {
+    if (await this.checkOne(name, def)) {
+      this.failCount.set(name, 0);
+      return;
+    }
+    const failures = (this.failCount.get(name) ?? 0) + 1;
+    this.failCount.set(name, failures);
+    const maxAttempts = def.maxRecreationAttempts ?? DEFAULT_MAX_RECREATION_ATTEMPTS;
+    log.warn(`resource "${name}" health check failed (${failures}/${maxAttempts})`);
+    if (this.stopped) {
+      return; // shutdown began while the check ran — don't recreate into teardown
+    }
+    if (await this.runtime.recreate(name)) {
+      this.failCount.set(name, 0); // a fresh instance starts with a clean slate
+    } else if (failures >= maxAttempts) {
+      this.runtime.markUnhealthy(name);
+    }
+  }
+
+  /** Run a single health check. A throwing check counts as unhealthy. */
+  private async checkOne(name: string, def: ResourceDefinition): Promise<boolean> {
+    const { healthCheck } = def;
+    if (!healthCheck) {
+      return true;
+    }
+    const built = this.runtime.builtWorkerInstance(name);
+    if (!built) {
+      return false; // nothing built yet — recreation will build it
+    }
+    try {
+      return Boolean(await healthCheck(await built));
+    } catch (error) {
+      log.warn(() => `health check for resource "${name}" threw`, error);
+      return false;
+    }
+  }
+}

sdks/node/src/resources/index.ts

@@ -1,7 +1,8 @@
 export { runWithResolver, useResource } from "./context";
+export { HealthChecker } from "./health";
 export { type MockResource, mockResource } from "./mock";
 export { ResourcePool } from "./pool";
-export { ResourceRuntime, type TaskScope } from "./runtime";
+export { type HealthState, ResourceRuntime, type TaskScope } from "./runtime";
 export type {
   PoolOptions,
   PoolStats,