From 2071a36e5d07f41f45f49f73d3af5d2045bf9703 Mon Sep 17 00:00:00 2001 From: Iain Sproat <68657+iainsproat@users.noreply.github.com> Date: Fri, 11 Apr 2025 13:25:19 +0100 Subject: [PATCH] fix(previews): disable previews is previews Redis is not reachable - exit preview-service process if Redis is not reachable - improve server healthcheck to include Redis client readiness check --- packages/preview-service/src/main.ts | 18 ++++++-- packages/preview-service/src/utils.ts | 35 +++++++++++++++ packages/server/healthchecks/redis.ts | 5 ++- packages/server/modules/previews/index.ts | 43 ++++++++++++++----- packages/server/modules/shared/redis/redis.ts | 34 +++++++++++++++ 5 files changed, 121 insertions(+), 14 deletions(-) create mode 100644 packages/preview-service/src/utils.ts diff --git a/packages/preview-service/src/main.ts b/packages/preview-service/src/main.ts index 6f70e016d..525474f6e 100644 --- a/packages/preview-service/src/main.ts +++ b/packages/preview-service/src/main.ts @@ -3,7 +3,7 @@ import puppeteer, { Browser } from 'puppeteer' import { createTerminus } from '@godaddy/terminus' import type { Logger } from 'pino' import { Redis, type RedisOptions } from 'ioredis' -import Bull from 'bull' +import Bull, { type QueueOptions } from 'bull' import { jobPayload } from '@speckle/shared/dist/esm/previews/job.js' @@ -22,6 +22,7 @@ import { jobProcessor } from '@/jobProcessor.js' import { AppState } from '@/const.js' import { initMetrics, initPrometheusRegistry } from '@/metrics.js' import { ensureError } from '@speckle/shared' +import { isRedisReady } from '@/utils.js' const app = express() const host = HOST @@ -38,7 +39,7 @@ await initMetrics({ app, registry: initPrometheusRegistry() }) let client: Redis let subscriber: Redis -const opts = { +const opts: QueueOptions = { // redisOpts here will contain at least a property of connectionName which will identify the queue based on its name createClient(type: string, redisOpts: RedisOptions) { switch (type) { @@ -116,11 +117,21 @@ const server = app.listen(port, host, async () => { try { const newQueue = new Bull(JobQueueName, opts) + + logger.info('Checking Redis connection is ready...') + + // Bull's Queue.isReady() does not actually check the Redis connection + // see https://github.com/OptimalBits/bull/issues/1873#issuecomment-953581143 + await isRedisReady(newQueue.client) + logger.info('Redis is ready') + jobQueue = await newQueue.isReady() } catch (e) { const err = ensureError(e, 'Unknown error creating job queue') logger.error({ err }, 'Error creating job queue') - throw err + + // the callback to server.listen has failed, so we need to exit the process and not just return + process.exit(1) } logger.debug(`Starting processing of "${JobQueueName}" message queue`) @@ -215,6 +226,7 @@ const beforeShutdown = async () => { await browser.close() browser = undefined } + // no need to close the job queue and redis client, when the process exits they will be closed automatically } const onShutdown = async () => { diff --git a/packages/preview-service/src/utils.ts b/packages/preview-service/src/utils.ts new file mode 100644 index 000000000..a0d6401fb --- /dev/null +++ b/packages/preview-service/src/utils.ts @@ -0,0 +1,35 @@ +import { ensureError } from '@speckle/shared' +import { Redis, type RedisOptions } from 'ioredis' + +// MIT Licensed: https://github.com/OptimalBits/bull/blob/develop/LICENSE.md +// Reference: https://github.com/OptimalBits/bull/blob/develop/lib/utils.js +export const isRedisReady = (client: Redis) => { + return new Promise((resolve, reject) => { + if (client.status === 'ready') { + resolve() + } else { + function handleReady() { + client.removeListener('end', handleEnd) + client.removeListener('error', handleError) + resolve() + } + + function handleError(e: unknown) { + const err = ensureError(e, 'Unknown error in Redis client') + client.removeListener('ready', handleReady) + client.removeListener('error', handleError) + reject(err) + } + + function handleEnd() { + client.removeListener('ready', handleReady) + client.removeListener('error', handleError) + reject(new Error('Redis connection ended')) + } + + client.once('ready', handleReady) + client.on('error', handleError) + client.once('end', handleEnd) + } + }) +} diff --git a/packages/server/healthchecks/redis.ts b/packages/server/healthchecks/redis.ts index a017b5455..a9af0e8d6 100644 --- a/packages/server/healthchecks/redis.ts +++ b/packages/server/healthchecks/redis.ts @@ -1,12 +1,15 @@ import type { CheckResponse, RedisCheck } from '@/healthchecks/types' +import { isRedisReady } from '@/modules/shared/redis/redis' export const isRedisAlive: RedisCheck = async (params): Promise => { const { client } = params + await isRedisReady(client) + let result: CheckResponse = { isAlive: true } try { const redisResponse = await client.ping() if (redisResponse !== 'PONG') { - result = { isAlive: false, err: new Error('Redis did not respond correctly.') } + throw new Error('Redis did not respond correctly.') } } catch (err) { result = { isAlive: false, err } diff --git a/packages/server/modules/previews/index.ts b/packages/server/modules/previews/index.ts index 2d7d0a9e5..0aec4559c 100644 --- a/packages/server/modules/previews/index.ts +++ b/packages/server/modules/previews/index.ts @@ -9,12 +9,12 @@ import { getRedisUrl, getServerOrigin } from '@/modules/shared/helpers/envHelper' -import Bull from 'bull' +import Bull, { type QueueOptions } from 'bull' import Redis, { type RedisOptions } from 'ioredis' import { createBullBoard } from 'bull-board' import { BullMQAdapter } from 'bull-board/bullMQAdapter' import { authMiddlewareCreator } from '@/modules/shared/middleware' -import { Roles, TIME } from '@speckle/shared' +import { ensureError, Roles, TIME } from '@speckle/shared' import { validateServerRoleBuilderFactory } from '@/modules/shared/authz' import { getRolesFactory } from '@/modules/shared/repositories/roles' import { previewRouterFactory } from '@/modules/previews/rest/router' @@ -31,14 +31,18 @@ import { PreviewJobDurationStep } from '@/modules/previews/observability/metrics' import { addRequestQueueListeners } from '@/modules/previews/queues/previews' +import { isRedisReady } from '@/modules/shared/redis/redis' -const getPreviewQueues = (params: { responseQueueName: string }) => { +const JobQueueName = 'preview-service-jobs' +const ResponseQueueNamePrefix = 'preview-service-results' + +const getPreviewQueues = async (params: { responseQueueName: string }) => { const { responseQueueName } = params let client: Redis let subscriber: Redis const redisUrl = getPreviewServiceRedisUrl() ?? getRedisUrl() - const opts = { + const opts: QueueOptions = { // redisOpts here will contain at least a property of connectionName which will identify the queue based on its name createClient(type: string, redisOpts: RedisOptions) { switch (type) { @@ -69,7 +73,8 @@ const getPreviewQueues = (params: { responseQueueName: string }) => { } // previews are requested on this queue - const previewRequestQueue = new Bull('preview-service-jobs', opts) + const previewRequestQueue = new Bull(JobQueueName, opts) + await isRedisReady(previewRequestQueue.client) addRequestQueueListeners({ logger, previewRequestQueue @@ -77,10 +82,16 @@ const getPreviewQueues = (params: { responseQueueName: string }) => { // rendered previews are sent back on this queue const previewResponseQueue = new Bull(responseQueueName, opts) + + await isRedisReady(previewResponseQueue.client) return { previewRequestQueue, previewResponseQueue } } -export const init: SpeckleModule['init'] = ({ app, isInitial, metricsRegister }) => { +export const init: SpeckleModule['init'] = async ({ + app, + isInitial, + metricsRegister +}) => { if (isInitial) { if (disablePreviews()) { moduleLogger.warn('📸 Object preview module is DISABLED') @@ -88,13 +99,25 @@ export const init: SpeckleModule['init'] = ({ app, isInitial, metricsRegister }) moduleLogger.info('📸 Init object preview module') } - const responseQueueName = `preview-service-results-${ + const responseQueueName = `${ResponseQueueNamePrefix}-${ new URL(getServerOrigin()).hostname }` - const { previewRequestQueue, previewResponseQueue } = getPreviewQueues({ - responseQueueName - }) + let previewRequestQueue: Bull.Queue + let previewResponseQueue: Bull.Queue + + try { + ;({ previewRequestQueue, previewResponseQueue } = await getPreviewQueues({ + responseQueueName + })) + } catch (e) { + const err = ensureError(e, 'Unknown error when creating preview queues') + moduleLogger.error( + { err }, + 'Could not create preview queues. Disabling previews.' + ) + return + } const { previewJobsProcessedSummary } = initializeMetrics({ registers: [metricsRegister], diff --git a/packages/server/modules/shared/redis/redis.ts b/packages/server/modules/shared/redis/redis.ts index 2a8022309..293f54062 100644 --- a/packages/server/modules/shared/redis/redis.ts +++ b/packages/server/modules/shared/redis/redis.ts @@ -5,6 +5,7 @@ import { MisconfiguredEnvironmentError } from '@/modules/shared/errors' import { getRedisUrl } from '@/modules/shared/helpers/envHelper' +import { ensureError } from '@speckle/shared' export function createRedisClient(redisUrl: string, redisOptions: RedisOptions): Redis { let redisClient: Redis @@ -34,3 +35,36 @@ export const getGenericRedis = (): Redis => { if (!redisClient) redisClient = createRedisClient(getRedisUrl(), {}) return redisClient } + +export const isRedisReady = (client: Redis) => { + // MIT Licensed: https://github.com/OptimalBits/bull/blob/develop/LICENSE.md + // Reference: https://github.com/OptimalBits/bull/blob/develop/lib/utils.js + return new Promise((resolve, reject) => { + if (client.status === 'ready') { + resolve() + } else { + function handleReady() { + client.removeListener('end', handleEnd) + client.removeListener('error', handleError) + resolve() + } + + function handleError(e: unknown) { + const err = ensureError(e, 'Unknown error in Redis client') + client.removeListener('ready', handleReady) + client.removeListener('error', handleError) + reject(err) + } + + function handleEnd() { + client.removeListener('ready', handleReady) + client.removeListener('error', handleError) + reject(new Error('Redis connection ended')) + } + + client.once('ready', handleReady) + client.on('error', handleError) + client.once('end', handleEnd) + } + }) +}