github-mirror/n8n
1
0
Fork 0
mirror of https://github.com/n8n-io/n8n.git synced 2026-10-19 16:55:02 +02:00
Code Issues Actions 12 Packages Projects Releases Wiki Activity Graph

refactor(core): Simplify turn lease loss and claim finalization

Browse Source 
Abort the thread lease signal with a reason in the coordinator, so the
orchestrator no longer names lease loss itself. Merge the three
finalization methods of the execution repository into one with an
optional stale guard. Remove the wake service suspension pre-check; the
orchestrator keeps the post-admission check. Drop the unused conflict
error field and the one-field waiter interface.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Robin Braumann 2026-09-12 12:07:30 +02:00
parent ff94b893ac
commit fb65257ca5
No known key found for this signature in database
11 changed files with 47 additions and 130 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
packages/cli/src/modules/agents/__tests__/agent-execution-orchestrator.service.test.ts
View File

@ -330,7 +330,7 @@ describe('AgentExecutionOrchestratorService', () => {
it('waits for a foreign claimed row to end and claims again on conflict', async () => {
const { service, executionService, turnCoordinator, permitFor } = makeService();
executionService.startClaimedExecutionRecording
.mockRejectedValueOnce(new AgentThreadClaimConflictError('thread-1'))
.mockRejectedValueOnce(new AgentThreadClaimConflictError())
.mockResolvedValueOnce({
executionId: 'execution-2',
claimLost: new AbortController().signal,
@ -365,7 +365,7 @@ describe('AgentExecutionOrchestratorService', () => {
const { service, executionService, turnCoordinator, permitFor } = makeService();
const permit = await permitFor('thread-1');
executionService.startClaimedExecutionRecording.mockRejectedValue(
new AgentThreadClaimConflictError('thread-1'),
new AgentThreadClaimConflictError(),
);
turnCoordinator.executionRepository.existsRunningByThread.mockImplementationOnce(async () => {
turnCoordinator.leases.at(-1)?.abort();
@ -388,7 +388,7 @@ describe('AgentExecutionOrchestratorService', () => {
permit,
}),
),
).rejects.toThrow('Agent thread lease was lost before the turn could claim its thread');
).rejects.toThrow('Agent thread lease was lost');
expect(executionService.startClaimedExecutionRecording).toHaveBeenCalledOnce();
expect(runtime.agent.stream).not.toHaveBeenCalled();

7
packages/cli/src/modules/agents/__tests__/agent-execution.service.test.ts
View File

@ -73,7 +73,6 @@ describe('AgentExecutionService', () => {
agentExecutionRepository = mock<AgentExecutionRepository>();
agentExecutionRepository.updateIfRunning.mockResolvedValue(true);
agentExecutionRepository.updateIfAbandoned.mockResolvedValue(true);
agentExecutionRepository.updateTimelineIfRunning.mockResolvedValue(true);
agentExecutionThreadRepository = mock<AgentExecutionThreadRepository>();
n8nMemory = mock<N8nMemory>();
@ -864,7 +863,7 @@ describe('AgentExecutionService', () => {
executionUpdateBroadcaster,
);
const partial = [{ type: 'text', content: 'Partial', timestamp: 1, endTime: 2 }] as const;
agentExecutionRepository.updateIfAbandoned.mockResolvedValue(true);
agentExecutionRepository.updateIfRunning.mockResolvedValue(true);
agentExecutionThreadRepository.findOneBy.mockResolvedValue(makeThread());
await service.finalizeInterruptedExecution({
@ -883,9 +882,8 @@ describe('AgentExecutionService', () => {
executionId: 'execution-1',
}),
);
expect(agentExecutionRepository.updateIfAbandoned).toHaveBeenCalledWith(
expect(agentExecutionRepository.updateIfRunning).toHaveBeenCalledWith(
'execution-1',
expect.any(Date),
expect.objectContaining({
status: 'interrupted',
timeline: partial,
@ -901,6 +899,7 @@ describe('AgentExecutionService', () => {
},
},
}),
expect.any(Date),
);
expect(agentExecutionLogStore.write).not.toHaveBeenCalled();
});

6
packages/cli/src/modules/agents/__tests__/agent-thread-turn-coordinator.test.ts
View File

@ -127,8 +127,8 @@ describe('AgentThreadTurnCoordinator', () => {
const run = coordinator.run('t1', undefined, body);
await flush();
leases[0].abort(new Error('lock lost'));
await expect(run).rejects.toThrow('lock lost');
leases[0].abort();
await expect(run).rejects.toThrow('Agent thread lease was lost');
expect(body).not.toHaveBeenCalled();
executionRepository.existsRunningByThread.mockResolvedValue(false);
@ -157,7 +157,7 @@ describe('AgentThreadTurnCoordinator', () => {
const stream = coordinator.stream('t1', undefined, async function* (permit) {
expect(permit.threadId).toBe('t1');
yield 'one';
leases[0].abort(new Error('lock lost'));
leases[0].abort();
expect(permit.leaseLost.aborted).toBe(true);
yield 'two';
});

13
packages/cli/src/modules/agents/agent-execution-orchestrator.service.ts
View File

@ -902,6 +902,9 @@ export class AgentExecutionOrchestratorService {
const integrationType = isDraft ? N8N_CHAT_INTEGRATION_TYPE : identity.integrationType;
return await this.turnCoordinator.run(memory.threadId, abortSignal, async (permit) => {
// A suspended thread belongs to its pending resume, which may have
// started while this wake waited. Execution rows keep `suspended`
// after a resume, so the checkpoint store decides.
if (
(await this.agentExecutionService.hasSuspendedRun(memory.threadId)) &&
(await this.n8nCheckpointStorage.findSuspendedForThread(agentId, memory.threadId)) !== null
@ -1236,14 +1239,10 @@ export class AgentExecutionOrchestratorService {
permit: AgentThreadTurnPermit,
abortSignal?: AbortSignal,
): Promise<{ executionId: string; abortSignal: AbortSignal }> {
// The lock service aborts `leaseLost` without a reason; name the cause here.
const leaseLostError = () =>
new OperationalError('Agent thread lease was lost before the turn could claim its thread');
const waitSignal = AbortSignal.any(
[abortSignal, permit.leaseLost].filter((s) => s !== undefined),
);
for (;;) {
if (permit.leaseLost.aborted) throw leaseLostError();
waitSignal.throwIfAborted();
try {
const { executionId, claimLost } =
@ -1260,11 +1259,7 @@ export class AgentExecutionOrchestratorService {
agentId: params.agentId,
threadId: params.threadId,
});
try {
await this.turnCoordinator.waitUntilThreadIdle(params.threadId, waitSignal);
} catch (waitError) {
throw permit.leaseLost.aborted ? leaseLostError() : waitError;
}
await this.turnCoordinator.waitUntilThreadIdle(params.threadId, waitSignal);
}
}
}

5
packages/cli/src/modules/agents/agent-execution.service.ts
View File

@ -265,10 +265,8 @@ export class AgentExecutionService {
const duration = execution.startedAt
? Math.max(0, stoppedAt.getTime() - execution.startedAt.getTime())
: 0;
const staleBefore = new Date(Date.now() - AgentExecutionService.livenessGraceMs);
const finalized = await this.agentExecutionRepository.updateIfAbandoned(
const finalized = await this.agentExecutionRepository.updateIfRunning(
execution.id,
staleBefore,
{
status: 'interrupted',
stoppedAt,
@ -283,6 +281,7 @@ export class AgentExecutionService {
stoppedAt: stoppedAt.getTime(),
}),
},
new Date(Date.now() - AgentExecutionService.livenessGraceMs),
);
if (finalized) void this.notifyInterruptedExecution(execution);
return finalized;

38
packages/cli/src/modules/agents/agent-thread-turn-coordinator.ts
View File

@ -3,7 +3,7 @@ import { Service } from '@n8n/di';
import { ensureError } from '@n8n/utils/errors/ensure-error';
import { createDeferredPromise } from '@n8n/utils/promise/deferred-promise';
import { sleep } from '@n8n/utils/sleep';
import { UserError } from 'n8n-workflow';
import { OperationalError, UserError } from 'n8n-workflow';
import { AgentExecutionRepository } from './repositories/agent-execution.repository';
@ -44,10 +44,6 @@ export interface AgentThreadTurnLease {
release: () => Promise<void>;
}
interface Waiter {
resolve: () => void;
}
/**
* Admits one agent turn per durable thread at a time.
*
@ -61,7 +57,7 @@ interface Waiter {
@Service()
export class AgentThreadTurnCoordinator {
/** Threads with an active turn in this process, with their waiting turns in arrival order. */
private readonly waiters = new Map<string, Waiter[]>();
private readonly waiters = new Map<string, Array<() => void>>();
constructor(
private readonly lockService: LockService,
@ -111,8 +107,7 @@ export class AgentThreadTurnCoordinator {
[signal, lease.leaseLost].filter((s) => s !== undefined),
);
await this.waitUntilThreadIdle(threadId, idleWaitSignal);
signal?.throwIfAborted();
lease.leaseLost.throwIfAborted();
idleWaitSignal.throwIfAborted();
} catch (error) {
await lease?.release();
this.releaseLocally(threadId);
@ -152,11 +147,9 @@ export class AgentThreadTurnCoordinator {
if (index !== -1) queue.splice(index, 1);
reject(ensureError(signal?.reason));
};
const waiter: Waiter = {
resolve: () => {
signal?.removeEventListener('abort', onAbort);
resolve();
},
const waiter = () => {
signal?.removeEventListener('abort', onAbort);
resolve();
};
queue.push(waiter);
signal?.addEventListener('abort', onAbort, { once: true });
@ -166,28 +159,35 @@ export class AgentThreadTurnCoordinator {
/** Hand the turn to the next local waiter, or forget the thread when idle. */
private releaseLocally(threadId: string): void {
const next = this.waiters.get(threadId)?.shift();
if (next) next.resolve();
if (next) next();
else this.waiters.delete(threadId);
}
/**
* Hold the distributed lease as a handle instead of a callback scope, so a
* generator can keep it until its consumer is done.
* generator can keep it until its consumer is done. The lock service aborts
* without a reason, so `leaseLost` carries one for every consumer.
*/
private async acquireLease(
threadId: string,
): Promise<{ leaseLost: AbortSignal; release: () => Promise<void> }> {
const acquired = createDeferredPromise<AbortSignal>();
const acquired = createDeferredPromise();
const released = createDeferredPromise();
const lost = new AbortController();
const held = this.lockService
.withLease(LockNamespace.KNOWN_LOCKS, `agent-thread-turn:${threadId}`, async (signal) => {
acquired.resolve(signal);
signal.addEventListener(
'abort',
() => lost.abort(new OperationalError('Agent thread lease was lost')),
{ once: true },
);
acquired.resolve();
await released.promise;
})
.catch((error: Error) => acquired.reject(error));
const leaseLost = await acquired.promise;
await acquired.promise;
return {
leaseLost,
leaseLost: lost.signal,
release: async () => {
released.resolve();
await held;

43
packages/cli/src/modules/agents/background/__tests__/agent-wake.service.test.ts
View File

@ -13,9 +13,7 @@ import { hashAgentSandboxPrincipal } from '../../agent-sandbox-principal';
import { AgentThreadQueueFullError } from '../../agent-thread-turn-coordinator';
import type { AgentBackgroundJob } from '../../entities/agent-background-job.entity';
import type { ChatIntegrationRegistry } from '../../integrations/agent-chat-integration';
import type { N8NCheckpointStorage } from '../../integrations/n8n-checkpoint-storage';
import type { AgentBackgroundJobRepository } from '../../repositories/agent-background-job.repository';
import type { AgentExecutionRepository } from '../../repositories/agent-execution.repository';
import type { AgentRepository } from '../../repositories/agent.repository';
import {
AgentWakeService,
@ -60,10 +58,8 @@ function makeJob(overrides: Partial<AgentBackgroundJob> = {}): AgentBackgroundJo
function setup(options: { worker?: boolean; enabled?: boolean } = {}) {
const jobRepository = mock<AgentBackgroundJobRepository>();
const executionRepository = mock<AgentExecutionRepository>();
const agentRepository = mock<AgentRepository>();
const userRepository = mock<UserRepository>();
const checkpointStorage = mock<N8NCheckpointStorage>();
const integrationRegistry = mock<ChatIntegrationRegistry>();
const orchestrator = mock<AgentExecutionOrchestratorService>();
const lockService = mock<LockService>();
@ -74,9 +70,6 @@ function setup(options: { worker?: boolean; enabled?: boolean } = {}) {
logger.scoped.mockReturnValue(logger);
jobRepository.findWakeableUnconsumedSettled.mockResolvedValue([makeJob()]);
executionRepository.existsRunningByThread.mockResolvedValue(false);
executionRepository.hasSuspendedRun.mockResolvedValue(false);
checkpointStorage.findSuspendedForThread.mockResolvedValue(null);
agentRepository.findById.mockResolvedValue({ id: 'agent-1', projectId: 'project-1' } as never);
userRepository.findByIdWithRole.mockResolvedValue(user as never);
integrationRegistry.get.mockReturnValue({} as never);
@ -95,10 +88,8 @@ function setup(options: { worker?: boolean; enabled?: boolean } = {}) {
const service = new AgentWakeService(
jobRepository,
executionRepository,
agentRepository,
userRepository,
checkpointStorage,
integrationRegistry,
orchestrator,
lockService,
@ -111,10 +102,8 @@ function setup(options: { worker?: boolean; enabled?: boolean } = {}) {
return {
service,
jobRepository,
executionRepository,
agentRepository,
userRepository,
checkpointStorage,
orchestrator,
lockService,
publisher,
@ -357,21 +346,6 @@ describe('AgentWakeService', () => {
}
});
it('wakes behind a running parent turn but not a suspended one', async () => {
// The orchestrator queues the wake behind the running turn; the service no longer skips it.
const running = setup();
running.executionRepository.existsRunningByThread.mockResolvedValue(true);
await running.service.attemptWake('thread-1');
expect(running.orchestrator.executeForWake).toHaveBeenCalledTimes(1);
expect(running.jobRepository.markMailConsumed).toHaveBeenCalledWith('thread-1', ['job-1']);
const suspended = setup();
suspended.executionRepository.hasSuspendedRun.mockResolvedValue(true);
suspended.checkpointStorage.findSuspendedForThread.mockResolvedValue({} as never);
await suspended.service.attemptWake('thread-1');
expect(suspended.orchestrator.executeForWake).not.toHaveBeenCalled();
});
it('leaves results pending without recording a failure when the admitted wake is skipped', async () => {
const { service, orchestrator, jobRepository, logger } = setup();
orchestrator.executeForWake.mockResolvedValue('skipped');
@ -396,23 +370,6 @@ describe('AgentWakeService', () => {
expect(logger.warn).not.toHaveBeenCalled();
});
it('checks checkpoints only for threads with a recorded suspension', async () => {
const never = setup();
await never.service.attemptWake('thread-1');
expect(never.checkpointStorage.findSuspendedForThread).not.toHaveBeenCalled();
expect(never.orchestrator.executeForWake).toHaveBeenCalledTimes(1);
// A resumed run keeps its suspended status. Only an active checkpoint blocks the wake.
const resumed = setup();
resumed.executionRepository.hasSuspendedRun.mockResolvedValue(true);
await resumed.service.attemptWake('thread-1');
expect(resumed.checkpointStorage.findSuspendedForThread).toHaveBeenCalledWith(
'agent-1',
'thread-1',
);
expect(resumed.orchestrator.executeForWake).toHaveBeenCalledTimes(1);
});
it('does not mark results as delivered after the wake loses its lease', async () => {
const { service, lockService, orchestrator, jobRepository } = setup();
const controller = new AbortController();

15
packages/cli/src/modules/agents/background/agent-wake.service.ts
View File

@ -22,9 +22,7 @@ import {
} from './background-job-messages';
import type { AgentBackgroundJob } from '../entities/agent-background-job.entity';
import { ChatIntegrationRegistry } from '../integrations/agent-chat-integration';
import { N8NCheckpointStorage } from '../integrations/n8n-checkpoint-storage';
import { AgentBackgroundJobRepository } from '../repositories/agent-background-job.repository';
import { AgentExecutionRepository } from '../repositories/agent-execution.repository';
import { AgentRepository } from '../repositories/agent.repository';
import {
integrationTypeFromMemoryResourceId,
@ -51,10 +49,8 @@ export class AgentWakeService {
constructor(
private readonly jobRepository: AgentBackgroundJobRepository,
private readonly executionRepository: AgentExecutionRepository,
private readonly agentRepository: AgentRepository,
private readonly userRepository: UserRepository,
private readonly checkpointStorage: N8NCheckpointStorage,
private readonly integrationRegistry: ChatIntegrationRegistry,
private readonly orchestrator: AgentExecutionOrchestratorService,
private readonly lockService: LockService,
@ -166,17 +162,6 @@ export class AgentWakeService {
return;
}
// A running parent is no reason to skip: the wake queues behind it as a
// thread turn. A suspended parent is, because the resume owns the thread.
// Execution records keep their suspended status after a resume, so check
// the checkpoint store to determine whether the thread is still suspended.
if (
(await this.executionRepository.hasSuspendedRun(threadId)) &&
(await this.checkpointStorage.findSuspendedForThread(first.parentAgentId, threadId)) !== null
) {
return;
}
const agent = await this.agentRepository.findById(first.parentAgentId);
if (!agent) {
this.recordFailure(threadId, generation, 'Background job parent agent no longer exists');

26
packages/cli/src/modules/agents/repositories/agent-execution.repository.ts
View File

@ -14,7 +14,7 @@ export type RunningAgentExecution = Pick<
/** Another running turn already claims this thread. The caller waits and retries. */
export class AgentThreadClaimConflictError extends OperationalError {
constructor(readonly threadId: string) {
constructor() {
super('Another agent turn already holds this thread', { level: 'info' });
}
}
@ -66,7 +66,7 @@ export class AgentExecutionRepository extends Repository<AgentExecution> {
return await this.save(this.create(values));
} catch (error) {
if (values.activeThreadId !== null && isUniqueConstraintError(error)) {
throw new AgentThreadClaimConflictError(values.activeThreadId);
throw new AgentThreadClaimConflictError();
}
throw error;
}
@ -96,26 +96,14 @@ export class AgentExecutionRepository extends Repository<AgentExecution> {
return result.affected === 1;
}
/** Move a running row to a terminal status. This also releases its thread claim. */
/**
* Move a running row to a terminal status. This also releases its thread
* claim. With `staleBefore`, only a row without a heartbeat since then
* matches, so a run that is alive again keeps its claim.
*/
async updateIfRunning(
executionId: string,
values: AgentExecutionFinalizationValues,
): Promise<boolean> {
return await this.finalizeRunning(executionId, values);
}
/** Finalize a running row only while its liveness timestamp remains stale. */
async updateIfAbandoned(
executionId: string,
staleBefore: Date,
values: AgentExecutionFinalizationValues,
): Promise<boolean> {
return await this.finalizeRunning(executionId, values, staleBefore);
}
private async finalizeRunning(
executionId: string,
values: AgentExecutionFinalizationValues,
staleBefore?: Date,
): Promise<boolean> {
const result = await this.update(

6
packages/cli/test/integration/database/repositories/agent-background-job.repository.test.ts
View File

@ -15,7 +15,6 @@ import { AgentBackgroundJobService } from '@/modules/agents/background/agent-bac
import { AgentWakeService, WAKE_DEBOUNCE_MS } from '@/modules/agents/background/agent-wake.service';
import type { AgentBackgroundJob } from '@/modules/agents/entities/agent-background-job.entity';
import type { Agent } from '@/modules/agents/entities/agent.entity';
import type { N8NCheckpointStorage } from '@/modules/agents/integrations/n8n-checkpoint-storage';
import type { ChatIntegrationRegistry } from '@/modules/agents/integrations/agent-chat-integration';
import { AgentBackgroundJobRepository } from '@/modules/agents/repositories/agent-background-job.repository';
import type { AgentExecutionRepository } from '@/modules/agents/repositories/agent-execution.repository';
@ -175,9 +174,6 @@ describe('AgentBackgroundJobRepository', () => {
});
const executionRepository = mock<AgentExecutionRepository>();
executionRepository.existsRunningByThread.mockResolvedValue(false);
const checkpointStorage = mock<N8NCheckpointStorage>();
checkpointStorage.findSuspendedForThread.mockResolvedValue(null);
const orchestrator = mock<AgentExecutionOrchestratorService>();
let firstWakeStarted!: () => void;
const firstWake = new Promise<void>((resolve) => (firstWakeStarted = resolve));
@ -202,10 +198,8 @@ describe('AgentBackgroundJobRepository', () => {
});
const wakeService = new AgentWakeService(
repository,
executionRepository,
agentRepository,
userRepository,
checkpointStorage,
mock<ChatIntegrationRegistry>(),
orchestrator,
lockService,

12
packages/cli/test/integration/database/repositories/agent-execution.repository.test.ts
View File

@ -295,9 +295,9 @@ describe('AgentExecutionRepository', () => {
{ id: execution.id },
{ updatedAt: new Date('2026-01-02T00:00:01Z') },
);
expect(
await repository.updateIfAbandoned(execution.id, staleBefore, finalizationValues),
).toBe(false);
expect(await repository.updateIfRunning(execution.id, finalizationValues, staleBefore)).toBe(
false,
);
expect(await repository.findOneByOrFail({ id: execution.id })).toMatchObject({
status: 'running',
activeThreadId: thread.id,
@ -308,9 +308,9 @@ describe('AgentExecutionRepository', () => {
{ id: execution.id },
{ updatedAt: new Date('2026-01-01T23:59:59Z') },
);
expect(
await repository.updateIfAbandoned(execution.id, staleBefore, finalizationValues),
).toBe(true);
expect(await repository.updateIfRunning(execution.id, finalizationValues, staleBefore)).toBe(
true,
);
expect(await repository.findOneByOrFail({ id: execution.id })).toMatchObject({
status: 'interrupted',
activeThreadId: null,