fix: resolve Redis null client crashes on startup

After #5600 merged tasks/websocket into the Next.js process, `createRedisConnection()` returns `null` because `CI=true` is baked into the bundle at build time. This causes `TypeError: Cannot read properties of null` on every `.publish()` / `.get()` call during cron job startup. - Remove the CI/DISABLE_REDIS_LOGS null guard from createRedisConnection - Lazy-init all Redis clients to avoid build-time evaluation - Add a Redis readiness retry loop before starting cron jobs
2026-05-21 16:33:43 +02:00
1 parent dade477b32
commit a6ec5f789d
@@ -1,11 +1,33 @@
 // This import has to be the first import in the file so that the agent is overridden before any other modules are imported.
 import "./overrides";

+import { createLogger } from "@homarr/core/infrastructure/logs";
 import { jobGroup } from "@homarr/cron-jobs";
+import { handshakeAsync } from "@homarr/redis";

 import { onStartAsync } from "./on-start";

+const logger = createLogger({ module: "tasks" });
+
+const MAX_RETRIES = 10;
+const RETRY_DELAY_MS = 500;
+
+const waitForRedisAsync = async () => {
+  for (let attempt = 1; attempt <= MAX_RETRIES; attempt++) {
+    try {
+      await handshakeAsync();
+      logger.info("Redis connection established");
+      return;
+    } catch {
+      logger.warn(`Redis not ready, retrying (${attempt}/${MAX_RETRIES})...`);
+      await new Promise((resolve) => setTimeout(resolve, RETRY_DELAY_MS));
+    }
+  }
+  logger.error("Redis did not become ready after retries, proceeding anyway");
+};
+
 void (async () => {
+  await waitForRedisAsync();
  await onStartAsync();
  await jobGroup.initializeAsync();
  await jobGroup.startAllAsync();
@@ -2,6 +2,7 @@ import { randomUUID } from "crypto";

 import type { MaybePromise } from "@homarr/common/types";
 import { createLogger } from "@homarr/core/infrastructure/logs";
+import type { RedisClient } from "@homarr/core/infrastructure/redis";

 import { createRedisConnection } from "./connection";

@@ -18,7 +19,10 @@ type SubscriptionCallback = (message: string) => MaybePromise<void>;
 */
 export class ChannelSubscriptionTracker {
  private static subscriptions = new Map<string, Map<string, SubscriptionCallback>>();
-  private static redis = createRedisConnection();
+  private static _redis: RedisClient | undefined;
+  private static get redis() {
+    return (this._redis ??= createRedisConnection());
+  }
  private static listenerActive = false;

  /**
@@ -2,6 +2,7 @@ import superjson from "superjson";

 import { createId, hashObjectBase64 } from "@homarr/common";
 import { createLogger } from "@homarr/core/infrastructure/logs";
+import type { RedisClient } from "@homarr/core/infrastructure/redis";
 import type { WidgetKind } from "@homarr/definitions";

 import { ChannelSubscriptionTracker } from "./channel-subscription-tracker";
@@ -9,8 +10,11 @@ import { createRedisConnection } from "./connection";

 const logger = createLogger({ module: "redisChannel" });

-const publisher = createRedisConnection();
-const lastDataClient = createRedisConnection();
+let _publisher: RedisClient | undefined;
+const getPublisher = () => (_publisher ??= createRedisConnection());
+
+let _lastDataClient: RedisClient | undefined;
+const getLastDataClient = () => (_lastDataClient ??= createRedisConnection());

 /**
 * Creates a new pub/sub channel.
@@ -27,34 +31,33 @@ export const createSubPubChannel = <TData>(name: string, { persist }: { persist:
     */
    subscribe: (callback: (data: TData) => void) => {
      if (persist) {
-        void lastDataClient.get(lastChannelName).then((data) => {
-          if (data) {
-            callback(superjson.parse(data));
-          }
-        });
+        void getLastDataClient()
+          .get(lastChannelName)
+          .then((data) => {
+            if (data) {
+              callback(superjson.parse(data));
+            }
+          });
      }
      return ChannelSubscriptionTracker.subscribe(channelName, (message) => {
        callback(superjson.parse(message));
      });
    },
-    /**
-     * Publish data to the channel with last data saved.
-     * @param data data to be published
-     */
    publishAsync: async (data: TData) => {
      if (persist) {
-        await lastDataClient.set(lastChannelName, superjson.stringify(data));
+        await getLastDataClient().set(lastChannelName, superjson.stringify(data));
      }
-      await publisher.publish(channelName, superjson.stringify(data));
+      await getPublisher().publish(channelName, superjson.stringify(data));
    },
    getLastDataAsync: async () => {
-      const data = await lastDataClient.get(lastChannelName);
+      const data = await getLastDataClient().get(lastChannelName);
      return data ? superjson.parse<TData>(data) : null;
    },
  };
 };

-const getSetClient = createRedisConnection();
+let _getSetClient: RedisClient | undefined;
+const getSetClient = () => (_getSetClient ??= createRedisConnection());

 /**
 * Creates a new redis channel for a list
@@ -69,7 +72,7 @@ export const createListChannel = <TItem>(name: string) => {
     * @returns an array of all items
     */
    getAllAsync: async () => {
-      const items = await getSetClient.lrange(listChannelName, 0, -1);
+      const items = await getSetClient().lrange(listChannelName, 0, -1);
      return items.map((item) => superjson.parse<TItem>(item));
    },
    /**
@@ -77,20 +80,20 @@ export const createListChannel = <TItem>(name: string) => {
     * @param item item to remove
     */
    removeAsync: async (item: TItem) => {
-      await getSetClient.lrem(listChannelName, 0, superjson.stringify(item));
+      await getSetClient().lrem(listChannelName, 0, superjson.stringify(item));
    },
    /**
     * Clear all items from the channels list
     */
    clearAsync: async () => {
-      await getSetClient.del(listChannelName);
+      await getSetClient().del(listChannelName);
    },
    /**
     * Add an item to the channels list
     * @param item item to add
     */
    addAsync: async (item: TItem) => {
-      await getSetClient.lpush(listChannelName, superjson.stringify(item));
+      await getSetClient().lpush(listChannelName, superjson.stringify(item));
    },
  };
 };
@@ -106,7 +109,7 @@ export const createGetSetChannel = <TData>(name: string) => {
     * @returns data or null if not found
     */
    getAsync: async () => {
-      const data = await getSetClient.get(name);
+      const data = await getSetClient().get(name);
      return data ? superjson.parse<TData>(data) : null;
    },
    /**
@@ -114,13 +117,13 @@ export const createGetSetChannel = <TData>(name: string) => {
     * @param data data to be stored in the channel
     */
    setAsync: async (data: TData) => {
-      await getSetClient.set(name, superjson.stringify(data));
+      await getSetClient().set(name, superjson.stringify(data));
    },
    /**
     * Remove data from the channel
     */
    removeAsync: async () => {
-      await getSetClient.del(name);
+      await getSetClient().del(name);
    },
  };
 };
@@ -140,7 +143,7 @@ export const createCacheChannel = <TData>(name: string, cacheDurationMs: number
     * @returns data or null if not found or expired
     */
    getAsync: async () => {
-      const data = await getSetClient.get(cacheChannelName);
+      const data = await getSetClient().get(cacheChannelName);
      if (!data) return null;

      const parsedData = superjson.parse<{ data: TData; timestamp: Date }>(data);
@@ -156,13 +159,13 @@ export const createCacheChannel = <TData>(name: string, cacheDurationMs: number
     * @returns data or new data if not present or expired
     */
    consumeAsync: async (callback: () => Promise<TData>) => {
-      const data = await getSetClient.get(cacheChannelName);
+      const data = await getSetClient().get(cacheChannelName);

      const getNewDataAsync = async () => {
        logger.debug(`Cache miss for channel '${cacheChannelName}'`);
        const newData = await callback();
        const result = { data: newData, timestamp: new Date() };
-        await getSetClient.set(cacheChannelName, superjson.stringify(result));
+        await getSetClient().set(cacheChannelName, superjson.stringify(result));
        logger.debug(`Cache updated for channel '${cacheChannelName}'`);
        return result;
      };
@@ -187,14 +190,14 @@ export const createCacheChannel = <TData>(name: string, cacheDurationMs: number
     * Invalidate the cache channels data.
     */
    invalidateAsync: async () => {
-      await getSetClient.del(cacheChannelName);
+      await getSetClient().del(cacheChannelName);
    },
    /**
     * Set the data in the cache channel.
     * @param data data to be stored in the cache channel
     */
    setAsync: async (data: TData) => {
-      await getSetClient.set(cacheChannelName, superjson.stringify({ data, timestamp: new Date() }));
+      await getSetClient().set(cacheChannelName, superjson.stringify({ data, timestamp: new Date() }));
    },
  };
 };
@@ -236,12 +239,12 @@ export const createChannelEventHistory = <TData>(channelName: string, maxElement
      });
    },
    pushAsync: async (data: TData, options = { publish: false }) => {
-      if (options.publish) await publisher.publish(channelName, superjson.stringify(data));
-      await getSetClient.lpush(channelName, superjson.stringify({ data, timestamp: new Date() }));
-      await getSetClient.ltrim(channelName, 0, maxElements);
+      if (options.publish) await getPublisher().publish(channelName, superjson.stringify(data));
+      await getSetClient().lpush(channelName, superjson.stringify({ data, timestamp: new Date() }));
+      await getSetClient().ltrim(channelName, 0, maxElements);
    },
    clearAsync: async () => {
-      await getSetClient.del(channelName);
+      await getSetClient().del(channelName);
    },
    /**
     * Returns a slice of the available data in the channel.
@@ -250,17 +253,17 @@ export const createChannelEventHistory = <TData>(channelName: string, maxElement
     * @param endIndex End index of the slice, negative values are counted from the end, defaults at end of range.
     */
    getSliceAsync: async (startIndex = 0, endIndex = -1) => {
-      const range = await getSetClient.lrange(channelName, startIndex, endIndex);
+      const range = await getSetClient().lrange(channelName, startIndex, endIndex);
      return range.map((item) => superjson.parse<{ data: TData; timestamp: Date }>(item));
    },
    getSliceUntilTimeAsync: async (time: Date) => {
-      const itemsInCollection = await getSetClient.lrange(channelName, 0, -1);
+      const itemsInCollection = await getSetClient().lrange(channelName, 0, -1);
      return itemsInCollection
        .map((item) => superjson.parse<{ data: TData; timestamp: Date }>(item))
        .filter((item) => item.timestamp < time);
    },
    getLengthAsync: async () => {
-      return await getSetClient.llen(channelName);
+      return await getSetClient().llen(channelName);
    },
    name: channelName,
  };
@@ -271,11 +274,11 @@ export const createChannelEventHistory = <TData>(channelName: string, maxElement
 */
 export const createChannelEventHistoryOld = <TData>(channelName: string, maxElements = 15) => {
  const popElementsOverMaxAsync = async () => {
-    const length = await getSetClient.llen(channelName);
+    const length = await getSetClient().llen(channelName);
    if (length <= maxElements) {
      return;
    }
-    await getSetClient.ltrim(channelName, 0, maxElements - 1);
+    await getSetClient().ltrim(channelName, 0, maxElements - 1);
  };

  return {
@@ -285,33 +288,33 @@ export const createChannelEventHistoryOld = <TData>(channelName: string, maxElem
      });
    },
    publishAndPushAsync: async (data: TData) => {
-      await publisher.publish(channelName, superjson.stringify(data));
-      await getSetClient.lpush(channelName, superjson.stringify({ data, timestamp: new Date() }));
+      await getPublisher().publish(channelName, superjson.stringify(data));
+      await getSetClient().lpush(channelName, superjson.stringify({ data, timestamp: new Date() }));
      await popElementsOverMaxAsync();
    },
    pushAsync: async (data: TData) => {
-      await getSetClient.lpush(channelName, superjson.stringify({ data, timestamp: new Date() }));
+      await getSetClient().lpush(channelName, superjson.stringify({ data, timestamp: new Date() }));
      await popElementsOverMaxAsync();
    },
    clearAsync: async () => {
-      await getSetClient.del(channelName);
+      await getSetClient().del(channelName);
    },
    getLastAsync: async () => {
-      const length = await getSetClient.llen(channelName);
-      const data = await getSetClient.lrange(channelName, length - 1, length);
+      const length = await getSetClient().llen(channelName);
+      const data = await getSetClient().lrange(channelName, length - 1, length);
      if (data.length !== 1) return null;

      // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
      return superjson.parse<{ data: TData; timestamp: Date }>(data[0]!);
    },
    getSliceAsync: async (startIndex: number, endIndex: number) => {
-      const range = await getSetClient.lrange(channelName, startIndex, endIndex);
+      const range = await getSetClient().lrange(channelName, startIndex, endIndex);
      return range.map((item) => superjson.parse<{ data: TData; timestamp: Date }>(item));
    },
    getSliceUntilTimeAsync: async (time: Date) => {
-      const length = await getSetClient.llen(channelName);
+      const length = await getSetClient().llen(channelName);
      const items: TData[] = [];
-      const itemsInCollection = await getSetClient.lrange(channelName, 0, length - 1);
+      const itemsInCollection = await getSetClient().lrange(channelName, 0, length - 1);

      for (let i = 0; i < length - 1; i++) {
        // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
@@ -324,7 +327,7 @@ export const createChannelEventHistoryOld = <TData>(channelName: string, maxElem
      return items;
    },
    getLengthAsync: async () => {
-      return await getSetClient.llen(channelName);
+      return await getSetClient().llen(channelName);
    },
    name: channelName,
  };
@@ -338,14 +341,14 @@ export const createChannelWithLatestAndEvents = <TData>(channelName: string) =>
      });
    },
    publishAndUpdateLastStateAsync: async (data: TData) => {
-      await publisher.publish(channelName, superjson.stringify(data));
-      await getSetClient.set(channelName, superjson.stringify({ data, timestamp: new Date() }));
+      await getPublisher().publish(channelName, superjson.stringify(data));
+      await getSetClient().set(channelName, superjson.stringify({ data, timestamp: new Date() }));
    },
    setAsync: async (data: TData) => {
-      await getSetClient.set(channelName, superjson.stringify({ data, timestamp: new Date() }));
+      await getSetClient().set(channelName, superjson.stringify({ data, timestamp: new Date() }));
    },
    getAsync: async () => {
-      const data = await getSetClient.get(channelName);
+      const data = await getSetClient().get(channelName);
      if (!data) return null;

      return superjson.parse<{ data: TData; timestamp: Date }>(data);
@@ -354,7 +357,8 @@ export const createChannelWithLatestAndEvents = <TData>(channelName: string) =>
  };
 };

-const queueClient = createRedisConnection();
+let _queueClient: RedisClient | undefined;
+const getQueueClient = () => (_queueClient ??= createRedisConnection());

 type WithId<TItem> = TItem & { _id: string };

@@ -366,11 +370,11 @@ type WithId<TItem> = TItem & { _id: string };
 export const createQueueChannel = <TItem>(name: string) => {
  const queueChannelName = `queue:${name}`;
  const getDataAsync = async () => {
-    const data = await queueClient.get(queueChannelName);
+    const data = await getQueueClient().get(queueChannelName);
    return data ? superjson.parse<WithId<TItem>[]>(data) : [];
  };
  const setDataAsync = async (data: WithId<TItem>[]) => {
-    await queueClient.set(queueChannelName, superjson.stringify(data));
+    await getQueueClient().set(queueChannelName, superjson.stringify(data));
  };

  return {
@@ -417,5 +421,5 @@ export const createQueueChannel = <TItem>(name: string) => {
 };

 export const handshakeAsync = async () => {
-  await getSetClient.hello();
+  await getSetClient().hello();
 };
@@ -1,15 +1,3 @@
-import type { RedisClient } from "@homarr/core/infrastructure/redis";
 import { createRedisClient } from "@homarr/core/infrastructure/redis";

-/**
- * Creates a new Redis connection
- * @returns redis client
- */
-export const createRedisConnection = () => {
-  if (Boolean(process.env.CI) || Boolean(process.env.DISABLE_REDIS_LOGS)) {
-    // Return null if we are in CI as we don't want to connect to Redis
-    return null as unknown as RedisClient;
-  }
-
-  return createRedisClient();
-};
+export const createRedisConnection = () => createRedisClient();