Files
twenty/packages/twenty-server/src/main.ts
T
Charles Bochet 1c3ae92c04 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.
2026-06-10 10:05:55 +00:00

112 lines
3.9 KiB
TypeScript

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';
import session from 'express-session';
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';
import './instrument';
import { settings } from './engine/constants/settings';
import { generateFrontConfig } from './utils/generate-front-config';
// Trigger
const bootstrap = async () => {
setPgDateTypeParser();
const app = await NestFactory.create<NestExpressApplication>(AppModule, {
// Expose WWW-Authenticate so browser-based MCP clients can read the
// resource_metadata pointer on 401. Required by MCP authorization spec.
cors: { exposedHeaders: ['WWW-Authenticate'] },
bufferLogs: process.env.LOGGER_IS_BUFFER_ENABLED === 'true',
rawBody: true,
snapshot: process.env.NODE_ENV === NodeEnvironment.DEVELOPMENT,
...(process.env.SSL_KEY_PATH && process.env.SSL_CERT_PATH
? {
httpsOptions: {
key: fs.readFileSync(process.env.SSL_KEY_PATH),
cert: fs.readFileSync(process.env.SSL_CERT_PATH),
},
}
: {}),
});
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)
? Number(trustProxyRaw)
: (configTransformers.boolean(trustProxyRaw) ?? trustProxyRaw);
app.set('trust proxy', trustProxy);
app.use(session(getSessionStorageOptions(twentyConfigService)));
// Apply class-validator container so that we can use injection in validators
useContainer(app.select(AppModule), { fallbackOnErrors: true });
// Use our logger
app.useLogger(logger);
app.useGlobalFilters(new UnhandledExceptionFilter());
app.useBodyParser('json', { limit: settings.storage.maxFileSize });
app.useBodyParser('urlencoded', {
limit: settings.storage.maxFileSize,
extended: true,
});
app.useBodyParser('text', { type: 'text/plain', limit: '1024kb' });
// Graphql file upload
app.use(
'/graphql',
graphqlUploadExpress({
maxFieldSize: bytes(settings.storage.maxFileSize)!,
maxFiles: 10,
}),
);
app.use(
'/metadata',
graphqlUploadExpress({
maxFieldSize: bytes(settings.storage.maxFileSize)!,
maxFiles: 10,
}),
);
// Inject the server url in the frontend page
generateFrontConfig();
await app.listen(twentyConfigService.get('NODE_PORT'));
};
void bootstrap();