Skip to content

test(agents): await native dispatch in delivery nonce regressions #2631

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
lilyshen0722 merged 1 commit into from
Oct 24, 2026
Merged

test(agents): await native dispatch in delivery nonce regressions #2631

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
85 changes: 45 additions & 40 deletions backend/__tests__/unit/services/agentEventService.deliveryNonce.test.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,33 @@ const AGENT = 'pixel';
const INSTANCE = 'default';
const POD_ID = new mongoose.Types.ObjectId();

// Dispatch loads digest state before invoking the runner and does not await
// the turn. Observe that boundary; a returned enqueue/sweep is not a run.
async function waitForNativeCalls(runAgent, count) {
const deadline = Date.now() + 5000;
while (runAgent.mock.calls.length < count && Date.now() < deadline) {
await new Promise((resolve) => setTimeout(resolve, 10));
}
expect(runAgent).toHaveBeenCalledTimes(count);
}

async function settleNativeRun(settle, result) {
let resolveAck;
const acknowledged = new Promise((resolve) => { resolveAck = resolve; });
const original = AgentEventService.acknowledge.bind(AgentEventService);
const ack = jest.spyOn(AgentEventService, 'acknowledge').mockImplementation(async (...args) => {
const result = original(...args);
resolveAck(result);
return result;
});
try {
settle(result);
await acknowledged;
} finally {
ack.mockRestore();
}
}

beforeAll(async () => {
mongod = await MongoMemoryServer.create();
await mongoose.connect(mongod.getUri());
Expand Down Expand Up @@ -643,10 +670,9 @@ describe('the native path carries its own delivery nonce', () => {
const replacement = await claim();
expect(replacement.payload.deliveryId).not.toBe(born.deliveryNonce);

// The slow native run finally finishes and acks.
settle('done');
await new Promise((resolve) => setImmediate(resolve));
await new Promise((resolve) => setImmediate(resolve));
// Wait for both the run and its real acknowledgement, not two event-loop ticks.
await waitForNativeCalls(runAgent, 1);
await settleNativeRun(settle, 'done');

const after = await AgentEvent.findById(event._id).lean();
expect(after.status).toBe('delivered');
Expand Down Expand Up @@ -682,10 +708,7 @@ describe('the sweep re-runs a stuck in-process event', () => {
// Native dispatch now gathers pending ambient digest before calling the
// runner. Wait for that asynchronous read so it cannot leak into the next
// test or make an immediate runAgent assertion race the database.
for (let i = 0; i < 50 && runAgent.mock.calls.length === 0; i += 1) {
// eslint-disable-next-line no-await-in-loop
await new Promise((resolve) => setTimeout(resolve, 10));
}
await waitForNativeCalls(runAgent, 1);
const born = await AgentEvent.findById(event._id).lean();
return { event, born, settles };
};
Expand All @@ -699,22 +722,13 @@ describe('the sweep re-runs a stuck in-process event', () => {
expect(rerun.deliveryNonce).toEqual(expect.any(String));
expect(rerun.deliveryNonce).not.toBe(born.deliveryNonce);
expect(rerun.attempts).toBe(born.attempts + 1);
expect(runAgent).toHaveBeenCalledTimes(2);
await waitForNativeCalls(runAgent, 2);

settles[0]('done'); // the dead turn's late settle
await new Promise((resolve) => setImmediate(resolve));
await new Promise((resolve) => setImmediate(resolve));
await settleNativeRun(settles[0], 'done'); // the dead turn's late settle
expect((await AgentEvent.findById(event._id).lean()).status).toBe('delivered');

settles[1]('done'); // the re-run's settle
let status;
for (let i = 0; i < 50 && status !== 'acked'; i += 1) {
// eslint-disable-next-line no-await-in-loop
await new Promise((resolve) => setTimeout(resolve, 10));
// eslint-disable-next-line no-await-in-loop
({ status } = await AgentEvent.findById(event._id).lean());
}
expect(status).toBe('acked');
await settleNativeRun(settles[1], 'done');
expect((await AgentEvent.findById(event._id).lean()).status).toBe('acked');
});

test('two sweeps racing (two replicas) re-run it once', async () => {
Expand All @@ -726,13 +740,8 @@ describe('the sweep re-runs a stuck in-process event', () => {
AgentEventService.redispatchStuckInProcess(threshold, 3),
]);

for (let i = 0; i < 50 && runAgent.mock.calls.length < 2; i += 1) {
// eslint-disable-next-line no-await-in-loop
await new Promise((resolve) => setTimeout(resolve, 10));
}

expect(results.reduce((sum, r) => sum + r.redispatched, 0)).toBe(1);
expect(runAgent).toHaveBeenCalledTimes(2); // the original turn plus one re-run
await waitForNativeCalls(runAgent, 2); // the original turn plus one re-run
});

test('acks instead of answering twice when the run had already succeeded', async () => {
Expand All @@ -741,7 +750,7 @@ describe('the sweep re-runs a stuck in-process event', () => {
await requeue(event._id);

expect((await AgentEvent.findById(event._id).lean()).status).toBe('acked');
expect(runAgent).toHaveBeenCalledTimes(1);
await waitForNativeCalls(runAgent, 1);
});

test('marks a run its process abandoned as interrupted, and leaves live and finished runs alone', async () => {
Expand All @@ -766,7 +775,7 @@ describe('the sweep re-runs a stuck in-process event', () => {
await requeue(event._id);

expect((await AgentEvent.findById(event._id).lean()).status).toBe('failed');
expect(runAgent).toHaveBeenCalledTimes(1);
await waitForNativeCalls(runAgent, 1);
});
});

Expand Down Expand Up @@ -797,7 +806,7 @@ describe('the sweep runs a pending event stranded by a move in-process', () => {
beforeEach(async () => {
await AgentInstallation.deleteMany({});
runAgent.mockReset();
runAgent.mockResolvedValue({ status: 'succeeded' });
runAgent.mockImplementation(() => new Promise(() => {}));
});
afterEach(async () => {
await AgentInstallation.deleteMany({});
Expand Down Expand Up @@ -825,10 +834,10 @@ describe('the sweep runs a pending event stranded by a move in-process', () => {
expect(ran.status).toBe('delivered');
expect(ran.deliveryNonce).toEqual(expect.any(String));
expect(ran.attempts).toBe(1);
expect(runAgent).toHaveBeenCalledTimes(1);
await waitForNativeCalls(runAgent, 1);

await sweep();
expect(runAgent).toHaveBeenCalledTimes(1);
await waitForNativeCalls(runAgent, 1);
});

test.each([
Expand Down Expand Up @@ -897,7 +906,7 @@ describe('the sweep runs a pending event stranded by a move in-process', () => {
const result = await sweep();

expect(result.rescuedPendingInProcess).toBe(1);
expect(runAgent).toHaveBeenCalledTimes(1);
await waitForNativeCalls(runAgent, 1);
expect((await AgentEvent.findById(event._id).lean()).status).not.toBe('pending');
expect(await AgentEvent.countDocuments({ agentName: 'stuck-agent', status: 'pending', attempts: 0 })).toBe(50);
});
Expand Down Expand Up @@ -935,7 +944,7 @@ describe('the sweep runs a pending event stranded by a move in-process', () => {
expect(runAgent).not.toHaveBeenCalled();
releaseDigest();
const [installation, trigger] = await runStarted;
expect(runAgent).toHaveBeenCalledTimes(1);
await waitForNativeCalls(runAgent, 1);
expect(String(installation.podId)).toBe(String(POD_ID));
expect(installation.config.hosted.migration.state).toBe('done');
expect(trigger.eventId).toBe(String(event._id));
Expand All @@ -955,13 +964,8 @@ describe('the sweep runs a pending event stranded by a move in-process', () => {
AgentEventService.rescueStrandedPendingInProcess(new Date(), 3),
]);

for (let i = 0; i < 50 && runAgent.mock.calls.length === 0; i += 1) {
// eslint-disable-next-line no-await-in-loop
await new Promise((resolve) => setTimeout(resolve, 10));
}

expect(counts[0].rescued + counts[1].rescued).toBe(1);
expect(runAgent).toHaveBeenCalledTimes(1);
await waitForNativeCalls(runAgent, 1);
});

test("a backlog of external agents' pending events does not crowd it out", async () => {
Expand All @@ -978,6 +982,7 @@ describe('the sweep runs a pending event stranded by a move in-process', () => {
await age(event._id, 2 * 60 * 1000);

expect((await AgentEventService.rescueStrandedPendingInProcess(new Date(), 3)).rescued).toBe(1);
await waitForNativeCalls(runAgent, 1);
expect((await AgentEvent.findById(event._id).lean()).status).not.toBe('pending');
});
});
Expand Down
Loading