diff --git a/resources/IterableEventQueue.ts b/resources/IterableEventQueue.ts index ffc24a25ab..e2a9f0aec1 100644 --- a/resources/IterableEventQueue.ts +++ b/resources/IterableEventQueue.ts @@ -39,8 +39,21 @@ export class IterableEventQueue extends EventEmitter return new Promise((resolve) => { if (!this.queue || this.queue.length === 0) resolve(true); else { - this.once('drained', () => resolve(true)); - this.currentDrainResolver = resolve; + // The queue can also empty through paths that never emit 'drained' (the on('data') + // attach loop and the resolveNext bypass), so a waiter relying on the event alone + // can hang forever on an already-empty queue. Poll as a fallback wakeup. + const settle = (drained: boolean) => { + clearInterval(poll); + this.removeListener('drained', onDrained); + resolve(drained); + }; + const onDrained = () => settle(true); + this.once('drained', onDrained); + this.currentDrainResolver = settle; + const poll = setInterval(() => { + if (!this.queue || this.queue.length === 0) settle(true); + }, 100); + poll.unref?.(); if (!this.drainCloseListener) { this.drainCloseListener = true; this.on('close', () => { diff --git a/resources/Table.ts b/resources/Table.ts index 382afaebc1..200e95ea0f 100644 --- a/resources/Table.ts +++ b/resources/Table.ts @@ -327,6 +327,9 @@ export function makeTable(options) { static get expirationMS() { return expirationMs; } + static get evictionMS() { + return evictionMs; + } static dbisDB = dbisDb; static schemaDefined = schemaDefined; /** diff --git a/unitTests/resources/IterableEventQueue.test.js b/unitTests/resources/IterableEventQueue.test.js index 9a74961e9d..981bb3313c 100644 --- a/unitTests/resources/IterableEventQueue.test.js +++ b/unitTests/resources/IterableEventQueue.test.js @@ -49,4 +49,14 @@ describe('IterableEventQueue', () => { q.off('close', close); assert.equal(q.hasDataListeners, true, "removing a 'close' listener leaves data state intact"); }); + + it('waitForDrain resolves when the on(data) attach loop empties the queue without emitting drained', async () => { + const q = new IterableEventQueue(); + q.send({ n: 1 }); + const drained = q.waitForDrain(); + q.on('data', () => {}); // attach loop drains the buffered queue synchronously, no 'drained' emit + const result = await Promise.race([drained, new Promise((r) => setTimeout(() => r('hung'), 1000))]); + assert.equal(result, true, 'waiter must observe the queue emptying through the attach path'); + assert.equal(q.listenerCount('drained'), 0, 'poll-path settle must remove the drained listener'); + }); });