houseassassin/twenty
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Remove DATABASE_EVENT_JOBS_CHUNK_SIZE and Promise.all from logic function trigger jobs (#23205)

Browse Source 
## What

- `LogicFunctionTriggerJob` now processes a single
`LogicFunctionTriggerJobData` payload instead of an array processed with
`Promise.all`.
- Removed `DATABASE_EVENT_JOBS_CHUNK_SIZE` and the `lodash.chunk` usage
in `CallDatabaseEventTriggerJobsJob`.
- Added `bulkAdd` to `MessageQueueService` and both drivers (BullMQ
driver uses native `queue.addBulk`, sync driver processes payloads
sequentially). `CallDatabaseEventTriggerJobsJob` uses it to enqueue all
payloads in one call.
- Updated the other producers (`ServerRouteTriggerService`,
`ApplicationInstallService`, `ConnectionProviderOauthFlowService`,
`CronTriggerCronJob`) to enqueue a single payload instead of a
one-element array, and updated the corresponding specs.

## Why

Each logic function execution now gets its own queue job, so a failing
execution only retries itself instead of re-running the whole chunk, and
job-level retry/metrics apply per execution.

---
_Generated by [Claude
Code](https://claude.ai/code/session_018VCs2kopnDiZCL41eToxQF)_

<!-- This is an auto-generated description by cubic. -->
<a
href="https://cubic.dev/pr/twentyhq/twenty/pull/23205?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. -->
This commit is contained in:
martmull
2026-07-23 17:50:43 +02:00
committed by GitHub
parent 65d399c70d
commit 4d09c400a4
12 changed files with 164 additions and 106 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/twenty-server/src/engine/core-modules/application/application-install/application-install.service.ts
+6 -8
View File
@@ -557,15 +557,13 @@ export class ApplicationInstallService {
);
if (!shouldRunSynchronously) {
await this.messageQueueService.add<LogicFunctionTriggerJobData[]>(
await this.messageQueueService.add<LogicFunctionTriggerJobData>(
LogicFunctionTriggerJob.name,
[
{
logicFunctionId: flatLogicFunction.id,
workspaceId,
payload,
},
],
{
logicFunctionId: flatLogicFunction.id,
workspaceId,
payload,
},
{ retryLimit: 3 },
);
return;
packages/twenty-server/src/engine/core-modules/application/connection-provider/__tests__/connection-provider-oauth-flow.service.spec.ts
+8 -10
View File
@@ -501,17 +501,15 @@ describe('ConnectionProviderOAuthFlowService', () => {
);
expect(messageQueueService.add).toHaveBeenCalledWith(
'LogicFunctionTriggerJob',
[
{
logicFunctionId: 'logic-function-1',
workspaceId: 'workspace-1',
payload: {
connectionProviderId: 'provider-1',
connectionProviderName: 'linear',
connectedAccountId: result.connectedAccountId,
},
{
logicFunctionId: 'logic-function-1',
workspaceId: 'workspace-1',
payload: {
connectionProviderId: 'provider-1',
connectionProviderName: 'linear',
connectedAccountId: result.connectedAccountId,
},
],
},
{ retryLimit: 3 },
);
});
packages/twenty-server/src/engine/core-modules/application/connection-provider/connection-provider-oauth-flow.service.ts
+9 -11
View File
@@ -252,19 +252,17 @@ export class ConnectionProviderOAuthFlowService {
);
}
await this.messageQueueService.add<LogicFunctionTriggerJobData[]>(
await this.messageQueueService.add<LogicFunctionTriggerJobData>(
LogicFunctionTriggerJob.name,
[
{
logicFunctionId: flatLogicFunction.id,
workspaceId,
payload: {
connectionProviderId: provider.id,
connectionProviderName: provider.name,
connectedAccountId,
},
{
logicFunctionId: flatLogicFunction.id,
workspaceId,
payload: {
connectionProviderId: provider.id,
connectionProviderName: provider.name,
connectedAccountId,
},
],
},
{ retryLimit: 3 },
);
} catch (error) {
packages/twenty-server/src/engine/core-modules/logic-function/logic-function-trigger/jobs/logic-function-trigger.job.ts
+26 -23
View File
@@ -27,30 +27,33 @@ export class LogicFunctionTriggerJob {
) {}
@Process(LogicFunctionTriggerJob.name)
async handle(logicFunctionPayloads: LogicFunctionTriggerJobData[]) {
await Promise.all(
logicFunctionPayloads.map(async (logicFunctionPayload) => {
try {
await this.logicFunctionExecutorService.execute({
logicFunctionId: logicFunctionPayload.logicFunctionId,
workspaceId: logicFunctionPayload.workspaceId,
payload: logicFunctionPayload.payload ?? {},
userId: logicFunctionPayload.userId,
userWorkspaceId: logicFunctionPayload.userWorkspaceId,
});
} catch (error) {
// A stopped application must not fail the job: failing would make
// the queue retry an execution that is intentionally blocked.
if (
error instanceof LogicFunctionException &&
error.code === LogicFunctionExceptionCode.LOGIC_FUNCTION_DISABLED
) {
return;
}
async handle(
jobData: LogicFunctionTriggerJobData | LogicFunctionTriggerJobData[],
) {
// Jobs enqueued in version <=2.24.x carry arrays, remove this case once those jobs are drained
const logicFunctionPayloads = Array.isArray(jobData) ? jobData : [jobData];
throw error;
for (const logicFunctionPayload of logicFunctionPayloads) {
try {
await this.logicFunctionExecutorService.execute({
logicFunctionId: logicFunctionPayload.logicFunctionId,
workspaceId: logicFunctionPayload.workspaceId,
payload: logicFunctionPayload.payload ?? {},
userId: logicFunctionPayload.userId,
userWorkspaceId: logicFunctionPayload.userWorkspaceId,
});
} catch (error) {
// A stopped application must not fail the job: failing would make
// the queue retry an execution that is intentionally blocked.
if (
error instanceof LogicFunctionException &&
error.code === LogicFunctionExceptionCode.LOGIC_FUNCTION_DISABLED
) {
continue;
}
}),
);
throw error;
}
}
}
}
packages/twenty-server/src/engine/core-modules/logic-function/logic-function-trigger/triggers/cron/cron-trigger.cron.job.ts
+6 -8
View File
@@ -85,15 +85,13 @@ export class CronTriggerCronJob {
continue;
}
await this.messageQueueService.add<LogicFunctionTriggerJobData[]>(
await this.messageQueueService.add<LogicFunctionTriggerJobData>(
LogicFunctionTriggerJob.name,
[
{
logicFunctionId: logicFunction.id,
workspaceId: activeWorkspace.id,
payload: {},
},
],
{
logicFunctionId: logicFunction.id,
workspaceId: activeWorkspace.id,
payload: {},
},
{ retryLimit: 10 },
);
}
packages/twenty-server/src/engine/core-modules/logic-function/logic-function-trigger/triggers/database-event/call-database-event-trigger-jobs.job.ts
+3 -17
View File
@@ -1,4 +1,3 @@
import chunk from 'lodash.chunk';
import { isDefined } from 'twenty-shared/utils';
import type { ObjectRecordEvent } from 'twenty-shared/database-events';
@@ -16,8 +15,6 @@ import {
import { WorkspaceCacheService } from 'src/engine/workspace-cache/services/workspace-cache.service';
import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event-batch.type';
const DATABASE_EVENT_JOBS_CHUNK_SIZE = 20;
@Processor(MessageQueue.triggerQueue)
export class CallDatabaseEventTriggerJobsJob {
constructor(
@@ -59,22 +56,11 @@ export class CallDatabaseEventTriggerJobsJob {
workspaceEventBatch,
});
if (logicFunctionPayloads.length === 0) {
return;
}
const logicFunctionPayloadsChunks = chunk(
await this.messageQueueService.bulkAdd<LogicFunctionTriggerJobData>(
LogicFunctionTriggerJob.name,
logicFunctionPayloads,
DATABASE_EVENT_JOBS_CHUNK_SIZE,
{ retryLimit: 3 },
);
for (const logicFunctionPayloadsChunk of logicFunctionPayloadsChunks) {
await this.messageQueueService.add<LogicFunctionTriggerJobData[]>(
LogicFunctionTriggerJob.name,
logicFunctionPayloadsChunk,
{ retryLimit: 3 },
);
}
}
private shouldTriggerJob({
packages/twenty-server/src/engine/core-modules/message-queue/drivers/bullmq.driver.ts
+57 -14
View File
@@ -327,6 +327,30 @@ export class BullMQDriver
);
}
private buildJobsOptions({
queueName,
options,
}: {
queueName: MessageQueue;
options?: QueueJobOptions;
}): JobsOptions {
return {
// We suffix the id with V4() to make sure ids are unique so we can add a waiting job when a job related with the same option.id is running
jobId: options?.id ? `${options.id}-${v4()}` : undefined,
priority: options?.priority ?? MESSAGE_QUEUE_PRIORITY[queueName],
attempts: 1 + (options?.retryLimit || 0),
removeOnComplete: {
age: QUEUE_RETENTION.completedMaxAge,
count: QUEUE_RETENTION.completedMaxCount,
},
removeOnFail: {
age: QUEUE_RETENTION.failedMaxAge,
count: QUEUE_RETENTION.failedMaxCount,
},
delay: options?.delay,
};
}
async add<T>(
queueName: MessageQueue,
jobName: string,
@@ -352,24 +376,43 @@ export class BullMQDriver
}
}
const queueOptions: JobsOptions = {
jobId: options?.id ? `${options.id}-${v4()}` : undefined, // We add V4() to id to make sure ids are uniques so we can add a waiting job when a job related with the same option.id is running
priority: options?.priority ?? MESSAGE_QUEUE_PRIORITY[queueName],
attempts: 1 + (options?.retryLimit || 0),
removeOnComplete: {
age: QUEUE_RETENTION.completedMaxAge,
count: QUEUE_RETENTION.completedMaxCount,
},
removeOnFail: {
age: QUEUE_RETENTION.failedMaxAge,
count: QUEUE_RETENTION.failedMaxCount,
},
delay: options?.delay,
};
const queueOptions = this.buildJobsOptions({ queueName, options });
await this.queueMap[queueName].add(jobName, data, queueOptions);
}
async bulkAdd<T>(
queueName: MessageQueue,
jobName: string,
dataItems: T[],
options?: QueueJobOptions,
): Promise<void> {
if (!this.queueMap[queueName]) {
throw new Error(
`Queue ${queueName} is not registered, make sure you have added it as a queue provider`,
);
}
if (dataItems.length === 0) {
return;
}
const queueOptions = this.buildJobsOptions({ queueName, options });
await this.queueMap[queueName].addBulk(
dataItems.map((data, index) => ({
name: jobName,
data,
opts: {
...queueOptions,
jobId: queueOptions.jobId
? `${queueOptions.jobId}-${index}`
: undefined,
},
})),
);
}
async getInFlightJobs<T extends MessageQueueJobData>(
queueName: MessageQueue,
): Promise<InFlightQueueJob<T>[]> {
packages/twenty-server/src/engine/core-modules/message-queue/drivers/interfaces/message-queue-driver.interface.ts
+6
View File
@@ -14,6 +14,12 @@ export interface MessageQueueDriver {
data: T,
options?: QueueJobOptions,
): Promise<void>;
bulkAdd<T extends MessageQueueJobData>(
queueName: MessageQueue,
jobName: string,
dataItems: T[],
options?: QueueJobOptions,
): Promise<void>;
work<T extends MessageQueueJobData>(
queueName: MessageQueue,
handler: ({ data, id }: { data: T; id: string }) => Promise<void> | void,
packages/twenty-server/src/engine/core-modules/message-queue/drivers/sync.driver.ts
+24
View File
@@ -1,5 +1,7 @@
import { Logger } from '@nestjs/common';
import { isDefined } from 'twenty-shared/utils';
import { type MessageQueueDriver } from 'src/engine/core-modules/message-queue/drivers/interfaces/message-queue-driver.interface';
import {
type MessageQueueJob,
@@ -25,6 +27,28 @@ export class SyncDriver implements MessageQueueDriver {
await this.processJob(queueName, { id: '', name: jobName, data });
}
async bulkAdd<T extends MessageQueueJobData>(
queueName: MessageQueue,
jobName: string,
dataItems: T[],
): Promise<void> {
let firstError: unknown = undefined;
// Each payload is an independent job in BullMQ, so a failing one must not
// prevent the others from being processed
for (const data of dataItems) {
try {
await this.processJob(queueName, { id: '', name: jobName, data });
} catch (error) {
firstError = firstError ?? error;
}
}
if (isDefined(firstError)) {
throw firstError;
}
}
async addCron<T extends MessageQueueJobData | undefined>({
queueName,
jobName,
packages/twenty-server/src/engine/core-modules/message-queue/services/message-queue.service.ts
+8
View File
@@ -38,6 +38,14 @@ export class MessageQueueService {
return this.driver.add(this.queueName, jobName, data, options);
}
bulkAdd<T extends MessageQueueJobData>(
jobName: string,
dataItems: T[],
options?: QueueJobOptions,
): Promise<void> {
return this.driver.bulkAdd(this.queueName, jobName, dataItems, options);
}
getInFlightJobs<T extends MessageQueueJobData>(): Promise<
InFlightQueueJob<T>[]
> {
packages/twenty-server/src/engine/core-modules/server-route-trigger/__tests__/server-route-trigger.service.spec.ts
+5 -7
View File
@@ -146,13 +146,11 @@ describe('ServerRouteTriggerService', () => {
);
expect(messageQueueService.add).toHaveBeenCalledWith(
LogicFunctionTriggerJob.name,
[
{
logicFunctionId: 'target-id',
workspaceId: 'target-ws',
payload: { from: 'resolver' },
},
],
{
logicFunctionId: 'target-id',
workspaceId: 'target-ws',
payload: { from: 'resolver' },
},
{ retryLimit: 3 },
);
expect(result).toEqual(
packages/twenty-server/src/engine/core-modules/server-route-trigger/server-route-trigger.service.ts
+6 -8
View File
@@ -189,15 +189,13 @@ export class ServerRouteTriggerService {
applicationRegistrationId,
});
await this.messageQueueService.add<LogicFunctionTriggerJobData[]>(
await this.messageQueueService.add<LogicFunctionTriggerJobData>(
LogicFunctionTriggerJob.name,
[
{
logicFunctionId: logicFunction.id,
workspaceId,
payload,
},
],
{
logicFunctionId: logicFunction.id,
workspaceId,
payload,
},
{ retryLimit: QUEUED_TARGET_RETRY_LIMIT },
);