From 2d25e2791c346e79f8b6f66a0fb65c1c1d3c0712 Mon Sep 17 00:00:00 2001 From: Patrik Oldsberg Date: Fri, 24 May 2024 15:33:33 +0200 Subject: [PATCH] events-backend: refactor event bus notifications and add timeout Signed-off-by: Patrik Oldsberg --- .../src/service/hub/DatabaseEventBusStore.ts | 22 ++++++-- .../src/service/hub/MemoryEventBusStore.ts | 20 ++++--- .../src/service/hub/createEventBusRouter.ts | 55 ++++++++++++++----- .../events-backend/src/service/hub/types.ts | 5 +- 4 files changed, 72 insertions(+), 30 deletions(-) diff --git a/plugins/events-backend/src/service/hub/DatabaseEventBusStore.ts b/plugins/events-backend/src/service/hub/DatabaseEventBusStore.ts index fc2128e2cf..c32e2393a7 100644 --- a/plugins/events-backend/src/service/hub/DatabaseEventBusStore.ts +++ b/plugins/events-backend/src/service/hub/DatabaseEventBusStore.ts @@ -378,27 +378,37 @@ export class DatabaseEventBusStore implements EventBusStore { async listen( subscriptionId: string, - listeners: { + options: { + signal: AbortSignal; onNotify: (topicId: string) => void; onError: () => void; }, - ): Promise<{ cancel(): void }> { + ): Promise { const result = await this.#db(TABLE_SUBSCRIPTIONS) .select('topics') .where({ id: subscriptionId }) .first(); - console.log(`DEBUG: result=`, result); + if (!result) { throw new NotFoundError( `Subscription with ID '${subscriptionId}' not found`, ); } + + if (options.signal.aborted) { + return; + } + const topics = new Set(result.topics ?? []); const cancel = await this.#listener.listen( topics, - listeners.onNotify, - listeners.onError, + options.onNotify, + options.onError, ); - return { cancel }; + if (options.signal.aborted) { + cancel(); + } else { + options.signal.addEventListener('abort', cancel); + } } } diff --git a/plugins/events-backend/src/service/hub/MemoryEventBusStore.ts b/plugins/events-backend/src/service/hub/MemoryEventBusStore.ts index 8a03c2961a..98739fdb33 100644 --- a/plugins/events-backend/src/service/hub/MemoryEventBusStore.ts +++ b/plugins/events-backend/src/service/hub/MemoryEventBusStore.ts @@ -99,21 +99,25 @@ export class MemoryEventBusStore implements EventBusStore { async listen( subscriptionId: string, - listeners: { + options: { + signal: AbortSignal; onNotify(topicId: string): void; onError(): void; }, - ): Promise<{ cancel(): void }> { + ): Promise { + if (options.signal.aborted) { + return; + } + const sub = this.#subscribers.get(subscriptionId); if (!sub) { throw new Error(`Subscription not found`); } - const listener = { topics: sub.topics, notify: listeners.onNotify }; + const listener = { topics: sub.topics, notify: options.onNotify }; this.#listeners.add(listener); - return { - cancel: () => { - this.#listeners.delete(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 23aadacadc..84d04a00fb 100644 --- a/plugins/events-backend/src/service/hub/createEventBusRouter.ts +++ b/plugins/events-backend/src/service/hub/createEventBusRouter.ts @@ -27,12 +27,19 @@ import { DatabaseEventBusStore } from './DatabaseEventBusStore'; import { EventBusStore } from './types'; import { EventParams } from '@backstage/plugin-events-node'; +const DEFAULT_NOTIFY_TIMEOUT_MS = 55_000; // Just below 60s, which is a common HTTP timeout + export async function createEventBusRouter(options: { logger: LoggerService; database: DatabaseService; httpAuth: HttpAuthService; + notifyTimeoutMs?: number; // for testing }): Promise { - const { database, httpAuth } = options; + const { + database, + httpAuth, + notifyTimeoutMs = DEFAULT_NOTIFY_TIMEOUT_MS, + } = options; const logger = options.logger.child({ type: 'EventBus' }); const router = Router(); @@ -92,28 +99,48 @@ 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(resolve => { resolveShouldNotify = resolve; }); - const { cancel } = await store.listen(id, { + 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, { + signal: controller.signal, onNotify() { - shouldNotifyPromise.then(shouldNotify => { - if (shouldNotify) { - res.status(204).end(); - } - }); + notify(204); }, onError() { - shouldNotifyPromise.then(shouldNotify => { - if (shouldNotify) { - res.status(500).end(); - } - }); + notify(500); }, }); - req.on('end', cancel); + + // 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); + controller.abort(); + }, notifyTimeoutMs); try { const { events } = await store.readSubscription(id); @@ -125,11 +152,11 @@ export async function createEventBusRouter(options: { if (events.length > 0) { res.json({ events }); - resolveShouldNotify!(false); } else { resolveShouldNotify!(true); } } finally { + clearTimeout(timeout); resolveShouldNotify!(false); } }, diff --git a/plugins/events-backend/src/service/hub/types.ts b/plugins/events-backend/src/service/hub/types.ts index f7a162a3a6..55159352b4 100644 --- a/plugins/events-backend/src/service/hub/types.ts +++ b/plugins/events-backend/src/service/hub/types.ts @@ -28,9 +28,10 @@ export type EventBusStore = { listen( subscriptionId: string, - listeners: { + options: { + signal: AbortSignal; onNotify(topicId: string): void; onError(): void; }, - ): Promise<{ cancel(): void }>; + ): Promise; };