From 4fab4d5d5d8e3118cf57c87b7e5e1ada7755ed0e Mon Sep 17 00:00:00 2001
From: Golo Roden <golo.roden@thenativeweb.io>
Date: Mon, 19 Oct 2026 17:46:11 +0200
Subject: [PATCH] chore: Stop observing in the tests once the expected events
 arrived, instead of after a fixed time.

Four tests of observeEvents aborted observing after 100 milliseconds and then
counted the events, so on a busy machine they could stop before the events
arrived. They now stop once the expected number of events has arrived, and
fail with the number they observed if the events have not arrived within 10
seconds.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QykegdHe7vhHMBN3kWxgfH
---
 src/observeEvents.test.ts | 156 +++++++++++++++++++++-----------------
 1 file changed, 86 insertions(+), 70 deletions(-)

diff --git a/src/observeEvents.test.ts b/src/observeEvents.test.ts
index 0bb4cef..9e731c7 100644
--- a/src/observeEvents.test.ts
+++ b/src/observeEvents.test.ts
@@ -7,6 +7,42 @@ import { getImageTag } from './getImageTag.js';
 import { containerStartTimeout } from './testing/containerStartTimeout.js';
 import { integrationTimeout } from './testing/integrationTimeout.js';
 
+// The events of a test are written before it observes them, so they arrive
+// at once. A test stops observing once the expected number of events has
+// arrived, and fails if they have not arrived within this many milliseconds.
+const observeTimeout = 10_000;
+
+const observeUntil = async (
+	observe: (signal: AbortSignal) => AsyncGenerator<Event, void, void>,
+	expectedCount: number,
+): Promise<Event[]> => {
+	const events: Event[] = [];
+	const controller = new AbortController();
+	const timer = setTimeout(() => {
+		controller.abort();
+	}, observeTimeout);
+
+	try {
+		for await (const event of observe(controller.signal)) {
+			events.push(event);
+
+			if (events.length === expectedCount) {
+				break;
+			}
+		}
+	} finally {
+		clearTimeout(timer);
+	}
+
+	if (events.length < expectedCount) {
+		throw new Error(
+			`Observed ${events.length} of ${expectedCount} events within ${observeTimeout} milliseconds.`,
+		);
+	}
+
+	return events;
+};
+
 suite('observeEvents', () => {
 	let container: Container;
 
@@ -72,22 +108,17 @@ suite('observeEvents', () => {
 
 			await client.writeEvents([firstEvent, secondEvent]);
 
-			const controller = new AbortController();
-
-			setTimeout(() => {
-				controller.abort();
-			}, 100);
-
-			const eventsObserved: Event[] = [];
-			for await (const event of client.observeEvents(
-				'/test',
-				{
-					recursive: false,
-				},
-				controller.signal,
-			)) {
-				eventsObserved.push(event);
-			}
+			const eventsObserved = await observeUntil(
+				signal =>
+					client.observeEvents(
+						'/test',
+						{
+							recursive: false,
+						},
+						signal,
+					),
+				2,
+			);
 
 			assert.equal(eventsObserved.length, 2);
 		},
@@ -116,22 +147,17 @@ suite('observeEvents', () => {
 
 		await client.writeEvents([firstEvent, secondEvent]);
 
-		const controller = new AbortController();
-
-		setTimeout(() => {
-			controller.abort();
-		}, 100);
-
-		const eventsObserved: Event[] = [];
-		for await (const event of client.observeEvents(
-			'/',
-			{
-				recursive: true,
-			},
-			controller.signal,
-		)) {
-			eventsObserved.push(event);
-		}
+		const eventsObserved = await observeUntil(
+			signal =>
+				client.observeEvents(
+					'/',
+					{
+						recursive: true,
+					},
+					signal,
+				),
+			2,
+		);
 
 		assert.equal(eventsObserved.length, 2);
 	});
@@ -159,23 +185,18 @@ suite('observeEvents', () => {
 
 		await client.writeEvents([firstEvent, secondEvent]);
 
-		const controller = new AbortController();
-
-		setTimeout(() => {
-			controller.abort();
-		}, 100);
-
-		const eventsObserved: Event[] = [];
-		for await (const event of client.observeEvents(
-			'/test',
-			{
-				recursive: false,
-				lowerBound: { id: '1', type: 'inclusive' },
-			},
-			controller.signal,
-		)) {
-			eventsObserved.push(event);
-		}
+		const eventsObserved = await observeUntil(
+			signal =>
+				client.observeEvents(
+					'/test',
+					{
+						recursive: false,
+						lowerBound: { id: '1', type: 'inclusive' },
+					},
+					signal,
+				),
+			1,
+		);
 
 		assert.equal(eventsObserved.length, 1);
 		const [eventObserved] = eventsObserved;
@@ -206,27 +227,22 @@ suite('observeEvents', () => {
 
 		await client.writeEvents([firstEvent, secondEvent]);
 
-		const controller = new AbortController();
-
-		setTimeout(() => {
-			controller.abort();
-		}, 100);
-
-		const eventsObserved: Event[] = [];
-		for await (const event of client.observeEvents(
-			'/test',
-			{
-				recursive: false,
-				fromLatestEvent: {
-					subject: '/test',
-					type: 'io.eventsourcingdb.test.bar',
-					ifEventIsMissing: 'read-everything',
-				},
-			},
-			controller.signal,
-		)) {
-			eventsObserved.push(event);
-		}
+		const eventsObserved = await observeUntil(
+			signal =>
+				client.observeEvents(
+					'/test',
+					{
+						recursive: false,
+						fromLatestEvent: {
+							subject: '/test',
+							type: 'io.eventsourcingdb.test.bar',
+							ifEventIsMissing: 'read-everything',
+						},
+					},
+					signal,
+				),
+			1,
+		);
 
 		assert.equal(eventsObserved.length, 1);
 		const [eventObserved] = eventsObserved;