diff --git a/lib/local-ledger.js b/lib/local-ledger.js index 9c2d211..5b21569 100644 --- a/lib/local-ledger.js +++ b/lib/local-ledger.js @@ -2,6 +2,7 @@ // Copyright (c) 2026 EvoMap import { execFile } from 'node:child_process'; +import { stat } from 'node:fs/promises'; import { createRequire } from 'node:module'; import { join } from 'node:path'; import { pathToFileURL } from 'node:url'; @@ -76,6 +77,13 @@ export async function recordReuseLocally({ assetId, outcome, sessionId }, { engi const ingestor = await loadIngestor(enginePrefix); if (!ingestor) return false; + // Check the target before ingest, where a failure cannot follow an append. + try { + if (!(await stat(ingestor.path)).isFile()) return false; + } catch (error) { + if (error?.code !== 'ENOENT') return false; + } + const cycleId = cycleIdOf(sessionId); const success = outcome === 'success'; try { @@ -87,8 +95,10 @@ export async function recordReuseLocally({ assetId, outcome, sessionId }, { engi }, payload: success ? { assetId, cycleId } : { assetId, cycleId, outcome }, }); - return true; - } catch { - return false; + } catch (error) { + // Core acquires the lock before appending. Other errors can follow a write. + if (error?.code === 'LOCK_TIMEOUT') return false; + throw error; } + return true; } diff --git a/lib/report-outbox.js b/lib/report-outbox.js index fb47a0f..f09680f 100644 --- a/lib/report-outbox.js +++ b/lib/report-outbox.js @@ -34,6 +34,10 @@ function settlementOf({ verdict, reason, status }, now) { export function withAttempt(outbox, key, attempt, now) { const entry = outbox[key]; if (!entry) return outbox; - const local = attempt.isLocal ? { localAt: now } : {}; + const local = { + ...(attempt.isLocal === true && { localAt: now }), + ...(attempt.isLocal === false && { localFailures: (entry.localFailures ?? 0) + 1 }), + ...(attempt.localUncertain && { localUncertainAt: now }), + }; return { ...outbox, [key]: { ...entry, attempts: entry.attempts + 1, ...local, ...settlementOf(attempt, now) } }; } diff --git a/lib/reuse.js b/lib/reuse.js index 57927ea..6bf4645 100644 --- a/lib/reuse.js +++ b/lib/reuse.js @@ -89,6 +89,7 @@ export function retentionReportOf(reuse, retentionClass) { }); } +const MAX_LOCAL_FAILURES = 3; const REJECTED_BY_HUB = /^hub (?:400|422)\b/; function isRejection(data) { @@ -113,26 +114,35 @@ async function postToHub(proxyFetch, report) { } } -function isLocalRecordDue({ report, localAt }, delivery) { - return !localAt && delivery.verdict !== REJECTED && report.evidence.tier !== 'none'; +function isLocalRecordDue({ report, localAt, localUncertainAt, localFailures = 0, rejectedAt }) { + return !localAt && !localUncertainAt && !rejectedAt && localFailures < MAX_LOCAL_FAILURES && report.evidence.tier !== 'none'; } async function recordedLocally(recordLocally, { report, sessionId }) { - return recordLocally({ assetId: report.asset_id, outcome: report.outcome, sessionId }).catch(() => false); + try { + return { isLocal: await recordLocally({ assetId: report.asset_id, outcome: report.outcome, sessionId }) }; + } catch { + return { localUncertain: true }; + } } async function attemptOf(ledgers, entry) { const delivery = await postToHub(ledgers.proxyFetch, entry.report); - const isLocal = isLocalRecordDue(entry, delivery) && (await recordedLocally(ledgers.recordLocally, entry)); - if (delivery.verdict === REJECTED) await ledgers.onRejected({ taskId: entry.report.task_id, assetId: entry.report.asset_id, ...delivery }); - return { ...delivery, isLocal }; + if (delivery.verdict === REJECTED) { + await ledgers.onRejected({ taskId: entry.report.task_id, assetId: entry.report.asset_id, ...delivery }); + } else if (isLocalRecordDue(entry)) { + return { ...delivery, ...await recordedLocally(ledgers.recordLocally, entry) }; + } + return delivery; } export async function deliverPending(ledgers, outbox, now, afterAttempt = () => {}) { let delivered = outbox; for (const [key, entry] of Object.entries(outbox)) { - if (isSettled(entry)) continue; - delivered = withAttempt(delivered, key, await attemptOf(ledgers, entry), now); + const localOnly = entry.sentAt && isLocalRecordDue(entry); + if (isSettled(entry) && !localOnly) continue; + const attempt = localOnly ? await recordedLocally(ledgers.recordLocally, entry) : await attemptOf(ledgers, entry); + delivered = withAttempt(delivered, key, attempt, now); afterAttempt(); } return delivered; diff --git a/test/reuse.test.js b/test/reuse.test.js index fd88137..0da7b56 100644 --- a/test/reuse.test.js +++ b/test/reuse.test.js @@ -1,6 +1,12 @@ import assert from 'node:assert/strict'; +import { mkdirSync, mkdtempSync, rmdirSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; import { test } from 'node:test'; +import { events } from '@evomap/evolver-core'; + +import { recordReuseLocally } from '../lib/local-ledger.js'; import { withQueued } from '../lib/report-outbox.js'; import { correctionReportOf, @@ -139,6 +145,119 @@ test('evidence-bearing reports are recorded locally once; tier none never is', a assert.equal(sinks.posts.length, 4); }); +test('a sent report retries a pre-write local failure without posting to the Hub again', async () => { + const down = ledgers({ local: false }); + const queued = withQueued({}, [{ report: immediate({ checks: [{ command: 'npm test', failed: false }] }), sessionId: 'claude-code:s1' }], NOW); + const sent = await deliverPending(down, queued, NOW); + assert.equal(down.posts.length, 1); + assert.equal(Object.values(sent)[0].sentAt, NOW); + assert.equal(Object.values(sent)[0].localAt, undefined); + const up = ledgers(); + const recorded = await deliverPending(up, sent, NOW + 1); + assert.equal(up.posts.length, 0); + assert.deepEqual(up.locals, [{ assetId: 'sha256:a', outcome: 'success', sessionId: 'claude-code:s1' }]); + assert.equal(Object.values(recorded)[0].sentAt, NOW, 'the original Hub delivery time stays intact'); + assert.equal(Object.values(recorded)[0].localAt, NOW + 1); + await deliverPending(up, recorded, NOW + 2); + assert.equal(up.locals.length, 1); + assert.equal(up.posts.length, 0); +}); + +function useTemporaryLedger(t) { + const home = mkdtempSync(join(tmpdir(), 'evolver-ledger-retry-')); + const previous = process.env.EVOLVER_HOME; + process.env.EVOLVER_HOME = home; + t.after(() => { + if (previous === undefined) delete process.env.EVOLVER_HOME; + else process.env.EVOLVER_HOME = previous; + rmSync(home, { recursive: true, force: true }); + }); +} + +test('a ledger lock timeout retries the local write after Hub delivery', async (t) => { + useTemporaryLedger(t); + let attempts = 0; + t.mock.method(events.Ingestor.prototype, 'ingest', async () => { + attempts += 1; + if (attempts === 1) { + throw Object.assign(new Error('ledger lock timeout'), { code: 'LOCK_TIMEOUT' }); + } + }); + const sinks = { ...ledgers(), recordLocally: recordReuseLocally }; + const report = immediate({ checks: [{ command: 'npm test', failed: false }] }); + const queued = withQueued({}, [{ report, sessionId: 'claude-code:s1' }], NOW); + const sent = await deliverPending(sinks, queued, NOW); + const [pending] = Object.values(sent); + assert.equal(pending.localFailures, 1); + assert.equal(pending.localUncertainAt, undefined); + assert.equal(pending.sentAt, NOW); + + const recorded = await deliverPending(sinks, sent, NOW + 1); + assert.equal(Object.values(recorded)[0].localAt, NOW + 1); + assert.equal(attempts, 2); + assert.equal(sinks.posts.length, 1); +}); + +test('a ledger path blocked by a directory retries locally after the path is repaired', async (t) => { + useTemporaryLedger(t); + const path = events.rootEventsPath(); + mkdirSync(path, { recursive: true }); + + const sinks = { ...ledgers(), recordLocally: recordReuseLocally }; + const report = immediate({ checks: [{ command: 'npm test', failed: false }] }); + const queued = withQueued({}, [{ report, sessionId: 'claude-code:s1' }], NOW); + const sent = await deliverPending(sinks, queued, NOW); + const [pending] = Object.values(sent); + assert.equal(pending.sentAt, NOW); + assert.equal(pending.localFailures, 1); + assert.equal(pending.localUncertainAt, undefined); + + rmdirSync(path); + const recorded = await deliverPending(sinks, sent, NOW + 1); + assert.equal(Object.values(recorded)[0].localAt, NOW + 1); + await deliverPending(sinks, recorded, NOW + 2); + assert.equal(sinks.posts.length, 1); + const entries = new events.Ingestor({ path }).readAllStrict(); + assert.deepEqual(entries.map(({ type, payload }) => ({ type, payload })), [ + { type: 'value.reuse_hit', payload: { assetId: ASSET.assetId, cycleId: 'claude-code:s1' } }, + ]); +}); + +test('an error after appending never retries either ledger', async (t) => { + useTemporaryLedger(t); + const ingest = events.Ingestor.prototype.ingest; + const attempted = t.mock.method(events.Ingestor.prototype, 'ingest', async function (event) { + await ingest.call(this, event); + throw Object.assign(new Error('ledger failed after appending'), { code: 'EISDIR' }); + }); + const sinks = { ...ledgers(), recordLocally: recordReuseLocally }; + const report = immediate({ checks: [{ command: 'npm test', failed: false }] }); + let outbox = withQueued({}, [{ report, sessionId: 'claude-code:s1' }], NOW); + for (let i = 0; i < 5; i += 1) outbox = await deliverPending(sinks, outbox, NOW + i); + + assert.equal(sinks.posts.length, 1); + assert.equal(attempted.mock.callCount(), 1); + assert.equal(new events.Ingestor({ path: events.rootEventsPath() }).readAllStrict().length, 1); + assert.equal(Object.values(outbox)[0].sentAt, NOW); + assert.equal(Object.values(outbox)[0].localUncertainAt, NOW); + assert.equal(Object.values(outbox)[0].localAt, undefined); +}); + +test('exhausting local attempts does not stop Hub delivery retries', async () => { + const down = ledgers({ hub: new Error('Hub offline'), local: false }); + const report = immediate({ checks: [{ command: 'npm test', failed: false }] }); + let outbox = withQueued({}, [{ report, sessionId: 'claude-code:s1' }], NOW); + for (let i = 0; i < 4; i += 1) outbox = await deliverPending(down, outbox, NOW + i); + + assert.equal(down.posts.length, 4); + assert.equal(down.locals.length, 3); + const up = ledgers(); + const sent = await deliverPending(up, outbox, NOW + 4); + assert.equal(up.posts.length, 1); + assert.equal(up.locals.length, 0); + assert.equal(Object.values(sent)[0].sentAt, NOW + 4); +}); + test('a report the Hub already holds counts as delivered and is not retried', async () => { const sinks = ledgers({ hub: { ok: true, data: { recorded: false, reason: 'duplicate_report' } } }); const outbox = await deliverPending(sinks, withQueued({}, [{ report: immediate(), sessionId: 'claude-code:s1' }], NOW), NOW);