Queued message is stranded when it is enqueued between the holder's last dequeue and its lock release

[salmundani]

Bug Description

This an AI-generated report, let me know if you need any help reproducing/understanding it

Summary

With the queue, burst or debounce concurrency strategies, a message can be left in the queue and never handled. This happens when it arrives while another handler holds the thread lock, and it gets enqueued after the holder's final (empty) dequeue but before the holder releases the lock.

Nothing dequeues that thread again. The message waits until another message arrives on the same thread, or until queueEntryTtlMs passes, at which point it is discarded. For a user this looks like the bot ignoring a message they sent just as the previous reply finished.

• Package: chat
• Version: 4.40.0. main has the same code as of 2026-09-14.
• Strategies affected: queue, burst, debounce
• State adapters: the bug is in core, so any adapter shows it. We hit it with a Postgres-backed adapter and reproduced it with gating over the Supabase adapter; the repro below uses @chat-adapter/state-memory.

Where it happens

The relevant code is in packages/chat/src/chat.ts.

The holder runs handleQueueOrDebounce → withHeldLock → drainQueue (or debounceLoop):

// drainQueue
const entry = await this._stateAdapter.dequeue(lockKey);   // (1) returns null
...
if (pending.length === 0) {
  return;                                                  // (2) leave the drain
}

// withHeldLock
} finally {
  await heartbeat.stop();
  await this._stateAdapter.releaseLock(lock);              // (3) release
}

The arriving message runs handleQueueOrDebounce too:

const lock = await this._stateAdapter.acquireLock(lockKey, DEFAULT_LOCK_TTL_MS); // (a) null: lock is held
if (!lock) {
  const depth = await this._stateAdapter.queueDepth(lockKey);                    // (b)
  await this._stateAdapter.enqueue(lockKey, { message, ... }, effectiveMaxSize); // (c)
  return;                                                                        // (d)
}

The losing interleaving is:

holder:   (1) dequeue → null ─ (2) return ─────────────── (3) releaseLock
arriving:       (a) acquireLock → null ─ (b) ─ (c) enqueue ─ (d) return

After this, no lock is held, the queue holds one live entry, and nobody is draining. debounceLoop has the same gap: its final iteration breaks on an empty dequeue before withHeldLock releases the lock.

The window is several round trips to the state backend: dequeue, heartbeat stop and release on one side, acquire, depth and enqueue on the other. With a networked backend that is tens of milliseconds. It is rare with human-paced messages, but it happens reliably when a message arrives as a turn finishes, for example a user replying the moment the answer appears.

Steps to Reproduce

This is deterministic: it gates the state adapter to force the interleaving above.

import {
  Chat,
  Message,
  type Adapter,
  type Lock,
  type QueueEntry,
  type StateAdapter,
} from "chat";
import { createMemoryState } from "@chat-adapter/state-memory";

function gate() {
  let open!: () => void;
  const opened = new Promise<void>((resolve) => (open = resolve));
  return { opened, open };
}

const adapter = {
  name: "fake",
  userName: "bot",
  channelIdFromThreadId: (id: string) => id.split(":").slice(0, 2).join(":"),
  isDM: () => true,
} as unknown as Adapter;

const dm = (threadId: string, id: string, text: string) =>
  new Message({
    id,
    threadId,
    text,
    formatted: { type: "root", children: [] },
    raw: null,
    author: { userId: "U1", userName: "user", fullName: "User", isBot: false, isMe: false },
    metadata: { dateSent: new Date(), edited: false },
    attachments: [],
  });

const threadId = "fake:D1:1";
const first = dm(threadId, "m1", "first");
const second = dm(threadId, "m2", "second");

const real = createMemoryState();
// processMessage skips the webhook path, which is what connects state.
await real.connect();
const holderDrained = gate(); // holder's post-turn dequeue came back empty
const secondQueued = gate();  // second message is in the queue
let turnDone = false;

// Layered over the real adapter so every method not overridden here still works.
const state: StateAdapter = Object.assign(Object.create(real), {
  async dequeue(t: string) {
    const entry = await real.dequeue(t);
    if (!entry && turnDone) holderDrained.open();
    return entry;
  },
  async enqueue(t: string, entry: QueueEntry, max: number) {
    if (entry.message.id === second.id) await holderDrained.opened; // enqueue after (1)
    const depth = await real.enqueue(t, entry, max);
    if (entry.message.id === second.id) secondQueued.open();
    return depth;
  },
  async releaseLock(lock: Lock) {
    await secondQueued.opened; // release after (c)
    await real.releaseLock(lock);
  },
});

const chat = new Chat({
  userName: "bot",
  adapters: { fake: adapter },
  state,
  concurrency: { strategy: "queue" }, // "burst" behaves the same
  logger: "silent",
});

const handled: string[] = [];
let secondTask: Promise<void> | undefined;

chat.onDirectMessage(async (_thread, message) => {
  handled.push(message.text);
  if (message.id === first.id) {
    // Arrives while `first` holds the lock, so it takes the enqueue path.
    secondTask = chat.processMessage(adapter, threadId, second);
    turnDone = true;
  }
});

await chat.processMessage(adapter, threadId, first);
await secondTask;

console.log(handled);                          // ["first"]  expected ["first", "second"]
console.log(await real.queueDepth(threadId));  // 1          expected 0

Expected Behavior

both messages are handled and the queue ends up empty.

Actual Behavior

only first is handled. second stays queued with no lock held, until another message arrives on the thread or the entry expires.

Code Sample

See above

Chat SDK Version

4.40.0

Node.js Version

24.16

Platform Adapter

WhatsApp, Slack

Operating System

Linux

Additional Context

I'm the code from Supabase Edge Functions (Deno), but I can repro the issue in Node 24.16