feat(node): resource health-check daemon#378
Merged
Merged
Conversation
📝 WalkthroughWalkthroughThis PR adds worker resource health checks, runtime recreation and unhealthy-state tracking, and worker heartbeat reporting. Native bindings now carry resource metadata and heartbeat updates, while the Node SDK adds health-check configuration, a periodic checker, and heartbeat emission from workers. ChangesRust native worker heartbeat and resource registration
Node SDK resource health checking
Estimated code review effort: 4 (Complex) | ~60 minutes 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🧹 Nitpick comments (3)
crates/taskito-node/src/worker.rs (1)
94-96: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winStale comment: lifecycle loop no longer heartbeats.
This comment still describes the lifecycle loop as "registers/heartbeats", but heartbeats have been fully moved to JS (per the new module doc at Line 2-4) and the loop now only registers, waits on
stop, then unregisters (Line 267). Worth updating to avoid confusing future readers about where heartbeats originate.📝 Proposed comment fix
- // The dispatcher reads cancel flags, and the lifecycle loop registers/heartbeats + // The dispatcher reads cancel flags, and the lifecycle loop registers/unregisters // — both need their own storage handle before `storage` moves into the scheduler.Also applies to: 267-267
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@crates/taskito-node/src/worker.rs` around lines 94 - 96, Update the stale comment near dispatcher_storage in worker.rs so it no longer says the lifecycle loop “registers/heartbeats”; the heartbeats now live in JS, and the Rust lifecycle loop only registers, waits on stop, then unregisters. Keep the comment aligned with the current behavior around the dispatcher storage clone and the lifecycle loop so future readers can locate the responsibility split correctly.crates/taskito-node/src/queue/inspect.rs (1)
129-149: 🚀 Performance & Scalability | 🔵 TrivialReaping on every heartbeat may not scale with worker count.
Each worker heartbeats independently (every ~5s), and each call unconditionally invokes
storage.reap_dead_workers(). With N live workers, that's N reap scans every heartbeat interval instead of one global periodic reap — redundant load that grows with fleet size.Consider throttling reaping (e.g., only reap if enough time elapsed since the last reap, tracked via a shared atomic/mutex on
JsQueueor storage), or moving it to a single background interval decoupled from per-worker heartbeats.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@crates/taskito-node/src/queue/inspect.rs` around lines 129 - 149, The worker heartbeat path in worker_heartbeat currently triggers storage.reap_dead_workers() on every call, which creates redundant reap scans as worker count grows. Update the JsQueue/inspect flow so reaping is throttled or centralized: either gate reaping with a shared last-reap timestamp/lock on JsQueue or storage, or move it to a single background interval. Keep heartbeat recording in worker_heartbeat intact and ensure the reap step remains opportunistic without blocking or failing the heartbeat.sdks/node/src/resources/health.ts (1)
8-8: 🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick winHealth-checked resource set is frozen at
start(); later registrations/replacements are invisible to the daemon.
checkedis captured once fromruntime.healthChecked()and reused for the timer's whole life.ResourceRuntime.register()explicitly supports replacing a definition on a live runtime, but if that happens afterstart()(e.g. a worker is already running), the newhealthCheck/healthCheckIntervalMs/maxRecreationAttempts— or a brand-new health-checked resource registered later — never gets picked up by this already-running daemon.Also,
TICK_MS = 500means no resource is ever checked more often than every 500ms, even ifhealthCheckIntervalMsis configured smaller (as several tests here do, e.g. 50/100ms) — worth documenting onResourceDefinition.healthCheckIntervalMs.♻️ Re-derive `checked` per tick
- 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); - } - this.timer = setInterval(() => void this.tick(checked), TICK_MS); + if (this.runtime.healthChecked().size === 0) { + return; + } + this.timer = setInterval(() => void this.tick(), TICK_MS); this.timer.unref();and inside
tick(), callconst checked = this.runtime.healthChecked();at the top, seedingnextDue/failCountlazily on first sight of a name.Also applies to: 32-49
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@sdks/node/src/resources/health.ts` at line 8, The health daemon in `HealthCheckDaemon` caches the result of `runtime.healthChecked()` once at `start()`, so later `ResourceRuntime.register()` changes and newly added health-checked resources are never observed; move the lookup into `tick()` and re-derive `checked` each tick, initializing `nextDue` and `failCount` lazily for newly seen names. Also account for `TICK_MS` as a hard polling floor by documenting on `ResourceDefinition.healthCheckIntervalMs` that checks cannot run more frequently than the daemon’s tick interval.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@sdks/node/src/resources/health.ts`:
- Around line 51-78: `HealthChecker.stop()` only clears the timer, so an
in-flight `tick()` can still run `checkAndRepair()` and call
`runtime.recreate()` after shutdown starts. Update the `HealthChecker` flow so
`stop()` also prevents any pending/active tick from doing work and `tick()`
exits early once stop has begun; then make `Worker.stop()` await
`resources.teardownWorker()` before returning so teardown and health checks
cannot race. Use the existing `stop()`, `tick()`, and `Worker.stop()` entry
points to wire this shutdown guard through the lifecycle.
In `@sdks/node/src/worker.ts`:
- Around line 200-227: Worker.start() is creating a separate HealthChecker per
worker even though the shared ResourceRuntime can already be leased
concurrently, which can lead to overlapping recreate() calls on the same
resources. Move HealthChecker lifecycle management to the shared runtime lease
path so a single checker is started and stopped with the ResourceRuntime instead
of per Worker instance, and update the Worker.start() / Worker constructor flow
to avoid creating multiple checkers for the same resources.
- Around line 213-224: The teardown path can still race with an in-flight
HealthChecker tick, allowing recreate() to run after the heartbeat timer is
cleared but before native.stop() and resources.teardownWorker() finish. Update
HealthChecker.stop() to wait for any active tick/recreate work to complete
instead of only clearing its interval, and make Worker.stop() await that stop()
result before releasing worker resources. Use the HealthChecker and
Worker.stop() paths to ensure no health-check activity remains before teardown
proceeds.
---
Nitpick comments:
In `@crates/taskito-node/src/queue/inspect.rs`:
- Around line 129-149: The worker heartbeat path in worker_heartbeat currently
triggers storage.reap_dead_workers() on every call, which creates redundant reap
scans as worker count grows. Update the JsQueue/inspect flow so reaping is
throttled or centralized: either gate reaping with a shared last-reap
timestamp/lock on JsQueue or storage, or move it to a single background
interval. Keep heartbeat recording in worker_heartbeat intact and ensure the
reap step remains opportunistic without blocking or failing the heartbeat.
In `@crates/taskito-node/src/worker.rs`:
- Around line 94-96: Update the stale comment near dispatcher_storage in
worker.rs so it no longer says the lifecycle loop “registers/heartbeats”; the
heartbeats now live in JS, and the Rust lifecycle loop only registers, waits on
stop, then unregisters. Keep the comment aligned with the current behavior
around the dispatcher storage clone and the lifecycle loop so future readers can
locate the responsibility split correctly.
In `@sdks/node/src/resources/health.ts`:
- Line 8: The health daemon in `HealthCheckDaemon` caches the result of
`runtime.healthChecked()` once at `start()`, so later
`ResourceRuntime.register()` changes and newly added health-checked resources
are never observed; move the lookup into `tick()` and re-derive `checked` each
tick, initializing `nextDue` and `failCount` lazily for newly seen names. Also
account for `TICK_MS` as a hard polling floor by documenting on
`ResourceDefinition.healthCheckIntervalMs` that checks cannot run more
frequently than the daemon’s tick interval.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 3cc1cf53-d00f-4acd-82c7-85f873b16bcb
📒 Files selected for processing (11)
crates/taskito-node/src/config.rscrates/taskito-node/src/convert/stats.rscrates/taskito-node/src/queue/inspect.rscrates/taskito-node/src/worker.rssdks/node/src/queue.tssdks/node/src/resources/health.tssdks/node/src/resources/index.tssdks/node/src/resources/runtime.tssdks/node/src/resources/types.tssdks/node/src/worker.tssdks/node/test/resources/health.test.ts
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
sdks/node/src/worker.ts (1)
213-221: 🩺 Stability & Availability | 🟡 Minor | ⚡ Quick winWrap
sendHeartbeat's synchronous body in try/catch.Only the async
queue.workerHeartbeat(...)call is guarded via.catch(). Ifresources.healthSnapshot()orJSON.stringify(snapshot)throws synchronously, the exception escapes thesetIntervalcallback uncaught, contradicting the intent stated in the comment above ("Failures are logged, never thrown").🛡️ Proposed fix
const sendHeartbeat = (): void => { - const snapshot = resources.healthSnapshot(); - void queue.workerHeartbeat(native.id, snapshot && JSON.stringify(snapshot)).catch((error) => { - log.debug(() => "worker heartbeat failed", error); - }); + try { + const snapshot = resources.healthSnapshot(); + void queue.workerHeartbeat(native.id, snapshot && JSON.stringify(snapshot)).catch((error) => { + log.debug(() => "worker heartbeat failed", error); + }); + } catch (error) { + log.debug(() => "worker heartbeat failed", error); + } };🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@sdks/node/src/worker.ts` around lines 213 - 221, The sendHeartbeat function in worker.ts only catches errors from queue.workerHeartbeat, so synchronous failures from resources.healthSnapshot() or JSON.stringify(snapshot) can still escape the setInterval callback. Wrap the entire body of sendHeartbeat in a try/catch, keep the existing log.debug error reporting for both sync and async failures, and ensure the function continues to match the “Failures are logged, never thrown” behavior.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Outside diff comments:
In `@sdks/node/src/worker.ts`:
- Around line 213-221: The sendHeartbeat function in worker.ts only catches
errors from queue.workerHeartbeat, so synchronous failures from
resources.healthSnapshot() or JSON.stringify(snapshot) can still escape the
setInterval callback. Wrap the entire body of sendHeartbeat in a try/catch, keep
the existing log.debug error reporting for both sync and async failures, and
ensure the function continues to match the “Failures are logged, never thrown”
behavior.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 74c20324-f042-4126-9497-6ed33f05f1ba
📒 Files selected for processing (4)
sdks/node/src/resources/health.tssdks/node/src/resources/runtime.tssdks/node/src/worker.tssdks/node/test/resources/health.test.ts
🚧 Files skipped from review as they are similar to previous changes (3)
- sdks/node/test/resources/health.test.ts
- sdks/node/src/resources/health.ts
- sdks/node/src/resources/runtime.ts
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Stacked on #377 — retargets to master automatically when it merges.
Summary
queue.resource(name, factory, { healthCheck, healthCheckIntervalMs, maxRecreationAttempts })(worker scope only — registering them on task/pooled scope throws). AHealthCheckerdaemon wakes every 500ms (unref'd, no-op when nothing asks for checks) and runs each resource's check on its own schedule; a failed or throwing check triggers recreation (dispose old instance, rebuild via the normal path); when recreation fails and the failure budget (default 3) is spent, the resource is marked permanently unhealthy for the rest of the worker run and every later resolve rejects withResourceUnavailableError.resource_health = NULLon every beat, so any JS writer would flap. The tick moves to JS — a 5s unref'd interval calls the new nativeworkerHeartbeat(workerId, resourceHealth)(heartbeat + best-effort dead-worker reap), carrying{name: "healthy"|"unhealthy"}per registered resource. The Rust lifecycle task now only registers (advertising sorted resource names on the cross-SDK wire shape) and unregisters.listWorkers()rows now surfaceresourcesandresourceHealth.Native API
JsQueue.workerHeartbeat(workerId, resourceHealth?) → Promise<string[]>(reaped worker ids)JsWorker.idgetter;WorkerOptions.resources?: string[];JsWorkerRow.resources?/resourceHealth?Notes
Test plan
test/resources/health.test.ts: recreate on failing check, exhaustion → permanently unhealthy, throwing check treated unhealthy, start() no-op without checks, registration scope guard, worker integration (listWorkers shows healthy resourceHealth JSON).cargo check/clippy -D warnings/fmt --checkon the native crate;pnpm typecheck,pnpm lint, full suite 269 tests green.Summary by CodeRabbit
healthCheck,healthCheckIntervalMs,maxRecreationAttempts) with automatic recreation on failure.