Skip to content
This repository was archived by the owner on Mar 17, 2026. It is now read-only.

Commit 52ea441

Browse files
authored
fix: don't crash if an already-drained/removed queue gets flushed again (#1747)
1 parent 2ec6dc9 commit 52ea441

2 files changed

Lines changed: 19 additions & 0 deletions

File tree

src/publisher/message-queues.ts

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -354,6 +354,14 @@ export class OrderedQueue extends MessageQueue {
354354
* @fires OrderedQueue#drain
355355
*/
356356
async publish(): Promise<void> {
357+
// If there's nothing to flush, don't try, just short-circuit to the drain event.
358+
// This can happen if we get a publish() call after already being drained, in
359+
// the case that topic.flush() pulls a reference to us before we get deleted.
360+
if (!this.batches.length) {
361+
this.emit('drain');
362+
return;
363+
}
364+
357365
this.inFlight = true;
358366

359367
if (this.pending) {

test/publisher/message-queues.ts

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -729,6 +729,17 @@ describe('Message Queues', () => {
729729

730730
assert.strictEqual(spy.callCount, 1);
731731
});
732+
733+
it('should emit "drain" if already empty on publish', async () => {
734+
const spy = sandbox.spy();
735+
sandbox.stub(queue, '_publish').resolves();
736+
737+
queue.on('drain', spy);
738+
await queue.publish();
739+
await queue.publish();
740+
741+
assert.strictEqual(spy.callCount, 2);
742+
});
732743
});
733744

734745
describe('resumePublishing', () => {

0 commit comments

Comments
 (0)