events-node: more resilient and configurable events service

Signed-off-by: Patrik Oldsberg <poldsberg@gmail.com>

.changeset/cyan-trees-bow.md

@@ -0,0 +1,7 @@
+---
+'@backstage/plugin-events-node': patch
+---
+
+Fixed an issue where subscribing to events threw an error and gave up too easily. Calling the subscribe method will cause the background polling loop to keep trying to connect to the events backend, even if the initial request fails.
+
+By default the events service will attempt to publish and subscribe to events from the events bus API in the events backend, but if it fails due to the events backend not being installed, it will bail and never try calling the API again. There is now a new `events.useEventBus` configuration and option for the `DefaultEventsService` that lets you control this behavior. You can set it to `'never'` to disabled API calls to the events backend completely, or `'always'` to never allow it to be disabled.

plugins/events-node/config.d.ts

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2024 The Backstage Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+export interface Config {
+  events?: {
+    /**
+     * Whether to use the event bus API in the events plugin backend to
+     * distribute events across multiple instances when publishing and
+     * subscribing to events.
+     *
+     * The default is 'auto', which means means that the event bus API will be
+     * used if it's available, but will be disabled if the events backend
+     * returns a 404.
+     *
+     * If set to 'never', the events service will only ever publish events
+     * locally to the same instance, while if set to 'always', the event bus API
+     * will never be disabled, even if the events backend returns a 404.
+     */
+    useEventBus?: 'never' | 'always' | 'auto';
+  };
+}

plugins/events-node/package.json

@@ -39,7 +39,8 @@
     }
   },
   "files": [
-    "dist"
+    "dist",
+    "config.d.ts"
   ],
   "scripts": {
     "build": "backstage-cli package build",
@@ -60,6 +61,8 @@
   "devDependencies": {
     "@backstage/backend-common": "^0.25.0",
     "@backstage/backend-test-utils": "workspace:^",
-    "@backstage/cli": "workspace:^"
-  }
+    "@backstage/cli": "workspace:^",
+    "msw": "^1.0.0"
+  },
+  "configSchema": "config.d.ts"
 }

plugins/events-node/report.api.md

@@ -7,13 +7,18 @@ import { AuthService } from '@backstage/backend-plugin-api';
 import { DiscoveryService } from '@backstage/backend-plugin-api';
 import { LifecycleService } from '@backstage/backend-plugin-api';
 import { LoggerService } from '@backstage/backend-plugin-api';
+import { RootConfigService } from '@backstage/backend-plugin-api';
 import { ServiceFactory } from '@backstage/backend-plugin-api';
 import { ServiceRef } from '@backstage/backend-plugin-api';
 
 // @public
 export class DefaultEventsService implements EventsService {
   // (undocumented)
-  static create(options: { logger: LoggerService }): DefaultEventsService;
+  static create(options: {
+    logger: LoggerService;
+    config?: RootConfigService;
+    useEventBus?: EventBusMode;
+  }): DefaultEventsService;
   forPlugin(
     pluginId: string,
     options?: {
@@ -37,6 +42,9 @@ export interface EventBroker {
   ): void;
 }
 
+// @public (undocumented)
+export type EventBusMode = 'never' | 'always' | 'auto';
+
 // @public (undocumented)
 export interface EventParams<TPayload = unknown> {
   eventPayload: TPayload;

plugins/events-node/src/api/DefaultEventsService.test.ts

@@ -16,12 +16,16 @@
 
 import { DefaultEventsService } from './DefaultEventsService';
 import { EventParams } from './EventParams';
-import { mockServices } from '@backstage/backend-test-utils';
-
-const logger = mockServices.logger.mock();
+import { rest } from 'msw';
+import { setupServer } from 'msw/node';
+import {
+  mockServices,
+  registerMswTestHooks,
+} from '@backstage/backend-test-utils';
 
 describe('DefaultEventsService', () => {
   it('passes events to interested subscribers', async () => {
+    const logger = mockServices.logger.mock();
     const events = DefaultEventsService.create({ logger });
     const eventsSubscriber1: EventParams[] = [];
     const eventsSubscriber2: EventParams[] = [];
@@ -70,6 +74,7 @@ describe('DefaultEventsService', () => {
   it('logs errors from subscribers', async () => {
     const topic = 'testTopic';
 
+    const logger = mockServices.logger.mock();
     const warnSpy = jest.spyOn(logger, 'warn');
     const events = DefaultEventsService.create({ logger });
 
@@ -108,4 +113,165 @@ describe('DefaultEventsService', () => {
       new Error('NOPE 2'),
     );
   });
+
+  describe('with event bus', () => {
+    const mswServer = setupServer();
+    registerMswTestHooks(mswServer);
+
+    it('should read events from events bus API', async () => {
+      const logger = mockServices.logger.mock();
+      const service = DefaultEventsService.create({ logger }).forPlugin('a', {
+        auth: mockServices.auth(),
+        logger,
+        discovery: mockServices.discovery(),
+        lifecycle: mockServices.lifecycle.mock(),
+      });
+
+      mswServer.use(
+        rest.put(
+          'http://localhost:0/api/events/bus/v1/subscriptions/a.tester',
+          (_req, res, ctx) => res(ctx.status(200)),
+        ),
+        rest.get(
+          'http://localhost:0/api/events/bus/v1/subscriptions/a.tester/events',
+          (_req, res, ctx) =>
+            res(
+              ctx.status(200),
+              ctx.json({
+                events: [{ topic: 'test', payload: { foo: 'bar' } }],
+              }),
+            ),
+        ),
+      );
+
+      const event = await new Promise(resolve => {
+        service.subscribe({
+          id: 'tester',
+          topics: ['test'],
+          async onEvent(newEvent) {
+            resolve(newEvent);
+          },
+        });
+      });
+
+      expect(event).toEqual({ topic: 'test', eventPayload: { foo: 'bar' } });
+
+      // Internal call to clean up subscriptions
+      await (service as any).shutdown();
+    });
+
+    it('should not read events from bus if disabled', async () => {
+      const logger = mockServices.logger.mock();
+      const service = DefaultEventsService.create({
+        logger,
+        useEventBus: 'never',
+      }).forPlugin('a', {
+        auth: mockServices.auth(),
+        logger,
+        discovery: mockServices.discovery(),
+        lifecycle: mockServices.lifecycle.mock(),
+      });
+
+      let calledApi = false;
+      mswServer.use(
+        rest.put(
+          'http://localhost:0/api/events/bus/v1/subscriptions/a.tester',
+          (_req, res, ctx) => {
+            calledApi = true;
+            res(ctx.status(200));
+          },
+        ),
+      );
+
+      await service.subscribe({
+        id: 'tester',
+        topics: ['test'],
+        async onEvent() {
+          expect('not').toBe('reached');
+        },
+      });
+
+      // Internal call to clean up subscriptions, and wait for the API call
+      await (service as any).shutdown();
+
+      expect(calledApi).toBe(false);
+    });
+
+    it('should deactivate event bus on 404', async () => {
+      expect.assertions(1);
+
+      const logger = mockServices.logger.mock();
+      const service = DefaultEventsService.create({ logger }).forPlugin('a', {
+        auth: mockServices.auth(),
+        logger,
+        discovery: mockServices.discovery(),
+        lifecycle: mockServices.lifecycle.mock(),
+      });
+
+      mswServer.use(
+        rest.put(
+          'http://localhost:0/api/events/bus/v1/subscriptions/a.tester',
+          (_req, res, ctx) => res(ctx.status(404)),
+        ),
+      );
+
+      service.subscribe({
+        id: 'tester',
+        topics: ['test'],
+        async onEvent() {
+          expect('not').toBe('reached');
+        },
+      });
+
+      const msg = await new Promise(resolve => {
+        logger.warn.mockImplementationOnce(resolve);
+      });
+
+      expect(msg).toMatch(/Event subscribe request failed with status 404/);
+
+      // Internal call to clean up subscriptions
+      await (service as any).shutdown();
+    });
+
+    it('should not deactivate event bus if configured to always be used', async () => {
+      expect.assertions(1);
+
+      const logger = mockServices.logger.mock();
+      const service = DefaultEventsService.create({
+        logger,
+        useEventBus: 'always',
+      }).forPlugin('a', {
+        auth: mockServices.auth(),
+        logger,
+        discovery: mockServices.discovery(),
+        lifecycle: mockServices.lifecycle.mock(),
+      });
+
+      mswServer.use(
+        rest.put(
+          'http://localhost:0/api/events/bus/v1/subscriptions/a.tester',
+          (_req, res, ctx) => res(ctx.status(404)),
+        ),
+      );
+
+      service.subscribe({
+        id: 'tester',
+        topics: ['test'],
+        async onEvent() {
+          expect('not').toBe('reached');
+        },
+      });
+
+      const msg = await new Promise(resolve => {
+        logger.warn.mockImplementationOnce(resolve);
+      });
+
+      expect(msg).toMatch(
+        'Poll failed for subscription "a.tester", retrying in 1000ms',
+      );
+
+      // Internal call to clean up subscriptions
+      await (service as any).shutdown();
+    });
+  });
 });

plugins/events-node/src/api/DefaultEventsService.ts

@@ -19,6 +19,7 @@ import {
   DiscoveryService,
   LifecycleService,
   LoggerService,
+  RootConfigService,
 } from '@backstage/backend-plugin-api';
 import { EventParams } from './EventParams';
 import { EventsService, EventsServiceSubscribeOptions } from './EventsService';
@@ -29,6 +30,13 @@ const POLL_BACKOFF_START_MS = 1_000;
 const POLL_BACKOFF_MAX_MS = 60_000;
 const POLL_BACKOFF_FACTOR = 2;
 
+const EVENT_BUS_MODES = ['never', 'always', 'auto'] as const;
+
+/**
+ * @public
+ */
+export type EventBusMode = 'never' | 'always' | 'auto';
+
 /**
  * Local event bus for subscribers within the same process.
  *
@@ -107,23 +115,28 @@ class PluginEventsService implements EventsService {
     private readonly pluginId: string,
     private readonly localBus: LocalEventBus,
     private readonly logger: LoggerService,
+    private readonly mode: EventBusMode,
     private client?: DefaultApiClient,
     private readonly auth?: AuthService,
   ) {}
 
   async publish(params: EventParams): Promise<void> {
     const lock = this.#getShutdownLock();
+    if (!lock) {
+      throw new Error('Service is shutting down');
+    }
     try {
       const { notifiedSubscribers } = await this.localBus.publish(params);
 
-      if (!this.client) {
+      const client = this.client;
+      if (!client) {
         return;
       }
       const token = await this.#getToken();
       if (!token) {
         return;
       }
-      const res = await this.client.postEvent(
+      const res = await client.postEvent(
         {
           body: {
             event: { payload: params.eventPayload, topic: params.topic },
@@ -134,7 +147,7 @@ class PluginEventsService implements EventsService {
       );
 
       if (!res.ok) {
-        if (res.status === 404) {
+        if (res.status === 404 && this.mode !== 'always') {
           this.logger.warn(
             `Event publish request failed with status 404, events backend not found. Future events will not be persisted.`,
           );
@@ -160,27 +173,6 @@ class PluginEventsService implements EventsService {
     if (!this.client) {
       return;
     }
-    const token = await this.#getToken();
-    if (!token) {
-      return;
-    }
-    const res = await this.client.putSubscription(
-      {
-        path: { subscriptionId },
-        body: { topics: options.topics },
-      },
-      { token },
-    );
-    if (!res.ok) {
-      if (res.status === 404) {
-        this.logger.warn(
-          `Event subscribe request failed with status 404, events backend not found. Will only receive events that were sent locally on this process.`,
-        );
-        delete this.client;
-        return;
-      }
-      throw await ResponseError.fromResponse(res);
-    }
 
     this.#startPolling(subscriptionId, options.topics, options.onEvent);
   }
@@ -190,75 +182,101 @@ class PluginEventsService implements EventsService {
     topics: string[],
     onEvent: EventsServiceSubscribeOptions['onEvent'],
   ) {
+    let hasSubscription = false;
     let backoffMs = POLL_BACKOFF_START_MS;
     const poll = async () => {
-      if (!this.client) {
+      const client = this.client;
+      if (!client) {
         return;
       }
       const lock = this.#getShutdownLock();
+      if (!lock) {
+        return; // shutting down
+      }
       try {
         const token = await this.#getToken();
         if (!token) {
           return;
         }
-        const res = await this.client.getSubscriptionEvents(
-          {
-            path: { subscriptionId },
-          },
-          { token },
-        );
 
-        if (!res.ok) {
-          if (res.status === 404) {
-            this.logger.info(
-              `Polling event subscription resulted in a 404, recreating subscription`,
-            );
-            const putRes = await this.client.putSubscription(
-              {
-                path: { subscriptionId },
-                body: { topics },
-              },
-              { token },
-            );
-            if (!putRes.ok) {
+        if (hasSubscription) {
+          const res = await client.getSubscriptionEvents(
+            {
+              path: { subscriptionId },
+            },
+            { token },
+          );
+
+          if (!res.ok) {
+            if (res.status === 404) {
+              this.logger.info(
+                `Polling event subscription resulted in a 404, recreating subscription`,
+              );
+              hasSubscription = false;
+            } else {
               throw await ResponseError.fromResponse(res);
             }
           }
-          throw await ResponseError.fromResponse(res);
-        }
-        backoffMs = POLL_BACKOFF_START_MS;
 
-        // 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 === 202) {
-          lock.release();
-          await res.body?.getReader()?.closed;
-          process.nextTick(poll);
-        } else if (res.status === 200) {
-          const data = await res.json();
-          if (data) {
-            for (const event of data.events ?? []) {
-              try {
-                await onEvent({
-                  topic: event.topic,
-                  eventPayload: event.payload,
-                });
-              } catch (error) {
-                this.logger.warn(
-                  `Subscriber "${subscriptionId}" failed to process event for topic "${event.topic}"`,
-                  error,
-                );
+          // Successful response, reset backoff
+          backoffMs = POLL_BACKOFF_START_MS;
+
+          // 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 === 202) {
+            lock.release();
+            await res.body?.getReader()?.closed;
+            process.nextTick(poll);
+          } else if (res.status === 200) {
+            const data = await res.json();
+            if (data) {
+              for (const event of data.events ?? []) {
+                try {
+                  await onEvent({
+                    topic: event.topic,
+                    eventPayload: event.payload,
+                  });
+                } catch (error) {
+                  this.logger.warn(
+                    `Subscriber "${subscriptionId}" failed to process event for topic "${event.topic}"`,
+                    error,
+                  );
+                }
               }
+            } else {
+              this.logger.warn(
+                `Unexpected response status ${res.status} from events backend for subscription "${subscriptionId}"`,
+              );
             }
           }
-          process.nextTick(poll);
-        } else {
-          this.logger.warn(
-            `Unexpected response status ${res.status} from events backend for subscription "${subscriptionId}"`,
-          );
         }
+
+        // If we haven't yet created the subscription, or if it was removed, create a new one
+        if (!hasSubscription) {
+          const res = await client.putSubscription(
+            {
+              path: { subscriptionId },
+              body: { topics },
+            },
+            { token },
+          );
+          hasSubscription = true;
+          if (!res.ok) {
+            if (res.status === 404 && this.mode !== 'always') {
+              this.logger.warn(
+                `Event subscribe request failed with status 404, events backend not found. Will only receive events that were sent locally on this process.`,
+              );
+              // Events backend is not present and not configured to always be used, bail out and stop polling
+              delete this.client;
+              return;
+            }
+            throw await ResponseError.fromResponse(res);
+          }
+        }
+
+        process.nextTick(poll);
       } catch (error) {
         this.logger.warn(
           `Poll failed for subscription "${subscriptionId}", retrying in ${backoffMs.toFixed(
@@ -292,7 +310,10 @@ class PluginEventsService implements EventsService {
     } catch (error) {
       // This is a bit hacky, but handles the case where new auth is used
       // without legacy auth fallback, and the events backend is not installed
-      if (String(error).includes('Unable to generate legacy token')) {
+      if (
+        String(error).includes('Unable to generate legacy token') &&
+        this.mode !== 'always'
+      ) {
         this.logger.warn(
           `The events backend is not available and neither is legacy auth. Future events will not be persisted.`,
         );
@@ -309,22 +330,25 @@ class PluginEventsService implements EventsService {
   }
 
   #isShuttingDown = false;
-  #shutdownLocks: Promise<void>[] = [];
+  #shutdownLocks = new Set<Promise<void>>();
 
   // This locking mechanism helps ensure that we are either idle or waiting for
   // a blocked events call before shutting down. It increases out changes of
   // never dropping any events on shutdown.
-  #getShutdownLock(): { release(): void } {
+  #getShutdownLock(): { release(): void } | undefined {
     if (this.#isShuttingDown) {
-      throw new Error('Service is shutting down');
+      return undefined;
     }
 
     let release: () => void;
-    this.#shutdownLocks.push(
-      new Promise<void>(resolve => {
-        release = resolve;
-      }),
-    );
+
+    const lock = new Promise<void>(resolve => {
+      release = () => {
+        resolve();
+        this.#shutdownLocks.delete(lock);
+      };
+    });
+    this.#shutdownLocks.add(lock);
     return { release: release! };
   }
 }
@@ -342,12 +366,30 @@ export class DefaultEventsService implements EventsService {
   private constructor(
     private readonly logger: LoggerService,
     private readonly localBus: LocalEventBus,
+    private readonly mode: EventBusMode,
   ) {}
 
-  static create(options: { logger: LoggerService }): DefaultEventsService {
+  static create(options: {
+    logger: LoggerService;
+    config?: RootConfigService;
+    useEventBus?: EventBusMode;
+  }): DefaultEventsService {
+    const eventBusMode =
+      options.useEventBus ??
+      ((options.config?.getOptionalString('events.useEventBus') ??
+        'auto') as EventBusMode);
+    if (!EVENT_BUS_MODES.includes(eventBusMode)) {
+      throw new Error(
+        `Invalid events.useEventBus config, must be one of ${EVENT_BUS_MODES.join(
+          ', ',
+        )}, got '${eventBusMode}'`,
+      );
+    }
+
     return new DefaultEventsService(
       options.logger,
       new LocalEventBus(options.logger),
+      eventBusMode,
     );
   }
 
@@ -367,16 +409,18 @@ export class DefaultEventsService implements EventsService {
     },
   ): EventsService {
     const client =
-      options &&
-      new DefaultApiClient({
-        discoveryApi: options.discovery,
-        fetchApi: { fetch }, // use native node fetch
-      });
+      options && this.mode !== 'never'
+        ? new DefaultApiClient({
+            discoveryApi: options.discovery,
+            fetchApi: { fetch }, // use native node fetch
+          })
+        : undefined;
     const logger = options?.logger ?? this.logger;
     const service = new PluginEventsService(
       pluginId,
       this.localBus,
       logger,
+      this.mode,
       client,
       options?.auth,
     );

plugins/events-node/src/api/index.ts

@@ -21,6 +21,9 @@ export type {
   EventsServiceSubscribeOptions,
   EventsServiceEventHandler,
 } from './EventsService';
-export { DefaultEventsService } from './DefaultEventsService';
+export {
+  DefaultEventsService,
+  type EventBusMode,
+} from './DefaultEventsService';
 export * from './http';
 export { SubTopicEventRouter } from './SubTopicEventRouter';

yarn.lock

@@ -6325,6 +6325,7 @@ __metadata:
     "@backstage/errors": "workspace:^"
     "@backstage/types": "workspace:^"
     cross-fetch: ^4.0.0
+    msw: ^1.0.0
     uri-template: ^2.0.0
   languageName: unknown
   linkType: soft