Agent this.queue is stuck

[cansirin]

Describe the bug

A single failing callback in _flushQueue() permanently poisons the entire queue. When a queued task's callback throws an error, the failing row is never dequeued from the SQLite table, and the error is silently swallowed. On the next flush, the same failing row is picked up again, creating an infinite retry loop that blocks all subsequent queued tasks.

To Reproduce

1. Create an Agent subclass with a method that can throw an error
2. Call this.queue("myMethod", payload) where myMethod will throw
3. Observe that _flushQueue() runs, the callback throws, and the error is caught by .catch() on line 712 with only a console.error
4. Observe that the failing row remains in cf_agents_queues because this.dequeue(row.id) on line 740 never executes
5. Next time _flushQueue() is triggered (e.g., by another queue() call), the same failing row is fetched first (ORDER BY created_at ASC), throws again, and blocks all items behind it

Expected behavior

• A failing callback should not block the entire queue. The failing item should either be dequeued (dead-lettered) or skipped so that subsequent items can still be processed.
• The error from _flushQueue() should propagate or be exposed in a meaningful way, rather than being silently swallowed with console.error.

Screenshots

N/A — the bug is in the source code of _flushQueue():

// line 717-745 in dist/src-i_UcyBYf.js
async _flushQueue() {
    if (this._flushingQueue) return;
    this._flushingQueue = true;
    while (true) {
        const result = this.sql`
          SELECT * FROM cf_agents_queues ORDER BY created_at ASC
        `;
        if (!result || result.length === 0) break;
        for (const row of result || []) {
            const callback = this[row.callback];
            if (!callback) {
                console.error(`callback ${row.callback} not found`);
                continue; // <-- missing callback is skipped, but...
            }
            const { connection, request, email } = agentContext.getStore() || {};
            await agentContext.run({ agent: this, connection, request, email }, async () => {
                await callback.bind(this)(JSON.parse(row.payload), row); // <-- if this throws...
                await this.dequeue(row.id); // <-- ...this never runs, row stays in the table
            });
        }
    }
    this._flushingQueue = false;
}

There are two issues:

1. No try-catch around the callback invocation — if callback.bind(this)(...) throws, this.dequeue(row.id) is skipped, the error breaks out of the while(true) loop, and the row permanently stays in cf_agents_queues.
2. Error is silently swallowed at the call site — this._flushQueue().catch((e) => { console.error(...) }) means the caller of queue() has no way to know the queue is stuck.

Version:

agents@0.3.10

Additional context

A minimal fix would be to wrap the callback invocation in a try-catch and dequeue (or mark as failed) the row regardless of whether the callback succeeds:

for (const row of result || []) {
    // ...
    try {
        await agentContext.run({...}, async () => {
            await callback.bind(this)(JSON.parse(row.payload), row);
        });
    } catch (e) {
        console.error(`Queue callback ${row.callback} failed for row ${row.id}:`, e);
    } finally {
        await this.dequeue(row.id);
    }
}

Ideally the SDK would also support a dead-letter / retry-with-limit mechanism so users can handle failed queue items gracefully rather than losing them silently. (linking issue #874 since its going to be resolved with this)

[threepointone]

can you try 0.4.0? I think I fixed this

[cansirin]

Yup, i'll try in a bit

[cansirin]

I just tried with my agent and checked the _flushQueue() implementation in agents@0.4.0 and it looks like the same issue is still present.

await agentContext.run({...}, async () => {
    await callback.bind(this)(JSON.parse(row.payload), row);
    await this.dequeue(row.id);
});

There's still no try-catch around the callback invocation, so if it throws:

1. this.dequeue(row.id) is skipped, leaving the failing row in cf_agents_queues
2. The error propagates out of the while(true) loop, so this._flushingQueue is never reset to false — which actually locks the queue permanently for the lifetime of that DO instance

[threepointone]

Ugh sorry, I think I have this in my retry proposal. I'll have a look tomorrow.

[threepointone]

/bonk can you try to make a pr with a fix?

[ask-bonk]

Created PR #890

github run