E2E 결함 수정: Agent 절대 번들 URL, HAL 작업 재사용(실패 숨김 방지), 도메인 탐침 재개 키, 작업 엔진 동시 실행(부모/자식 교착 해소), 외부 Agent 자리표시자, JobError 4xx, 포털 HAL 이벤트·health 수신, link-portal 공유 비밀
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Fable 5.1
parent
7662d06009
commit
98f4182bbb
+14
-12
@@ -35,7 +35,7 @@ CREATE INDEX IF NOT EXISTS jobs_parent ON jobs(parent_id);` }];
|
||||
|
||||
/** run() 안에서 흐름을 바꾸는 신호(예외로 전달) */
|
||||
class Signal extends Error { constructor(type, payload) { super(type); this.signal = type; this.payload = payload || {}; } }
|
||||
export class JobError extends Error { constructor(code, message, { retryable = false, detail } = {}) { super(message || code); this.code = code; this.retryable = retryable; this.detail = detail; } }
|
||||
export class JobError extends Error { constructor(code, message, { retryable = false, detail, status = 400 } = {}) { super(message || code); this.code = code; this.retryable = retryable; this.detail = detail; this.status = status; /* API 로 새면 400 계열 — 500 으로 숨기지 않는다 */ } }
|
||||
|
||||
const BACKOFF_BASE_MS = 30 * 1000; const BACKOFF_MAX_MS = 60 * 60 * 1000; const STALE_LOCK_MS = 10 * 60 * 1000;
|
||||
|
||||
@@ -58,6 +58,15 @@ export function createJobEngine(store, opts = {}) {
|
||||
store.run(`UPDATE jobs SET ${sets.join(', ')} WHERE id=?`, ...vals, id);
|
||||
const j = engine.get(id); emit(j); return j;
|
||||
}
|
||||
/** 실행 가능한 작업 하나를 잠근다(큐 → 실행 중). 없으면 null */
|
||||
function claim() {
|
||||
if (running >= concurrency) return null;
|
||||
const now = nowIso();
|
||||
const cand = store.get('SELECT id FROM jobs WHERE status=? AND (next_run_at IS NULL OR next_run_at<=?) ORDER BY next_run_at ASC, created_at ASC LIMIT 1', S.QUEUED, now);
|
||||
if (!cand) return null;
|
||||
const got = store.run('UPDATE jobs SET status=?, locked_by=?, locked_at=?, started_at=COALESCE(started_at, ?), attempt=attempt+1, updated_at=? WHERE id=? AND status=?', S.RUNNING, workerId, now, now, now, cand.id, S.QUEUED).changes;
|
||||
return got ? cand.id : null;
|
||||
}
|
||||
const engine = {
|
||||
workerId,
|
||||
register(kind, def) { kinds.set(kind, { maxAttempts: 1, steps: [], ...def }); return engine; },
|
||||
@@ -122,19 +131,12 @@ export function createJobEngine(store, opts = {}) {
|
||||
return stale.length;
|
||||
},
|
||||
/** 실행 가능한 작업 하나를 집어 실행(테스트에서도 직접 부름). 돌아온 값: 실행한 작업 또는 null */
|
||||
async tick() {
|
||||
if (running >= concurrency) return null;
|
||||
const now = nowIso();
|
||||
const cand = store.get('SELECT id FROM jobs WHERE status=? AND (next_run_at IS NULL OR next_run_at<=?) ORDER BY next_run_at ASC, created_at ASC LIMIT 1', S.QUEUED, now);
|
||||
if (!cand) return null;
|
||||
const got = store.run('UPDATE jobs SET status=?, locked_by=?, locked_at=?, started_at=COALESCE(started_at, ?), attempt=attempt+1, updated_at=? WHERE id=? AND status=?', S.RUNNING, workerId, now, now, now, cand.id, S.QUEUED).changes;
|
||||
if (!got) return null;
|
||||
running++;
|
||||
try { return await execute(cand.id); } finally { running--; }
|
||||
},
|
||||
async tick() { const id = claim(); if (!id) return null; running++; try { return await execute(id); } finally { running--; } },
|
||||
/** 루프용: 작업을 집어 **기다리지 않고** 실행을 시작한다(느린 작업이 다른 작업을 막지 않게, 부모/자식 작업 교착 방지). 집은 작업 id 또는 null */
|
||||
dispatch() { const id = claim(); if (!id) return null; running++; execute(id).catch((e) => log('[jobs] execute error', safeError(e))).finally(() => { running--; }); return id; },
|
||||
/** 모든 큐 작업을 비울 때까지 tick(테스트용) */
|
||||
async drain(max = 100) { let n = 0; while (n < max) { const r = await engine.tick(); if (!r) break; n++; } return n; },
|
||||
start() { stopped = false; engine.recover(); const loop = async () => { if (stopped) return; try { let n = 0; while (n < concurrency && (await engine.tick())) n++; } catch (e) { log('[jobs] tick error', safeError(e)); } timer = setTimeout(loop, opts.pollMs || 1000); }; loop(); return engine; },
|
||||
start() { stopped = false; engine.recover(); const loop = () => { if (stopped) return; try { let n = 0; while (running < concurrency && n < concurrency && engine.dispatch()) n++; } catch (e) { log('[jobs] dispatch error', safeError(e)); } timer = setTimeout(loop, opts.pollMs || 1000); }; loop(); return engine; },
|
||||
stop() { stopped = true; if (timer) clearTimeout(timer); },
|
||||
};
|
||||
function appendLog(job, line) { const l = (job.log || []).slice(-200); l.push({ at: nowIso(), line: String(line).slice(0, 500) }); return l; }
|
||||
|
||||
Reference in New Issue
Block a user