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

feat(server): queued-only server-route dispatch (#23134)

Browse Source 
## Context

Follow-up to the incident where 500s and latency spiked around 6pm until
the Recall webhook was disabled. Server routes
(`/webhooks/server/:resolverUid`) ran the resolver **and** the target
logic function synchronously inside the API request, so any handler
throw became a 500 and any slowdown past Svix's delivery timeout marked
the delivery failed — Svix redelivered, feeding load back into the API
in a self-sustaining storm.

## What changed

- Every server-route request acks with **202 `{ queued: true }`** as
soon as the resolver returns; the target runs on `logicFunctionQueue`.
Signature verification stays synchronous in the resolver and still
rejects with a non-2xx. External senders never observe target latency or
failures.
- The resolver contract is unchanged from main: `{ workspaceId,
targetLogicFunctionUniversalIdentifier, payload? }`.
- The target lookup still happens synchronously before enqueueing,
scoped to the resolver's application registration, so unknown targets
404 as before.
- Endpoints whose caller must read the response body (challenge
handshakes, Slack commands) should use `httpRouteTriggerSettings`
routes.

Final diff is 4 files: the server-route service, its spec, the
integration spec, and the docs page. Trigger jobs, fan-out, message
queue, shared types, and SDK are all untouched.

## Tests

- `server-route-trigger.service.spec.ts`: 202 ack + enqueue,
unknown-target 404 without enqueue, resolver auth/contract/error
mapping.
- Integration: `server-route-trigger-authorization.integration-spec.ts`
asserts the 202 queued ack (run locally against a seeded DB, green).
- `typecheck` + `lint:diff-with-main` + `oxfmt` clean.

## Notes

- **Breaking for existing server-route resolvers**: responses are always
202; the target's return value no longer reaches the caller. Existing
resolvers returning response bodies must move those endpoints to
`httpRouteTriggerSettings`.
- A queued target's handler failure is recorded in execution logs but
not retried (same as other queue-executed functions today); retry
semantics are deliberately out of scope here.
- Follow-up candidates: retry-on-failure semantics for queued
executions, `addBulk` for single-round-trip fan-out, declarative
signature verification to take resolver code out of the request path,
moving the call-recorder 250s artifacts import off the API request path.
- Companion PR #23135 (call-recorder): no app change needed for dispatch
— queued dispatch applies by default.
This commit is contained in:
martmull
2026-07-22 14:57:18 +02:00
committed by GitHub
parent 271f9f833b
commit f59bda1dbf
4 changed files with 121 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/twenty-server/src/engine/core-modules/server-route-trigger/__tests__/server-route-trigger.service.spec.ts
+54 -82
View File
@@ -7,6 +7,8 @@ import {
LogicFunctionExecutionExceptionCode,
type LogicFunctionExecutorService,
} from 'src/engine/core-modules/logic-function/logic-function-executor/logic-function-executor.service';
import { LogicFunctionTriggerJob } from 'src/engine/core-modules/logic-function/logic-function-trigger/jobs/logic-function-trigger.job';
import { type MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
import { ServerRouteTriggerExceptionCode } from 'src/engine/core-modules/server-route-trigger/exceptions/server-route-trigger.exception';
import { ServerRouteTriggerService } from 'src/engine/core-modules/server-route-trigger/server-route-trigger.service';
import { type LogicFunctionEntity } from 'src/engine/metadata-modules/logic-function/logic-function.entity';
@@ -72,6 +74,7 @@ describe('ServerRouteTriggerService', () => {
let logicFunctionExecutorService: jest.Mocked<
Pick<LogicFunctionExecutorService, 'execute'>
>;
let messageQueueService: jest.Mocked<Pick<MessageQueueService, 'add'>>;
let resolverRow: unknown;
let queryBuilder: QueryBuilderMock;
@@ -101,54 +104,57 @@ describe('ServerRouteTriggerService', () => {
createQueryBuilder: jest.fn(() => queryBuilder),
findOne: jest
.fn()
// resolver lookup inside runFunction
// resolver lookup
.mockResolvedValueOnce({ id: 'resolver-id' })
// target lookup inside runFunction
// target lookup before enqueueing
.mockResolvedValueOnce({ id: 'target-id' }),
};
logicFunctionExecutorService = {
execute: jest
.fn()
// resolver returns { workspaceId, targetLogicFunctionUniversalIdentifier, payload }
.mockResolvedValueOnce(
buildExecuteResult({
workspaceId: 'target-ws',
targetLogicFunctionUniversalIdentifier: TARGET_UID,
payload: { from: 'resolver' },
}),
)
// target returns the final response body
.mockResolvedValueOnce(buildExecuteResult({ ok: true })),
execute: jest.fn().mockResolvedValueOnce(
buildExecuteResult({
workspaceId: 'target-ws',
targetLogicFunctionUniversalIdentifier: TARGET_UID,
payload: { from: 'resolver' },
}),
),
};
messageQueueService = {
add: jest.fn().mockResolvedValue(undefined),
};
service = new ServerRouteTriggerService(
logicFunctionRepository as unknown as Repository<LogicFunctionEntity>,
logicFunctionExecutorService as unknown as LogicFunctionExecutorService,
messageQueueService as unknown as MessageQueueService,
);
});
it('runs the resolver in the owner workspace then the resolver-named target in the resolved workspace', async () => {
it('runs the resolver synchronously, enqueues the target, and acks with 202', async () => {
const result = await handle();
expect(logicFunctionExecutorService.execute).toHaveBeenNthCalledWith(
1,
expect(logicFunctionExecutorService.execute).toHaveBeenCalledTimes(1);
expect(logicFunctionExecutorService.execute).toHaveBeenCalledWith(
expect.objectContaining({
logicFunctionId: 'resolver-id',
workspaceId: 'owner-ws',
}),
);
expect(logicFunctionExecutorService.execute).toHaveBeenNthCalledWith(
2,
expect.objectContaining({
logicFunctionId: 'target-id',
workspaceId: 'target-ws',
payload: { from: 'resolver' },
}),
expect(messageQueueService.add).toHaveBeenCalledWith(
LogicFunctionTriggerJob.name,
[
{
logicFunctionId: 'target-id',
workspaceId: 'target-ws',
payload: { from: 'resolver' },
},
],
{ retryLimit: 3 },
);
expect(result).toEqual(
expect.objectContaining({
statusCode: 200,
body: { ok: true },
statusCode: 202,
body: { queued: true },
}),
);
});
@@ -184,6 +190,21 @@ describe('ServerRouteTriggerService', () => {
);
});
it('scopes the target lookup to the resolver application registration', async () => {
await handle();
expect(logicFunctionRepository.findOne).toHaveBeenNthCalledWith(
2,
expect.objectContaining({
where: expect.objectContaining({
universalIdentifier: TARGET_UID,
workspaceId: 'target-ws',
application: { applicationRegistrationId: 'reg-1' },
}),
}),
);
});
it('throws LOGIC_FUNCTION_NOT_FOUND and executes nothing when the query returns no server-route resolver', async () => {
resolverRow = null;
@@ -191,6 +212,7 @@ describe('ServerRouteTriggerService', () => {
code: ServerRouteTriggerExceptionCode.LOGIC_FUNCTION_NOT_FOUND,
});
expect(logicFunctionExecutorService.execute).not.toHaveBeenCalled();
expect(messageQueueService.add).not.toHaveBeenCalled();
});
it('rejects and executes nothing when the resolver requires authentication', async () => {
@@ -240,35 +262,16 @@ describe('ServerRouteTriggerService', () => {
});
});
it('throws LOGIC_FUNCTION_NOT_FOUND when the target named by the resolver is missing in the resolved workspace', async () => {
it('throws LOGIC_FUNCTION_NOT_FOUND and enqueues nothing when the target is missing', async () => {
logicFunctionRepository.findOne.mockReset();
logicFunctionRepository.findOne
// resolver lookup succeeds
.mockResolvedValueOnce({ id: 'resolver-id' })
// target lookup returns null
.mockResolvedValueOnce(null);
await expect(handle()).rejects.toMatchObject({
code: ServerRouteTriggerExceptionCode.LOGIC_FUNCTION_NOT_FOUND,
});
});
it('surfaces a target userError as a server-route exception', async () => {
logicFunctionExecutorService.execute.mockReset();
logicFunctionExecutorService.execute
.mockResolvedValueOnce(
buildExecuteResult({
workspaceId: 'target-ws',
targetLogicFunctionUniversalIdentifier: TARGET_UID,
}),
)
.mockResolvedValueOnce(
buildExecuteResult(null, { errorMessage: 'boom' }),
);
await expect(handle()).rejects.toMatchObject({
code: ServerRouteTriggerExceptionCode.SERVER_ROUTE_USER_UNCAUGHT_ERROR,
});
expect(messageQueueService.add).not.toHaveBeenCalled();
});
it('maps a LogicFunctionExecutionException(LOGIC_FUNCTION_NOT_FOUND) to the server-route not-found code', async () => {
@@ -285,15 +288,6 @@ describe('ServerRouteTriggerService', () => {
});
});
it('falls back to PLATFORM_ERROR for any other thrown executor error', async () => {
logicFunctionExecutorService.execute.mockReset();
logicFunctionExecutorService.execute.mockRejectedValue(new Error('boom'));
await expect(handle()).rejects.toMatchObject({
code: ServerRouteTriggerExceptionCode.SERVER_ROUTE_PLATFORM_ERROR,
});
});
it('maps a LogicFunctionExecutionException(RATE_LIMIT_EXCEEDED) to the server-route rate-limit code', async () => {
logicFunctionExecutorService.execute.mockReset();
logicFunctionExecutorService.execute.mockRejectedValue(
@@ -308,33 +302,11 @@ describe('ServerRouteTriggerService', () => {
});
});
it('scopes the target lookup to the resolver application registration', async () => {
await handle();
expect(logicFunctionRepository.findOne).toHaveBeenNthCalledWith(
2,
expect.objectContaining({
where: expect.objectContaining({
universalIdentifier: TARGET_UID,
workspaceId: 'target-ws',
application: { applicationRegistrationId: 'reg-1' },
}),
}),
);
});
it('does not leak the raw executor error message to the caller', async () => {
it('does not leak the raw resolver executor error message to the caller', async () => {
logicFunctionExecutorService.execute.mockReset();
logicFunctionExecutorService.execute
.mockResolvedValueOnce(
buildExecuteResult({
workspaceId: 'target-ws',
targetLogicFunctionUniversalIdentifier: TARGET_UID,
}),
)
.mockRejectedValueOnce(
new Error('internal: connection to lambda-internal:5000 refused'),
);
logicFunctionExecutorService.execute.mockRejectedValue(
new Error('internal: connection to lambda-internal:5000 refused'),
);
await expect(handle()).rejects.toMatchObject({
code: ServerRouteTriggerExceptionCode.SERVER_ROUTE_PLATFORM_ERROR,
packages/twenty-server/src/engine/core-modules/server-route-trigger/server-route-trigger.service.ts
+62 -16
View File
@@ -11,11 +11,15 @@ import {
LogicFunctionExecutionExceptionCode,
LogicFunctionExecutorService,
} from 'src/engine/core-modules/logic-function/logic-function-executor/logic-function-executor.service';
import { buildLogicFunctionEvent } from 'src/engine/core-modules/logic-function/logic-function-trigger/triggers/route/utils/build-logic-function-event.util';
import {
type RouteTriggerResponse,
buildRouteTriggerResponse,
} from 'src/engine/core-modules/logic-function/logic-function-trigger/triggers/route/utils/route-trigger-response.util';
LogicFunctionTriggerJob,
type LogicFunctionTriggerJobData,
} from 'src/engine/core-modules/logic-function/logic-function-trigger/jobs/logic-function-trigger.job';
import { buildLogicFunctionEvent } from 'src/engine/core-modules/logic-function/logic-function-trigger/triggers/route/utils/build-logic-function-event.util';
import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator';
import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
import { type RouteTriggerResponse } from 'src/engine/core-modules/logic-function/logic-function-trigger/triggers/route/utils/route-trigger-response.util';
import {
ServerRouteTriggerException,
ServerRouteTriggerExceptionCode,
@@ -28,6 +32,8 @@ type ResolverResult = {
payload?: object;
};
const QUEUED_TARGET_RETRY_LIMIT = 3;
@Injectable()
export class ServerRouteTriggerService {
private readonly logger = new Logger(ServerRouteTriggerService.name);
@@ -36,6 +42,8 @@ export class ServerRouteTriggerService {
@InjectRepository(LogicFunctionEntity)
private readonly logicFunctionRepository: Repository<LogicFunctionEntity>,
private readonly logicFunctionExecutorService: LogicFunctionExecutorService,
@InjectMessageQueue(MessageQueue.logicFunctionQueue)
private readonly messageQueueService: MessageQueueService,
) {}
async handle({
@@ -89,22 +97,13 @@ export class ServerRouteTriggerService {
});
const resolved = this.parseResolverResult(resolverResult);
const targetResult = await this.runFunction({
return await this.enqueueTargetFunction({
logicFunctionUniversalIdentifier:
resolved.targetLogicFunctionUniversalIdentifier,
workspaceId: resolved.workspaceId,
payload: resolved.payload ?? event,
applicationRegistrationId,
});
if (isDefined(targetResult.error)) {
throw new ServerRouteTriggerException(
targetResult.error.errorMessage,
ServerRouteTriggerExceptionCode.SERVER_ROUTE_USER_UNCAUGHT_ERROR,
);
}
return buildRouteTriggerResponse(targetResult.data);
}
private async findResolver({
@@ -169,7 +168,7 @@ export class ServerRouteTriggerService {
};
}
private async runFunction({
private async enqueueTargetFunction({
logicFunctionUniversalIdentifier,
workspaceId,
payload,
@@ -178,8 +177,38 @@ export class ServerRouteTriggerService {
logicFunctionUniversalIdentifier: string;
workspaceId: string;
payload: object;
applicationRegistrationId: string;
}): Promise<RouteTriggerResponse> {
const logicFunction = await this.findLogicFunctionOrFail({
logicFunctionUniversalIdentifier,
workspaceId,
applicationRegistrationId,
});
await this.messageQueueService.add<LogicFunctionTriggerJobData[]>(
LogicFunctionTriggerJob.name,
[
{
logicFunctionId: logicFunction.id,
workspaceId,
payload,
},
],
{ retryLimit: QUEUED_TARGET_RETRY_LIMIT },
);
return { statusCode: 202, headers: {}, body: { queued: true } };
}
private async findLogicFunctionOrFail({
logicFunctionUniversalIdentifier,
workspaceId,
applicationRegistrationId,
}: {
logicFunctionUniversalIdentifier: string;
workspaceId: string;
applicationRegistrationId?: string;
}): Promise<{ data: object | null; error?: { errorMessage: string } }> {
}): Promise<LogicFunctionEntity> {
const logicFunction = await this.logicFunctionRepository.findOne({
where: {
universalIdentifier: logicFunctionUniversalIdentifier,
@@ -200,6 +229,23 @@ export class ServerRouteTriggerService {
);
}
return logicFunction;
}
private async runFunction({
logicFunctionUniversalIdentifier,
workspaceId,
payload,
}: {
logicFunctionUniversalIdentifier: string;
workspaceId: string;
payload: object;
}): Promise<{ data: object | null; error?: { errorMessage: string } }> {
const logicFunction = await this.findLogicFunctionOrFail({
logicFunctionUniversalIdentifier,
workspaceId,
});
try {
return await this.logicFunctionExecutorService.execute({
logicFunctionId: logicFunction.id,