Serialize reconnect drains before global recovery

drainReconnectQueue only guards concurrency per accountId, but _drainReconnectQueueCore then calls recoverPendingDeliveries, which scans and delivers the entire queue (not just that account). If two WhatsApp accounts reconnect close together (or one reconnect drain overlaps the startup recovery in src/gateway/server.impl.ts), both workers can load the same pending entry and call deliver before either ackDelivery runs, resulting in duplicate outbound sends. The lock needs to be global for queue recovery, or recovery must be limited to the selected entry IDs.

Useful? React with 👍 / 👎.

Fixed in the follow-up commit. Replaced recoverPendingDeliveries() with a targeted delivery loop that only processes the eligible entries scoped to opts.accountId. This eliminates the cross-account delivery race and the startup-recovery overlap entirely — no global serialization needed since we never touch other accounts' entries.

recoverPendingDeliveries is not scoped to the draining account

After resetting the backoff for entries belonging to opts.accountId, recoverPendingDeliveries is called with no account filter — it processes every pending entry across all accounts and channels.

In a multi-account WhatsApp setup this creates a correctness hazard: reconnecting account A triggers delivery attempts for account B's queued entries (if their backoff period has already expired). Those attempts fail with "No active WhatsApp Web listener (account: B)", which calls failDelivery and increments account B's retryCount. If account A flaps rapidly (multiple reconnects), each reconnect fires another full recovery run, and account B's entries can exhaust MAX_RETRIES and be moved to failed/ before account B ever gets a chance to reconnect.

A targeted fix would be to filter recoverPendingDeliveries's candidate list to only the reset entry IDs, or to inline the delivery loop rather than delegating to the full recovery function:

// Instead of a full recoverPendingDeliveries call, only retry the entries
// that were just reset for this account.
for (const entry of eligible) {
  try {
    await deliver({ cfg: opts.cfg, ...entry, skipQueue: true });
    await ackDelivery(entry.id, opts.stateDir);
  } catch (err) {
    const errMsg = err instanceof Error ? err.message : String(err);
    if (isPermanentDeliveryError(errMsg)) {
      await moveToFailed(entry.id, opts.stateDir).catch(() => {});
    } else {
      await failDelivery(entry.id, errMsg, opts.stateDir).catch(() => {});
    }
  }
}

If delegating to recoverPendingDeliveries is preferred for its permanent-error and max-retries handling, consider adding a per-account or per-entry-id inclusion filter to that function so the drain stays scoped.

This is a comment left during a code review.
Path: src/infra/outbound/delivery-queue.ts
Line: 515-521

Comment:
**`recoverPendingDeliveries` is not scoped to the draining account**

After resetting the backoff for entries belonging to `opts.accountId`, `recoverPendingDeliveries` is called with no account filter — it processes every pending entry across all accounts and channels.

In a multi-account WhatsApp setup this creates a correctness hazard: reconnecting account A triggers delivery attempts for account B's queued entries (if their backoff period has already expired). Those attempts fail with `"No active WhatsApp Web listener (account: B)"`, which calls `failDelivery` and increments account B's `retryCount`. If account A flaps rapidly (multiple reconnects), each reconnect fires another full recovery run, and account B's entries can exhaust `MAX_RETRIES` and be moved to `failed/` before account B ever gets a chance to reconnect.

A targeted fix would be to filter `recoverPendingDeliveries`'s candidate list to only the reset entry IDs, or to inline the delivery loop rather than delegating to the full recovery function:

```ts
// Instead of a full recoverPendingDeliveries call, only retry the entries
// that were just reset for this account.
for (const entry of eligible) {
  try {
    await deliver({ cfg: opts.cfg, ...entry, skipQueue: true });
    await ackDelivery(entry.id, opts.stateDir);
  } catch (err) {
    const errMsg = err instanceof Error ? err.message : String(err);
    if (isPermanentDeliveryError(errMsg)) {
      await moveToFailed(entry.id, opts.stateDir).catch(() => {});
    } else {
      await failDelivery(entry.id, errMsg, opts.stateDir).catch(() => {});
    }
  }
}
```

If delegating to `recoverPendingDeliveries` is preferred for its permanent-error and max-retries handling, consider adding a per-account or per-entry-id inclusion filter to that function so the drain stays scoped.

How can I resolve this? If you propose a fix, please make it concise.

Fixed. The recoverPendingDeliveries call has been replaced with the targeted loop you suggested — delivering only the eligible entries for opts.accountId. Each entry is delivered with skipQueue: true, then acked or failed individually with permanent-error handling.

Unnecessary export of resolveQueueDir

resolveQueueDir is called only from within this same file (_drainReconnectQueueCore at line 492). The new test file constructs the queue path manually with path.join(tmpDir, "delivery-queue") rather than importing this function, so the export isn't consumed externally. Exporting it unnecessarily widens the module's public API surface.

This is a comment left during a code review.
Path: src/infra/outbound/delivery-queue.ts
Line: 56-59

Comment:
**Unnecessary export of `resolveQueueDir`**

`resolveQueueDir` is called only from within this same file (`_drainReconnectQueueCore` at line 492). The new test file constructs the queue path manually with `path.join(tmpDir, "delivery-queue")` rather than importing this function, so the export isn't consumed externally. Exporting it unnecessarily widens the module's public API surface.

```suggestion
function resolveQueueDir(stateDir?: string): string {
```

How can I resolve this? If you propose a fix, please make it concise.

Fixed. resolveQueueDir is no longer exported — the export keyword has been removed as suggested.

Remove hard 30s cutoff from reconnect queue drain

The reconnect drain always calls recoverPendingDeliveries with maxRecoveryMs: 30_000, so a larger backlog can hit the deadline and leave remaining entries in delivery-queue/ with no follow-up pass. Because this drain is only triggered on reconnect, those untouched messages can sit indefinitely until another reconnect or a process restart, which reintroduces the “stuck until restart” behavior for high-volume disconnect windows.

Useful? React with 👍 / 👎.

Fixed. As part of a previous commit, recoverPendingDeliveries is no longer called in the reconnect drain. It was replaced with a targeted per-entry delivery loop, so the hard 30s cutoff is no longer applicable here.

Revalidate entry before rewriting reconnect-reset file

drainReconnectQueue works from a one-time loadPendingDeliveries snapshot, then unconditionally rewrites each eligible {id}.json file here. Because startup recovery is still launched concurrently in src/gateway/server.impl.ts (recoverPendingDeliveries runs in a detached async task), another worker can deliver+ack the same entry between snapshot and this rename; this code will recreate the deleted queue file from stale data and then deliver it again in the later loop, causing duplicate outbound sends. Add a per-entry existence/ownership check (or shared lock) before rewriting/sending to avoid resurrecting already-acked entries.

Useful? React with 👍 / 👎.

The race you describe is real — between the loadPendingDeliveries snapshot and the atomic tmp → rename write, a concurrent recoverPendingDeliveries run could have already acked and deleted that {id}.json. Renaming our tmp file into the now-absent path resurrects a stale entry and causes a duplicate send.

The drainInProgress guard prevents two drain calls from racing each other, but doesn't cover the startup-recovery path running concurrently.

Fix: add a per-entry existence check before the rewrite, and skip if the file is already gone:

try {
  await fs.promises.access(filePath);
} catch {
  opts.log.info(`reconnect drain: entry ${entry.id} already acked by concurrent recovery, skipping`);
  continue;
}
// proceed with tmp write + rename

Also wrap ackDelivery in the delivery loop to tolerate ENOENT (the moveToFailed paths already use .catch(() => {}) — ack should be the same). Will address in a follow-up commit.

Handle missing expired entries without aborting the drain

This moveToFailed call is not guarded, so an ENOENT (for example, when another concurrent recovery already moved/acked the same entry after the initial snapshot) throws out of _drainReconnectQueueCore and aborts the whole reconnect drain. Since the caller in monitor.ts only logs and does not retry immediately, one disappearing expired entry can prevent all remaining eligible messages in that batch from being replayed until a later reconnect/restart.

Useful? React with 👍 / 👎.

Good catch. The moveToFailed call in the expired-entries loop has no error guard, so an ENOENT (entry already moved or acked by a concurrent recovery) throws out of _drainReconnectQueueCore and kills the rest of the drain — all remaining eligible entries in that batch are silently dropped until the next reconnect.

The delivery loop already uses .catch(() => {}) on its moveToFailed calls, so the fix is to apply the same pattern to the expired loop:

for (const entry of expired) {
  await moveToFailed(entry.id, opts.stateDir).catch((err) => {
    if ((err as NodeJS.ErrnoException).code !== 'ENOENT') throw err;
    // Already gone — another recovery handled it
  });
  opts.log.warn(`reconnect drain: expired entry ${entry.id} (TTL/max-retries)`);
}

This makes the drain resilient to disappearing entries without masking real I/O errors. Will include in the same follow-up commit as the P1 fix.

Skip ENOENT entries in the reconnect send pass

When rename(tmp, filePath) hits ENOENT, the code logs that the entry was already acked and continues, but the entry remains in the original eligible array and is still sent in this later loop. In the overlap case with startup recovery (src/gateway/server.impl.ts runs recoverPendingDeliveries concurrently), that path can re-send messages that were already delivered/acked by the other worker. Fresh evidence in this revision is that ENOENT handling was added, but eligible is never pruned before this deliver loop, so the skipped entry is still delivered.

Useful? React with 👍 / 👎.

Confirmed — this is a real bug. The continue in the rename-ENOENT branch skips the writeFile/rename but does not remove the entry from the eligible slice, so the subsequent deliver loop still sends it, creating a double-delivery risk when startup recoverPendingDeliveries has already handled the entry concurrently.

Fix: collect the IDs that hit ENOENT during the reset pass into a Set, then filter eligible before entering the deliver loop:

const skippedIds = new Set<string>();

for (const entry of eligible) {
  // ... writeFile + rename ...
  try {
    await fs.promises.rename(tmp, filePath);
  } catch (err: unknown) {
    await fs.promises.unlink(tmp).catch(() => {});
    if ((err as NodeJS.ErrnoException).code === "ENOENT") {
      opts.log.info(`reconnect drain: entry ${entry.id} already acked by concurrent recovery, skipping`);
      skippedIds.add(entry.id);
      continue;
    }
    throw err;
  }
}

// Only deliver entries that were successfully reset
for (const entry of eligible.filter(e => !skippedIds.has(e.id))) {
  await deliver({ cfg: opts.cfg, ...entry, skipQueue: true });
}

Will push this fix.

Revalidate queue entry before reconnect delivery

drainReconnectQueue sends every item from the initial loadPendingDeliveries snapshot without confirming the entry is still pending, so it can race with startup recovery (recoverPendingDeliveries in src/gateway/server.impl.ts) and deliver the same message twice when the other worker acks between snapshot and send. In that overlap, ackDelivery treats missing .json files as success, so duplicates are not detected. Fresh evidence in this revision is that the reconnect path now iterates matchingEntries directly and calls deliver(...) without any per-entry file revalidation step.

Useful? React with 👍 / 👎.

Re-attempt entries skipped by concurrent recovery

When drainReconnectQueue hits an entry already claimed by startup recovery, it logs and continues without revisiting that ID in the same drain pass. If the startup attempt later fails transiently (for example, during listener handoff/race windows), that entry stays queued but this reconnect-triggered drain has already moved on, so the message waits for a later reconnect or process restart instead of being recovered immediately. Consider re-queuing skipped IDs for a follow-up pass (or reloading pending entries once claims are released) so the drain does not silently drop recovery opportunities.

Useful? React with 👍 / 👎.