mirror of
https://github.com/n8n-io/n8n.git
synced 2026-10-17 21:49:36 +02:00
perf: Back off idle Codespaces polling (no-changelog) (#37751)
This commit is contained in:
parent
f8bea81da6
commit
5adfcaa37a
|
|
@ -86,6 +86,10 @@ If the Slack API fails, the turn still completes through the n8n resume URL.
|
|||
The worker does not export its dequeue or Slack credentials to OpenCode.
|
||||
Interactive sessions remove all worker-only credentials before they start.
|
||||
|
||||
An idle worker starts with a 3-second poll interval. After each empty dequeue,
|
||||
it doubles the interval and limits it to 30 seconds. Work resets the interval
|
||||
to 3 seconds. Queued turns run without a delay between them.
|
||||
|
||||
The log is at `/tmp/agent-worker.log`. The tmux session is `agent-worker`. To
|
||||
watch it, run `tmux attach -t agent-worker`. The worker does not start if a
|
||||
required secret or `$GITHUB_USER` is missing.
|
||||
|
|
|
|||
|
|
@ -15,7 +15,8 @@ const GITHUB_USER = codespaceEnv('GITHUB_USER');
|
|||
const BOX_ID = codespaceEnv('CODESPACE_NAME');
|
||||
const ROOT = '/workspaces';
|
||||
|
||||
const POLL_INTERVAL_MS = 3000;
|
||||
const INITIAL_POLL_INTERVAL_MS = 3000;
|
||||
const MAX_POLL_INTERVAL_MS = 30_000;
|
||||
const SLACK_UPDATE_INTERVAL_MS = 1500;
|
||||
const SLACK_TEXT_LIMIT = 3900;
|
||||
const OPENCODE_CONFIG_CONTENT = JSON.stringify({
|
||||
|
|
@ -31,6 +32,36 @@ function posNum(name, fallback) {
|
|||
return fallback;
|
||||
}
|
||||
|
||||
function nextIdlePollInterval(interval) {
|
||||
return Math.min(interval * 2, MAX_POLL_INTERVAL_MS);
|
||||
}
|
||||
|
||||
export async function pollOnce(
|
||||
interval,
|
||||
{ dequeueTurn = dequeue, handleTurn = handle, wait = sleep, logError = console.error } = {},
|
||||
) {
|
||||
let turn;
|
||||
try {
|
||||
turn = await dequeueTurn();
|
||||
} catch (error) {
|
||||
logError(`poll error: ${error.message}`);
|
||||
await wait(interval);
|
||||
return nextIdlePollInterval(interval);
|
||||
}
|
||||
if (!turn) {
|
||||
await wait(interval);
|
||||
return nextIdlePollInterval(interval);
|
||||
}
|
||||
try {
|
||||
await handleTurn(turn);
|
||||
return INITIAL_POLL_INTERVAL_MS;
|
||||
} catch (error) {
|
||||
logError(`poll error: ${error.message}`);
|
||||
await wait(INITIAL_POLL_INTERVAL_MS);
|
||||
return nextIdlePollInterval(INITIAL_POLL_INTERVAL_MS);
|
||||
}
|
||||
}
|
||||
|
||||
// This limit expires before n8n's Wait node so that the user receives a specific error.
|
||||
const TURN_TIMEOUT_MS = posNum('TURN_TIMEOUT_MS', 25 * 60_000);
|
||||
function turnTimeoutMessage(timeout) {
|
||||
|
|
@ -328,6 +359,9 @@ export async function startSlackProgress(
|
|||
}
|
||||
|
||||
async function handle(turn) {
|
||||
console.log(
|
||||
`${new Date().toISOString()} turn ${turn.turnId} by ${turn.author ?? 'unknown'}: ${turn.sessionId ? 'resume' : 'new'}`,
|
||||
);
|
||||
let result;
|
||||
let activeSessionId = turn.sessionId ?? '';
|
||||
const progress = await startSlackProgress(turn);
|
||||
|
|
@ -388,22 +422,11 @@ async function main() {
|
|||
'SLACK_BOT_TOKEN is not set. Turns will complete without Slack progress updates.',
|
||||
);
|
||||
|
||||
console.log(`agent-worker polling as ${GITHUB_USER} every ${POLL_INTERVAL_MS}ms`);
|
||||
for (;;) {
|
||||
try {
|
||||
const turn = await dequeue();
|
||||
if (turn) {
|
||||
console.log(
|
||||
`${new Date().toISOString()} turn ${turn.turnId} by ${turn.author ?? 'unknown'}: ${turn.sessionId ? 'resume' : 'new'}`,
|
||||
);
|
||||
await handle(turn);
|
||||
continue;
|
||||
}
|
||||
} catch (error) {
|
||||
console.error(`poll error: ${error.message}`);
|
||||
}
|
||||
await sleep(POLL_INTERVAL_MS);
|
||||
}
|
||||
console.log(
|
||||
`agent-worker polling as ${GITHUB_USER} every ${INITIAL_POLL_INTERVAL_MS}-${MAX_POLL_INTERVAL_MS}ms`,
|
||||
);
|
||||
let pollInterval = INITIAL_POLL_INTERVAL_MS;
|
||||
for (;;) pollInterval = await pollOnce(pollInterval);
|
||||
}
|
||||
|
||||
if (process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href) await main();
|
||||
|
|
|
|||
|
|
@ -5,7 +5,7 @@ import { PassThrough } from 'node:stream';
|
|||
import { test } from 'node:test';
|
||||
import { setTimeout as sleep } from 'node:timers/promises';
|
||||
|
||||
import { openCodeEnvironment, runOpenCode, startSlackProgress } from './agent-worker.mjs';
|
||||
import { openCodeEnvironment, pollOnce, runOpenCode, startSlackProgress } from './agent-worker.mjs';
|
||||
|
||||
const turn = {
|
||||
turnId: 'turn-1',
|
||||
|
|
@ -60,6 +60,67 @@ function completedChild(lines) {
|
|||
return child;
|
||||
}
|
||||
|
||||
test('backs off idle polls and resets when work arrives', async () => {
|
||||
let interval = 3000;
|
||||
const waits = [];
|
||||
const turns = [null, null, null, null, null, { turnId: '1' }, { turnId: '2' }];
|
||||
const handled = [];
|
||||
for (let attempt = 0; attempt < 5; attempt++) {
|
||||
interval = await pollOnce(interval, {
|
||||
dequeueTurn: async () => turns.shift(),
|
||||
handleTurn: async (turn) => handled.push(turn.turnId),
|
||||
wait: async (delay) => waits.push(delay),
|
||||
});
|
||||
}
|
||||
assert.deepEqual(waits, [3000, 6000, 12_000, 24_000, 30_000]);
|
||||
assert.equal(interval, 30_000);
|
||||
|
||||
interval = await pollOnce(interval, {
|
||||
dequeueTurn: async () => turns.shift(),
|
||||
handleTurn: async (turn) => handled.push(turn.turnId),
|
||||
wait: async (delay) => waits.push(delay),
|
||||
});
|
||||
interval = await pollOnce(interval, {
|
||||
dequeueTurn: async () => turns.shift(),
|
||||
handleTurn: async (turn) => handled.push(turn.turnId),
|
||||
wait: async (delay) => waits.push(delay),
|
||||
});
|
||||
assert.equal(interval, 3000);
|
||||
assert.deepEqual(handled, ['1', '2']);
|
||||
assert.equal(waits.length, 5);
|
||||
});
|
||||
|
||||
test('backs off after a dequeue error', async () => {
|
||||
const waits = [];
|
||||
const errors = [];
|
||||
const interval = await pollOnce(3000, {
|
||||
dequeueTurn: async () => {
|
||||
throw new Error('unavailable');
|
||||
},
|
||||
wait: async (delay) => waits.push(delay),
|
||||
logError: (error) => errors.push(error),
|
||||
});
|
||||
assert.equal(interval, 6000);
|
||||
assert.deepEqual(waits, [3000]);
|
||||
assert.deepEqual(errors, ['poll error: unavailable']);
|
||||
});
|
||||
|
||||
test('resets the interval before handling work', async () => {
|
||||
const waits = [];
|
||||
const errors = [];
|
||||
const interval = await pollOnce(30_000, {
|
||||
dequeueTurn: async () => ({ turnId: '1' }),
|
||||
handleTurn: async () => {
|
||||
throw new Error('failed');
|
||||
},
|
||||
wait: async (delay) => waits.push(delay),
|
||||
logError: (error) => errors.push(error),
|
||||
});
|
||||
assert.equal(interval, 6000);
|
||||
assert.deepEqual(waits, [3000]);
|
||||
assert.deepEqual(errors, ['poll error: failed']);
|
||||
});
|
||||
|
||||
test('streams tool progress but excludes reasoning', async () => {
|
||||
const slack = slackRecorder();
|
||||
const progress = await startSlackProgress(turn, {
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user