mirror of
https://github.com/n8n-io/n8n.git
synced 2026-10-17 05:29:44 +02:00
fix(core): Stabilise background child approval wait (#40834)
Co-authored-by: cubic-dev-ai[bot] <191113872+cubic-dev-ai[bot]@users.noreply.github.com>
This commit is contained in:
parent
cea974e770
commit
2ca4e55502
|
|
@ -380,7 +380,9 @@ describe('AgentExecutionRepository', () => {
|
|||
});
|
||||
}
|
||||
|
||||
describe('background child approvals', () => {
|
||||
describe('background child approvals', { timeout: 20_000 }, () => {
|
||||
const childSettleTimeoutMs = 8_000;
|
||||
|
||||
afterEach(async () => {
|
||||
vi.restoreAllMocks();
|
||||
await Container.get(AgentBackgroundJobRepository).delete({});
|
||||
|
|
@ -446,10 +448,6 @@ describe('AgentExecutionRepository', () => {
|
|||
},
|
||||
);
|
||||
const wake = mock<AgentWakeService>();
|
||||
const initialSuspensionPersisted = createDeferredPromise();
|
||||
wake.requestWake.mockImplementation(async () => {
|
||||
initialSuspensionPersisted.resolve();
|
||||
});
|
||||
const getService = Container.get.bind(Container);
|
||||
vi.spyOn(Container, 'get').mockImplementation((service) => {
|
||||
if (service === AgentRuntimeReconstructionService) return reconstruction;
|
||||
|
|
@ -530,8 +528,32 @@ describe('AgentExecutionRepository', () => {
|
|||
if (!approval) throw new Error('Expected a pending approval');
|
||||
return { job, approval };
|
||||
};
|
||||
await vi.waitFor(() => expect(wake.requestWake).toHaveBeenCalledWith(parent.id));
|
||||
await initialSuspensionPersisted.promise;
|
||||
// spawn returns before the child suspends. Read the job row before the
|
||||
// wake assertion so a failed child is distinct from a late wake.
|
||||
const deadline = Date.now() + childSettleTimeoutMs;
|
||||
let suspendedJob = await jobs.findById(receipt.jobId);
|
||||
while (suspendedJob?.status === 'running' && Date.now() < deadline) {
|
||||
await new Promise((resolve) => setTimeout(resolve, 50));
|
||||
suspendedJob = await jobs.findById(receipt.jobId);
|
||||
}
|
||||
const jobState = [
|
||||
`status=${suspendedJob?.status ?? 'missing'}`,
|
||||
`error=${suspendedJob?.error ?? 'null'}`,
|
||||
].join(', ');
|
||||
if (!suspendedJob || suspendedJob.status !== 'suspended' || suspendedJob.error !== null) {
|
||||
throw new Error(`Background child did not suspend (${jobState})`);
|
||||
}
|
||||
try {
|
||||
await vi.waitFor(() => expect(wake.requestWake).toHaveBeenCalledWith(parent.id), {
|
||||
timeout: childSettleTimeoutMs,
|
||||
interval: 50,
|
||||
});
|
||||
} catch (error) {
|
||||
throw new Error(
|
||||
`Background child is suspended but requestWake was not called for ${parent.id}`,
|
||||
{ cause: error },
|
||||
);
|
||||
}
|
||||
expect(await jobs.findById(receipt.jobId)).toMatchObject({
|
||||
status: 'suspended',
|
||||
error: null,
|
||||
|
|
@ -580,11 +602,13 @@ describe('AgentExecutionRepository', () => {
|
|||
{ token: approval.token, resumeData: { approved } },
|
||||
context,
|
||||
);
|
||||
await vi.waitFor(async () =>
|
||||
expect(await jobs.findById(job.id)).toMatchObject({
|
||||
status: 'completed',
|
||||
childThreadId: job.childThreadId,
|
||||
}),
|
||||
await vi.waitFor(
|
||||
async () =>
|
||||
expect(await jobs.findById(job.id)).toMatchObject({
|
||||
status: 'completed',
|
||||
childThreadId: job.childThreadId,
|
||||
}),
|
||||
{ timeout: childSettleTimeoutMs },
|
||||
);
|
||||
expect(fixture.action).toHaveBeenCalledTimes(approved ? 1 : 0);
|
||||
expect(fixture.workspaceService.getAgentWorkspace).toHaveBeenCalledWith(
|
||||
|
|
@ -621,11 +645,14 @@ describe('AgentExecutionRepository', () => {
|
|||
.runner.resume(job, { token: approval.token, resumeData: { approved: true } }, context),
|
||||
]);
|
||||
expect(outcomes.map(({ status }) => status).sort()).toEqual(['fulfilled', 'rejected']);
|
||||
await vi.waitFor(async () => {
|
||||
const current = await fixture.readApproval();
|
||||
expect(current.job.status).toBe('suspended');
|
||||
expect(current.approval.token).not.toBe(approval.token);
|
||||
});
|
||||
await vi.waitFor(
|
||||
async () => {
|
||||
const current = await fixture.readApproval();
|
||||
expect(current.job.status).toBe('suspended');
|
||||
expect(current.approval.token).not.toBe(approval.token);
|
||||
},
|
||||
{ timeout: childSettleTimeoutMs },
|
||||
);
|
||||
const next = await fixture.readApproval();
|
||||
expect(
|
||||
await fixture.service.settle(
|
||||
|
|
@ -649,8 +676,9 @@ describe('AgentExecutionRepository', () => {
|
|||
{ token: next.approval.token, resumeData: { approved: true } },
|
||||
context,
|
||||
);
|
||||
await vi.waitFor(async () =>
|
||||
expect(await jobs.findById(job.id)).toMatchObject({ status: 'completed' }),
|
||||
await vi.waitFor(
|
||||
async () => expect(await jobs.findById(job.id)).toMatchObject({ status: 'completed' }),
|
||||
{ timeout: childSettleTimeoutMs },
|
||||
);
|
||||
expect(fixture.action).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
|
@ -682,8 +710,9 @@ describe('AgentExecutionRepository', () => {
|
|||
{ token: retry.approval.token, resumeData: { approved: true } },
|
||||
context,
|
||||
);
|
||||
await vi.waitFor(async () =>
|
||||
expect(await jobs.findById(job.id)).toMatchObject({ status: 'completed' }),
|
||||
await vi.waitFor(
|
||||
async () => expect(await jobs.findById(job.id)).toMatchObject({ status: 'completed' }),
|
||||
{ timeout: childSettleTimeoutMs },
|
||||
);
|
||||
expect(action).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user