From 1c3ae92c04ceed75a47945befd31e167e147f119 Mon Sep 17 00:00:00 2001 From: Charles Bochet Date: Wed, 10 Jun 2026 12:05:55 +0200 Subject: [PATCH] fix(server): prevent SSE stream teardown errors from crashing all pods (#21395) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Context In prod-eu, **all `twenty-server` API pods crash simultaneously** several times per hour (then restart in lockstep) since ~Jun 2. Each crash is an **unhandled promise rejection** in SSE event-stream teardown — `exitCode=1`, identical stack on every one of the 7 pods: ``` Error: Failed to acquire lock for key: workspace::activeStreams at CacheLockService.withLock (cache-lock.service.ts:53) at async EventStreamService.destroyEventStream (event-stream.service.ts:75) at async cleanup (wrap-async-iterator-with-lifecycle.ts) at async Object.return (wrap-async-iterator-with-lifecycle.ts) at async Object.cancel (graphql-yoga/.../result-processor/sse.js:68) ``` ### Mechanism 1. A busy workspace's SSE clients reconnect (no client backoff), so connect/disconnect contend on a **single per-workspace Redis lock** `workspace::activeStreams`. 2. Under contention `CacheLockService.withLock` exhausts its retries and **throws**. 3. In the `destroyEventStream` teardown path the throw escapes `wrapAsyncIteratorWithLifecycle`'s `cleanup()` — `return()` does `try { await cleanup() } finally { … }` and does **not** catch a cleanup throw. 4. graphql-yoga invokes this from `cancel()` as a **fire-and-forget** `Promise.all` on connection abort. With **no global `unhandledRejection` handler**, Node's default policy terminates the process with **exit code 1**. 5. The crash drops all that pod's SSE clients → they reconnect to surviving pods → contention moves there → the whole fleet crashes together → restarts → reconnect storm → repeats (~14 min period, matching the metrics). ## Changes Crash-stopping hotfix (defense in depth). Does **not** change the locking design or client reconnect behavior — see follow-ups. - **`wrapAsyncIteratorWithLifecycle`**: `onCleanup()` is now best-effort — wrapped in try/catch so teardown can never reject out of `next()`/`return()`/`throw()`. The original iterator error is still rethrown unchanged. Adds an `onCleanupError` hook so the swallowed error is still reported. - **`EventStreamResolver`**: wires `onCleanupError` to `ExceptionHandlerService.captureExceptions` (→ Sentry) with workspace + channel context, so these failures stay visible. - **`main.ts`**: registers a global `process.on('unhandledRejection')` that reports via `ExceptionHandlerService` (Sentry) instead of letting Node terminate. Registering the listener also suppresses Node's default process-termination. Non-`Error` reasons are formatted with `util.inspect` (per Copilot review) so Sentry gets a readable message rather than `[object Object]`. ## Verification The fix was checked against the exact crash path — a wrapped iterator whose `onCleanup` rejects: - `return()` (graphql-yoga's `cancel()` path) **resolves** instead of rejecting → no unhandled rejection. - `next()` on a completed stream **resolves** despite a rejecting cleanup. - A genuine iterator error still surfaces on `next()` (cleanup failure doesn't mask it). - `onCleanupError` receives the original `Error`. All four pass. `nx lint twenty-server` + `oxfmt` clean; no new `tsc` errors in the changed files. ## Follow-ups (not in this PR) - Remove the unnecessary `withLock` around the already-atomic Redis `SADD`/`SREM` in `event-stream.service.ts` (the contention source). - Restore exponential backoff + jitter on the frontend SSE reconnect (regressed in #21061) to stop the thundering herd. - Infra: make `/healthz` a meaningful liveness signal and add `maxUnavailable` + a PodDisruptionBudget so pods can't all die together. --- .../engine/subscriptions/event-stream.resolver.ts | 7 +++++++ .../utils/wrap-async-iterator-with-lifecycle.ts | 15 +++++++++++++-- packages/twenty-server/src/main.ts | 15 +++++++++++++++ 3 files changed, 35 insertions(+), 2 deletions(-) diff --git a/packages/twenty-server/src/engine/subscriptions/event-stream.resolver.ts b/packages/twenty-server/src/engine/subscriptions/event-stream.resolver.ts index ad5e9ee544..3a04160fa2 100644 --- a/packages/twenty-server/src/engine/subscriptions/event-stream.resolver.ts +++ b/packages/twenty-server/src/engine/subscriptions/event-stream.resolver.ts @@ -5,6 +5,7 @@ import { isDefined } from 'twenty-shared/utils'; import { MetadataResolver } from 'src/engine/api/graphql/graphql-config/decorators/metadata-resolver.decorator'; import { type ApiKeyEntity } from 'src/engine/core-modules/api-key/api-key.entity'; +import { ExceptionHandlerService } from 'src/engine/core-modules/exception-handler/exception-handler.service'; import { PreventNestToAutoLogGraphqlErrorsFilter } from 'src/engine/core-modules/graphql/filters/prevent-nest-to-auto-log-graphql-errors.filter'; import { ResolverValidationPipe } from 'src/engine/core-modules/graphql/pipes/resolver-validation.pipe'; import { type AuthContextUser } from 'src/engine/core-modules/auth/types/auth-context.type'; @@ -40,6 +41,7 @@ export class EventStreamResolver { constructor( private readonly subscriptionService: SubscriptionService, private readonly eventStreamService: EventStreamService, + private readonly exceptionHandlerService: ExceptionHandlerService, ) {} @Subscription(() => EventSubscriptionDTO, { @@ -152,6 +154,11 @@ export class EventStreamResolver { workspaceId: workspace.id, eventStreamChannelId, }), + onCleanupError: (error) => + this.exceptionHandlerService.captureExceptions([error], { + workspace: { id: workspace.id }, + additionalData: { eventStreamChannelId }, + }), }); } diff --git a/packages/twenty-server/src/engine/subscriptions/utils/wrap-async-iterator-with-lifecycle.ts b/packages/twenty-server/src/engine/subscriptions/utils/wrap-async-iterator-with-lifecycle.ts index fb182cf5bd..9b172385fe 100644 --- a/packages/twenty-server/src/engine/subscriptions/utils/wrap-async-iterator-with-lifecycle.ts +++ b/packages/twenty-server/src/engine/subscriptions/utils/wrap-async-iterator-with-lifecycle.ts @@ -5,13 +5,20 @@ type AsyncIteratorLifecycleOptions = { onHeartbeat?: () => Promise; heartbeatIntervalMs?: number; onCleanup?: () => Promise; + onCleanupError?: (error: unknown) => void; }; export function wrapAsyncIteratorWithLifecycle( iterator: AsyncIterableIterator, options: AsyncIteratorLifecycleOptions, ): AsyncIterableIterator { - const { initialValue, onHeartbeat, heartbeatIntervalMs, onCleanup } = options; + const { + initialValue, + onHeartbeat, + heartbeatIntervalMs, + onCleanup, + onCleanupError, + } = options; let heartbeatInterval: NodeJS.Timeout | null = null; let hasYieldedInitialValue = false; @@ -33,7 +40,11 @@ export function wrapAsyncIteratorWithLifecycle( heartbeatInterval = null; } if (onCleanup) { - await onCleanup(); + try { + await onCleanup(); + } catch (error) { + onCleanupError?.(error); + } } }; diff --git a/packages/twenty-server/src/main.ts b/packages/twenty-server/src/main.ts index 9b65bfa59e..121505c376 100644 --- a/packages/twenty-server/src/main.ts +++ b/packages/twenty-server/src/main.ts @@ -2,6 +2,7 @@ import { NestFactory } from '@nestjs/core'; import { type NestExpressApplication } from '@nestjs/platform-express'; import fs from 'fs'; +import { inspect } from 'util'; import bytes from 'bytes'; import { useContainer } from 'class-validator'; @@ -11,10 +12,12 @@ import graphqlUploadExpress from 'graphql-upload/graphqlUploadExpress.mjs'; import { NodeEnvironment } from 'src/engine/core-modules/twenty-config/interfaces/node-environment.interface'; import { setPgDateTypeParser } from 'src/database/pg/set-pg-date-type-parser'; +import { ExceptionHandlerService } from 'src/engine/core-modules/exception-handler/exception-handler.service'; import { LoggerService } from 'src/engine/core-modules/logger/logger.service'; import { getSessionStorageOptions } from 'src/engine/core-modules/session-storage/session-storage.module-factory'; import { TwentyConfigService } from 'src/engine/core-modules/twenty-config/twenty-config.service'; import { configTransformers } from 'src/engine/core-modules/twenty-config/utils/config-transformers.util'; +import { shouldCaptureException } from 'src/engine/utils/global-exception-handler.util'; import { UnhandledExceptionFilter } from 'src/filters/unhandled-exception.filter'; import { AppModule } from './app.module'; @@ -45,6 +48,18 @@ const bootstrap = async () => { }); const logger = app.get(LoggerService); const twentyConfigService = app.get(TwentyConfigService); + const exceptionHandlerService = app.get(ExceptionHandlerService); + + process.on('unhandledRejection', (reason) => { + const error = + reason instanceof Error + ? reason + : new Error(typeof reason === 'string' ? reason : inspect(reason)); + + if (shouldCaptureException(error)) { + exceptionHandlerService.captureExceptions([error]); + } + }); const trustProxyRaw = twentyConfigService.get('TRUST_PROXY'); const trustProxy = /^\d+$/.test(trustProxyRaw)