fix(usage): page filtered index candidates

src/server/management/usage-aggregate-cache.ts

@@ -15,6 +15,7 @@ import {
   readUsageFilterCandidatesPassive,
   requestUsageFilterIndexRebuild,
   resetUsageFilterIndexForTests,
+  UsageFilterIndexPageError,
   type UsageFilterCandidateRead,
   type UsageFilterIndexQuery,
   type UsageFilterIndexSnapshot,
@@ -110,7 +111,6 @@ const MAX_REBUILD_ATTEMPTS = 2;
 const MAX_FILTER_INDEX_ATTEMPTS = 3;
 // Selected rows are folded in bounded batches, so a filter that matches most of a large ledger
 // costs the same peak memory as the streaming scan it replaces.
-const FILTER_ROW_READ_BATCH = 2_000;
 const MAX_RETAINED_FILTERED_AGGREGATES = 4;
 
 let retainedAggregate: RetainedUsageAggregate | null = null;
@@ -396,7 +396,7 @@ function filterStats(
 ): UsageFilterQueryStats {
   return {
     totalRows: read.snapshot.indexedRows,
-    candidateRows: read.offsets.length,
+    candidateRows: read.candidateRows,
     scope: update,
     ingestedRows: read.snapshot.ingestedRows,
     ingestPasses: read.snapshot.ingestPasses,
@@ -465,33 +465,45 @@ async function readCandidates(
     : readUsageFilterCandidates(query);
 }
 
+/**
+ * Folds every candidate page into the accumulator, yielding between pages and re-reading rows
+ * against the snapshot identity; closes the reader when done.
+ */
 async function accumulateCandidates(
   accumulator: UsageSummaryAccumulator,
   read: UsageFilterCandidateRead,
   filter: NormalizedUsageFilter,
 ): Promise<void> {
-  // Always read at least once, even with no candidates: the reader validates the snapshot
-  // against the ledger, and an empty selection must not skip that check.
-  const batchCount = Math.max(1, Math.ceil(read.offsets.length / FILTER_ROW_READ_BATCH));
-  for (let batch = 0; batch < batchCount; batch += 1) {
-    const start = batch * FILTER_ROW_READ_BATCH;
-    const rows = await readUsageLedgerRowsAtOffsets({
-      offsets: read.offsets.slice(start, start + FILTER_ROW_READ_BATCH),
-      expectedIdentityKey: read.snapshot.identityKey,
-      // The offsets belong to this index snapshot. If the ledger moved since the snapshot,
-      // the reader re-validates the boundary before trusting any of them, and a rewrite of
-      // the indexed tail becomes a rebuild instead of a silently different row set.
-      expectedRevisionKey: usageLogRevisionKey(read.snapshot.revision),
-      expectedProcessedThroughBytes: read.snapshot.indexedOffset,
-      expectedProcessedThroughDigest: read.snapshot.processedThroughDigest,
-      // The physical size at publish time; a torn suffix makes it larger than the boundary.
-      expectedFileSize: read.snapshot.revision?.size,
-    });
-    if (rows.unreadableOffsets.length > 0) {
-      // Serving only the rows that happened to parse would silently under-report the filter.
-      throw new UsageFilterIndexCorruptError();
+  try {
+    let first = true;
+    for (;;) {
+      // Candidate selection and row decoding are synchronous bounded chunks. A timer yield
+      // between them prevents a broad authenticated query from monopolizing the event loop.
+      if (!first) await new Promise<void>(resolve => setTimeout(resolve, 0));
+      const offsets = read.nextPage();
+      if (!first && offsets.length === 0) break;
+      first = false;
+      const rows = await readUsageLedgerRowsAtOffsets({
+        offsets,
+        expectedIdentityKey: read.snapshot.identityKey,
+        // The offsets belong to this index snapshot. If the ledger moved since the snapshot,
+        // the reader re-validates the boundary before trusting any of them, and a rewrite of
+        // the indexed tail becomes a rebuild instead of a silently different row set.
+        expectedRevisionKey: usageLogRevisionKey(read.snapshot.revision),
+        expectedProcessedThroughBytes: read.snapshot.indexedOffset,
+        expectedProcessedThroughDigest: read.snapshot.processedThroughDigest,
+        // The physical size at publish time; a torn suffix makes it larger than the boundary.
+        expectedFileSize: read.snapshot.revision?.size,
+      });
+      if (rows.unreadableOffsets.length > 0) {
+        // Serving only the rows that happened to parse would silently under-report the filter.
+        throw new UsageFilterIndexCorruptError();
+      }
+      for (const entry of rows.entries) accumulator.add(entry);
+      if (offsets.length === 0) break;
     }
-    for (const entry of rows.entries) accumulator.add(entry);
+  } finally {
+    read.close();
   }
   // Without a source filter the reported span covers the whole ledger, including rows this
   // filter never selected. A source filter derives its span from the rows that match it, so
@@ -626,57 +638,80 @@ async function appendFilteredAggregate(
   }
 }
 
+/**
+ * Serves the filtered report from the retained aggregate, appending new candidates or rebuilding
+ * as needed; a candidate-page failure clears the retained state and retries the streaming scan.
+ */
 async function refreshFilteredAggregate(
   key: string,
   filter: NormalizedUsageFilter,
   window?: UsageTimeWindow,
 ): Promise<UsageAggregateResult> {
-  return withFilterIndexRetry(async () => {
-    const overlayVersion = userCostOverlayVersion();
-    const timeZone = currentTimeZone();
-    const state = retainedFilteredAggregates.get(key);
-    if (!state
-      || state.overlayVersion !== overlayVersion
-      || state.timeZone !== timeZone
-      || state.indexGeneration === null) {
-      return rebuildFilteredAggregate(key, filter, window, overlayVersion, timeZone);
-    }
-    const read = await readCandidates({
-      ...filterQuery(filter, window),
-      afterOffset: state.processedThroughBytes,
+  try {
+    return await withFilterIndexRetry(async () => {
+      const overlayVersion = userCostOverlayVersion();
+      const timeZone = currentTimeZone();
+      const state = retainedFilteredAggregates.get(key);
+      if (!state
+        || state.overlayVersion !== overlayVersion
+        || state.timeZone !== timeZone
+        || state.indexGeneration === null) {
+        return rebuildFilteredAggregate(key, filter, window, overlayVersion, timeZone);
+      }
+      const read = await readCandidates({
+        ...filterQuery(filter, window),
+        afterOffset: state.processedThroughBytes,
+      });
+      if (!read) {
+        retainedFilteredAggregates.delete(key);
+        return rebuildFilteredAggregate(key, filter, window, overlayVersion, timeZone);
+      }
+      if (state.indexGeneration !== read.snapshot.generation
+        || state.identityKey !== read.snapshot.identityKey
+        || state.processedThroughBytes > read.snapshot.indexedOffset) {
+        // This branch never pages, so it owns closing the reader before the rebuild discards it.
+        read.close();
+        retainedFilteredAggregates.delete(key);
+        return rebuildFilteredAggregate(key, filter, window, overlayVersion, timeZone);
+      }
+      if (state.processedThroughBytes === read.snapshot.indexedOffset) {
+        // Even an unchanged retained projection validates the ledger against the snapshot, so a
+        // rewrite in the window between the index refresh and this return is not served stale.
+        try {
+          await readUsageLedgerRowsAtOffsets({
+            offsets: [],
+            expectedIdentityKey: read.snapshot.identityKey,
+            expectedRevisionKey: usageLogRevisionKey(read.snapshot.revision),
+            expectedProcessedThroughBytes: read.snapshot.indexedOffset,
+            expectedProcessedThroughDigest: read.snapshot.processedThroughDigest,
+            expectedFileSize: read.snapshot.revision?.size,
+          });
+        } finally {
+          // This branch never pages, so it owns closing the reader before publishing its stats.
+          read.close();
+        }
+        state.retainedAt = Date.now();
+        // Whole-ledger diagnostics can move without the boundary moving: a torn oversized or
+        // malformed suffix is observed by every append scan and never advances the offset.
+        state.usageIncomplete = read.snapshot.oversizedRows > 0;
+        state.usageInvalidRows = read.snapshot.invalidRows > 0;
+        state.revision = read.snapshot.revision;
+        state.revisionKey = usageLogRevisionKey(read.snapshot.revision);
+        return publishFilteredAggregate(key, state, "unchanged", read);
+      }
+      return appendFilteredAggregate(key, state, filter, read, overlayVersion, timeZone);
     });
-    if (!read) {
-      retainedFilteredAggregates.delete(key);
-      return rebuildFilteredAggregate(key, filter, window, overlayVersion, timeZone);
-    }
-    if (state.indexGeneration !== read.snapshot.generation
-      || state.identityKey !== read.snapshot.identityKey
-      || state.processedThroughBytes > read.snapshot.indexedOffset) {
+  } catch (error) {
+    if (error instanceof UsageFilterIndexPageError) {
+      // A page failed after its reader was handed out: the index cannot serve this query,
+      // so answer from the authoritative ledger scan instead of failing the request.
       retainedFilteredAggregates.delete(key);
-      return rebuildFilteredAggregate(key, filter, window, overlayVersion, timeZone);
-    }
-    if (state.processedThroughBytes === read.snapshot.indexedOffset) {
-      // Even an unchanged retained projection validates the ledger against the snapshot, so a
-      // rewrite in the window between the index refresh and this return is not served stale.
-      await readUsageLedgerRowsAtOffsets({
-        offsets: [],
-        expectedIdentityKey: read.snapshot.identityKey,
-        expectedRevisionKey: usageLogRevisionKey(read.snapshot.revision),
-        expectedProcessedThroughBytes: read.snapshot.indexedOffset,
-        expectedProcessedThroughDigest: read.snapshot.processedThroughDigest,
-        expectedFileSize: read.snapshot.revision?.size,
-      });
-      state.retainedAt = Date.now();
-      // Whole-ledger diagnostics can move without the boundary moving: a torn oversized or
-      // malformed suffix is observed by every append scan and never advances the offset.
-      state.usageIncomplete = read.snapshot.oversizedRows > 0;
-      state.usageInvalidRows = read.snapshot.invalidRows > 0;
-      state.revision = read.snapshot.revision;
-      state.revisionKey = usageLogRevisionKey(read.snapshot.revision);
-      return publishFilteredAggregate(key, state, "unchanged", read);
+      return withFilterIndexRetry(() =>
+        rebuildFilteredAggregateStreaming(key, filter, window, userCostOverlayVersion(), currentTimeZone()),
+      );
     }
-    return appendFilteredAggregate(key, state, filter, read, overlayVersion, timeZone);
-  });
+    throw error;
+  }
 }
 
 function jevStatsResultFrom(