## Why
Deploys kill workers mid-job: neither the queue worker nor the API
server ever calls `enableShutdownHooks()`, so NestJS never listens for
SIGTERM and the graceful close in `BullMQDriver.onModuleDestroy` is dead
code. On every rollout a worker dies instantly — an in-flight 10-minute
AI stream freezes for the watching user, and BullMQ silently re-runs the
half-executed job on another worker ~10 minutes later.
This is the root cause, not a symptom: the correct drain semantics
already exist in the driver (`worker.close()` waits for active jobs and
stops picking new ones, per the BullMQ graceful-shutdown docs) — the
process just never received the signal.
## What
- `queue-worker.ts` + `main.ts`: enable shutdown hooks. SIGTERM now runs
`onModuleDestroy` across providers: the BullMQ driver drains active
jobs, `RedisClientService` and the AI cancel subscriber quit their Redis
connections, TypeORM closes its pools, then the process exits on its
own.
- BullMQ close order: workers drain before queues close, so a job
finishing during the drain can still enqueue follow-ups (e.g. the AI
queue flushing the next queued message).
- API server: `forceCloseConnections` so long-lived subscription sockets
don't hold `close()` open until the pod is force-killed. They were
dropped abruptly on every deploy before this PR too — clients already
recover.
- Drain start/completion logs so pod terminations are debuggable.
## User impact
Deploys stop corrupting in-flight background work. Follow-ups build on
this: bounded drain-then-abort for AI stream jobs, and eliminating the
stalled-job zombie re-run.
## Validation
- Local: SIGTERM'd a running worker mid-job — drain log appears, the
active job completes, "Message queue shutdown complete" is logged,
process exits by itself. (Also verified with
`LOGGER_IS_BUFFER_ENABLED=true` that final logs are not swallowed.)
- The k8s side (termination grace period ≥ drain budget, exec'ing `node`
directly so PID 1 receives SIGTERM) lands separately in twenty-infra.
<!-- This is an auto-generated description by cubic. -->
<a
href="https://cubic.dev/pr/twentyhq/twenty/pull/22514?utm_source=github"
target="_blank" rel="noopener noreferrer"
data-no-image-dialog="true"><picture><source
media="(prefers-color-scheme: dark)"
srcset="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"><source
media="(prefers-color-scheme: light)"
srcset="https://www.cubic.dev/buttons/review-in-cubic-light.svg"><img
alt="Review in cubic"
src="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"></picture></a>
<!-- End of auto-generated description by cubic. -->
---------
Co-authored-by: cubic-dev-ai[bot] <191113872+cubic-dev-ai[bot]@users.noreply.github.com>
Follow-up on https://github.com/twentyhq/twenty/pull/12007
In this PR
- adding a filter on HttpExceptionHandlerService to filter out 4xx
errors from driver handling (as we do for graphQL errors: see
useGraphQLErrorHandler hook - only filteredIssues are sent to`
exceptionHandlerService.captureExceptions()`.)
- grouping together more missing metadata issues
- attempting to use error codes as issues names in sentry to improve UI;
for now it says "Error" all the time
Done :
- move metrics and health cache services from health module to metrics
module
- refactor metrics counter from specific method to set up from enum keys
- add OpenTelemetry (Otel) instrumentation for metrics
- set up Otel SDK to send metrics to Otel collector
To do later :
- implement Otel instrumentation for traces + plug Sentry on top
- Improve snackbar to enable displaying multi-line message (so far we
only displayed the first few words which was very frustrating)
- Followup on previous issue to enable tim@apple.dev on the demo
workspace (prefilled automatically)
- Fix sentry tracing which had been broken when migrating from v7 to v8
### Overview
This PR introduces significant enhancements to the MessageQueue module
by integrating `@Processor`, `@Process`, and `@InjectMessageQueue`
decorators. These changes streamline the process of defining and
managing queue processors and job handlers, and also allow for
request-scoped handlers, improving compatibility with services that rely
on scoped providers like TwentyORM repositories.
### Key Features
1. **Decorator-based Job Handling**: Use `@Processor` and `@Process`
decorators to define job handlers declaratively.
2. **Request Scope Support**: Job handlers can be scoped per request,
enhancing integration with request-scoped services.
### Usage
#### Defining Processors and Job Handlers
The `@Processor` decorator is used to define a class that processes jobs
for a specific queue. The `@Process` decorator is applied to methods
within this class to define specific job handlers.
##### Example 1: Specific Job Handlers
```typescript
import { Processor, Process, InjectMessageQueue } from 'src/engine/integrations/message-queue';
@Processor('taskQueue')
export class TaskProcessor {
@Process('taskA')
async handleTaskA(job: { id: string, data: any }) {
console.log(`Handling task A with data:`, job.data);
// Logic for task A
}
@Process('taskB')
async handleTaskB(job: { id: string, data: any }) {
console.log(`Handling task B with data:`, job.data);
// Logic for task B
}
}
```
In the example above, `TaskProcessor` is responsible for processing jobs
in the `taskQueue`. The `handleTaskA` method will only be called for
jobs with the name `taskA`, while `handleTaskB` will be called for
`taskB` jobs.
##### Example 2: General Job Handler
```typescript
import { Processor, Process, InjectMessageQueue } from 'src/engine/integrations/message-queue';
@Processor('generalQueue')
export class GeneralProcessor {
@Process()
async handleAnyJob(job: { id: string, name: string, data: any }) {
console.log(`Handling job ${job.name} with data:`, job.data);
// Logic for any job
}
}
```
In this example, `GeneralProcessor` handles all jobs in the
`generalQueue`, regardless of the job name. The `handleAnyJob` method
will be invoked for every job added to the `generalQueue`.
#### Adding Jobs to a Queue
You can use the `@InjectMessageQueue` decorator to inject a queue into a
service and add jobs to it.
##### Example:
```typescript
import { Injectable } from '@nestjs/common';
import { InjectMessageQueue, MessageQueue } from 'src/engine/integrations/message-queue';
@Injectable()
export class TaskService {
constructor(
@InjectMessageQueue('taskQueue') private readonly taskQueue: MessageQueue,
) {}
async addTaskA(data: any) {
await this.taskQueue.add('taskA', data);
}
async addTaskB(data: any) {
await this.taskQueue.add('taskB', data);
}
}
```
In this example, `TaskService` adds jobs to the `taskQueue`. The
`addTaskA` and `addTaskB` methods add jobs named `taskA` and `taskB`,
respectively, to the queue.
#### Using Scoped Job Handlers
To utilize request-scoped job handlers, specify the scope in the
`@Processor` decorator. This is particularly useful for services that
use scoped repositories like those in TwentyORM.
##### Example:
```typescript
import { Processor, Process, InjectMessageQueue, Scope } from 'src/engine/integrations/message-queue';
@Processor({ name: 'scopedQueue', scope: Scope.REQUEST })
export class ScopedTaskProcessor {
@Process('scopedTask')
async handleScopedTask(job: { id: string, data: any }) {
console.log(`Handling scoped task with data:`, job.data);
// Logic for scoped task, which might use request-scoped services
}
}
```
Here, the `ScopedTaskProcessor` is associated with `scopedQueue` and
operates with request scope. This setup is essential when the job
handler relies on services that need to be instantiated per request,
such as scoped repositories.
### Migration Notes
- **Decorators**: Refactor job handlers to use `@Processor` and
`@Process` decorators.
- **Request Scope**: Utilize the scope option in `@Processor` if your
job handlers depend on request-scoped services.
Fix#5628
---------
Co-authored-by: Weiko <corentin@twenty.com>
Query read timeouts happen when a remote server is not available. It
breaks:
- the remote server show page
- the record table page of imported remote tables
This PR will catch the exception so it does not go to Sentry in both
cases.
Also did 2 renaming.
## Context
JobsModule is hard to maintain because we provide all the jobs there,
including their dependencies. This PR aims to split jobs in dedicated
modules.
Félix Malfait
