From 22e6caf06f95899074c8ab7865401100f63d941d Mon Sep 17 00:00:00 2001 From: Mara Nikola Kiefer <8320933+mnkiefer@users.noreply.github.com> Date: Fri, 23 Oct 2026 21:36:29 +0200 Subject: [PATCH] Reclaim resolved failure log excerpts from canonical caches Compact diagnostic-only failureLog payloads after same-scope recovery while retaining Run summaries and metric/comparison evidence. Protect pruning against stale imports and refresh subscribers after excerpt-only maintenance. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: bf47d592-6917-4408-96ae-d8486602439a --- activity/cli/ingestion.mjs | 24 +- dashboard/site/src/data-worker.js | 2 +- dashboard/site/src/data/ingest/coordinator.js | 8 +- dashboard/site/src/data/storage/indexeddb.js | 72 +--- dashboard/site/src/data/storage/retention.js | 60 +++- dashboard/site/src/data/storage/shards.js | 135 +++++++- .../site/test/e2e/indexeddb-stress.spec.js | 53 +++ .../unit/data-failure-log-retention.test.js | 316 ++++++++++++++++++ .../test/unit/data-worker-retention.test.js | 44 ++- docs/dashboard-data-ingestion.md | 15 + specs/dashboard-data.md | 31 ++ 11 files changed, 661 insertions(+), 99 deletions(-) create mode 100644 dashboard/site/test/unit/data-failure-log-retention.test.js diff --git a/activity/cli/ingestion.mjs b/activity/cli/ingestion.mjs index 07109cd3c2..5a9d96b82d 100644 --- a/activity/cli/ingestion.mjs +++ b/activity/cli/ingestion.mjs @@ -82,12 +82,10 @@ export async function ingestJsonlShardDirectory(indexedDB, shardDirectory, optio debug('ingested shard %s: committedRecords=%d', name, result.committedRecords ?? 0); shards.push({ shard: name, skipped: Boolean(result.skipped), committedRecords: result.committedRecords ?? 0 }); } - if (!await isAuditCurationCurrent(indexedDB)) { - const maintenance = await finalizeNormalizedJsonlIngestion(indexedDB, { - ...options, repairAuditCuration: true - }); - updated ||= maintenance.deletedRecords > 0; - } + const maintenance = await finalizeNormalizedJsonlIngestion(indexedDB, { + ...options, repairAuditCuration: !await isAuditCurationCurrent(indexedDB) + }); + updated ||= maintenance.deletedRecords > 0 || maintenance.prunedFailureLogs > 0; return { ...totals, updated, committedRecords, shards }; } @@ -146,14 +144,12 @@ export async function ingestNormalizedShardDirectories(indexedDB, directories, o ); } const auditCurationCurrent = await isAuditCurationCurrent(indexedDB); - if (updated || !auditCurationCurrent) { - const maintenanceStartedAt = Date.now(); - const maintenance = await finalizeNormalizedJsonlIngestion(indexedDB, { - ...options, repairAuditCuration: !updated && !auditCurationCurrent - }); - updated ||= maintenance.deletedRecords > 0; - debug('applied deferred canonical maintenance durationMs=%d', Date.now() - maintenanceStartedAt); - } + const maintenanceStartedAt = Date.now(); + const maintenance = await finalizeNormalizedJsonlIngestion(indexedDB, { + ...options, repairAuditCuration: !updated && !auditCurationCurrent + }); + updated ||= maintenance.deletedRecords > 0 || maintenance.prunedFailureLogs > 0; + debug('applied deferred canonical maintenance durationMs=%d', Date.now() - maintenanceStartedAt); return { updated, committedRecords, shards }; } diff --git a/dashboard/site/src/data-worker.js b/dashboard/site/src/data-worker.js index c454058737..594c9e5fe2 100644 --- a/dashboard/site/src/data-worker.js +++ b/dashboard/site/src/data-worker.js @@ -1287,7 +1287,7 @@ export function processDataRequest(request, signal) { onLockWait: () => progress.log(INGESTION_LOCK_WAIT_MESSAGE), signal }); - changed ||= maintenance.deletedRecords > 0; + changed ||= maintenance.deletedRecords > 0 || maintenance.prunedFailureLogs > 0; progress.reportImportProgress(shardCount + 2 * preparationStepWeight, importSteps, 'queries'); } else { progress.log('Normalizing dashboard source data.'); diff --git a/dashboard/site/src/data/ingest/coordinator.js b/dashboard/site/src/data/ingest/coordinator.js index 177d99f674..016ee7244b 100644 --- a/dashboard/site/src/data/ingest/coordinator.js +++ b/dashboard/site/src/data/ingest/coordinator.js @@ -213,7 +213,7 @@ export async function isAuditCurationCurrent(indexedDB) { async function skippedIngestion(indexedDB, options) { const cleanup = await maintainNormalizedJsonlDatabase(indexedDB, options); return { - updated: (cleanup?.deletedRecords ?? 0) > 0, + updated: (cleanup?.deletedRecords ?? 0) > 0 || (cleanup?.prunedFailureLogs ?? 0) > 0, skipped: true, committedBatches: 0, committedRecords: 0 @@ -539,9 +539,9 @@ async function maintainNormalizedJsonlDatabase(indexedDB, options) { * Runs the retention and size maintenance pass that `ingestNormalizedJsonl` * skips when called with `deferMaintenance`. * - * Maintenance uses persisted shard accounting and indexes to visit only expired - * records and dirty audit shards. Multi-shard callers defer it until inventory - * and all activity shards have committed. + * Maintenance uses shard accounting and indexes for expiry and audit curation, + * and bounded Run reads for failure-log recovery. Multi-shard callers defer it + * until inventory and all activity shards have committed. * CLI callers can explicitly repair legacy audits when the curation receipt is missing. * * @param {IDBFactory} indexedDB diff --git a/dashboard/site/src/data/storage/indexeddb.js b/dashboard/site/src/data/storage/indexeddb.js index 8f164e7ad9..68f3d87627 100644 --- a/dashboard/site/src/data/storage/indexeddb.js +++ b/dashboard/site/src/data/storage/indexeddb.js @@ -179,7 +179,6 @@ const INGESTION_LOCK_ACQUIRE_TIMEOUT_MS = INGESTION_LOCK_LEASE_MS + 30_000; const INGESTION_LOCK_RETRY_DELAY_MS = 25; const INGESTION_LOCK_WAITING_NOTICE_DELAY_MS = 500; const MAX_QUERY_INDEX_LOOKUPS = 32; -const RECORD_OVERHEAD_BYTES = 512; const RETENTION_TIMESTAMPS = new Set([ 'runs', 'domains', @@ -668,14 +667,9 @@ export async function upsertCanonicalBatch(indexedDB, batch, options = {}) { } } - /** @param {Record} record */ -function estimatedRecordBytes(record) { - return new TextEncoder().encode(JSON.stringify(record)).byteLength + RECORD_OVERHEAD_BYTES; -} - /** - * Applies retention and size limits using indexed, bounded shard cleanup. - * Full scans are reserved for explicitly requested relationship or legacy audit repair. + * Applies retention and size limits using indexed shard cleanup and bounded Run pages. + * Full evidence scans are reserved for explicit relationship or legacy audit repair. * * @param {IDBFactory} indexedDB * @param {{ now?: number, retentionWindowMs?: number, retentionWindowMsByStore?: Record, maxDatabaseBytes: number, usageBytes?: number | null, reconcileRelationships?: boolean, repairAuditCuration?: boolean, preserveEntityIds?: { repositories?: string[], workflows?: string[] }, onMaintenanceProgress?: (completed: number, total: number) => void, signal?: AbortSignal }} options @@ -704,22 +698,12 @@ async function reconcileCanonicalDatabase(indexedDB, options) { const defaultWindow = Number.isFinite(options.retentionWindowMs) ? Math.max(0, Number(options.retentionWindowMs)) : 30 * 24 * 60 * 60 * 1000; - const targetBytes = Math.floor(Math.max(0, options.maxDatabaseBytes) * 0.75); - let estimatedBytes = 0; let deletedRecords = 0; - let retainedRecords = 0; let prunedAudits = 0; /** @type {Map>} */ const auditRunFacts = new Map(); - /** @type {{ id: string, timestamp: number, bytes: number }[]} */ - const runs = []; /** @type {string[]} */ const expiredRunIds = []; - /** @type {Map} */ - const linkedBytesByRun = new Map(); - /** @type {Map} */ - const linkedRecordsByRun = new Map(); - const retainedRunIds = new Set(); const campaignIds = new Set(); const repositoryIds = new Set(); /** @type {Map} */ @@ -836,21 +820,6 @@ async function reconcileCanonicalDatabase(indexedDB, options) { } return; } - const bytes = estimatedRecordBytes(record); - estimatedBytes += bytes; - retainedRecords += 1; - if (storeName === 'runs') { - retainedRunIds.add(id); - runs.push({ - id: String(record.id), - timestamp: timestamp ?? Number.NEGATIVE_INFINITY, - bytes - }); - } else if (RUN_LINKED_STORES.includes(/** @type {typeof RUN_LINKED_STORES[number]} */ (storeName))) { - const runId = String(record.runId); - linkedBytesByRun.set(runId, (linkedBytesByRun.get(runId) ?? 0) + bytes); - linkedRecordsByRun.set(runId, (linkedRecordsByRun.get(runId) ?? 0) + 1); - } }; if (typeof store.openCursor !== 'function') { for (const record of await requestResult(store.getAll())) { @@ -914,7 +883,6 @@ async function reconcileCanonicalDatabase(indexedDB, options) { if (/** @type {Set} */ (retainedIds).has(String(id))) return; remove(); deletedRecords += 1; - retainedRecords -= 1; }; if (typeof store.openCursor !== 'function') { for (const record of await requestResult(store.getAll())) { @@ -939,31 +907,8 @@ async function reconcileCanonicalDatabase(indexedDB, options) { } } - for (const runId of expiredRunIds) { - estimatedBytes -= linkedBytesByRun.get(runId) ?? 0; - retainedRecords -= linkedRecordsByRun.get(runId) ?? 0; - } - const usageTarget = Number.isFinite(options.usageBytes) - && Number(options.usageBytes) > options.maxDatabaseBytes - ? Math.floor(estimatedBytes * (options.maxDatabaseBytes / Number(options.usageBytes)) * 0.9) - : targetBytes; - const effectiveTarget = Math.min(targetBytes, usageTarget); - const evictedRunIds = [...expiredRunIds]; - if (estimatedBytes > effectiveTarget) { - runs.sort((left, right) => left.timestamp - right.timestamp || left.id.localeCompare(right.id)); - for (const run of runs) { - if (estimatedBytes <= effectiveTarget) break; - evictedRunIds.push(run.id); - estimatedBytes -= run.bytes + (linkedBytesByRun.get(run.id) ?? 0); - retainedRecords -= 1 + (linkedRecordsByRun.get(run.id) ?? 0); - } - } - for (const runId of expiredRunIds) { - if (retainedRunIds.has(runId)) retainedRecords -= 1; - } - - for (let offset = 0; offset < evictedRunIds.length; offset += DEFAULT_WRITE_BATCH_SIZE) { - const ids = evictedRunIds.slice(offset, offset + DEFAULT_WRITE_BATCH_SIZE); + for (let offset = 0; offset < expiredRunIds.length; offset += DEFAULT_WRITE_BATCH_SIZE) { + const ids = expiredRunIds.slice(offset, offset + DEFAULT_WRITE_BATCH_SIZE); const transaction = readwriteTransaction(database, ['runs', ...RUN_LINKED_STORES]); const done = transactionDone(transaction); const runStore = transaction.objectStore('runs'); @@ -1009,8 +954,15 @@ async function reconcileCanonicalDatabase(indexedDB, options) { }); await receiptDone; await rebuildStorageShards(database, ENTITY_STORES); + const maintenance = await maintainStorageShards(database, { + ...options, entityStores: ENTITY_STORES, runLinkedStores: RUN_LINKED_STORES + }); debug('curated canonical audits', { prunedAudits, version: AUDIT_CURATION_VERSION }); - return { deletedRecords, estimatedBytes, retainedRecords, prunedAudits }; + return { + ...maintenance, + deletedRecords: deletedRecords + maintenance.deletedRecords, + prunedAudits: prunedAudits + maintenance.prunedAudits + }; } finally { database.close(); } diff --git a/dashboard/site/src/data/storage/retention.js b/dashboard/site/src/data/storage/retention.js index 4044ae3549..e92892c836 100644 --- a/dashboard/site/src/data/storage/retention.js +++ b/dashboard/site/src/data/storage/retention.js @@ -1,5 +1,5 @@ import { orderRunRecords } from '../normalize/index.js'; -import { EVIDENCE_DEFINITION_STORES, mergeEvidenceDefinition } from '../model/schema.js'; +import { EVIDENCE_DEFINITION_STORES, mergeEvidenceDefinition, timestampMilliseconds } from '../model/schema.js'; import { createDebug } from '../../debug.js'; import { mergeOperationalValueEvidence } from '../model/operational-value-evidence.js'; @@ -210,6 +210,64 @@ export function recordTimestamp(storeName, record) { return null; } +const FAILURE_CONCLUSIONS = new Set(['failure', 'startup-failure', 'startup_failure', 'stale', 'timed-out', 'timed_out']); + +/** @typedef {{ scope: string, timestamp: number, githubRunId: string, attempt: number }} FailureLogBoundary */ + +/** + * Uses the runtime-health run order, with stricter scope matching for destructive + * retention. Unknown worker targets cannot establish a recovery boundary. + * @param {Record} run + * @param {Record | undefined} workflow + * @returns {FailureLogBoundary | null} + */ +export function failureLogBoundary(run, workflow) { + if (!workflow || workflow.id !== run.workflowId || workflow.repositoryId !== run.repositoryId + || typeof run.repositoryId !== 'string' || typeof run.workflowId !== 'string' + || typeof run.event !== 'string' || !run.event || run.event === 'unknown' + || !Number.isSafeInteger(run.attempt) || Number(run.attempt) < 1) return null; + const target = run.targetRepositoryId ?? run.targetRepository; + const knownTarget = typeof target === 'string' && target.length > 0 && target !== 'unknown'; + const branch = run.branch ?? run.headBranch; + if (typeof branch !== 'string' || !branch || branch === 'unknown') return null; + if ((workflow.role === 'worker' || workflow.role == null || workflow.role === '' || workflow.role === 'unknown') + && !knownTarget) return null; + if ((workflow.role === 'worker' || knownTarget) + && (typeof run.rolloutMode !== 'string' || !run.rolloutMode || run.rolloutMode === 'unknown')) return null; + const timestamp = ['startedAt', 'createdAt', 'updatedAt'] + .map((field) => timestampMilliseconds(run[field])).find((value) => value !== null); + const githubRunId = String(run.githubRunId ?? ''); + if (timestamp === undefined || !/^[1-9][0-9]*$/.test(githubRunId) + || (typeof run.githubRunId === 'number' && !Number.isSafeInteger(run.githubRunId))) return null; + return { + scope: JSON.stringify([ + run.repositoryId, run.workflowId, workflow.campaignId ?? null, workflow.role ?? null, + run.targetRepositoryId ?? null, run.targetRepository ?? null, + run.event, branch, run.rolloutMode ?? null + ]), + timestamp, githubRunId, attempt: Number(run.attempt) + }; +} + +/** @param {FailureLogBoundary} left @param {FailureLogBoundary} right */ +export function compareFailureLogBoundaries(left, right) { + return left.timestamp - right.timestamp + || left.githubRunId.localeCompare(right.githubRunId, undefined, { numeric: true }) + || left.attempt - right.attempt; +} + +/** @param {Record} run */ +export function isFailureLogRecovery(run) { + return run.status === 'completed' && run.conclusion === 'success' + && (run.terminalOutcome == null || run.terminalOutcome === '') + && (run.failureKind == null || run.failureKind === ''); +} + +/** @param {Record} run */ +export function hasFailedConclusion(run) { + return run.status === 'completed' && FAILURE_CONCLUSIONS.has(String(run.conclusion)); +} + /** * @param {import('../model/schema.js').CanonicalBatch} batch * @returns {number | null} diff --git a/dashboard/site/src/data/storage/shards.js b/dashboard/site/src/data/storage/shards.js index b0ee5bbed3..164dcc94aa 100644 --- a/dashboard/site/src/data/storage/shards.js +++ b/dashboard/site/src/data/storage/shards.js @@ -1,10 +1,13 @@ -import { EVIDENCE_DEFINITION_STORES, mergeEvidenceDefinition } from '../model/schema.js'; +import { EVIDENCE_DEFINITION_STORES, mergeEvidenceDefinition, timestampMilliseconds } from '../model/schema.js'; import { mergeCanonicalRun } from '../model/fields.js'; import { mergeOperationalValueEvidence } from '../model/operational-value-evidence.js'; import { auditCurationRunFacts, discardAudit, AUDIT_CURATION_TRANSACTION_ID, AUDIT_CURATION_VERSION } from '../model/audit-curation.js'; -import { recordTimestamp, RETENTION_TIMESTAMPS, retentionWindowForStore } from './retention.js'; +import { + compareFailureLogBoundaries, failureLogBoundary, hasFailedConclusion, isFailureLogRecovery, + recordTimestamp, RETENTION_TIMESTAMPS, retentionWindowForStore +} from './retention.js'; import { createDebug } from '../../debug.js'; const debug = createDebug('data:retention'); @@ -13,8 +16,8 @@ const TOTALS_ID = 'totals'; const BATCH_SIZE = 1000; const DAY_MS = 86400000; -/** @typedef {{ shard: string, timestamp: number, bytes: number }} StorageMetadata */ -/** @typedef {{ id: string, storeName?: string, day?: number, bytes: number, count: number, dirty?: boolean, runId?: string, auditId?: string }} Shard */ +/** @typedef {{ shard: string, timestamp: number, bytes: number, failureLogPruned?: boolean }} StorageMetadata */ +/** @typedef {{ id: string, storeName?: string, day?: number, bytes: number, count: number, dirty?: boolean, runId?: string, auditId?: string, failureLogs?: boolean }} Shard */ /** @template T @param {IDBRequest} request @returns {Promise} */ function result(request) { @@ -41,8 +44,8 @@ export function withoutStorageMetadata(record) { return canonical; } -/** @param {string} storeName @param {Record} record */ -export function withStorageMetadata(storeName, record) { +/** @param {string} storeName @param {Record} record @param {boolean} [failureLogPruned] */ +export function withStorageMetadata(storeName, record, failureLogPruned = false) { const stored = { ...record }; delete stored._storage; const timestamp = recordTimestamp(storeName, stored) ?? -Number.MAX_SAFE_INTEGER; @@ -53,11 +56,27 @@ export function withStorageMetadata(storeName, record) { _storage: { shard: `${storeName}:${day}`, timestamp, - bytes: new TextEncoder().encode(JSON.stringify(stored)).byteLength + 512 + bytes: new TextEncoder().encode(JSON.stringify(stored)).byteLength + 512, + ...(failureLogPruned ? { failureLogPruned: true } : {}) } }; } +/** @param {Record | undefined} existing @param {Record} incoming */ +function mergeStoredRun(existing, incoming) { + if (existing && Number.isSafeInteger(existing.attempt) && Number(existing.attempt) > 0) { + if (!Number.isSafeInteger(incoming.attempt) || Number(incoming.attempt) < Number(existing.attempt)) return existing; + const before = timestampMilliseconds(existing.observedAt); + const after = timestampMilliseconds(incoming.observedAt); + if (existing.attempt === incoming.attempt && before !== null && after !== null && before > after) return existing; + } + const merged = mergeCanonicalRun(existing ? withoutStorageMetadata(existing) : undefined, withoutStorageMetadata(incoming)); + const pruned = metadata(incoming).failureLogPruned === true + || (existing?.attempt === merged.attempt && existing && metadata(existing).failureLogPruned === true); + if (pruned) delete merged.failureLog; + return withStorageMetadata('runs', merged, Boolean(pruned)); +} + /** @param {Record} record @returns {StorageMetadata} */ function metadata(record) { const value = /** @type {StorageMetadata | undefined} */ (record._storage); @@ -83,9 +102,7 @@ export async function writeAccountedRecords(transaction, storeName, records, rem for (const record of records) { const previous = unique.get(String(record.id)); unique.set(String(record.id), previous && storeName === 'runs' - ? withStorageMetadata(storeName, mergeCanonicalRun( - withoutStorageMetadata(previous), withoutStorageMetadata(record) - )) + ? mergeStoredRun(previous, record) : previous && EVIDENCE_DEFINITION_STORES.has(storeName) ? withStorageMetadata(storeName, mergeEvidenceDefinition( withoutStorageMetadata(previous), withoutStorageMetadata(record) @@ -104,6 +121,7 @@ export async function writeAccountedRecords(transaction, storeName, records, rem /** @type {string[]} */ const dirtyRuns = []; const dirtyAudits = new Set(); + let failureLogs = false; const referencesAudits = ['experimentAssignments', 'graderObservations', 'evalObservations'].includes(storeName); /** @param {Record} record @param {number} sign */ const account = (record, sign) => { @@ -128,9 +146,7 @@ export async function writeAccountedRecords(transaction, storeName, records, rem dirtyAudits.add(existing.auditId); } if (storeName === 'runs' && existing) { - const incomingRun = withoutStorageMetadata(record); - const merged = mergeCanonicalRun(withoutStorageMetadata(existing), incomingRun); - if (merged !== incomingRun) record = withStorageMetadata(storeName, merged); + record = mergeStoredRun(existing, record); } else if (storeName === 'operationalValueEvidence') { record = withStorageMetadata(storeName, mergeOperationalValueEvidence( existing ? withoutStorageMetadata(existing) : undefined, withoutStorageMetadata(record) @@ -141,6 +157,7 @@ export async function writeAccountedRecords(transaction, storeName, records, rem )); } account(record, 1); + if (storeName === 'runs' && typeof record.failureLog === 'string' && record.failureLog.length > 0) failureLogs = true; if (storeName === 'runs' && existing && JSON.stringify(auditCurationRunFacts(existing)) !== JSON.stringify(auditCurationRunFacts(record))) dirtyRuns.push(String(record.id)); store.put(record); @@ -150,7 +167,8 @@ export async function writeAccountedRecords(transaction, storeName, records, rem const changes = [...deltas.values()]; const existing = await Promise.all([TOTALS_ID, ...changes.map(({ id }) => id)] .map((id) => result(shards.get(id)))); - const totals = /** @type {Shard} */ (existing[0] ?? { id: TOTALS_ID, bytes: 0, count: 0 }); + const totals = /** @type {Shard} */ (existing[0] ?? { id: TOTALS_ID, bytes: 0, count: 0, failureLogs: false }); + if (failureLogs) totals.failureLogs = true; changes.forEach((delta, index) => { const previous = /** @type {Shard | undefined} */ (existing[index + 1]); const next = { @@ -207,7 +225,8 @@ async function readShards(database, options) { .objectStore(STORAGE_SHARD_STORE).getAll(range, BATCH_SIZE))); if (rows.some((shard) => typeof shard.id !== 'string' || !Number.isFinite(shard.bytes) || shard.bytes < 0 || !Number.isInteger(shard.count) || shard.count < 0 - || (shard.storeName !== undefined && !Number.isFinite(shard.day)))) { + || (shard.failureLogs !== undefined && typeof shard.failureLogs !== 'boolean') + || (shard.storeName !== undefined && !Number.isFinite(shard.day)))) { throw new Error('Canonical storage shard accounting is invalid; rebuild the derived database'); } shards.push(...rows); @@ -318,6 +337,84 @@ async function curateDirtyAudits(database, shards, options) { return pruned; } +/** + * Reads bounded Run batches only when excerpts may exist. No run-owned evidence + * is read or deleted. Legacy totals without the hint get one discovery pass. + * @param {IDBDatabase} database + * @param {Shard[]} shards + * @param {Parameters[1]} options + */ +async function pruneFailureLogs(database, shards, options) { + if (shards.find(({ id }) => id === TOTALS_ID)?.failureLogs === false) return 0; + const windowMs = retentionWindowForStore('audits', options); + if (windowMs === Number.MAX_SAFE_INTEGER) return 0; + /** @type {Map} */ + const successes = new Map(); + /** @type {Map | undefined>} */ + const workflows = new Map(); + const logShards = new Set(); + /** @type {IDBKeyRange | undefined} */ + let range; + for (;;) { + options.signal?.throwIfAborted(); + const transaction = database.transaction(['runs', 'workflows']); + const done = completed(transaction); + const rows = await result(transaction.objectStore('runs').getAll(range, BATCH_SIZE)); + const ids = [...new Set(rows.map((run) => String(run.workflowId)))].filter((id) => !workflows.has(id)); + const definitions = await Promise.all(ids.map((id) => result(transaction.objectStore('workflows').get(id)))); + ids.forEach((id, index) => workflows.set(id, definitions[index])); + await done; + for (const run of rows) { + if (typeof run.failureLog === 'string' && run.failureLog.length > 0) logShards.add(metadata(run).shard); + if (!isFailureLogRecovery(run)) continue; + const point = failureLogBoundary(run, workflows.get(String(run.workflowId))); + const previous = point && successes.get(point.scope); + if (point && (!previous || compareFailureLogBoundaries(point, previous) > 0)) successes.set(point.scope, point); + } + if (rows.length < BATCH_SIZE) break; + range = IDBKeyRange.lowerBound(rows.at(-1)?.id, true); + } + let pruned = 0; + let retained = false; + const horizon = (options.now ?? Date.now()) - windowMs; + for (const shard of shards.filter(({ id }) => logShards.has(id))) { + let range = IDBKeyRange.bound([Number(shard.day) * DAY_MS], [(Number(shard.day) + 1) * DAY_MS], false, true); + for (;;) { + options.signal?.throwIfAborted(); + const page = await write(database, ['runs'], async (transaction) => { + const rows = await result(transaction.objectStore('runs').index('byStorageTimestamp').getAll(range, BATCH_SIZE)); + const updates = []; + for (const run of rows) { + if (typeof run.failureLog !== 'string' || run.failureLog.length === 0) continue; + const point = failureLogBoundary(run, workflows.get(String(run.workflowId))); + const success = point && successes.get(point.scope); + const resolved = hasFailedConclusion(run) && point && success + && compareFailureLogBoundaries(point, success) < 0; + if (!resolved && metadata(run).timestamp >= horizon) { + retained = true; + continue; + } + const compact = withoutStorageMetadata(run); + delete compact.failureLog; + updates.push(withStorageMetadata('runs', compact, true)); + } + await writeAccountedRecords(transaction, 'runs', updates); + return { rows, pruned: updates.length }; + }); + pruned += page.pruned; + if (page.rows.length < BATCH_SIZE) break; + const last = page.rows.at(-1); + range = IDBKeyRange.bound([metadata(last).timestamp, last.id], [(Number(shard.day) + 1) * DAY_MS], true, true); + } + } + await write(database, [], async (transaction) => { + const store = transaction.objectStore(STORAGE_SHARD_STORE); + const totals = await result(store.get(TOTALS_ID)); + if (totals) store.put({ ...totals, failureLogs: retained }); + }); + return pruned; +} + /** * @param {IDBDatabase} database * @param {Parameters[1] & { @@ -363,6 +460,7 @@ export async function maintainStorageShards(database, options) { shards = await readShards(database, options); const prunedAudits = await curateDirtyAudits(database, shards, options); deletedRecords += prunedAudits; + const prunedFailureLogs = await pruneFailureLogs(database, shards, options); const readTotals = async () => /** @type {Shard} */ ( await result(database.transaction(STORAGE_SHARD_STORE).objectStore(STORAGE_SHARD_STORE).get(TOTALS_ID)) ?? { id: TOTALS_ID, bytes: 0, count: 0 } @@ -390,10 +488,10 @@ export async function maintainStorageShards(database, options) { await receiptDone; debug({ event: 'maintained-storage-shards', durationMs: Math.round(performance.now() - startedAt), - shardCount: shards.length, deletedRecords, prunedAudits, + shardCount: shards.length, deletedRecords, prunedAudits, prunedFailureLogs, estimatedBytes: totals.bytes, retainedRecords: totals.count }); - return { deletedRecords, prunedAudits, estimatedBytes: totals.bytes, retainedRecords: totals.count }; + return { deletedRecords, prunedAudits, prunedFailureLogs, estimatedBytes: totals.bytes, retainedRecords: totals.count }; } /** @@ -414,7 +512,8 @@ export async function rebuildStorageShards(database, entityStores) { // Remove before accounting so rebuilding cannot subtract obsolete totals. for (const record of records) store.delete(record.id); await writeAccountedRecords(transaction, storeName, - records.map((record) => withStorageMetadata(storeName, record))); + records.map((record) => withStorageMetadata(storeName, record, + /** @type {StorageMetadata | undefined} */ (record._storage)?.failureLogPruned === true))); }); if (records.length < BATCH_SIZE) break; range = IDBKeyRange.lowerBound(records.at(-1)?.id, true); diff --git a/dashboard/site/test/e2e/indexeddb-stress.spec.js b/dashboard/site/test/e2e/indexeddb-stress.spec.js index a1624285a4..d8dd7f651d 100644 --- a/dashboard/site/test/e2e/indexeddb-stress.spec.js +++ b/dashboard/site/test/e2e/indexeddb-stress.spec.js @@ -142,3 +142,56 @@ test('native retention cleans expired shards without scanning retained entities' expect(measured.repeatedMs).toBeLessThan(1000); console.log(`Native shard retention: ${Math.round(measured.cleanupMs)}ms cleanup; ${Math.round(measured.repeatedMs)}ms unchanged`); }); + +test('native retention compacts resolved logs across bounded pages and keeps replay compact', async ({ page }) => { + const measured = await page.evaluate(async () => { + const storage = await import(`${location.origin}/src/data/storage/indexeddb.js`); + const batch = { + repositories: [{ id: 'repo:1', owner: 'octo', name: 'control' }], + workflows: [{ id: 'workflow:1', repositoryId: 'repo:1', role: 'orchestrator' }], + runs: Array.from({ length: 1007 }, (_, index) => ({ + id: `github:run:octo/control:${index + 1}`, githubRunId: index + 1, attempt: 1, + repositoryId: 'repo:1', workflowId: 'workflow:1', event: 'push', branch: 'main', + startedAt: '2026-10-20T12:00:00Z', status: 'completed', + conclusion: index === 1005 ? 'success' : 'failure', + ...(index === 1005 ? {} : { failureLog: 'diagnostic '.repeat(200) }) + })) + }; + await storage.upsertCanonicalBatch(indexedDB, batch); + const original = IDBObjectStore.prototype.getAll; + /** @type {(number | undefined)[]} */ + const reads = []; + IDBObjectStore.prototype.getAll = function ( + /** @type {IDBValidKey | IDBKeyRange | null | undefined} */ query, + /** @type {number | undefined} */ count + ) { + if (this.name === 'runs') reads.push(count); + return original.call(this, query, count); + }; + const options = { now: Date.parse('2026-10-20T13:00:00Z'), maxDatabaseBytes: Number.MAX_SAFE_INTEGER }; + try { + const startedAt = performance.now(); + const first = await storage.maintainCanonicalDatabase(indexedDB, options); + const durationMs = performance.now() - startedAt; + await storage.upsertCanonicalBatch(indexedDB, batch); + const replayed = await storage.maintainCanonicalDatabase(indexedDB, options); + IDBObjectStore.prototype.getAll = original; + const runs = await storage.readCollection(indexedDB, 'runs'); + return { + first, replayed, durationMs, reads, runCount: runs.length, + logs: runs.filter((/** @type {Record} */ run) => run.failureLog) + .map((/** @type {Record} */ run) => run.githubRunId) + }; + } finally { + IDBObjectStore.prototype.getAll = original; + } + }); + expect(measured.first).toMatchObject({ deletedRecords: 0, prunedFailureLogs: 1005 }); + expect(measured.replayed).toMatchObject({ deletedRecords: 0, prunedFailureLogs: 0 }); + expect(measured.first.estimatedBytes).toBe(measured.replayed.estimatedBytes); + expect(measured.reads.length).toBeGreaterThan(2); + expect(measured.reads.every((count) => count === 1000)).toBe(true); + expect(measured.runCount).toBe(1007); + expect(measured.logs).toEqual([1007]); + console.log(`Native failure-log retention: ${Math.round(measured.durationMs)}ms for 1007 Runs`); +}); diff --git a/dashboard/site/test/unit/data-failure-log-retention.test.js b/dashboard/site/test/unit/data-failure-log-retention.test.js new file mode 100644 index 0000000000..c16856f901 --- /dev/null +++ b/dashboard/site/test/unit/data-failure-log-retention.test.js @@ -0,0 +1,316 @@ +import 'fake-indexeddb/auto'; +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { IDBFactory } from 'fake-indexeddb'; +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { normalize } from '../../src/data/normalize/index.js'; +import { CANONICAL_SCHEMA_VERSION, relationshipErrors } from '../../src/data/model/schema.js'; +import { finalizeNormalizedJsonlIngestion, ingestNormalizedJsonl } from '../../src/data/ingest/coordinator.js'; +import { + maintainCanonicalDatabase, openCanonicalDatabase, prepareCanonicalRecord, + readCanonicalBatch, readCollection, upsertCanonicalBatch +} from '../../src/data/storage/indexeddb.js'; +import { createSqliteIndexedDB, SqliteIDBKeyRange } from '../../src/data/storage/sqlite-indexeddb.js'; +import { processDataRequest } from '../../src/data-worker.js'; +import sharedEvents from '../../dashboard-fragments/shared-events.json' with { type: 'json' }; + +const now = Date.parse('2026-10-20T12:00:00Z'); +const options = { now, maxDatabaseBytes: Number.MAX_SAFE_INTEGER }; +const directories = /** @type {string[]} */ ([]); + +afterEach(() => { + vi.unstubAllGlobals(); + for (const directory of directories.splice(0)) rmSync(directory, { recursive: true, force: true }); +}); + +/** @param {string[]} conclusions */ +function fixture(conclusions = ['success', 'failure', 'failure', 'success', 'failure', 'failure']) { + const batch = normalize([]); + batch.repositories = [{ id: 'repo:1', owner: 'octo', name: 'control' }]; + batch.workflows = [{ + id: 'workflow:1', repositoryId: 'repo:1', role: 'worker', path: '.github/workflows/worker.md' + }]; + batch.runs = conclusions.map((conclusion, index) => { + const timestamp = new Date(now - (10 - index) * 60_000).toISOString(); + return { + id: `github:run:octo/control:${index + 1}`, githubRunId: index + 1, attempt: 1, + repositoryId: 'repo:1', workflowId: 'workflow:1', owner: 'octo', repository: 'control', + workflowPath: '.github/workflows/worker.md', targetRepository: 'octo/target', + event: 'workflow_dispatch', branch: 'main', rolloutMode: 'review', + status: 'completed', conclusion, startedAt: timestamp, completedAt: timestamp, observedAt: timestamp, + aicTotal: index + 1, inputTokens: index * 100, operationalGrader: index === 2 ? null : 0.8, + failureKind: conclusion === 'failure' ? 'driver_exit' : null, + failureMessage: conclusion === 'failure' ? 'Agent exited' : null, + ...(conclusion === 'failure' ? { failureLog: `run ${index + 1}\n${'diagnostic line\n'.repeat(300)}` } : {}) + }; + }); + batch.audits = batch.runs.map((run) => ({ + id: `audit:${run.githubRunId}`, runId: run.id, timestamp: run.startedAt, source: 'audit', + type: 'audit.finding', status: 'high', summary: 'Specific diagnostic', code: 'driver_exit' + })); + return batch; +} + +/** @param {IDBFactory} indexedDB */ +async function storedTotals(indexedDB) { + const database = await openCanonicalDatabase(indexedDB); + try { + return await new Promise((resolve, reject) => { + const request = database.transaction('storageShards').objectStore('storageShards').get('totals'); + request.onsuccess = () => resolve(request.result); + request.onerror = () => reject(request.error); + }); + } finally { + database.close(); + } +} + +/** @param {IDBFactory} indexedDB @param {Record[]} runs @param {string} identity */ +async function importRuns(indexedDB, runs, identity) { + const content = [ + { kind: 'metadata', schemaVersion: CANONICAL_SCHEMA_VERSION, ingestionVersion: 4, phase: 'runs', records: runs.length }, + ...runs.map((record) => ({ kind: 'record', collection: 'runs', record })) + ].map((row) => JSON.stringify(row)).join('\n'); + async function* chunks() { yield content; } + await ingestNormalizedJsonl(indexedDB, chunks(), { + ...options, payloadIdentity: identity, payloadScope: identity, expectedPhase: 'runs', deferMaintenance: true + }); + return finalizeNormalizedJsonlIngestion(indexedDB, options); +} + +describe.each(['IndexedDB', 'SQLite'])('resolved failure logs in %s', (backend) => { + function database() { + if (backend === 'IndexedDB') return new IDBFactory(); + const directory = mkdtempSync(join(tmpdir(), 'cao-failure-retention-')); + directories.push(directory); + vi.stubGlobal('IDBKeyRange', SqliteIDBKeyRange); + return createSqliteIndexedDB(join(directory, 'dashboard.sqlite')); + } + + it('reclaims persisted log bytes without changing statistical or comparison evidence', async () => { + const indexedDB = database(); + vi.stubGlobal('indexedDB', indexedDB); + const input = fixture(); + const runId = input.runs[1].id; + const timestamp = input.runs[1].startedAt; + const provenance = { source: 'fixture', sourceId: 'comparison', observedAt: timestamp }; + input.tools.push({ id: 'tool:1', runId, timestamp, type: 'mcp.tool_call', source: 'mcp', + mcpServer: 'github', mcpTool: 'search', requestBytes: 10, responseBytes: 20 }); + input.domains.push({ id: 'domain:1', runId, timestamp, type: 'firewall.allow', domain: 'github.com' }); + input.friction.push({ id: 'friction:1', runId, timestamp, aic: 0.25, measurementState: 'measured' }); + input.experiments?.push({ id: 'experiment:1', workflowId: 'workflow:1', name: 'prompt', observedAt: timestamp, provenance }); + input.experimentAssignments?.push({ id: 'assignment:1', runId, experimentId: 'experiment:1', + variant: 'control', timestamp, observedAt: timestamp, provenance, auditId: 'audit:2' }); + input.graders?.push({ id: 'grader:1', workflowId: 'workflow:1', name: 'quality', observedAt: timestamp, provenance }); + input.graderObservations?.push({ id: 'grade:1', runId, graderId: 'grader:1', value: null, + timestamp, observedAt: timestamp, provenance, auditId: 'audit:2', evaluatorDigest: 'sha256:frozen' }); + input.evals?.push({ id: 'eval:1', workflowId: 'workflow:1', name: 'correctness', observedAt: timestamp, provenance }); + input.evalObservations?.push({ id: 'eval-result:1', runId, evalId: 'eval:1', evalResult: 'UNKNOWN', + timestamp, observedAt: timestamp, provenance, auditId: 'audit:2' }); + input.audits.push({ id: 'comparison:1', runId, timestamp, source: 'token-intervention-lifecycle', + type: 'optimization.intervention.updated', interventionState: 'awaiting-evidence', comparisonId: 'comparison:1' }); + await upsertCanonicalBatch(indexedDB, input); + const statistics = () => processDataRequest({ + operation: 'load-dashboard-query-sources', sources: {}, queries: sharedEvents.queries, + sourceNames: ['workflow-run-totals', 'workflow-aic-run-totals', 'grader-observations', + 'eval-observations', 'experiment-assignments', 'friction', 'tools', 'domains', 'audits'] + }); + const before = await statistics(); + expect(before).toMatchObject({ + 'workflow-run-totals': { + rows: [{ runs: 6, 'successful-runs': 2, 'failed-runs': 4, 'imported-runs': 6 }], + metadata: { availability: 'available' } + }, + 'workflow-aic-run-totals': { rows: [{ runs: 6, 'aic-runs': 6, aic: 21 }] } + }); + const bytes = (await storedTotals(indexedDB)).bytes; + const result = await maintainCanonicalDatabase(indexedDB, options); + expect(result).toMatchObject({ prunedFailureLogs: 2, deletedRecords: 0 }); + const retained = await readCanonicalBatch(indexedDB); + expect(retained.runs.filter((run) => typeof run.failureLog === 'string').map((run) => run.githubRunId)).toEqual([5, 6]); + const expected = structuredClone(input); + delete expected.runs[1].failureLog; + delete expected.runs[2].failureLog; + expect(retained).toEqual(expected); + expect(relationshipErrors(retained)).toEqual([]); + expect(await statistics()).toEqual(before); + const connection = await openCanonicalDatabase(indexedDB); + try { + const stored = await new Promise((resolve, reject) => { + const request = connection.transaction('runs').objectStore('runs').get(String(input.runs[1].id)); + request.onsuccess = () => resolve(request.result); + request.onerror = () => reject(request.error); + }); + expect(stored).not.toHaveProperty('failureLog'); + expect(stored._storage.failureLogPruned).toBe(true); + } finally { + connection.close(); + } + const totals = await storedTotals(indexedDB); + expect(bytes - totals.bytes).toBe(2 * (JSON.stringify({ failureLog: input.runs[1].failureLog }).length - 1)); + expect(totals.bytes).toBe(Object.entries(retained).reduce((sum, [store, rows]) => + sum + rows.reduce((bytes, row) => bytes + prepareCanonicalRecord(store, row)._storage.bytes, 0), 0)); + expect((await maintainCanonicalDatabase(indexedDB, options)).prunedFailureLogs).toBe(0); + }); + + it('handles incremental recovery, out-of-order imports, cached replay and stale attempts', async () => { + const indexedDB = database(); + const input = fixture(); + await upsertCanonicalBatch(indexedDB, { ...input, runs: [], audits: [] }); + await importRuns(indexedDB, input.runs.slice(0, 3), 'before-recovery'); + expect((await readCollection(indexedDB, 'runs')).filter((run) => run.failureLog)).toHaveLength(2); + expect((await importRuns(indexedDB, [input.runs[3]], 'recovery')).prunedFailureLogs).toBe(2); + await importRuns(indexedDB, input.runs, 'older-and-current'); + await importRuns(indexedDB, input.runs, 'older-and-current'); + expect((await readCollection(indexedDB, 'runs')).filter((run) => run.failureLog).map((run) => run.githubRunId)).toEqual([5, 6]); + await importRuns(indexedDB, [{ ...input.runs[3], attempt: 2 }], 'winning-attempt'); + await importRuns(indexedDB, [{ ...input.runs[3], conclusion: 'failure', failureLog: 'obsolete attempt' }], 'stale-attempt'); + expect((await readCollection(indexedDB, 'runs')).find((run) => run.githubRunId === 4)) + .toMatchObject({ attempt: 2, conclusion: 'success' }); + await importRuns(indexedDB, [{ ...input.runs[1], attempt: 2, startedAt: new Date(now).toISOString() }], 'new-failure-attempt'); + expect((await readCollection(indexedDB, 'runs')).find((run) => run.githubRunId === 2)).toHaveProperty('failureLog'); + }); + + it('prunes late-delivered failures and never restores their excerpts after recovery evidence changes', async () => { + const indexedDB = database(); + const input = fixture(['failure', 'success']); + await upsertCanonicalBatch(indexedDB, { ...input, runs: [input.runs[1]], audits: [] }); + await importRuns(indexedDB, [input.runs[0]], 'late-failure'); + expect((await readCollection(indexedDB, 'runs'))[0]).not.toHaveProperty('failureLog'); + await importRuns(indexedDB, [{ ...input.runs[1], attempt: 2, conclusion: 'failure' }], 'recovery-rerun'); + await importRuns(indexedDB, [input.runs[0]], 'failure-replay'); + expect((await readCollection(indexedDB, 'runs'))[0]).not.toHaveProperty('failureLog'); + await importRuns(indexedDB, [{ ...input.runs[0], attempt: undefined }], 'unknown-attempt-replay'); + expect((await readCollection(indexedDB, 'runs'))[0]).toMatchObject({ attempt: 1 }); + expect((await readCollection(indexedDB, 'runs'))[0]).not.toHaveProperty('failureLog'); + const fresh = { ...input.runs[0], id: 'github:run:octo/control:3', githubRunId: 3 }; + await importRuns(indexedDB, [fresh], 'unresolved-failure'); + await importRuns(indexedDB, [input.runs[1]], 'obsolete-success'); + expect((await readCollection(indexedDB, 'runs')).find((run) => run.githubRunId === 3)).toHaveProperty('failureLog'); + }); + + it('does not use an older observation of the winning attempt as recovery', async () => { + const indexedDB = database(); + const input = fixture(['failure', 'success']); + await upsertCanonicalBatch(indexedDB, { + ...input, runs: [input.runs[0], { ...input.runs[1], conclusion: 'failure', observedAt: new Date(now).toISOString() }] + }); + await importRuns(indexedDB, [input.runs[1]], 'older-observation'); + expect((await readCollection(indexedDB, 'runs'))[0]).toHaveProperty('failureLog'); + expect((await readCollection(indexedDB, 'runs'))[1].conclusion).toBe('failure'); + }); + + it.each([ + { repositoryId: 'repo:other' }, { workflowId: 'workflow:other' }, + { targetRepository: 'octo/another' }, { targetRepository: null }, + { event: 'push' }, { branch: 'feature' }, { rolloutMode: 'live' }, + { conclusion: 'cancelled' }, { conclusion: null }, { status: 'in_progress' }, + { terminalOutcome: 'report_incomplete' }, { startedAt: null, createdAt: null, updatedAt: null } + ])('does not mistake unrelated or incomplete evidence for recovery: %j', async (different) => { + const indexedDB = database(); + const input = fixture(['failure', 'success']); + Object.assign(input.runs[1], different); + await upsertCanonicalBatch(indexedDB, input, { validateRelationships: false }); + await maintainCanonicalDatabase(indexedDB, options); + expect((await readCollection(indexedDB, 'runs'))[0]).toHaveProperty('failureLog'); + }); + + it.each(['branch', 'rolloutMode', 'unknown-role'])('does not equate two unknown scope dimensions: %s', async (dimension) => { + const indexedDB = database(); + const input = fixture(['failure', 'success']); + if (dimension === 'unknown-role') { + delete input.workflows[0].role; + input.runs.forEach((run) => { delete run.targetRepository; }); + } else { + input.runs.forEach((run) => { delete run[dimension]; }); + } + await upsertCanonicalBatch(indexedDB, input); + expect((await maintainCanonicalDatabase(indexedDB, options)).prunedFailureLogs).toBe(0); + }); + + it('uses run-date and numeric run-ID ordering, not arrival or completion time', async () => { + const indexedDB = database(); + const input = fixture(['failure', 'success']); + input.runs[0].githubRunId = 9; + input.runs[1].githubRunId = 10; + input.runs[0].completedAt = new Date(now).toISOString(); + input.runs[1].startedAt = input.runs[0].startedAt; + await upsertCanonicalBatch(indexedDB, input); + expect((await maintainCanonicalDatabase(indexedDB, options)).prunedFailureLogs).toBe(1); + }); + + it('keeps no-success and current streaks within existing detail and database limits', async () => { + const indexedDB = database(); + const input = fixture(['failure', 'failure']); + const old = new Date(now - 8 * 86_400_000).toISOString(); + Object.assign(input.runs[0], { startedAt: old, completedAt: old, observedAt: old }); + input.audits = []; + await upsertCanonicalBatch(indexedDB, input); + expect((await maintainCanonicalDatabase(indexedDB, options)).prunedFailureLogs).toBe(1); + const retained = await readCollection(indexedDB, 'runs'); + expect(retained).toHaveLength(2); + expect(retained[0]).not.toHaveProperty('failureLog'); + expect(retained[1]).toHaveProperty('failureLog'); + await maintainCanonicalDatabase(indexedDB, { ...options, maxDatabaseBytes: 0 }); + expect(await readCollection(indexedDB, 'runs')).toEqual([]); + }); + + it.each([false, true])('reclaims excerpts before whole-Run size eviction (repair=%s)', async (repairAuditCuration) => { + const indexedDB = database(); + const input = fixture(['failure', 'success']); + await upsertCanonicalBatch(indexedDB, input); + const bytes = (await storedTotals(indexedDB)).bytes; + const savings = JSON.stringify({ failureLog: input.runs[0].failureLog }).length - 2; + const result = await maintainCanonicalDatabase(indexedDB, { + ...options, repairAuditCuration, maxDatabaseBytes: Math.ceil((bytes - savings + 10) / 0.75) + }); + expect(result).toMatchObject({ deletedRecords: 0, prunedFailureLogs: 1 }); + expect(await readCollection(indexedDB, 'runs')).toHaveLength(2); + }); + + it('discovers excerpts in an existing cache without requiring a changed source receipt', async () => { + const indexedDB = database(); + const input = fixture(['failure', 'success']); + await upsertCanonicalBatch(indexedDB, input); + const totals = await storedTotals(indexedDB); + delete totals.failureLogs; + const connection = await openCanonicalDatabase(indexedDB); + try { + await new Promise((resolve, reject) => { + const transaction = connection.transaction('storageShards', 'readwrite'); + transaction.objectStore('storageShards').put(totals); + transaction.oncomplete = resolve; + transaction.onabort = () => reject(transaction.error); + }); + } finally { + connection.close(); + } + expect((await finalizeNormalizedJsonlIngestion(indexedDB, options)).prunedFailureLogs).toBe(1); + }); + + it('preserves the explicitly unlimited detail profile for a separate historical archive', async () => { + const indexedDB = database(); + const input = fixture(['failure', 'success']); + await upsertCanonicalBatch(indexedDB, input); + const result = await maintainCanonicalDatabase(indexedDB, { + ...options, retentionWindowMs: Number.MAX_SAFE_INTEGER + }); + expect(result.prunedFailureLogs).toBe(0); + expect((await readCollection(indexedDB, 'runs'))[0]).toHaveProperty('failureLog'); + }); + + it('reports excerpt-only changes when a direct normalized import is already cached', async () => { + const indexedDB = database(); + const input = fixture(['failure', 'success']); + await upsertCanonicalBatch(indexedDB, { ...input, runs: [input.runs[0]], audits: [] }); + await importRuns(indexedDB, [input.runs[0]], 'cached-failure'); + await upsertCanonicalBatch(indexedDB, { ...input, runs: [input.runs[1]], audits: [] }); + async function* unread() { yield Promise.reject(new Error('Cached input must not be parsed')); } + const result = await ingestNormalizedJsonl(indexedDB, unread(), { + ...options, payloadIdentity: 'cached-failure', payloadScope: 'cached-failure', expectedPhase: 'runs' + }); + expect(result).toMatchObject({ skipped: true, updated: true, committedRecords: 0 }); + expect((await readCollection(indexedDB, 'runs'))[0]).not.toHaveProperty('failureLog'); + }); +}); diff --git a/dashboard/site/test/unit/data-worker-retention.test.js b/dashboard/site/test/unit/data-worker-retention.test.js index fa37184fbb..ccbc637bff 100644 --- a/dashboard/site/test/unit/data-worker-retention.test.js +++ b/dashboard/site/test/unit/data-worker-retention.test.js @@ -2,7 +2,8 @@ import 'fake-indexeddb/auto'; import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { CANONICAL_SCHEMA_VERSION } from '../../src/data/model/schema.js'; -import { DATABASE_NAME, readTransactions } from '../../src/data/storage/indexeddb.js'; +import { normalize } from '../../src/data/normalize/index.js'; +import { DATABASE_NAME, readTransactions, upsertCanonicalBatch } from '../../src/data/storage/indexeddb.js'; const metadata = { 'as-of': '2026-09-09T05:00:00Z', 'artifact-generation': 'generation-a' }; @@ -255,6 +256,47 @@ describe('canonical dashboard worker retention updates', () => { expect((await settled((message) => message.id === 6))?.data).toMatchObject({ changed: false }); expect(jsonlRequests).toHaveLength(4); + await upsertCanonicalBatch(indexedDB, { + ...normalize([]), + repositories: [{ id: 'repo:retention', owner: 'octo', name: 'control' }], + workflows: [{ id: 'workflow:retention', repositoryId: 'repo:retention', role: 'orchestrator', + path: '.github/workflows/test.md' }], + runs: ['failure', 'success'].map((conclusion, index) => ({ + id: `github:run:octo/control:${index + 1}`, githubRunId: index + 1, attempt: 1, + repositoryId: 'repo:retention', workflowId: 'workflow:retention', + owner: 'octo', repository: 'control', workflowPath: '.github/workflows/test.md', + event: 'push', branch: 'main', status: 'completed', conclusion, + startedAt: `2026-09-09T23:0${index}:00Z`, + ...(index === 0 ? { failureLog: 'resolved diagnostic excerpt' } : {}) + })) + }); + const failureContext = { ...context, queries: [{ + name: 'failure-excerpts', from: 'runs', select: [{ field: 'failure-log' }] + }] }; + dispatch({ + id: 10, operation: 'load-dashboard-query-sources', sources: {}, + sourceNames: ['failure-excerpts'], queries: failureContext.queries + }); + expect(JSON.stringify(await settled((message) => message.id === 10))).toContain('resolved diagnostic excerpt'); + dispatch({ + operation: 'subscribe-canonical-dashboard', subscriptionId: 'failures', + sourceNames: ['failure-excerpts'], context: failureContext + }); + const previousFailures = await settled((message) => message.subscriptionId === 'failures'); + dispatch({ + id: 9, operation: 'load-canonical-dashboard', + sourceUrl: 'https://dashboard.example/payload-hashes.json', + sourceNames: ['failure-excerpts'], context: failureContext, reportActivation: true + }); + expect((await settled((message) => message.id === 9))?.data).toMatchObject({ changed: true }); + const failures = await settled((message) => message.subscriptionId === 'failures' + && Number(message.revision) > Number(previousFailures?.revision)); + expect(failures).toMatchObject({ + data: { 'failure-excerpts': { rows: [{}, {}, {}] } } + }); + expect(JSON.stringify(failures)).not.toContain('resolved diagnostic excerpt'); + expect(jsonlRequests).toHaveLength(4); + globalThis.fetch = /** @type {typeof fetch} */ (async (input) => { if (String(input).endsWith('/payload-hashes.json')) return new Response(null, { status: 404 }); if (String(input).endsWith('/inventory-sources.json')) return new Response(null, { status: 404 }); diff --git a/docs/dashboard-data-ingestion.md b/docs/dashboard-data-ingestion.md index 1dcea80af2..b88c924348 100644 --- a/docs/dashboard-data-ingestion.md +++ b/docs/dashboard-data-ingestion.md @@ -181,6 +181,21 @@ structural parents. The effective retention horizon is the later of the browser clock and the newest incoming observation, so a slow clock cannot prune current producer data. +Browser IndexedDB and Node.js SQLite ingestion also discard a failed Run's raw +`failureLog` excerpt once a later completed success is known in the same +repository/workflow, target, event, branch, and rollout mode. Cancellation and +incomplete outcomes do not prove recovery. Only the excerpt is removed: Run +summaries, failure signatures/messages, cost and reliability facts, and all +run-linked metric and comparison evidence keep their existing retention. +Re-importing an older shard cannot restore a pruned excerpt for the same attempt. +Unresolved excerpts follow the existing detail window and database size cap. +A separate archive with `--retention-days all` preserves full detail. This does +not prune hosted PostgreSQL, published shards, or GitHub artifacts. +The saving is retained diagnostic bytes and reduced storage pressure, not fewer +GitHub API calls, downloaded artifacts, or workflow executions. Cleanup itself +adds a bounded maintenance scan when excerpts are present; no execution-time +reduction is assumed. + Every merged batch must satisfy these relationships: - Workflow to Repository diff --git a/specs/dashboard-data.md b/specs/dashboard-data.md index ff60c58c00..354c918049 100644 --- a/specs/dashboard-data.md +++ b/specs/dashboard-data.md @@ -3231,6 +3231,37 @@ Token-optimization retention SHALL also satisfy Section 5.5.5 so an active intervention is not pruned before maturation and compact terminal evidence remains bounded. +## 41.1 Resolved failure log excerpts + +Canonical IndexedDB and its Node.js SQLite adapter SHALL reclaim the optional +Run `failureLog` excerpt after a later completed native success in the same +execution Repository, Workflow, campaign/role, target, event, branch and rollout +mode. Ordering SHALL use the first valid `startedAt`, `createdAt`, or `updatedAt`, +then numeric GitHub run ID and attempt, as in the runtime-health contract. +Unknown ordering or branch, missing worker target or rollout mode, unknown +workflow role without a target association, cancellation, missing conclusions, +and explicitly incomplete outcomes MUST NOT establish recovery. +Lower attempts and older observations MUST NOT overwrite the retained winner. + +This first increment SHALL remove only the excerpt, not the Run, its failure +signature/message, accounting, or any run-linked record. Audit summaries and +counts, tool/domain activity, friction, safe outputs, experiment assignments, +grader/eval observations, and intervention/comparison evidence remain subject to +their existing retention rules. Missing excerpts MUST remain missing, not +successful outcomes or zero diagnostic activity. + +Unresolved excerpts SHALL use the existing Audit detail window and whole-Run +size cap; a separate full-detail archive explicitly using unlimited detail +retention SHALL remain exempt. A storage-local, attempt-scoped pruning marker +SHALL prevent replay from restoring a discarded excerpt while its Run survives, +without suppressing a new attempt's diagnostic evidence. + +Maintenance SHALL use bounded Run reads and accounted writes before size-based +Run eviction, including when source receipts are unchanged. It MUST NOT load +run-owned evidence collections or persist statistical aggregates. Excerpt-only +changes SHALL refresh active query results. This policy does not alter hosted +PostgreSQL, published JSONL, or GitHub runs and artifacts. + --- # 42. Search