fix(server): prevent SSE stream teardown errors from crashing all pods (#21395)

## 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:<id>: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:<id>: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.

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 },
+        }),
     });
   }
 

packages/twenty-server/src/engine/subscriptions/utils/wrap-async-iterator-with-lifecycle.ts

@@ -5,13 +5,20 @@ type AsyncIteratorLifecycleOptions<T> = {
   onHeartbeat?: () => Promise<boolean>;
   heartbeatIntervalMs?: number;
   onCleanup?: () => Promise<void>;
+  onCleanupError?: (error: unknown) => void;
 };
 
 export function wrapAsyncIteratorWithLifecycle<T>(
   iterator: AsyncIterableIterator<T>,
   options: AsyncIteratorLifecycleOptions<T>,
 ): AsyncIterableIterator<T> {
-  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<T>(
       heartbeatInterval = null;
     }
     if (onCleanup) {
-      await onCleanup();
+      try {
+        await onCleanup();
+      } catch (error) {
+        onCleanupError?.(error);
+      }
     }
   };
 

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)