Merge pull request #125 from native-federation/fix/sse-bounded-pool

fix(sse): bound the build-notification connection pool

Author: Aukevanoost

src/builders/build/federation-build-notifier.spec.ts

@@ -0,0 +1,197 @@
+import type { IncomingMessage, ServerResponse } from "http";
+import { afterEach, describe, expect, it, vi } from "vitest";
+
+vi.mock("@softarc/native-federation/internal", () => ({
+  logger: { info: vi.fn(), verbose: vi.fn(), error: vi.fn(), warn: vi.fn() },
+}));
+
+import { BuildNotificationType } from "@softarc/native-federation";
+
+import { federationBuildNotifier } from "./federation-build-notifier.js";
+
+const ENDPOINT = "/@angular-architects/native-federation:build-notifications";
+
+// Minimal stand-ins for the node req/res pair: the notifier only writes to the response
+// and subscribes to the request's close/error events.
+function createClient() {
+  const listeners = new Map<string, () => void>();
+  const written: string[] = [];
+
+  const req = {
+    destroyed: false,
+    on(event: string, callback: () => void) {
+      listeners.set(event, callback);
+      return req;
+    },
+  } as unknown as IncomingMessage;
+
+  const res = {
+    destroyed: false,
+    writable: true,
+    writeHead: vi.fn(),
+    write: vi.fn((chunk: string) => {
+      written.push(chunk);
+      return true;
+    }),
+    end: vi.fn(() => {
+      res.destroyed = true;
+      res.writable = false;
+    }),
+  };
+
+  return {
+    req,
+    res: res as unknown as ServerResponse,
+    written,
+    ended: () => res.end.mock.calls.length > 0,
+    fire: (event: string) => listeners.get(event)?.(),
+  };
+}
+
+function connect(clients: number) {
+  const middleware = federationBuildNotifier.createEventMiddleware(() => ENDPOINT);
+  const next = vi.fn();
+
+  return Array.from({ length: clients }, () => {
+    const client = createClient();
+    middleware(client.req, client.res, next);
+    return client;
+  });
+}
+
+// The notifier is a module-level singleton, so each test has to hand back a clean pool.
+afterEach(() => {
+  federationBuildNotifier.stopEventServer();
+  vi.clearAllMocks();
+});
+
+describe("createEventMiddleware", () => {
+  it("passes the request along when the notifier is inactive", () => {
+    const middleware = federationBuildNotifier.createEventMiddleware(() => ENDPOINT);
+    const client = createClient();
+    const next = vi.fn();
+
+    middleware(client.req, client.res, next);
+
+    expect(next).toHaveBeenCalled();
+    expect(client.res.writeHead).not.toHaveBeenCalled();
+  });
+
+  it("passes the request along when the url is not the endpoint", () => {
+    federationBuildNotifier.initialize(ENDPOINT);
+    const middleware = federationBuildNotifier.createEventMiddleware(() => "/main.js");
+    const client = createClient();
+    const next = vi.fn();
+
+    middleware(client.req, client.res, next);
+
+    expect(next).toHaveBeenCalled();
+    expect(client.res.writeHead).not.toHaveBeenCalled();
+  });
+
+  it("opens an event stream on the endpoint", () => {
+    federationBuildNotifier.initialize(ENDPOINT);
+
+    const [client] = connect(1);
+
+    expect(client!.res.writeHead).toHaveBeenCalledWith(
+      200,
+      expect.objectContaining({ "Content-Type": "text/event-stream" }),
+    );
+    expect(federationBuildNotifier.activeConnections).toBe(1);
+  });
+
+  it("declares the reconnect delay before the first event", () => {
+    federationBuildNotifier.initialize(ENDPOINT);
+
+    const [client] = connect(1);
+
+    expect(client!.written[0]).toBe("retry: 5000\n");
+    expect(client!.written[1]).toContain('"type":"connected"');
+  });
+
+  it("drops a connection from the pool when the request closes", () => {
+    federationBuildNotifier.initialize(ENDPOINT);
+    const [client] = connect(1);
+
+    client!.fire("close");
+
+    expect(federationBuildNotifier.activeConnections).toBe(0);
+  });
+});
+
+describe("connection limit", () => {
+  it("holds at most 16 connections", () => {
+    federationBuildNotifier.initialize(ENDPOINT);
+
+    connect(20);
+
+    expect(federationBuildNotifier.activeConnections).toBe(16);
+  });
+
+  it("evicts the oldest connection rather than refusing the newest", () => {
+    federationBuildNotifier.initialize(ENDPOINT);
+
+    const clients = connect(17);
+
+    expect(clients[0]!.ended()).toBe(true);
+    expect(clients[16]!.ended()).toBe(false);
+  });
+
+  it("stops broadcasting to an evicted connection", () => {
+    federationBuildNotifier.initialize(ENDPOINT);
+    const clients = connect(17);
+    const evicted = clients[0]!;
+    const writesBefore = evicted.written.length;
+
+    federationBuildNotifier.broadcastBuildCompletion();
+
+    expect(evicted.written.length).toBe(writesBefore);
+    expect(clients[16]!.written.at(-1)).toContain(BuildNotificationType.COMPLETED);
+  });
+});
+
+describe("broadcasts", () => {
+  it("sends the completion event to every connection", () => {
+    federationBuildNotifier.initialize(ENDPOINT);
+    const clients = connect(3);
+
+    federationBuildNotifier.broadcastBuildCompletion();
+
+    for (const client of clients) {
+      expect(client.written.at(-1)).toContain(BuildNotificationType.COMPLETED);
+    }
+  });
+
+  it("sends the error message with the error event", () => {
+    federationBuildNotifier.initialize(ENDPOINT);
+    const [client] = connect(1);
+
+    federationBuildNotifier.broadcastBuildError(new Error("boom"));
+
+    expect(client!.written.at(-1)).toContain(BuildNotificationType.ERROR);
+    expect(client!.written.at(-1)).toContain("boom");
+  });
+
+  it("does nothing when the notifier is inactive", () => {
+    const [client] = connect(1);
+    const writesBefore = client!.written.length;
+
+    federationBuildNotifier.broadcastBuildCancellation();
+
+    expect(client!.written.length).toBe(writesBefore);
+  });
+});
+
+describe("stopEventServer", () => {
+  it("closes every connection and empties the pool", () => {
+    federationBuildNotifier.initialize(ENDPOINT);
+    const clients = connect(3);
+
+    federationBuildNotifier.stopEventServer();
+
+    expect(clients.every((client) => client.ended())).toBe(true);
+    expect(federationBuildNotifier.activeConnections).toBe(0);
+    expect(federationBuildNotifier.isRunning).toBe(false);
+  });
+});

src/builders/build/federation-build-notifier.ts

@@ -21,6 +21,9 @@ interface FederationEvent {
 type NextFunction = (error?: Error) => void;
 type MiddlewareFunction = (req: IncomingMessage, res: ServerResponse, next: NextFunction) => void;
 
+const MAX_CONNECTIONS = 16;
+const RECONNECT_DELAY_MS = 5000;
+
 /**
  * Manages Server-Sent Events for federation hot reload in local development
  * Only active when running in development mode with dev server
@@ -71,6 +74,8 @@ class FederationBuildNotifier {
    * Sets up a new SSE connection
    */
   private _setupSSEConnection(req: IncomingMessage, res: ServerResponse): void {
+    this._evictOverflow();
+
     res.writeHead(200, {
       'Content-Type': 'text/event-stream',
       'Cache-Control': 'no-cache',
@@ -79,6 +84,9 @@ class FederationBuildNotifier {
       'Access-Control-Allow-Headers': 'Cache-Control',
     });
 
+    // Pins the reconnect backoff instead of leaving it to the browser default.
+    res.write(`retry: ${RECONNECT_DELAY_MS}\n`);
+
     // Send initial connection event
     this._sendEvent(res, {
       type: 'connected',
@@ -98,6 +106,28 @@ class FederationBuildNotifier {
     );
   }
 
+  /**
+   * Drops the oldest connections once the pool is full
+   *
+   * Behind a reverse proxy the peer is the proxy rather than the browser, so a stream the
+   * client already abandoned still looks writable here and cannot be detected as stale.
+   * Bounding the pool caps how many such connections can pile up.
+   */
+  private _evictOverflow(): void {
+    while (this.connections.length >= MAX_CONNECTIONS) {
+      const oldest = this.connections.shift();
+      if (!oldest) return;
+
+      try {
+        oldest.response.end();
+      } catch {
+        // Connection might already be closed
+      }
+
+      logger.info('[Federation SSE] Connection limit reached, dropped the oldest connection');
+    }
+  }
+
   /**
    * Removes a connection from the pool
    */

src/builders/build/schema.json

@@ -71,11 +71,12 @@
     },
     "buildNotifications": {
       "type": "object",
+      "description": "Opt in to build completion notifications. Omitting this option leaves them off; the defaults below only apply once it is present.",
       "properties": {
         "enable": {
           "type": "boolean",
           "default": true,
-          "description": "Enable build completion notifications for local development. It will send events to notify when the federation build is complete."
+          "description": "Enable build completion notifications for local development. It will send events to notify when the federation build is complete. Each connected browser tab holds an event stream open, so the runtime also has to opt in with 'sse: true'."
         },
         "endpoint": {
           "type": "string",