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 d49083cfd1..8012b74d2f 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; /** @@ -3366,7 +3369,7 @@ export function makeTable(options) { logger.error?.('Error getting history entry', auditRecord.localTime, error); } } - for (let i = history.length; i > 0; ) { + for (let i = history.length; i > 0;) { send(history[--i]); } // Use the latest record cursor saw (history[0] = most recent due to reverse @@ -3475,7 +3478,7 @@ export function makeTable(options) { } else break; if (count) count--; } while (nextTime > startTime && count !== 0); - for (let i = history.length; i > 0; ) { + for (let i = history.length; i > 0;) { send(history[--i]); } } @@ -3701,12 +3704,10 @@ export function makeTable(options) { ); break; case 'ID': - if ( - !( - typeof value === 'string' || - (value?.length > 0 && value.every?.((value) => typeof value === 'string')) - ) - ) + if (!( + typeof value === 'string' || + (value?.length > 0 && value.every?.((value) => typeof value === 'string')) + )) (validationErrors || (validationErrors = [])).push( `Value ${stringify(value)} in property ${name} must be a string, or an array of strings` ); diff --git a/unitTests/resources/IterableEventQueue.test.js b/unitTests/resources/IterableEventQueue.test.js index 9f2efe96b9..fb6e6ecf98 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'); + }); });