cherry-pick: fix(queue): waitForDrain poll fallback for drains that never emit (conflicts → v5.1)

resources/IterableEventQueue.ts

@@ -39,8 +39,21 @@ export class IterableEventQueue<Event extends object = any> 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', () => {

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`
 									);

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');
+	});
 });