Reclaim resolved failure log excerpts without dropping statistics

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 };
 }
 

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.');

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

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<string, unknown>} 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<string, number>, 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<string, Record<string, unknown>>} */
     const auditRunFacts = new Map();
-    /** @type {{ id: string, timestamp: number, bytes: number }[]} */
-    const runs = [];
     /** @type {string[]} */
     const expiredRunIds = [];
-    /** @type {Map<string, number>} */
-    const linkedBytesByRun = new Map();
-    /** @type {Map<string, number>} */
-    const linkedRecordsByRun = new Map();
-    const retainedRunIds = new Set();
     const campaignIds = new Set();
     const repositoryIds = new Set();
     /** @type {Map<string, string>} */
@@ -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<string>} */ (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();
     }

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<string, unknown>} run
+ * @param {Record<string, unknown> | 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<string, unknown>} run */
+export function isFailureLogRecovery(run) {
+  return run.status === 'completed' && run.conclusion === 'success'
+    && (run.terminalOutcome == null || run.terminalOutcome === '')
+    && (run.failureKind == null || run.failureKind === '');
+}
+
+/** @param {Record<string, unknown>} 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}