From a6ec5f789dfe6f738178a5eaa27295bd419e8f5e Mon Sep 17 00:00:00 2001 From: Thomas Camlong Date: Thu, 21 May 2026 16:27:53 +0200 Subject: [PATCH] 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 --- apps/tasks/src/main.ts | 22 ++++ .../src/lib/channel-subscription-tracker.ts | 6 +- packages/redis/src/lib/channel.ts | 112 +++++++++--------- packages/redis/src/lib/connection.ts | 14 +-- 4 files changed, 86 insertions(+), 68 deletions(-) diff --git a/apps/tasks/src/main.ts b/apps/tasks/src/main.ts index 55416e689..40496a39b 100644 --- a/apps/tasks/src/main.ts +++ b/apps/tasks/src/main.ts @@ -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(); diff --git a/packages/redis/src/lib/channel-subscription-tracker.ts b/packages/redis/src/lib/channel-subscription-tracker.ts index 02adc6897..f3644a62d 100644 --- a/packages/redis/src/lib/channel-subscription-tracker.ts +++ b/packages/redis/src/lib/channel-subscription-tracker.ts @@ -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; */ export class ChannelSubscriptionTracker { private static subscriptions = new Map>(); - private static redis = createRedisConnection(); + private static _redis: RedisClient | undefined; + private static get redis() { + return (this._redis ??= createRedisConnection()); + } private static listenerActive = false; /** diff --git a/packages/redis/src/lib/channel.ts b/packages/redis/src/lib/channel.ts index 193f551e3..29c484259 100644 --- a/packages/redis/src/lib/channel.ts +++ b/packages/redis/src/lib/channel.ts @@ -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 = (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(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 = (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(item)); }, /** @@ -77,20 +80,20 @@ export const createListChannel = (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 = (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(data) : null; }, /** @@ -114,13 +117,13 @@ export const createGetSetChannel = (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 = (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 = (name: string, cacheDurationMs: number * @returns data or new data if not present or expired */ consumeAsync: async (callback: () => Promise) => { - 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 = (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 = (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 = (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 = (channelName: string, maxElement */ export const createChannelEventHistoryOld = (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 = (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 = (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 = (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 = (channelName: string) => }; }; -const queueClient = createRedisConnection(); +let _queueClient: RedisClient | undefined; +const getQueueClient = () => (_queueClient ??= createRedisConnection()); type WithId = TItem & { _id: string }; @@ -366,11 +370,11 @@ type WithId = TItem & { _id: string }; export const createQueueChannel = (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[]>(data) : []; }; const setDataAsync = async (data: WithId[]) => { - await queueClient.set(queueChannelName, superjson.stringify(data)); + await getQueueClient().set(queueChannelName, superjson.stringify(data)); }; return { @@ -417,5 +421,5 @@ export const createQueueChannel = (name: string) => { }; export const handshakeAsync = async () => { - await getSetClient.hello(); + await getSetClient().hello(); }; diff --git a/packages/redis/src/lib/connection.ts b/packages/redis/src/lib/connection.ts index ba945de33..b01b43bba 100644 --- a/packages/redis/src/lib/connection.ts +++ b/packages/redis/src/lib/connection.ts @@ -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();