fix(beszel): prevent live subscription memory growth (#6448)

* fix(beszel): bound live stats subscriptions

* fix(beszel): avoid blocking subscription teardown

* fix(import): map new clock options

packages/api/src/router/widgets/__tests__/bounded-async-queue.spec.ts

@@ -0,0 +1,45 @@
+import { describe, expect, test } from "vitest";
+
+import { BoundedAsyncQueue } from "../bounded-async-queue";
+
+describe("BoundedAsyncQueue", () => {
+  test("keeps only the newest values when the queue is full", async () => {
+    const queue = new BoundedAsyncQueue<number>(2);
+    queue.push(1);
+    queue.push(2);
+    queue.push(3);
+    queue.close();
+
+    await expect(queue.next()).resolves.toEqual({ value: 2, done: false });
+    await expect(queue.next()).resolves.toEqual({ value: 3, done: false });
+    await expect(queue.next()).resolves.toEqual({ value: undefined, done: true });
+  });
+
+  test("resolves a pending read as soon as a value arrives", async () => {
+    const queue = new BoundedAsyncQueue<string>(1);
+    const next = queue.next();
+
+    queue.push("latest");
+
+    await expect(next).resolves.toEqual({ value: "latest", done: false });
+  });
+
+  test("resolves a pending read when the producer closes", async () => {
+    const queue = new BoundedAsyncQueue<number>(1);
+    const next = queue.next();
+
+    queue.close();
+
+    await expect(next).resolves.toEqual({ value: undefined, done: true });
+  });
+
+  test("rejects a pending read when the producer fails", async () => {
+    const queue = new BoundedAsyncQueue<number>(1);
+    const next = queue.next();
+    const error = new Error("upstream failed");
+
+    queue.fail(error);
+
+    await expect(next).rejects.toBe(error);
+  });
+});

packages/api/src/router/widgets/beszel.ts

@@ -1,5 +1,4 @@
 import { TRPCError } from "@trpc/server";
-import { observable } from "@trpc/server/observable";
 import { z } from "zod/v4";
 
 import { createLogger } from "@homarr/core/infrastructure/logs";
@@ -14,8 +13,10 @@ import {
 import { settleIntegrationQueries } from "../../settle-integrations";
 import { createManyIntegrationMiddleware } from "../../middlewares/integration";
 import { createTRPCRouter, publicProcedure } from "../../trpc";
+import { BoundedAsyncQueue } from "./bounded-async-queue";
 
 const logger = createLogger({ module: "beszelRouter" });
+const MAX_PENDING_LIVE_EVENTS = 4;
 
 const errorMessage = (error: unknown) => (error instanceof Error ? error.message : String(error));
 
@@ -203,98 +204,94 @@ export const beszelRouter = createTRPCRouter({
         systemId: z.string(),
       }),
     )
-    .subscription(({ ctx, input }) => {
-      return observable<LiveStatsEvent>((emit) => {
-        const controller = new AbortController();
-        let isActive = true;
-        let emittedEventCount = 0;
+    .subscription(async function* ({ ctx, input, signal }) {
+      const integration = ctx.integrations[0];
+      if (!integration) {
+        throw new TRPCError({ code: "BAD_REQUEST", message: "At least one Beszel integrationId is required" });
+      }
 
-        logger.debug("Beszel realtime subscription started", {
-          userId: ctx.session?.user?.id,
-          integrationIds: ctx.integrations.map((integration) => integration.id),
-          systemId: input.systemId,
-        });
+      const queue = new BoundedAsyncQueue<LiveStatsEvent>(MAX_PENDING_LIVE_EVENTS);
+      const controller = new AbortController();
+      let emittedEventCount = 0;
 
+      const stop = () => {
+        controller.abort();
+        void queue.return();
+      };
+      if (signal?.aborted) return;
+      signal?.addEventListener("abort", stop, { once: true });
+
+      logger.debug("Beszel realtime subscription started", {
+        userId: ctx.session?.user?.id,
+        integrationIds: ctx.integrations.map((candidate) => candidate.id),
+        systemId: input.systemId,
+      });
+
+      try {
+        const instance = await createIntegrationAsync(integration);
+        if (controller.signal.aborted) return;
         void (async () => {
-          const integration = ctx.integrations[0];
-          if (!integration) {
-            emit.error(
-              new TRPCError({ code: "BAD_REQUEST", message: "At least one Beszel integrationId is required" }),
-            );
-            return;
-          }
-
           try {
-            const instance = await createIntegrationAsync(integration);
-            if (typeof instance.subscribeRealtimeMetrics === "function") {
-              await instance.subscribeRealtimeMetrics(
-                input.systemId,
-                (event) => {
-                  if (!isActive) return;
-                  emittedEventCount += 1;
-                  if (emittedEventCount <= 2 || emittedEventCount % 60 === 0) {
-                    logger.debug("Forwarding Beszel realtime events", {
-                      userId: ctx.session?.user?.id,
-                      integrationId: integration.id,
-                      systemId: input.systemId,
-                      eventType: event.type,
-                      emittedEventCount,
-                      statsCount: Array.isArray(event.record.stats) ? event.record.stats.length : undefined,
-                    });
-                  }
-                  emit.next(event);
-                },
-                controller.signal,
-              );
-            } else {
-              emit.error(
-                new TRPCError({
-                  code: "METHOD_NOT_SUPPORTED",
-                  message: `Integration ${integration.kind} does not support realtime metrics`,
-                }),
-              );
-            }
-          } catch (error) {
-            if (isActive) {
-              logger.warn("Beszel realtime subscription failed", {
-                userId: ctx.session?.user?.id,
-                integrationId: integration.id,
-                systemId: input.systemId,
-                emittedEventCount,
-                error: errorMessage(error),
+            if (typeof instance.subscribeRealtimeMetrics !== "function") {
+              throw new TRPCError({
+                code: "METHOD_NOT_SUPPORTED",
+                message: `Integration ${integration.kind} does not support realtime metrics`,
               });
-              emit.error(
-                error instanceof TRPCError
-                  ? error
-                  : new TRPCError({
-                      code: "INTERNAL_SERVER_ERROR",
-                      message: error instanceof Error ? error.message : String(error),
-                    }),
-              );
             }
-          }
 
-          if (isActive) {
-            logger.debug("Beszel realtime subscription completed", {
+            await instance.subscribeRealtimeMetrics(
+              input.systemId,
+              (event) => {
+                emittedEventCount += 1;
+                if (emittedEventCount <= 2 || emittedEventCount % 60 === 0) {
+                  logger.debug("Forwarding Beszel realtime events", {
+                    userId: ctx.session?.user?.id,
+                    integrationId: integration.id,
+                    systemId: input.systemId,
+                    eventType: event.type,
+                    emittedEventCount,
+                    statsCount: Array.isArray(event.record.stats) ? event.record.stats.length : undefined,
+                  });
+                }
+                queue.push(event);
+              },
+              controller.signal,
+            );
+            queue.close();
+          } catch (error) {
+            if (controller.signal.aborted) {
+              queue.close();
+              return;
+            }
+
+            logger.warn("Beszel realtime subscription failed", {
               userId: ctx.session?.user?.id,
               integrationId: integration.id,
               systemId: input.systemId,
               emittedEventCount,
+              error: errorMessage(error),
             });
-            emit.complete();
+            queue.fail(
+              error instanceof TRPCError
+                ? error
+                : new TRPCError({
+                    code: "INTERNAL_SERVER_ERROR",
+                    message: error instanceof Error ? error.message : String(error),
+                  }),
+            );
           }
         })();
 
-        return () => {
-          isActive = false;
-          logger.debug("Beszel realtime subscription cancelled", {
-            userId: ctx.session?.user?.id,
-            integrationIds: ctx.integrations.map((integration) => integration.id),
-            systemId: input.systemId,
-            emittedEventCount,
-          });
-          controller.abort();
-        };
-      });
+        for await (const event of queue) yield event;
+      } finally {
+        stop();
+        signal?.removeEventListener("abort", stop);
+        logger.debug("Beszel realtime subscription stopped", {
+          userId: ctx.session?.user?.id,
+          integrationId: integration.id,
+          systemId: input.systemId,
+          emittedEventCount,
+        });
+      }
     }),
 });

packages/api/src/router/widgets/bounded-async-queue.ts

@@ -0,0 +1,88 @@
+type QueueResult<T> = IteratorResult<T, undefined>;
+
+type PendingRead<T> = {
+  resolve: (result: QueueResult<T>) => void;
+  reject: (reason?: unknown) => void;
+};
+
+/**
+ * A small async queue for live streams.
+ *
+ * Live metrics are snapshots, so retaining every snapshot while a client is
+ * slow is both unnecessary and dangerous. When the queue is full, discard
+ * the oldest pending snapshot and keep the newest one.
+ */
+export class BoundedAsyncQueue<T> implements AsyncIterableIterator<T> {
+  private readonly values: T[] = [];
+  private pendingRead: PendingRead<T> | undefined;
+  private closed = false;
+  private failed = false;
+  private failure: unknown;
+
+  public constructor(private readonly maxSize: number) {
+    if (!Number.isInteger(maxSize) || maxSize < 1) {
+      throw new Error("A bounded async queue requires a positive integer max size");
+    }
+  }
+
+  public push(value: T): void {
+    if (this.closed) return;
+
+    if (this.pendingRead) {
+      const pendingRead = this.pendingRead;
+      this.pendingRead = undefined;
+      pendingRead.resolve({ value, done: false });
+      return;
+    }
+
+    if (this.values.length >= this.maxSize) this.values.shift();
+    this.values.push(value);
+  }
+
+  public close(): void {
+    if (this.closed) return;
+    this.closed = true;
+
+    if (this.pendingRead) {
+      const pendingRead = this.pendingRead;
+      this.pendingRead = undefined;
+      pendingRead.resolve({ value: undefined, done: true });
+    }
+  }
+
+  public fail(error: unknown): void {
+    if (this.closed) return;
+    this.closed = true;
+    this.failed = true;
+    this.failure = error;
+    this.values.length = 0;
+
+    if (this.pendingRead) {
+      const pendingRead = this.pendingRead;
+      this.pendingRead = undefined;
+      pendingRead.reject(error);
+    }
+  }
+
+  public next(): Promise<QueueResult<T>> {
+    if (this.values.length > 0) {
+      return Promise.resolve({ value: this.values.shift() as T, done: false });
+    }
+    if (this.failed) return Promise.reject(this.failure);
+    if (this.closed) return Promise.resolve({ value: undefined, done: true });
+
+    return new Promise<QueueResult<T>>((resolve, reject) => {
+      this.pendingRead = { resolve, reject };
+    });
+  }
+
+  public return(): Promise<QueueResult<T>> {
+    this.values.length = 0;
+    this.close();
+    return Promise.resolve({ value: undefined, done: true });
+  }
+
+  public [Symbol.asyncIterator](): AsyncIterableIterator<T> {
+    return this;
+  }
+}