From 157751133392f1eeb664fd59dfc130dc4d6f5f07 Mon Sep 17 00:00:00 2001 From: Phil Kuang Date: Wed, 23 Oct 2024 21:30:07 -0400 Subject: [PATCH] feat(events): allow setting a timeout for event bus polling Signed-off-by: Phil Kuang --- .changeset/old-taxis-exist.md | 11 +++ plugins/events-backend/config.d.ts | 6 ++ .../src/service/EventsPlugin.ts | 5 + .../src/service/hub/createEventBusRouter.ts | 9 +- plugins/events-node/report.api.md | 3 + .../src/api/DefaultEventsService.test.ts | 96 +++++++++++++++++++ .../src/api/DefaultEventsService.ts | 32 ++++++- plugins/events-node/src/api/EventsService.ts | 5 + plugins/events-node/src/api/index.ts | 1 + 9 files changed, 162 insertions(+), 6 deletions(-) create mode 100644 .changeset/old-taxis-exist.md diff --git a/.changeset/old-taxis-exist.md b/.changeset/old-taxis-exist.md new file mode 100644 index 0000000000..f2c7ac1f02 --- /dev/null +++ b/.changeset/old-taxis-exist.md @@ -0,0 +1,11 @@ +--- +'@backstage/plugin-events-backend': patch +'@backstage/plugin-events-node': patch +--- + +Allow configuring a timeout for event bus polling requests. This can be set like so in your app-config: + +```yaml +events: + notifyTimeoutMs: 30000 +``` diff --git a/plugins/events-backend/config.d.ts b/plugins/events-backend/config.d.ts index 3d26661855..ffeafdcbd5 100644 --- a/plugins/events-backend/config.d.ts +++ b/plugins/events-backend/config.d.ts @@ -16,6 +16,12 @@ export interface Config { events?: { + /** + * Timeout in milliseconds for how long to wait before closing subscription events + * requests to ensure they don't stall or that events get stuck. Defaults to 55 seconds. + */ + notifyTimeoutMs?: number; + http?: { /** * Topics for which a route has to be registered diff --git a/plugins/events-backend/src/service/EventsPlugin.ts b/plugins/events-backend/src/service/EventsPlugin.ts index 89f50d835b..a837e69c32 100644 --- a/plugins/events-backend/src/service/EventsPlugin.ts +++ b/plugins/events-backend/src/service/EventsPlugin.ts @@ -113,6 +113,10 @@ export const eventsPlugin = createBackendPlugin({ // that is used there as part of the middleware stack. httpRouter.use(eventsRouter); + const notifyTimeoutMs = config.getOptionalNumber( + 'events.notifyTimeoutMs', + ); + httpRouter.use( await createEventBusRouter({ database, @@ -120,6 +124,7 @@ export const eventsPlugin = createBackendPlugin({ logger, httpAuth, scheduler, + notifyTimeoutMs, }), ); diff --git a/plugins/events-backend/src/service/hub/createEventBusRouter.ts b/plugins/events-backend/src/service/hub/createEventBusRouter.ts index 4a332d5ce5..1ac95be4b6 100644 --- a/plugins/events-backend/src/service/hub/createEventBusRouter.ts +++ b/plugins/events-backend/src/service/hub/createEventBusRouter.ts @@ -26,7 +26,10 @@ import { createOpenApiRouter } from '../../schema/openapi.generated'; import { MemoryEventBusStore } from './MemoryEventBusStore'; import { DatabaseEventBusStore } from './DatabaseEventBusStore'; import { EventBusStore } from './types'; -import { EventParams } from '@backstage/plugin-events-node'; +import { + EVENTS_NOTIFY_TIMEOUT_HEADER, + EventParams, +} from '@backstage/plugin-events-node'; const DEFAULT_NOTIFY_TIMEOUT_MS = 55_000; // Just below 60s, which is a common HTTP timeout @@ -229,6 +232,10 @@ export async function createEventBusRouter(options: { })), }); } else { + res.setHeader( + EVENTS_NOTIFY_TIMEOUT_HEADER, + notifyTimeoutMs.toString(), + ); res.status(202); res.flushHeaders(); diff --git a/plugins/events-node/report.api.md b/plugins/events-node/report.api.md index 0b9c31090a..f53bb251ec 100644 --- a/plugins/events-node/report.api.md +++ b/plugins/events-node/report.api.md @@ -74,6 +74,9 @@ export abstract class EventRouter { subscribe(): Promise; } +// @public (undocumented) +export const EVENTS_NOTIFY_TIMEOUT_HEADER = 'backstage-events-notify-timeout'; + // @public export interface EventsService { publish(params: EventParams): Promise; diff --git a/plugins/events-node/src/api/DefaultEventsService.test.ts b/plugins/events-node/src/api/DefaultEventsService.test.ts index a7fc7ba50b..13a6a68fa4 100644 --- a/plugins/events-node/src/api/DefaultEventsService.test.ts +++ b/plugins/events-node/src/api/DefaultEventsService.test.ts @@ -16,6 +16,7 @@ import { DefaultEventsService } from './DefaultEventsService'; import { EventParams } from './EventParams'; +import { EVENTS_NOTIFY_TIMEOUT_HEADER } from './EventsService'; import { rest } from 'msw'; import { setupServer } from 'msw/node'; import { @@ -248,6 +249,101 @@ describe('DefaultEventsService', () => { blockingController!.close(); }); + it('should timeout polling if the response has a timeout header', async () => { + const logger = mockServices.logger.mock(); + const service = DefaultEventsService.create({ logger }).forPlugin('a', { + auth: mockServices.auth(), + logger, + discovery: mockServices.discovery(), + lifecycle: mockServices.lifecycle.mock(), + }); + + let callCount = 0; + + let blockingController: ReadableStreamDefaultController; + const blockingStream = new ReadableStream({ + start(controller) { + blockingController = controller; + }, + }); + + mswServer.use( + rest.put( + 'http://localhost:0/api/events/bus/v1/subscriptions/a.tester', + (_req, res, ctx) => res(ctx.status(200)), + ), + // The first and third calls result in a blocking 202 that is resolved after 100ms + // The second and fourth calls result in a 200 with an event + // The fifth call blocks until the end of the test + // No more than 5 calls should be made + rest.get( + 'http://localhost:0/api/events/bus/v1/subscriptions/a.tester/events', + (_req, res, ctx) => { + callCount += 1; + if (callCount === 1 || callCount === 3) { + return res( + ctx.status(202), + ctx.body( + new ReadableStream({ + start(controller) { + setTimeout(() => controller.close(), 100); + }, + }), + ), + ); + } else if (callCount === 2 || callCount === 4) { + return res( + ctx.status(200), + ctx.json({ + events: [{ topic: 'test', payload: { callCount } }], + }), + ); + } else if (callCount === 5) { + // 5th call has a timeout header so polling should proceed to the next call + return res( + ctx.set(EVENTS_NOTIFY_TIMEOUT_HEADER, '100'), + ctx.status(202), + ctx.body(blockingStream), + ); + } else if (callCount === 6) { + return res(ctx.status(202), ctx.body(blockingStream)); + } + throw new Error(`events endpoint called too many times`); + }, + ), + ); + + const event = await new Promise(resolve => { + const events = new Array(); + service.subscribe({ + id: 'tester', + topics: ['test'], + async onEvent(newEvent) { + events.push(newEvent); + if (events.length === 2) { + resolve(events); + } + }, + }); + }); + + expect(event).toEqual([ + { topic: 'test', eventPayload: { callCount: 2 } }, + { topic: 'test', eventPayload: { callCount: 4 } }, + ]); + + // Wait to allow timeout to trigger and make sure no additional calls happen + await new Promise(resolve => setTimeout(resolve, 2000)); + + expect(callCount).toBe(6); + + // Internal call to clean up subscriptions + await (service as any).shutdown(); + + // Close the stream for the 5th call so that we don't leave the request hanging + blockingController!.close(); + }); + it('should not read events from bus if disabled', async () => { const logger = mockServices.logger.mock(); const service = DefaultEventsService.create({ diff --git a/plugins/events-node/src/api/DefaultEventsService.ts b/plugins/events-node/src/api/DefaultEventsService.ts index 7a35ed305a..3ceed45a26 100644 --- a/plugins/events-node/src/api/DefaultEventsService.ts +++ b/plugins/events-node/src/api/DefaultEventsService.ts @@ -22,7 +22,11 @@ import { RootConfigService, } from '@backstage/backend-plugin-api'; import { EventParams } from './EventParams'; -import { EventsService, EventsServiceSubscribeOptions } from './EventsService'; +import { + EVENTS_NOTIFY_TIMEOUT_HEADER, + EventsService, + EventsServiceSubscribeOptions, +} from './EventsService'; import { DefaultApiClient } from '../generated'; import { ResponseError } from '@backstage/errors'; @@ -214,10 +218,28 @@ class PluginEventsService implements EventsService { // immediately again lock.release(); - // We don't actually expect any response body here, but waiting for - // an empty body to be returned has been more reliable that waiting - // for the response body stream to close. - await res.text(); + + const notifyTimeoutHeader = res.headers.get( + EVENTS_NOTIFY_TIMEOUT_HEADER, + ); + + // Add 1s to the timeout to allow the server to potentially timeout first + const notifyTimeoutMs = + notifyTimeoutHeader && !isNaN(parseInt(notifyTimeoutHeader, 10)) + ? Number(notifyTimeoutHeader) + 1_000 + : null; + + await Promise.race( + [ + // We don't actually expect any response body here, but waiting for + // an empty body to be returned has been more reliable that waiting + // for the response body stream to close. + res.text(), + notifyTimeoutMs + ? new Promise(resolve => setTimeout(resolve, notifyTimeoutMs)) + : null, + ].filter(Boolean), + ); } else if (res.status === 200) { const data = await res.json(); if (data) { diff --git a/plugins/events-node/src/api/EventsService.ts b/plugins/events-node/src/api/EventsService.ts index 7af13f9b07..3426f646e7 100644 --- a/plugins/events-node/src/api/EventsService.ts +++ b/plugins/events-node/src/api/EventsService.ts @@ -55,3 +55,8 @@ export type EventsServiceSubscribeOptions = { * @public */ export type EventsServiceEventHandler = (params: EventParams) => Promise; + +/** + * @public + */ +export const EVENTS_NOTIFY_TIMEOUT_HEADER = 'backstage-events-notify-timeout'; diff --git a/plugins/events-node/src/api/index.ts b/plugins/events-node/src/api/index.ts index 6f3817cedc..ef2946396c 100644 --- a/plugins/events-node/src/api/index.ts +++ b/plugins/events-node/src/api/index.ts @@ -16,6 +16,7 @@ export type { EventParams } from './EventParams'; export { EventRouter } from './EventRouter'; +export { EVENTS_NOTIFY_TIMEOUT_HEADER } from './EventsService'; export type { EventsService, EventsServiceSubscribeOptions,