diff --git a/plugins/events-backend/src/service/hub/DatabaseEventBusStore.ts b/plugins/events-backend/src/service/hub/DatabaseEventBusStore.ts
index a6260cd0be..2146df9b84 100644
--- a/plugins/events-backend/src/service/hub/DatabaseEventBusStore.ts
+++ b/plugins/events-backend/src/service/hub/DatabaseEventBusStore.ts
@@ -82,8 +82,8 @@ class DatabaseEventBusListener {
 
   readonly #listeners = new Set<{
     topics: Set<string>;
-    onNotify: (topicId: string) => void;
-    onError: () => void;
+    resolve: (result: { topic: string }) => void;
+    reject: (error: Error) => void;
   }>();
 
   #connPromise?: Promise<InternalDbConnection>;
@@ -95,43 +95,37 @@ class DatabaseEventBusListener {
     this.#logger = logger.child({ type: 'DatabaseEventBusListener' });
   }
 
-  async listen(
+  async waitForUpdate(
     topics: Set<string>,
-    onNotify: (topicId: string) => void,
-    onError: () => void,
-  ): Promise<() => void> {
+    signal: AbortSignal,
+  ): Promise<{ topic: string }> {
     if (this.#connTimeout) {
       clearTimeout(this.#connTimeout);
       this.#connTimeout = undefined;
     }
+
     await this.#ensureConnection();
 
-    const listener = { topics, onNotify, onError };
-    this.#listeners.add(listener);
+    return new Promise<{ topic: string }>((resolve, reject) => {
+      const listener = { topics, resolve, reject };
+      this.#listeners.add(listener);
 
-    return () => {
-      this.#listeners.delete(listener);
-
-      // If we don't have any listeners, destroy the connection after a timeout
-      if (this.#listeners.size === 0) {
-        this.#connTimeout = setTimeout(() => {
-          this.#connPromise?.then(conn => {
-            this.#logger.info('Listener connection timed out, destroying');
-            this.#connPromise = undefined;
-            this.#destroyConnection(conn);
-          });
-        }, LISTENER_CONNECTION_TIMEOUT_MS);
-      }
-    };
+      signal.addEventListener('abort', () => {
+        this.#listeners.delete(listener);
+        this.#maybeTimeoutConnection();
+      });
+    });
   }
 
   #handleNotify(topic: string) {
-    this.#logger.info(`Listener received notification for topic '${topic}'`);
+    this.#logger.debug(`Listener received notification for topic '${topic}'`);
     for (const l of this.#listeners) {
       if (l.topics.has(topic)) {
-        l.onNotify(topic);
+        l.resolve({ topic });
+        this.#listeners.delete(l);
       }
     }
+    this.#maybeTimeoutConnection();
   }
 
   // We don't try to reconnect on error, instead we notify all listeners and let
@@ -142,7 +136,23 @@ class DatabaseEventBusListener {
       error,
     );
     for (const l of this.#listeners) {
-      l.onError();
+      l.reject(new Error('Listener connection failed'));
+    }
+    this.#listeners.clear();
+    this.#maybeTimeoutConnection();
+  }
+
+  #maybeTimeoutConnection() {
+    // If we don't have any listeners, destroy the connection after a timeout
+    if (this.#listeners.size === 0 && !this.#connTimeout) {
+      this.#connTimeout = setTimeout(() => {
+        this.#connTimeout = undefined;
+        this.#connPromise?.then(conn => {
+          this.#logger.info('Listener connection timed out, destroying');
+          this.#connPromise = undefined;
+          this.#destroyConnection(conn);
+        });
+      }, LISTENER_CONNECTION_TIMEOUT_MS);
     }
   }
 
@@ -437,14 +447,12 @@ export class DatabaseEventBusStore implements EventBusStore {
     };
   }
 
-  async listen(
+  async setupListener(
     subscriptionId: string,
     options: {
       signal: AbortSignal;
-      onNotify: (topicId: string) => void;
-      onError: () => void;
     },
-  ): Promise<void> {
+  ): Promise<{ waitForUpdate(): Promise<{ topic: string }> }> {
     const result = await this.#db<SubscriptionsRow>(TABLE_SUBSCRIPTIONS)
       .select('topics')
       .where({ id: subscriptionId })
@@ -456,21 +464,12 @@ export class DatabaseEventBusStore implements EventBusStore {
       );
     }
 
-    if (options.signal.aborted) {
-      return;
-    }
+    options.signal.throwIfAborted();
 
     const topics = new Set(result.topics ?? []);
-    const cancel = await this.#listener.listen(
-      topics,
-      options.onNotify,
-      options.onError,
-    );
-    if (options.signal.aborted) {
-      cancel();
-    } else {
-      options.signal.addEventListener('abort', cancel);
-    }
+    return {
+      waitForUpdate: () => this.#listener.waitForUpdate(topics, options.signal),
+    };
   }
 
   #cleanup = async () => {
diff --git a/plugins/events-backend/src/service/hub/MemoryEventBusStore.ts b/plugins/events-backend/src/service/hub/MemoryEventBusStore.ts
index 98739fdb33..0fc9a2b14f 100644
--- a/plugins/events-backend/src/service/hub/MemoryEventBusStore.ts
+++ b/plugins/events-backend/src/service/hub/MemoryEventBusStore.ts
@@ -28,19 +28,19 @@ export class MemoryEventBusStore implements EventBusStore {
   >();
   #listeners = new Set<{
     topics: Set<string>;
-    notify(topicId: string): void;
+    resolve(result: { topic: string }): void;
   }>();
 
   async publish(options: {
     params: EventParams;
     subscriberIds: string[];
   }): Promise<{ id: string } | undefined> {
-    const topicId = options.params.topic;
+    const topic = options.params.topic;
     const subscriberIds = new Set(options.subscriberIds);
 
     let hasOtherSubscribers = false;
     for (const sub of this.#subscribers.values()) {
-      if (sub.topics.has(topicId) && !subscriberIds.has(sub.id)) {
+      if (sub.topics.has(topic) && !subscriberIds.has(sub.id)) {
         hasOtherSubscribers = true;
         break;
       }
@@ -53,8 +53,9 @@ export class MemoryEventBusStore implements EventBusStore {
     this.#events.push({ ...options.params, subscriberIds, seq: nextSeq });
 
     for (const listener of this.#listeners) {
-      if (listener.topics.has(topicId)) {
-        listener.notify(topicId);
+      if (listener.topics.has(topic)) {
+        listener.resolve({ topic });
+        this.#listeners.delete(listener);
       }
     }
     return { id: String(nextSeq) };
@@ -97,27 +98,30 @@ export class MemoryEventBusStore implements EventBusStore {
     return { events: events.map(event => ({ ...event, seq: undefined })) };
   }
 
-  async listen(
+  async setupListener(
     subscriptionId: string,
     options: {
       signal: AbortSignal;
-      onNotify(topicId: string): void;
-      onError(): void;
     },
-  ): Promise<void> {
-    if (options.signal.aborted) {
-      return;
-    }
+  ): Promise<{ waitForUpdate(): Promise<{ topic: string }> }> {
+    return {
+      waitForUpdate: async () => {
+        options.signal.throwIfAborted();
 
-    const sub = this.#subscribers.get(subscriptionId);
-    if (!sub) {
-      throw new Error(`Subscription not found`);
-    }
-    const listener = { topics: sub.topics, notify: options.onNotify };
-    this.#listeners.add(listener);
+        const sub = this.#subscribers.get(subscriptionId);
+        if (!sub) {
+          throw new Error(`Subscription not found`);
+        }
 
-    options.signal.addEventListener('abort', () => {
-      this.#listeners.delete(listener);
-    });
+        return new Promise<{ topic: string }>(resolve => {
+          const listener = { topics: sub.topics, resolve };
+          this.#listeners.add(listener);
+
+          options.signal.addEventListener('abort', () => {
+            this.#listeners.delete(listener);
+          });
+        });
+      },
+    };
   }
 }
diff --git a/plugins/events-backend/src/service/hub/createEventBusRouter.ts b/plugins/events-backend/src/service/hub/createEventBusRouter.ts
index 30cd137104..d9c2bdf3dc 100644
--- a/plugins/events-backend/src/service/hub/createEventBusRouter.ts
+++ b/plugins/events-backend/src/service/hub/createEventBusRouter.ts
@@ -103,53 +103,22 @@ export async function createEventBusRouter(options: {
       });
       const id = req.params.subscriptionId;
 
-      // Don't notify until we know the outcome of reading events
-      let resolveShouldNotify: (shouldNotify: boolean) => void;
-      const shouldNotifyPromise = new Promise<boolean>(resolve => {
-        resolveShouldNotify = resolve;
-      });
-
       const controller = new AbortController();
       req.on('end', () => controller.abort());
 
-      let timeout: NodeJS.Timeout | undefined = undefined;
-
-      let notified = false;
-      const notify = async (status: number) => {
-        if (!notified) {
-          clearTimeout(timeout);
-          notified = true;
-          if (await shouldNotifyPromise) {
-            res.status(status).end();
-          }
-        }
-      };
-
       // By setting up the listener first we make sure we don't miss any events
       // that are published while reading. If an event is published we'll receive
       // a notification, which depending on the outcome of the read we may ignore
-      await store.listen(id, {
+      const listener = await store.setupListener(id, {
         signal: controller.signal,
-        onNotify() {
-          notify(204);
-        },
-        onError() {
-          notify(500);
-        },
       });
 
       // By timing out requests we make sure they don't stall or that events get stuck.
       // For the caller there's no difference between a timeout and a
       // notifications, either way they should try reading again.
-      timeout = setTimeout(() => {
-        notify(204);
+      const timeout = setTimeout(() => {
         controller.abort();
       }, notifyTimeoutMs);
-      shouldNotifyPromise.then(shouldNotify => {
-        if (!shouldNotify) {
-          clearTimeout(timeout);
-        }
-      });
 
       try {
         const { events } = await store.readSubscription(id);
@@ -162,10 +131,27 @@ export async function createEventBusRouter(options: {
         if (events.length > 0) {
           res.json({ events });
         } else {
-          resolveShouldNotify!(true);
+          res.status(202);
+          res.flushHeaders();
+
+          try {
+            const { topic } = await listener.waitForUpdate();
+            logger.info(
+              `Received notification for subscription '${id}' for topic '${topic}'`,
+              { subject: credentials.principal.subject },
+            );
+          } finally {
+            // A small extra delay ensures a more even spread of events across
+            // consumers in case some consumers are faster than others
+            await new Promise(resolve =>
+              setTimeout(resolve, 1 + Math.random() * 9),
+            );
+            res.end();
+          }
         }
       } finally {
-        resolveShouldNotify!(false);
+        controller.abort();
+        clearTimeout(timeout);
       }
     },
   );
diff --git a/plugins/events-backend/src/service/hub/types.ts b/plugins/events-backend/src/service/hub/types.ts
index 55159352b4..87e62049a9 100644
--- a/plugins/events-backend/src/service/hub/types.ts
+++ b/plugins/events-backend/src/service/hub/types.ts
@@ -26,12 +26,10 @@ export type EventBusStore = {
 
   readSubscription(id: string): Promise<{ events: EventParams[] }>;
 
-  listen(
+  setupListener(
     subscriptionId: string,
     options: {
       signal: AbortSignal;
-      onNotify(topicId: string): void;
-      onError(): void;
     },
-  ): Promise<void>;
+  ): Promise<{ waitForUpdate(): Promise<{ topic: string }> }>;
 };
diff --git a/plugins/events-node/src/api/DefaultEventsService.ts b/plugins/events-node/src/api/DefaultEventsService.ts
index d1d26b1723..7811349680 100644
--- a/plugins/events-node/src/api/DefaultEventsService.ts
+++ b/plugins/events-node/src/api/DefaultEventsService.ts
@@ -205,10 +205,12 @@ class PluginEventsService implements EventsService {
         }
         backoffMs = POLL_BACKOFF_START_MS;
 
-        // 204 arrives after a blocking response and means there are new events
-        // available or we timed out, either way should should try to read events
+        // 202 means there were no immediately available events, but the
+        // response will block until either new events are available or the
+        // request times out. In both cases we should should try to read events
         // immediately again
-        if (res.status === 204) {
+        if (res.status === 202) {
+          await res.body?.getReader()?.closed;
           process.nextTick(poll);
         } else if (res.status === 200) {
           const data = await res.json();
@@ -316,6 +318,7 @@ export class DefaultEventsService implements EventsService {
       options &&
       new DefaultApiClient({
         discoveryApi: options.discovery,
+        fetchApi: { fetch }, // use native node fetch
       });
     const logger = options?.logger ?? this.logger;
     return new PluginEventsService(