fix(stability): close 8 audit + destruction-test bugs (RED→GREEN repros)

End-to-end audit + adversarial destruction test of v2.8.13. The data plane was already bulletproof (exactly-once through a SIGKILL flood, zero corruption, lossless crash recovery, bad-input isolation); these are feature-conditional control-plane / resource-hygiene defects. Each fix ships a committed reproduction test.

- concurrency slot leak on lock expiry -> setConcurrency(N) queue wedged to 0 (lockManager: releaseJobResources on requeue/DLQ)
- dependency children orphaned for undefined-result (across restart) and removeOnComplete parents (depCompletions set + state='completed' recovery)
- addBulk/PUSHB ignored durable -> batch durable now written immediately in one transaction (push.ts, sqlite.insertJobsBatch)
- pool socket drop re-dispatched in-flight jobs (double exec) -> releaseClientJobs skips renewalCount>0; worker renews just-pulled locks immediately
- Worker.close() hang on group-limited buffered jobs -> requeue buffered, drain on activeJobs only, force-close pre-empts
- worker not re-registered after TCP reconnect -> tcpPool.onReconnect + worker re-register (visibility only)
- moveToWaitingChildren stranded job -> getJob/getJobByCustomId/cancelJob consult waitingChildren
- perQueueMetrics unbounded -> LRU-bounded + freed by obliterate

Verified: tsc clean; unit 5640/5641, TCP 59/59, embedded 272/273 (2 pre-existing timing flakes pass in isolation); 9 repro files (20 tests) green; 161/161 disconnect-related suites; skeptic PASS.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: egeominotti

docs/src/content/docs/changelog.md

@@ -10,6 +10,21 @@ head:
 
 All notable changes to bunqueue are documented here.
 
+## [2.8.14] - 2026-06-15
+
+### Fixed — 8 stability bugs from an end-to-end audit + destruction test (each with a RED→GREEN reproduction test)
+
+The data plane was already bulletproof under the destruction test (exactly-once held through a SIGKILL flood, zero corruption, lossless crash recovery, bad-input isolation). These fixes close feature-conditional defects in the control plane and resource hygiene. No change to data correctness or process stability for the default producer/consumer path.
+
+- **Concurrency slot leak on lock expiry** (`lockManager.ts`): `requeueExpiredJob` / `handleMaxStallsExceeded` now call `shard.releaseJobResources()` before re-queue/DLQ, mirroring the stall-detection paths. Previously a queue with `setConcurrency(N)` permanently wedged (throughput → 0) after N lock expiries under worker churn.
+- **Dependency children orphaned** (`ack.ts`, `ackHelpers.ts`, `dependencyProcessor.ts`, `push.ts`, `backgroundTasks.ts`, `sqlite.ts`): a child `dependsOn` a parent that returned `undefined` (across a restart) or had `removeOnComplete: true` was silently never run and dropped after 1h. Added a bounded `depCompletions` set for removeOnComplete parents and made dependency recovery recognize `state='completed'` rows (not only `job_results`). Fixes late-dependent ordering too.
+- **`addBulk` / PUSHB ignored `durable`** (`push.ts`, `sqlite.ts`): durable batch jobs sat in the 10ms write buffer instead of being written immediately like a single durable push. `insertJobsBatch(jobs, durable)` now writes the durable subset to disk atomically (single transaction), bypassing the buffer.
+- **Pool socket drop re-dispatched in-flight jobs** (`clientTracking.ts`, `worker.ts`): with `poolSize > 1`, dropping the connection that pulled a job re-queued a job a live worker was still running (double execution). `releaseClientJobs` now skips jobs whose lock was renewed (`renewalCount > 0`); the worker renews just-pulled locks immediately so the window cannot open.
+- **`Worker.close()` hang on buffered jobs** (`worker.ts`): a graceful close with group-limited buffered jobs hung forever; `close(true)` could not pre-empt it. Buffered (pulled-but-unstarted) jobs are now requeued on close, the drain waits only on genuinely in-flight jobs, and a force close pre-empts an in-progress graceful close.
+- **Worker not re-registered after a TCP reconnect** (`tcpPool.ts`, `worker.ts`): after a transient drop the worker vanished from `ListWorkers` / `getForQueue` while still consuming jobs. The pool now exposes `onReconnect()` and the worker re-registers on reconnect. (Visibility only — no data loss.)
+- **`moveToWaitingChildren` stranded the job** (`queryOperations.ts`, `jobManagement.ts`): a job moved to waiting-children was invisible to `getJob` and uncancellable. `getJob` / `getJobByCustomId` / `cancelJob` now consult `waitingChildren`.
+- **`perQueueMetrics` unbounded growth** (`queueManager.ts`, `cleanupTasks.ts`): the per-queue metrics map grew one permanent entry per distinct queue name and was not freed by `obliterate()`. It is now LRU-bounded and freed by `obliterate()`; cumulative counters survive a transient drain.
+
 ## [2.8.13] - 2026-06-15
 
 ### Fixed — explicit `job.moveToFailed(err)` now carries the stacktrace (#74 follow-up)

package.json

@@ -1,6 +1,6 @@
 {
   "name": "bunqueue",
-  "version": "2.8.13",
+  "version": "2.8.14",
   "description": "High-performance job queue for Bun & AI agents. SQLite persistence, cron scheduling, priorities, retries, DLQ, webhooks, native MCP server. Zero external dependencies.",
   "type": "module",
   "main": "dist/main.js",

src/application/cleanupTasks.ts

@@ -247,6 +247,11 @@ function cleanEmptyQueues(ctx: BackgroundContext): void {
       shard.clearQueueLimiters(queueName);
       shard.stallConfig.delete(queueName);
       shard.dlqConfig.delete(queueName);
+      // NOTE: perQueueMetrics is intentionally NOT pruned here — it is an
+      // LRU-bounded map and these counters are cumulative, so they must survive
+      // a transient drain (a busy queue momentarily empty must not reset to 0).
+      // obliterate() reclaims it explicitly; the LRU cap bounds growth for
+      // ephemeral/dynamically-named queues.
       ctx.dashboardEmit?.('queue:removed', { queue: queueName });
       ctx.unregisterQueueName(queueName);
     }

src/application/clientTracking.ts

@@ -62,6 +62,17 @@ export async function releaseClientJobs(clientId: string, ctx: LockContext): Pro
     const loc = ctx.jobIndex.get(jobId);
     if (loc?.type !== 'processing') continue;
 
+    // A job whose lock has been renewed since pull (renewalCount > 0) is being
+    // actively heartbeated by a live worker. With a pooled client, heartbeats
+    // travel on a DIFFERENT connection than the one that pulled, so THIS socket
+    // closing does not mean the worker died — re-queuing here would re-dispatch
+    // (double-execute) a job the worker still holds. Leave it; lock expiry /
+    // stall detection reclaims it if the worker truly stops heartbeating.
+    // A never-renewed lock (renewalCount === 0) keeps the original fast-recovery
+    // behavior: requeue immediately on disconnect.
+    const lock = ctx.jobLocks.get(jobId);
+    if (lock && lock.renewalCount > 0) continue;
+
     const procIdx = loc.shardIdx;
     const job = ctx.processingShards[procIdx].get(jobId);
     if (!job) continue;

src/application/contextFactory.ts

@@ -8,7 +8,7 @@ import type { JobLogEntry } from '../domain/types/worker';
 import type { Shard } from '../domain/queue/shard';
 import type { SqliteStorage } from '../infrastructure/persistence/sqlite';
 import type { RWLock } from '../shared/lock';
-import type { LRUMap, BoundedSet, BoundedMap } from '../shared/lru';
+import type { LRUMap, BoundedSet, BoundedMap, MapLike } from '../shared/lru';
 import type { WebhookManager } from './webhookManager';
 import type { WorkerManager } from './workerManager';
 import type { EventsManager } from './eventsManager';
@@ -34,6 +34,7 @@ export interface ContextDependencies {
   jobIndex: Map<JobId, JobLocation>;
   completedJobs: BoundedSet<JobId>;
   completedJobsData: BoundedMap<JobId, Job>;
+  depCompletions?: BoundedSet<JobId>;
   jobResults: LRUMap<JobId, unknown>;
   customIdMap: LRUMap<string, JobId>;
   jobLogs: LRUMap<JobId, JobLogEntry[]>;
@@ -55,7 +56,7 @@ export interface ContextDependencies {
   };
   startTime: number;
   maxLogsPerJob: number;
-  perQueueMetrics: Map<string, { totalCompleted: bigint; totalFailed: bigint }>;
+  perQueueMetrics: MapLike<string, { totalCompleted: bigint; totalFailed: bigint }>;
 }
 
 /** Callbacks needed for some contexts */
@@ -105,6 +106,7 @@ export class ContextFactory {
       processingLocks: this.deps.processingLocks,
       jobIndex: this.deps.jobIndex,
       completedJobs: this.deps.completedJobs,
+      depCompletions: this.deps.depCompletions,
       jobResults: this.deps.jobResults,
       customIdMap: this.deps.customIdMap,
       jobLogs: this.deps.jobLogs,
@@ -154,6 +156,7 @@ export class ContextFactory {
       shardLocks: this.deps.shardLocks,
       completedJobs: this.deps.completedJobs,
       completedJobsData: this.deps.completedJobsData,
+      depCompletions: this.deps.depCompletions,
       jobResults: this.deps.jobResults,
       customIdMap: this.deps.customIdMap,
       jobIndex: this.deps.jobIndex,
@@ -186,6 +189,7 @@ export class ContextFactory {
       processingLocks: this.deps.processingLocks,
       completedJobs: this.deps.completedJobs,
       completedJobsData: this.deps.completedJobsData,
+      depCompletions: this.deps.depCompletions,
       jobResults: this.deps.jobResults,
       jobIndex: this.deps.jobIndex,
       customIdMap: this.deps.customIdMap,

src/application/dependencyProcessor.ts

@@ -46,10 +46,16 @@ export async function processPendingDependencies(ctx: BackgroundContext): Promis
         const shard = ctx.shards[i];
         const jobsToPromote: Job[] = [];
 
-        // Check which jobs have all dependencies satisfied (inside lock)
+        // Check which jobs have all dependencies satisfied (inside lock).
+        // A removeOnComplete parent is not in completedJobs (its full record was
+        // dropped to bound memory), so also honor its bare-id depCompletions entry.
         for (const jobId of jobIdsToCheck) {
           const job = shard.waitingDeps.get(jobId);
-          if (job?.dependsOn.every((dep) => ctx.completedJobs.has(dep))) {
+          if (
+            job?.dependsOn.every(
+              (dep) => ctx.completedJobs.has(dep) || (ctx.depCompletions?.has(dep) ?? false)
+            )
+          ) {
             jobsToPromote.push(job);
           }
         }
@@ -61,6 +67,12 @@ export async function processPendingDependencies(ctx: BackgroundContext): Promis
       });
     })
   );
+
+  // NOTE: depCompletions is intentionally NOT pruned here. It is a FIFO
+  // BoundedSet (same cap as completedJobs), so it self-bounds. Pruning eagerly
+  // once "no waiters remain" would orphan a dependent pushed AFTER a
+  // removeOnComplete parent completed — exactly the symmetry completedJobs
+  // provides for normal parents (readiness holds for the whole bounded window).
 }
 
 /** Move jobs from waitingDeps to the active queue */

src/application/lockManager.ts

@@ -166,6 +166,10 @@ interface MaxStallsOptions {
 /** Move job to DLQ when max stalls exceeded */
 function handleMaxStallsExceeded(opts: MaxStallsOptions): void {
   const { jobId, job, lock, shard, ctx, now } = opts;
+  // Release the concurrency slot (+group+uniqueKey) acquired at pull before
+  // moving to DLQ — otherwise the slot leaks (mirrors
+  // stallDetection.moveStalliedJobToDlq).
+  shard.releaseJobResources(job.queue, job.uniqueKey, job.groupId);
   shard.addToDlq(job, FailureReason.Stalled, `Lock expired after ${lock.renewalCount} renewals`);
   ctx.jobIndex.set(jobId, { type: 'dlq', queueName: job.queue });
 
@@ -193,6 +197,10 @@ interface RequeueOptions {
 function requeueExpiredJob(opts: RequeueOptions): void {
   const { jobId, job, queue, idx, ctx, now } = opts;
   const shard = ctx.shards[idx];
+  // Release the concurrency slot (+group+uniqueKey) acquired at pull before
+  // re-pushing — otherwise the slot leaks and the queue wedges (mirrors
+  // stallDetection.retryStalliedJob).
+  shard.releaseJobResources(job.queue, job.uniqueKey, job.groupId);
   queue.push(job);
   const isDelayed = job.runAt > now;
   shard.incrementQueued(jobId, isDelayed, job.createdAt, job.queue, job.runAt);

src/application/operations/ack.ts

@@ -40,12 +40,14 @@ export interface AckContext {
   processingLocks: RWLock[];
   completedJobs: SetLike<JobId>;
   completedJobsData: MapLike<JobId, Job>;
+  /** Bare completion ids for removeOnComplete jobs so dependents can unblock */
+  depCompletions?: SetLike<JobId>;
   jobResults: MapLike<JobId, unknown>;
   jobIndex: Map<JobId, JobLocation>;
   customIdMap?: MapLike<string, JobId>;
   totalCompleted: { value: bigint };
   totalFailed: { value: bigint };
-  perQueueMetrics?: Map<string, { totalCompleted: bigint; totalFailed: bigint }>;
+  perQueueMetrics?: MapLike<string, { totalCompleted: bigint; totalFailed: bigint }>;
   broadcast: (event: {
     eventType: EventType;
     queue: string;
@@ -113,6 +115,11 @@ export async function ackJob(jobId: JobId, result: unknown, ctx: AckContext): Pr
   } else {
     ctx.jobIndex.delete(jobId);
     ctx.storage?.deleteJob(jobId);
+    // removeOnComplete drops the full job (index + data + persisted row) to bound
+    // memory, but dependent jobs gate readiness on completedJobs.has(parentId).
+    // Record the bare completion id (no payload) so dependents still unblock,
+    // without making the job appear in state/stats queries.
+    ctx.depCompletions?.add(jobId);
   }
 
   ctx.totalCompleted.value++;

src/application/operations/ackHelpers.ts

@@ -157,11 +157,13 @@ export interface FinalizeContext {
   storage: SqliteStorage | null;
   completedJobs: SetLike<JobId>;
   completedJobsData: MapLike<JobId, Job>;
+  /** Bare completion ids for removeOnComplete jobs so dependents can unblock */
+  depCompletions?: SetLike<JobId>;
   jobResults: MapLike<JobId, unknown>;
   jobIndex: Map<JobId, JobLocation>;
   customIdMap?: MapLike<string, JobId>;
   totalCompleted: { value: bigint };
-  perQueueMetrics?: Map<string, { totalCompleted: bigint; totalFailed: bigint }>;
+  perQueueMetrics?: MapLike<string, { totalCompleted: bigint; totalFailed: bigint }>;
   broadcast: (event: {
     eventType: EventType;
     queue: string;
@@ -241,6 +243,11 @@ export function finalizeBatchAck<T>(
     } else {
       ctx.jobIndex.delete(jobId);
       if (hasStorage) storage.deleteJob(jobId);
+      // removeOnComplete drops the full job to bound memory, but dependents gate
+      // readiness on completedJobs.has(parentId). Record the bare completion id
+      // (no payload) so dependent jobs still unblock, without surfacing the job
+      // in state/stats queries.
+      ctx.depCompletions?.add(jobId);
     }
   }
 

src/application/operations/jobManagement.ts

@@ -43,6 +43,16 @@ export async function cancelJob(jobId: JobId, ctx: JobManagementContext): Promis
         ctx.storage?.deleteJob(jobId);
         return { success: true, queueName: location.queueName };
       }
+      // Not in the run queue — it may be parked in waitingChildren (moved via
+      // moveToWaitingChildren, which already released its resources and does not
+      // track it in the queued counter, so do NOT decrement/release here).
+      const parked = shard.waitingChildren.get(jobId);
+      if (parked) {
+        shard.waitingChildren.delete(jobId);
+        ctx.jobIndex.delete(jobId);
+        ctx.storage?.deleteJob(jobId);
+        return { success: true, queueName: location.queueName };
+      }
       return { success: false, queueName: location.queueName };
     });