feat(node): operator admin surface - #557
Conversation
Worker.stop() is now idempotent: a repeated stop used to release a second resource lease and tear down a sibling worker's resources.
📝 WalkthroughWalkthroughThe Node SDK adds APIs for requeuing stuck jobs, analyzing interceptor outcomes, reloading resources, and shutting down queue-owned workers. Native bindings, resource dependency tracking, worker lifecycle handling, exports, documentation, changelogs, and tests are updated. ChangesNode queue administration
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)Resource reloadsequenceDiagram
participant Queue
participant ResourceRuntime
participant WorkerResource
Queue->>ResourceRuntime: reloadResources(names?)
ResourceRuntime->>ResourceRuntime: determine dependency order
ResourceRuntime->>WorkerResource: recreate or evict cached resources
WorkerResource-->>ResourceRuntime: reload result
ResourceRuntime-->>Queue: return result map
Queue shutdownsequenceDiagram
participant Queue
participant Worker
participant ResourceRuntime
Queue->>Worker: stop all tracked workers
Worker->>ResourceRuntime: release worker leases
ResourceRuntime-->>Worker: dispose scoped resources
Worker-->>Queue: resolve stop promise
Queue-->>Queue: resolve shutdown()
Possibly related PRs
Suggested labels: Suggested reviewers: 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Comment |
![coderabbitai[bot]](https://cdn.github.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
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)
399-417: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winInstall the stop promise before invoking teardown callbacks.
On Line 400,
stoppedis assigned only afterrunStop()returns.runStop()synchronously invokesonStoppedand emitsworker.stopped; either can reenterstop()whilestoppedis still unset. That starts a second teardown (and can recursively re-emitworker.stopped), defeating idempotency and potentially releasing another worker’s resource lease.Proposed fix
stop(): Promise<void> { - this.stopped ??= this.runStop(); - return this.stopped; + if (this.stopped) { + return this.stopped; + } + + let resolve!: () => void; + const stopped = new Promise<void>((done) => { + resolve = done; + }); + this.stopped = stopped; + + try { + void this.runStop().then(resolve, (error) => { + log.debug(() => "worker stop failed", error); + resolve(); + }); + } catch (error) { + log.debug(() => "worker stop failed", error); + resolve(); + } + return stopped; }🤖 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 399 - 417, Update Worker.stop() and runStop() so the shared stopped promise is created and assigned before any teardown callbacks or events can reenter stop(). Preserve idempotent behavior by having stop() return the already-installed promise, while moving synchronous teardown execution behind that promise’s initialization.
🧹 Nitpick comments (1)
sdks/node/src/interception.ts (1)
49-64: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winKeep
InterceptionAnalysisas a single canonical type.
sdks/node/src/queue.ts:49-64declares the same interface, while this file adds the exported copy.Queue.analyzeArguments()is therefore typed against a different declaration than the root SDK export, allowing the two shapes to diverge. Remove the duplicate fromqueue.tsand import this type there.🤖 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/interception.ts` around lines 49 - 64, Remove the duplicate InterceptionAnalysis interface declaration from Queue and update Queue.analyzeArguments() to import and use the canonical exported InterceptionAnalysis type from interception.ts, keeping the existing analysis shape and behavior unchanged.
🤖 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/queue.ts`:
- Around line 1643-1652: The Queue.shutdown method currently resolves before
worker handler invocations finish, despite documenting that tasks drain. Update
the shutdown lifecycle around Queue.shutdown and Worker.stop to track active
handler invocations and await their completion before resolving, while
preserving the existing no-op and repeated-stop behavior.
- Around line 1597-1601: Update analyzeArguments to pass a shallow copy of the
caller-provided args array into applyInterceptors, matching the copying behavior
used by real enqueue. Preserve the existing analysis result and outcomes
handling while ensuring mutating interceptors cannot alter the original args.
In `@sdks/node/src/resources/runtime.ts`:
- Around line 415-484: Update reload to await any already-in-flight worker
builds for the requested targets before calling dependencyOrder. Ensure
resolveWorker’s dependency tracking has completed before ordering, so
dependencies are visited before dependents during the same reload sweep.
Preserve existing reloadOne behavior and result handling.
In `@sdks/node/test/core/requeue.test.ts`:
- Around line 9-12: Update the afterEach cleanup hook to be asynchronous and
await worker.stop() before clearing the worker reference, ensuring teardown
completes before the hook finishes.
---
Outside diff comments:
In `@sdks/node/src/worker.ts`:
- Around line 399-417: Update Worker.stop() and runStop() so the shared stopped
promise is created and assigned before any teardown callbacks or events can
reenter stop(). Preserve idempotent behavior by having stop() return the
already-installed promise, while moving synchronous teardown execution behind
that promise’s initialization.
---
Nitpick comments:
In `@sdks/node/src/interception.ts`:
- Around line 49-64: Remove the duplicate InterceptionAnalysis interface
declaration from Queue and update Queue.analyzeArguments() to import and use the
canonical exported InterceptionAnalysis type from interception.ts, keeping the
existing analysis shape and behavior unchanged.
🪄 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: 67bc4d49-f81d-4e87-8988-5642674bb96d
📒 Files selected for processing (17)
CHANGELOG.mdcrates/taskito-node/src/queue/admin.rsdocs/content/docs/node/api-reference/queue/queues.mdxdocs/content/docs/node/api-reference/queue/resources.mdxdocs/content/docs/node/api-reference/queue/workers.mdxdocs/content/docs/resources/changelog.mdxdocs/content/docs/shared/guides/operations/troubleshooting.mdxsdks/node/src/index.tssdks/node/src/interception.tssdks/node/src/queue.tssdks/node/src/resources/runtime.tssdks/node/src/resources/types.tssdks/node/src/worker.tssdks/node/test/core/interceptors.test.tssdks/node/test/core/requeue.test.tssdks/node/test/resources/reload.test.tssdks/node/test/worker/shutdown.test.ts
onStopped and the worker.stopped listeners fire synchronously inside runStop, so a reentrant stop() saw no memoized promise and started a second teardown.
It stops dispatch and disposes worker resources; handlers already mid-flight are not awaited.
|
Addressed the review — 5 commits pushed (fd299ab). Outside-diff, Nitpick, duplicate The other four are answered on their threads: args copy (3b0ba6d), reload ordering race (5604b5c), shutdown drain wording (8075d3c), test teardown await (fd299ab). 631 tests / 89 files green, biome + tsc clean. |
kartikeya-27
changed the title
2 participants
Closes #523.
The core already had
Storage::requeue_stuck, but no Node binding reached it — so a Nodeoperator had no recovery path for a job stuck
Running(hung-but-heartbeating worker, hugetimeout, or a claim-less Running row). That is the functional gap here; the rest is wrapper
parity with the cross-SDK admin contract.
predicateStats()was already shipped in #543, so itneeded no work.
What's new
queue.requeueJob(id)runningjob back topending, releasing its execution claim atomically and preserving the retry budget.falsewhen the job is missing or isn't running.queue.reloadResources(names?)reloadable: trueresource; an unknown name reportsfalserather than throwing.queue.analyzeArguments(taskName, args)rejectedinstead of throwing, and the run leavesinterceptionStats()untouched.queue.shutdown()Worker.requeueJobcarries the double-execution caveat in its docstring, the API reference, and thetroubleshooting guide: only use it once the owning worker is confirmed dead or hung.
Notes for review
Worker.stop()was not idempotent. A second call released a second resource lease and couldtear down a sibling worker's resources. It is now memoized. This is a prerequisite for a
queue-level shutdown that may race a caller's own
worker.stop(), so it rides in that commit;the added regression test fails without the fix.
ResourceRuntime.workerCacheinsertion order isdependent-before-dependency —
resolveWorkercaches the pending promise before the factorybody runs — so reversing it is only correct for chains, not multi-root graphs.
reload()insteadrecords dependency edges in the
ctx.useclosure and walks them post-order, rebuilding adependency before the resources that resolved it.
builds fresh (idle instances disposed now, checked-out ones on release); task/request scopes are
a successful no-op, since nothing is cached to replace.
<SdkOnly>blocks ratherthan a
<CodeTabs>:SdkOnlymatches exactly one SDK id, and every existingCodeTabscarriesall three, so a two-tab one would render an empty panel for Java. It collapses into a
CodeTabsonce Java lands the same API.
Verification
pnpm test— 630 tests across 89 files green (16 new).cargo test --workspace,cargo clippy --workspace,cargo fmt --checkgreen.pnpm run typecheck+pnpm run lint(biome) green; docs lint, typecheck, and full build green.Summary by CodeRabbit
runningjobs back topending.reloadableresources with dependency-aware rebuild order.analyzeArgumentsto dry-run enqueue interceptors/guards, including redirects and structured rejection results.shutdown()to stop all queue workers together.