fix: harden retained log streaming

apps/nextjs/src/app/[locale]/manage/tools/logs/client.tsx

@@ -1,15 +1,14 @@
 "use client";
 
-import type { ComponentType } from "react";
 import dynamic from "next/dynamic";
 
 interface ClientSideTerminalComponentProps {
   focusTimestamp?: number;
 }
 
-export const ClientSideTerminalComponent = dynamic(
+export const ClientSideTerminalComponent = dynamic<ClientSideTerminalComponentProps>(
   () => import("./terminal").then(({ TerminalComponent }) => TerminalComponent),
   {
     ssr: false,
   },
-) as ComponentType<ClientSideTerminalComponentProps>;
+);

apps/nextjs/src/app/[locale]/manage/tools/logs/page.tsx

@@ -39,14 +39,9 @@ export default async function LogsManagementPage({ searchParams }: LogsManagemen
     notFound();
   }
 
-  const { focus } = await searchParams;
-  let focusTimestamp: number | undefined;
-  if (focus !== undefined) {
-    const parsedFocusTimestamp = Number(focus);
-    if (Number.isSafeInteger(parsedFocusTimestamp) && parsedFocusTimestamp > 0) {
-      focusTimestamp = parsedFocusTimestamp;
-    }
-  }
+  const parsedFocusTimestamp = Number((await searchParams).focus);
+  const focusTimestamp =
+    Number.isSafeInteger(parsedFocusTimestamp) && parsedFocusTimestamp > 0 ? parsedFocusTimestamp : undefined;
 
   return (
     <LogContextProvider defaultLevel={logsEnv.LEVEL}>

apps/nextjs/src/app/[locale]/manage/tools/logs/terminal.tsx

@@ -17,37 +17,22 @@ import classes from "./terminal.module.css";
 const ALL_LOG_LEVELS = [...logLevels];
 const TERMINAL_SCROLLBACK_LINES = LOG_HISTORY_MAX_ENTRIES * 4;
 
-interface FocusMarker {
-  index: number;
-  text: string;
-}
-
 const getFocusMarker = (
   messages: LoggerMessage[],
   visibleMessages: LoggerMessage[],
   focusTimestamp: number | undefined,
   markerText: string,
   expiredMarkerText: string,
-): FocusMarker | null => {
+) => {
   if (focusTimestamp === undefined) return null;
 
-  let oldestTimestamp = Number.POSITIVE_INFINITY;
-  for (const message of messages) {
-    oldestTimestamp = Math.min(oldestTimestamp, message.timestamp.getTime());
-  }
-  if (oldestTimestamp === Number.POSITIVE_INFINITY || focusTimestamp < oldestTimestamp) {
+  const oldestTimestamp = messages[0]?.timestamp.getTime();
+  if (oldestTimestamp === undefined || focusTimestamp < oldestTimestamp) {
     return { index: 0, text: expiredMarkerText };
   }
 
-  let markerIndex = 0;
-  for (const [index, message] of visibleMessages.entries()) {
-    if (message.timestamp.getTime() <= focusTimestamp) {
-      markerIndex = index + 1;
-    }
-  }
-
   return {
-    index: markerIndex,
+    index: visibleMessages.findLastIndex((message) => message.timestamp.getTime() <= focusTimestamp) + 1,
     text: markerText,
   };
 };
@@ -104,10 +89,7 @@ const renderMessages = ({
     output.push(`\x1b[33m--- ${focusMarker.text} ---\x1b[0m`);
   }
 
-  terminal.reset();
-  if (output.length === 0) return;
-
-  terminal.write(`${output.join("\r\n")}\r\n`, () => {
+  terminal.write(`\x1bc${output.join("\r\n")}${output.length > 0 ? "\r\n" : ""}`, () => {
     if (!focusMarker) {
       terminal.scrollToBottom();
       return;
@@ -134,8 +116,8 @@ export const TerminalComponent = ({ focusTimestamp }: TerminalComponentProps) =>
   const messagesRef = useRef<LoggerMessage[]>([]);
   const initialFontSizeRef = useRef(fontSize);
   const activeLevelsRef = useRef(activeLevels);
-  const markerTextRef = useRef(t("log.context.widgetError"));
-  const expiredMarkerTextRef = useRef(t("log.context.expiredWidgetError"));
+  const markerTextRef = useRef("");
+  const expiredMarkerTextRef = useRef("");
   const focusTimestampRef = useRef(focusTimestamp);
 
   activeLevelsRef.current = activeLevels;

packages/api/src/router/test/log.spec.ts

@@ -0,0 +1,79 @@
+import { describe, expect, test, vi } from "vitest";
+
+import type { Session } from "@homarr/auth";
+import type { LoggerMessage } from "@homarr/redis";
+
+import { logRouter } from "../log";
+
+const mocks = vi.hoisted(() => ({
+  getRecentAsync: vi.fn(),
+  subscribeAsync: vi.fn(),
+}));
+
+vi.mock("@homarr/redis", () => ({
+  loggingChannel: {
+    getRecentAsync: mocks.getRecentAsync,
+    subscribeAsync: mocks.subscribeAsync,
+  },
+}));
+
+const session = {
+  user: {
+    id: "log-viewer",
+    permissions: ["other-view-logs"],
+    colorScheme: "light",
+  },
+  expires: new Date(0).toISOString(),
+} satisfies Session;
+
+const createMessage = (id: string, timestamp: number): LoggerMessage => ({
+  id,
+  timestamp: new Date(timestamp),
+  level: "info",
+  message: id,
+});
+
+describe("log.subscribe", () => {
+  test("emits history before buffered live messages without duplicates", async () => {
+    const historyDeferred = Promise.withResolvers<LoggerMessage[]>();
+    let onMessage: ((message: LoggerMessage) => void) | undefined;
+    const unsubscribe = vi.fn();
+
+    mocks.subscribeAsync.mockImplementation(async (callback: (message: LoggerMessage) => void) => {
+      onMessage = callback;
+      return unsubscribe;
+    });
+    mocks.getRecentAsync.mockReturnValue(historyDeferred.promise);
+
+    const caller = logRouter.createCaller({
+      db: null as never,
+      deviceType: undefined,
+      session,
+    });
+    const subscription = await caller.subscribe({ levels: ["info"] });
+    const events: object[] = [];
+    const observer = subscription.subscribe({
+      next(event) {
+        events.push(event);
+      },
+    });
+
+    await vi.waitFor(() => expect(mocks.getRecentAsync).toHaveBeenCalledOnce());
+
+    const retainedMessage = createMessage("retained", 1);
+    const liveMessage = createMessage("live", 2);
+    onMessage?.(retainedMessage);
+    onMessage?.(liveMessage);
+    historyDeferred.resolve([retainedMessage]);
+
+    await vi.waitFor(() =>
+      expect(events).toEqual([
+        { type: "history", messages: [retainedMessage] },
+        { type: "message", message: liveMessage },
+      ]),
+    );
+
+    observer.unsubscribe();
+    expect(unsubscribe).toHaveBeenCalledOnce();
+  });
+});

packages/core/src/infrastructure/logs/constants.ts

@@ -1,12 +1,15 @@
+import z from "zod/v4";
+
 export const logLevels = ["error", "warn", "info", "debug"] as const;
 export type LogLevel = (typeof logLevels)[number];
 
-export interface LoggerMessage {
-  id: string;
-  timestamp: Date;
-  message: string;
-  level: LogLevel;
-}
+export const loggerMessageSchema = z.object({
+  id: z.string(),
+  timestamp: z.date(),
+  message: z.string(),
+  level: z.enum(logLevels),
+});
+export type LoggerMessage = z.infer<typeof loggerMessageSchema>;
 
 export const LOG_HISTORY_MAX_ENTRIES = 5_000;
 export const LOG_HISTORY_KEY = "list:logging:history";

packages/core/src/infrastructure/logs/transports/redis-transport.ts

@@ -25,9 +25,6 @@ export class RedisTransport extends Transport {
       this.emit("logged", info);
     });
 
-    // Is only initialized here because it did not work when initialized in the constructor or outside the class
-    this.redis ??= createRedisClient();
-
     const message: LoggerMessage = {
       id: randomUUID(),
       timestamp: new Date(),
@@ -36,15 +33,24 @@ export class RedisTransport extends Transport {
     };
     const serializedMessage = stringify(message);
 
-    this.redis
-      .multi()
-      .lpush(LOG_HISTORY_KEY, serializedMessage)
-      .ltrim(LOG_HISTORY_KEY, 0, LOG_HISTORY_MAX_ENTRIES - 1)
-      .publish(LOG_PUBLISH_CHANNEL, serializedMessage)
-      .exec()
-      .catch(() => {
-        // Ignore errors
+    callback();
+
+    void Promise.resolve()
+      .then(() => {
+        // Is only initialized here because it did not work when initialized in the constructor or outside the class
+        if (!this.redis) {
+          this.redis = createRedisClient();
+          this.redis.options.maxRetriesPerRequest = 1;
+          this.redis.on("error", () => undefined);
+        }
+
+        return this.redis
+          .multi()
+          .lpush(LOG_HISTORY_KEY, serializedMessage)
+          .ltrim(LOG_HISTORY_KEY, 0, LOG_HISTORY_MAX_ENTRIES - 1)
+          .publish(LOG_PUBLISH_CHANNEL, serializedMessage)
+          .exec();
       })
-      .finally(callback);
+      .catch(() => undefined);
   }
 }

packages/redis/src/index.ts

@@ -1,9 +1,15 @@
 import { parse } from "superjson";
 
 import type { LoggerMessage } from "@homarr/core/infrastructure/logs/constants";
-import { LOG_HISTORY_KEY, LOG_HISTORY_MAX_ENTRIES } from "@homarr/core/infrastructure/logs/constants";
+import {
+  loggerMessageSchema,
+  LOG_HISTORY_KEY,
+  LOG_HISTORY_MAX_ENTRIES,
+  LOG_PUBLISH_CHANNEL,
+} from "@homarr/core/infrastructure/logs/constants";
 
 import { createListChannel, createSubPubChannel } from "./lib/channel";
+import { ChannelSubscriptionTracker } from "./lib/channel-subscription-tracker";
 import { createRedisConnection } from "./lib/connection";
 
 export {
@@ -21,14 +27,29 @@ export const pingUrlChannel = createListChannel<string>("ping-url");
 
 export type { LoggerMessage } from "@homarr/core/infrastructure/logs/constants";
 
-const liveLoggingChannel = createSubPubChannel<LoggerMessage>("logging", { persist: false });
 const loggingHistoryClient = createRedisConnection();
 
+const parseLoggerMessage = (message: string): LoggerMessage | null => {
+  try {
+    const result = loggerMessageSchema.safeParse(parse<unknown>(message));
+    return result.success ? result.data : null;
+  } catch {
+    return null;
+  }
+};
+
 export const loggingChannel = {
-  subscribe: liveLoggingChannel.subscribe,
-  subscribeAsync: liveLoggingChannel.subscribeAsync,
+  subscribeAsync: (callback: (message: LoggerMessage) => void) =>
+    ChannelSubscriptionTracker.subscribeAsync(LOG_PUBLISH_CHANNEL, (message) => {
+      const parsedMessage = parseLoggerMessage(message);
+      if (parsedMessage) callback(parsedMessage);
+    }),
   getRecentAsync: async () => {
-    const messages = await loggingHistoryClient.lrange(LOG_HISTORY_KEY, 0, LOG_HISTORY_MAX_ENTRIES - 1);
-    return messages.map((message) => parse<LoggerMessage>(message)).toReversed();
+    const messages =
+      (await loggingHistoryClient?.lrange(LOG_HISTORY_KEY, 0, LOG_HISTORY_MAX_ENTRIES - 1).catch(() => [])) ?? [];
+    return messages
+      .map(parseLoggerMessage)
+      .filter((message): message is LoggerMessage => message !== null)
+      .toReversed();
   },
 };

packages/redis/src/lib/channel-subscription-tracker.ts

@@ -58,14 +58,18 @@ export class ChannelSubscriptionTracker {
 
     const channelSubscriptions = this.subscriptions.get(channelName) ?? new Map<string, SubscriptionCallback>();
     const id = randomUUID();
-    let ready = this.subscriptionReadiness.get(channelName) ?? Promise.resolve();
+    let ready = this.subscriptionReadiness.get(channelName);
 
-    // If there are no subscriptions to the channel, subscribe to it
-    if (channelSubscriptions.size === 0) {
+    // Subscribe when this channel has no active or pending Redis subscription.
+    if (!ready) {
       logger.debug("Subscribing to redis channel", { channel: channelName });
       ready = this.redis.subscribe(channelName).then(() => undefined);
       this.subscriptionReadiness.set(channelName, ready);
-      void ready.catch(() => undefined);
+      void ready.catch(() => {
+        if (this.subscriptionReadiness.get(channelName) === ready) {
+          this.subscriptionReadiness.delete(channelName);
+        }
+      });
     }
 
     logger.debug("Adding redis channel callback", { channel: channelName, id });

packages/redis/src/lib/channel.ts

@@ -1,4 +1,4 @@
-import superjson, { parse } from "superjson";
+import superjson from "superjson";
 
 import { createId } from "@homarr/common";
 
@@ -8,12 +8,6 @@ import { createRedisConnection } from "./connection";
 const publisher = createRedisConnection();
 const lastDataClient = createRedisConnection();
 
-const deserializeCallback =
-  <TData>(callback: (data: TData) => void) =>
-  (message: string) => {
-    callback(parse(message));
-  };
-
 /**
  * Creates a new pub/sub channel.
  * @param name name of the channel
@@ -35,19 +29,9 @@ export const createSubPubChannel = <TData>(name: string, { persist }: { persist:
           }
         });
       }
-      return ChannelSubscriptionTracker.subscribe(channelName, deserializeCallback(callback));
-    },
-    /**
-     * Subscribes to the channel and resolves once Redis confirms the subscription.
-     */
-    subscribeAsync: async (callback: (data: TData) => void) => {
-      if (persist) {
-        const data = await lastDataClient.get(lastChannelName);
-        if (data) {
-          callback(superjson.parse(data));
-        }
-      }
-      return await ChannelSubscriptionTracker.subscribeAsync(channelName, deserializeCallback(callback));
+      return ChannelSubscriptionTracker.subscribe(channelName, (message) => {
+        callback(superjson.parse(message));
+      });
     },
     /**
      * Publish data to the channel with last data saved.

packages/widgets/src/errors/base-component.tsx

@@ -19,14 +19,13 @@ export const BaseWidgetError = (props: BaseWidgetErrorProps) => {
   const t = useI18n();
   const { data: session } = useSession();
   const [errorTimestamp] = useState(Date.now);
-  const canViewLogs = session?.user.permissions.includes("other-view-logs") ?? false;
 
   return (
     <Stack h="100%" align="center" justify="center" gap="md" data-homarr-widget-error>
       <props.icon size={40} />
       <Stack gap={0}>
         <Text ta="center">{translateIfNecessary(t, props.message)}</Text>
-        {props.showLogsLink && canViewLogs && (
+        {props.showLogsLink && session?.user.permissions.includes("other-view-logs") && (
           <Anchor
             component={Link}
             href={`/manage/tools/logs?focus=${errorTimestamp}`}

packages/widgets/src/errors/component.spec.ts

@@ -11,6 +11,10 @@ vi.mock("@homarr/translation/client", () => ({
   useI18n: () => (key: string) => key,
 }));
 
+vi.mock("@homarr/auth/client", () => ({
+  useSession: () => ({ data: null }),
+}));
+
 const matchMedia = (query: string) => ({
   matches: false,
   media: query,