perf: batch and resume path history generation

2026-09-27 01:48:04 +00:00 · 2026-08-09 13:20:05 +00:00
parent c3f64d5be3
commit 3e26280ed4
10 changed files with 799 additions and 130 deletions
@@ -28,3 +28,33 @@ test('bounded segment counter retains heavy hitters with conservative counts', (
  assert.ok((candidates.get('common-a') ?? 0) <= 150);
  assert.ok((candidates.get('common-b') ?? 0) <= 100);
 });
+
+test('bounded segment counter resumes with byte-equivalent continuation state', () => {
+  const uninterrupted = new BoundedSegmentCounter(8);
+  const checkpointed = new BoundedSegmentCounter(8);
+  const before = Array.from({ length: 50 }, (_, index) => `before-${index % 13}`);
+  const after = Array.from({ length: 75 }, (_, index) => `after-${index % 17}`);
+
+  for (const key of before) {
+    uninterrupted.observe(key);
+    checkpointed.observe(key);
+  }
+  const resumed = BoundedSegmentCounter.fromSnapshot(8, checkpointed.snapshot());
+  for (const key of after) {
+    uninterrupted.observe(key);
+    resumed.observe(key);
+  }
+
+  assert.deepEqual(resumed.snapshot(), uninterrupted.snapshot());
+  assert.deepEqual(resumed.candidates(1), uninterrupted.candidates(1));
+  assert.equal(resumed.replacementCount(), uninterrupted.replacementCount());
+});
+
+test('bounded segment counter rejects incompatible checkpoint state', () => {
+  const counter = new BoundedSegmentCounter(4);
+  counter.observe('a');
+  assert.throws(
+    () => BoundedSegmentCounter.fromSnapshot(5, counter.snapshot()),
+    /INVALID_SEGMENT_COUNTER_SNAPSHOT/,
+  );
+});
@@ -5,6 +5,17 @@ type CounterEntry = {
  heapIndex: number;
 };

+export type BoundedSegmentCounterSnapshot = {
+  version: 1;
+  capacity: number;
+  replacements: number;
+  entries: Array<{
+    key: string;
+    count: number;
+    error: number;
+  }>;
+};
+
 /**
 * Space-Saving heavy-hitter counter.
 *
@@ -23,6 +34,55 @@ export class BoundedSegmentCounter {
    }
  }

+  static fromSnapshot(
+    capacity: number,
+    snapshot: BoundedSegmentCounterSnapshot,
+  ): BoundedSegmentCounter {
+    if (snapshot.version !== 1 || snapshot.capacity !== capacity) {
+      throw new Error('INVALID_SEGMENT_COUNTER_SNAPSHOT');
+    }
+    if (!Number.isSafeInteger(snapshot.replacements) || snapshot.replacements < 0) {
+      throw new Error('INVALID_SEGMENT_COUNTER_SNAPSHOT');
+    }
+    if (!Array.isArray(snapshot.entries) || snapshot.entries.length > capacity) {
+      throw new Error('INVALID_SEGMENT_COUNTER_SNAPSHOT');
+    }
+
+    const counter = new BoundedSegmentCounter(capacity);
+    const seen = new Set<string>();
+    for (const saved of snapshot.entries) {
+      if (
+        typeof saved.key !== 'string'
+        || saved.key.length === 0
+        || seen.has(saved.key)
+        || !Number.isSafeInteger(saved.count)
+        || saved.count < 1
+        || !Number.isSafeInteger(saved.error)
+        || saved.error < 0
+        || saved.error > saved.count
+      ) {
+        throw new Error('INVALID_SEGMENT_COUNTER_SNAPSHOT');
+      }
+      seen.add(saved.key);
+      const entry: CounterEntry = {
+        key: saved.key,
+        count: saved.count,
+        error: saved.error,
+        heapIndex: counter.heap.length,
+      };
+      counter.entries.set(entry.key, entry);
+      counter.heap.push(entry);
+    }
+    for (let index = 1; index < counter.heap.length; index += 1) {
+      const parent = Math.floor((index - 1) / 2);
+      if (counter.less(counter.heap[index]!, counter.heap[parent]!)) {
+        throw new Error('INVALID_SEGMENT_COUNTER_SNAPSHOT');
+      }
+    }
+    counter.replacements = snapshot.replacements;
+    return counter;
+  }
+
  observe(key: string): void {
    const existing = this.entries.get(key);
    if (existing) {
@@ -72,6 +132,19 @@ export class BoundedSegmentCounter {
    return this.replacements;
  }

+  snapshot(): BoundedSegmentCounterSnapshot {
+    return {
+      version: 1,
+      capacity: this.capacity,
+      replacements: this.replacements,
+      entries: this.heap.map((entry) => ({
+        key: entry.key,
+        count: entry.count,
+        error: entry.error,
+      })),
+    };
+  }
+
  private less(left: CounterEntry, right: CounterEntry): boolean {
    return left.count < right.count || (left.count === right.count && left.key < right.key);
  }
@@ -328,6 +328,83 @@ export async function updateAnalysisRunTotalItems(
  if (result.rows.length !== 1) throw new AnalysisRunLeaseLostError(handle.runId);
 }

+export async function checkpointAnalysisRun(
+  handle: AnalysisRunHandle,
+  checkpoint: number,
+  metadata: unknown,
+  signal?: AbortSignal,
+): Promise<void> {
+  const result = await query(
+    `UPDATE analysis_runs run
+        SET checkpoint = $4,
+            metadata = $5::jsonb,
+            heartbeat_at = clock_timestamp()
+       FROM analysis_workload_state state
+      WHERE state.workload = $1
+        AND state.scope = $2
+        AND state.active_run_id = run.run_id
+        AND state.active_run_id = $3
+        AND state.active_lease_token = $6
+        AND state.active_lease_expires_at > clock_timestamp()
+        AND state.active_run_deadline_at > clock_timestamp()
+        AND run.lease_token = $6
+        AND run.status = 'running'
+      RETURNING run.run_id`,
+    [
+      handle.workload,
+      handle.scope,
+      handle.runId,
+      Math.max(0, Math.trunc(checkpoint)),
+      JSON.stringify(metadata ?? {}),
+      handle.leaseToken,
+    ],
+    signal,
+  );
+  if (result.rows.length !== 1) throw new AnalysisRunLeaseLostError(handle.runId);
+}
+
+export async function getLatestTimedOutAnalysisCheckpoint(
+  workload: string,
+  scope: string,
+): Promise<{
+  checkpoint: number;
+  metadata: unknown;
+  windowStart: string;
+  windowEnd: string;
+  privacyGeneration: number | null;
+  modelGeneration: string | null;
+} | null> {
+  const result = await query<{
+    status: string;
+    checkpoint: string;
+    metadata: unknown;
+    window_start: string;
+    window_end: string;
+    privacy_generation: string | null;
+    model_generation: string | null;
+  }>(
+    `SELECT status, checkpoint::text, metadata,
+            window_start::text, window_end::text,
+            privacy_generation::text, model_generation
+       FROM analysis_runs
+      WHERE workload = $1
+        AND scope = $2
+      ORDER BY started_at DESC
+      LIMIT 1`,
+    [workload, scope],
+  );
+  const row = result.rows[0];
+  if (!row || row.status !== 'timed_out' || Number(row.checkpoint) <= 0) return null;
+  return {
+    checkpoint: Number(row.checkpoint),
+    metadata: row.metadata,
+    windowStart: row.window_start,
+    windowEnd: row.window_end,
+    privacyGeneration: row.privacy_generation == null ? null : Number(row.privacy_generation),
+    modelGeneration: row.model_generation,
+  };
+}
+
 export type AnalysisRunHeartbeat = (() => Promise<void>) & {
  signal: AbortSignal;
  assertOwned: () => void;
@@ -92,7 +92,7 @@ function observerRegionFromTopic(topic: string): string | null {
 async function queryPool<T extends pg.QueryResultRow = pg.QueryResultRow>(
  targetPool: pg.Pool,
  poolName: 'oltp' | 'analytics',
-  text: string,
+  querySpec: string | pg.QueryConfig<unknown[]>,
  params?: unknown[],
  signal?: AbortSignal,
 ): Promise<pg.QueryResult<T>> {
@@ -100,7 +100,12 @@ async function queryPool<T extends pg.QueryResultRow = pg.QueryResultRow>(
  let outcome = 'success';
  try {
    updateDbPoolMetrics(poolName, targetPool);
-    if (!signal) return await targetPool.query<T>(text, params);
+    const execute = (target: pg.Pool | pg.PoolClient) => (
+      typeof querySpec === 'string'
+        ? target.query<T>(querySpec, params)
+        : target.query<T>(querySpec)
+    );
+    if (!signal) return await execute(targetPool);
    signal.throwIfAborted();
    const client = await targetPool.connect();
    let destroyed = false;
@@ -125,7 +130,7 @@ async function queryPool<T extends pg.QueryResultRow = pg.QueryResultRow>(
    try {
      signal.throwIfAborted();
      const result = await Promise.race([
-        client.query<T>(text, params),
+        execute(client),
        aborted,
      ]);
      signal.throwIfAborted();
@@ -153,6 +158,16 @@ export async function query<T extends pg.QueryResultRow = pg.QueryResultRow>(
  return queryPool<T>(pool, 'oltp', text, params, signal);
 }

+export async function namedQuery<T extends pg.QueryResultRow = pg.QueryResultRow>(
+  name: string,
+  text: string,
+  params?: unknown[],
+  signal?: AbortSignal,
+): Promise<pg.QueryResult<T>> {
+  if (!/^[a-z0-9_-]{1,63}$/.test(name)) throw new Error('INVALID_PREPARED_STATEMENT_NAME');
+  return queryPool<T>(pool, 'oltp', { name, text, values: params }, undefined, signal);
+}
+
 export async function analyticsQuery<T extends pg.QueryResultRow = pg.QueryResultRow>(
  text: string,
  params?: unknown[],
@@ -1144,32 +1159,43 @@ export async function getPublicVisibilityGeneration(signal?: AbortSignal): Promi
 }

 export async function getRecentPathHistoryPacketHashes(
-  hours = 1,
+  windowStart: Date,
+  windowEnd: Date,
  network?: string,
  limit = 1200,
  minPathHashSizeBytes = 1,
+  signal?: AbortSignal,
 ): Promise<string[]> {
  const normalizedMinPathHashSizeBytes = Number.isFinite(minPathHashSizeBytes)
    ? Math.max(1, Math.floor(minPathHashSizeBytes))
    : 1;
-  const scope = buildScopePlaceholders(4, network);
-  const params: unknown[] = [hours, limit, normalizedMinPathHashSizeBytes, ...scope.params];
-  const res = await pool.query<{ packet_hash: string }>(
+  const scope = buildScopePlaceholders(5, network);
+  const params: unknown[] = [
+    windowStart,
+    windowEnd,
+    limit,
+    normalizedMinPathHashSizeBytes,
+    ...scope.params,
+  ];
+  const res = await namedQuery<{ packet_hash: string }>(
+    `path-history-selection-${scope.networkIsMulti ? 'multi' : 'single'}-v1`,
    `SELECT packet_hash
     FROM (
       SELECT p.packet_hash, MAX(p.time) AS last_seen
       FROM packets p
-        WHERE p.time > NOW() - INTERVAL '1 hour' * $1
+        WHERE p.time > $1
+          AND p.time <= $2
          AND p.path_hashes IS NOT NULL
          AND cardinality(p.path_hashes) > 0
-          AND COALESCE(p.path_hash_size_bytes, 1) >= $3
+          AND COALESCE(p.path_hash_size_bytes, 1) >= $4
          ${buildPacketScopeClause(scope, 'p', network)}
          ${buildPublicPacketPrivacyClause('p')}
       GROUP BY p.packet_hash
     ) recent
     ORDER BY last_seen DESC
-     LIMIT $2`,
+     LIMIT $3`,
    params,
+    signal,
  );
  return res.rows.map((row) => row.packet_hash).filter(Boolean);
 }
@@ -1,5 +1,6 @@
 import {
  getPublicVisibilityGeneration,
+  namedQuery,
  query,
  touchNodesPredictedOnline,
 } from '../db/index.js';
@@ -211,12 +212,14 @@ async function loadContext(
  options?: {
    pinForBatch?: boolean;
    requiredVisibilityGeneration?: number;
+    currentVisibilityGeneration?: number;
    signal?: AbortSignal;
  },
 ): Promise<BetaResolveContext> {
  options?.signal?.throwIfAborted();
  const now = Date.now();
-  const visibilityGeneration = await getPublicVisibilityGeneration(options?.signal);
+  const visibilityGeneration = options?.currentVisibilityGeneration
+    ?? await getPublicVisibilityGeneration(options?.signal);
  if (options?.requiredVisibilityGeneration != null
      && visibilityGeneration !== options.requiredVisibilityGeneration) {
    throw new Error('PUBLIC_VISIBILITY_CHANGED_DURING_RESOLUTION');
@@ -317,7 +320,8 @@ async function loadContext(
    mlPrefixScores,
    learningModel,
  };
-  const confirmedGeneration = await getPublicVisibilityGeneration(options?.signal);
+  const confirmedGeneration = options?.currentVisibilityGeneration
+    ?? await getPublicVisibilityGeneration(options?.signal);
  if (confirmedGeneration !== visibilityGeneration) {
    if (visibilityRetry >= 1) throw new Error('PUBLIC_VISIBILITY_CHANGED_DURING_CONTEXT_LOAD');
    return loadContext(network, visibilityRetry + 1, options);
@@ -811,56 +815,17 @@ function buildRegionLinks(
  return links;
 }

-export async function resolveMultiObserverBetaPath(
+async function resolveMultiObserverBetaPathFromRows(
  packetHash: string,
  network: string,
+  rows: readonly PathPacket[],
+  context: BetaResolveContext,
  stickyMap?: Map<string, string>,
  stickyAgeFraction?: number,
  options?: PathResolutionOptions,
 ): Promise<MultiObserverResolvedPayload | null> {
-  throwIfResolutionAborted(options);
-  const allResult = await query<PathPacket>(
-    `SELECT packet_hash, rx_node_id, src_node_id, packet_type, hop_count, path_hashes, path_hash_size_bytes
-     FROM packets
-     WHERE packet_hash = $1
-       AND ($2 = 'all' OR network = $2)
-       AND time >= NOW() - ($3::int * INTERVAL '1 hour')
-       AND rx_node_id IS NOT NULL
-       AND NOT EXISTS (
-         SELECT 1 FROM nodes private_node
-         WHERE private_node.name LIKE '%🚫%'
-           AND ${privateNodePacketNetworkMatchSql('private_node', 'packets')}
-           AND (
-             private_node.node_id IN (packets.rx_node_id, packets.src_node_id)
-             OR EXISTS (
-               SELECT 1
-               FROM unnest(COALESCE(packets.path_hashes, ARRAY[]::text[])) AS path_hash
-               WHERE packets.path_hash_size_bytes BETWEEN 1 AND 3
-                 AND UPPER(private_node.node_id) LIKE UPPER(path_hash) || '%'
-             )
-           )
-       )
-     ORDER BY COALESCE(cardinality(path_hashes), 0) DESC,
-              COALESCE(path_hash_size_bytes, 0) DESC,
-              CASE WHEN src_node_id IS NOT NULL THEN 1 ELSE 0 END DESC,
-              hop_count ASC NULLS LAST,
-              time ASC
-     LIMIT $4`,
-    [packetHash, network, PATH_MULTI_HISTORY_WINDOW_HOURS, PATH_MULTI_MAX_SCAN_ROWS + 1],
-    options?.signal,
-  );
-  if (allResult.rows.length === 0) return null;
-  if (allResult.rows.length > PATH_MULTI_MAX_SCAN_ROWS) throw new Error('PATH_HISTORY_LIMIT');
-
-  const context = await loadContext(network, 0, {
-    pinForBatch: options?.pinContextForBatch,
-    requiredVisibilityGeneration: options?.requiredVisibilityGeneration,
-    signal: options?.signal,
-  });
-  throwIfResolutionAborted(options);
-
  const byObserver = new Map<string, PreparedPacketObservation>();
-  for (const row of allResult.rows) {
+  for (const row of rows) {
    if (!row.rx_node_id) continue;
    const prepared = preparePacketObservation(row, context.nodesById.get(row.rx_node_id) ?? null);
    if (prepared.ignoreForPathing) continue;
@@ -955,3 +920,178 @@ export async function resolveMultiObserverBetaPath(
    ...(Object.keys(stickyUpdates).length > 0 ? { stickyUpdates } : {}),
  };
 }
+
+export async function resolveMultiObserverBetaPath(
+  packetHash: string,
+  network: string,
+  stickyMap?: Map<string, string>,
+  stickyAgeFraction?: number,
+  options?: PathResolutionOptions,
+): Promise<MultiObserverResolvedPayload | null> {
+  throwIfResolutionAborted(options);
+  const allResult = await query<PathPacket>(
+    `SELECT packet_hash, rx_node_id, src_node_id, packet_type, hop_count, path_hashes, path_hash_size_bytes
+     FROM packets
+     WHERE packet_hash = $1
+       AND ($2 = 'all' OR network = $2)
+       AND time >= NOW() - ($3::int * INTERVAL '1 hour')
+       AND rx_node_id IS NOT NULL
+       AND NOT EXISTS (
+         SELECT 1 FROM nodes private_node
+         WHERE private_node.name LIKE '%🚫%'
+           AND ${privateNodePacketNetworkMatchSql('private_node', 'packets')}
+           AND (
+             private_node.node_id IN (packets.rx_node_id, packets.src_node_id)
+             OR EXISTS (
+               SELECT 1
+               FROM unnest(COALESCE(packets.path_hashes, ARRAY[]::text[])) AS path_hash
+               WHERE packets.path_hash_size_bytes BETWEEN 1 AND 3
+                 AND UPPER(private_node.node_id) LIKE UPPER(path_hash) || '%'
+             )
+           )
+       )
+     ORDER BY COALESCE(cardinality(path_hashes), 0) DESC,
+              COALESCE(path_hash_size_bytes, 0) DESC,
+              CASE WHEN src_node_id IS NOT NULL THEN 1 ELSE 0 END DESC,
+              hop_count ASC NULLS LAST,
+              time ASC
+     LIMIT $4`,
+    [packetHash, network, PATH_MULTI_HISTORY_WINDOW_HOURS, PATH_MULTI_MAX_SCAN_ROWS + 1],
+    options?.signal,
+  );
+  if (allResult.rows.length === 0) return null;
+  if (allResult.rows.length > PATH_MULTI_MAX_SCAN_ROWS) throw new Error('PATH_HISTORY_LIMIT');
+
+  const context = await loadContext(network, 0, {
+    pinForBatch: options?.pinContextForBatch,
+    requiredVisibilityGeneration: options?.requiredVisibilityGeneration,
+    signal: options?.signal,
+  });
+  throwIfResolutionAborted(options);
+  return resolveMultiObserverBetaPathFromRows(
+    packetHash,
+    network,
+    allResult.rows,
+    context,
+    stickyMap,
+    stickyAgeFraction,
+    options,
+  );
+}
+
+export type MultiObserverPathBatch = {
+  results: Map<string, MultiObserverResolvedPayload | null>;
+  limitedPacketHashes: Set<string>;
+};
+
+export async function createMultiObserverPathBatchResolver(
+  network: string,
+  visibilityGeneration: number,
+  signal?: AbortSignal,
+): Promise<{
+  resolveBatch: (
+    packetHashes: readonly string[],
+    windowStart: Date,
+    windowEnd: Date,
+    signal?: AbortSignal,
+  ) => Promise<MultiObserverPathBatch>;
+}> {
+  const context = await loadContext(network, 0, {
+    pinForBatch: true,
+    requiredVisibilityGeneration: visibilityGeneration,
+    currentVisibilityGeneration: visibilityGeneration,
+    signal,
+  });
+
+  return {
+    resolveBatch: async (packetHashes, windowStart, windowEnd, batchSignal) => {
+      batchSignal?.throwIfAborted();
+      const results = new Map<string, MultiObserverResolvedPayload | null>();
+      const limitedPacketHashes = new Set<string>();
+      if (packetHashes.length === 0) return { results, limitedPacketHashes };
+
+      const packetResult = await namedQuery<PathPacket>(
+        'path-history-observations-v1',
+        `SELECT observation.packet_hash, observation.rx_node_id, observation.src_node_id,
+                observation.packet_type, observation.hop_count, observation.path_hashes,
+                observation.path_hash_size_bytes
+           FROM unnest($1::text[]) WITH ORDINALITY AS requested(packet_hash, selection_order)
+           CROSS JOIN LATERAL (
+             SELECT p.packet_hash, p.rx_node_id, p.src_node_id, p.packet_type,
+                    p.hop_count, p.path_hashes, p.path_hash_size_bytes, p.time
+               FROM packets p
+              WHERE p.packet_hash = requested.packet_hash
+                AND ($2 = 'all' OR p.network = $2)
+                AND p.time >= $3
+                AND p.time <= $4
+                AND p.rx_node_id IS NOT NULL
+                AND NOT EXISTS (
+                  SELECT 1 FROM nodes private_node
+                   WHERE private_node.name LIKE '%🚫%'
+                     AND ${privateNodePacketNetworkMatchSql('private_node', 'p')}
+                     AND (
+                       private_node.node_id IN (p.rx_node_id, p.src_node_id)
+                       OR EXISTS (
+                         SELECT 1
+                           FROM unnest(COALESCE(p.path_hashes, ARRAY[]::text[])) AS path_hash
+                          WHERE p.path_hash_size_bytes BETWEEN 1 AND 3
+                            AND UPPER(private_node.node_id) LIKE UPPER(path_hash) || '%'
+                       )
+                     )
+                )
+              ORDER BY COALESCE(cardinality(p.path_hashes), 0) DESC,
+                       COALESCE(p.path_hash_size_bytes, 0) DESC,
+                       CASE WHEN p.src_node_id IS NOT NULL THEN 1 ELSE 0 END DESC,
+                       p.hop_count ASC NULLS LAST,
+                       p.time ASC
+              LIMIT $5
+           ) observation
+          ORDER BY requested.selection_order,
+                   COALESCE(cardinality(observation.path_hashes), 0) DESC,
+                   COALESCE(observation.path_hash_size_bytes, 0) DESC,
+                   CASE WHEN observation.src_node_id IS NOT NULL THEN 1 ELSE 0 END DESC,
+                   observation.hop_count ASC NULLS LAST,
+                   observation.time ASC`,
+        [packetHashes, network, windowStart, windowEnd, PATH_MULTI_MAX_SCAN_ROWS + 1],
+        batchSignal,
+      );
+      batchSignal?.throwIfAborted();
+
+      const rowsByHash = new Map<string, PathPacket[]>();
+      for (const row of packetResult.rows) {
+        const rows = rowsByHash.get(row.packet_hash) ?? [];
+        rows.push(row);
+        rowsByHash.set(row.packet_hash, rows);
+      }
+      for (const packetHash of packetHashes) {
+        batchSignal?.throwIfAborted();
+        const rows = rowsByHash.get(packetHash) ?? [];
+        if (rows.length > PATH_MULTI_MAX_SCAN_ROWS) {
+          limitedPacketHashes.add(packetHash);
+          continue;
+        }
+        try {
+          results.set(packetHash, await resolveMultiObserverBetaPathFromRows(
+            packetHash,
+            network,
+            rows,
+            context,
+            undefined,
+            undefined,
+            {
+              touchPredictedOnline: false,
+              log: false,
+              pinContextForBatch: true,
+              requiredVisibilityGeneration: visibilityGeneration,
+              signal: batchSignal,
+            },
+          ));
+        } catch (error) {
+          if ((error as Error).message !== 'PATH_HISTORY_LIMIT') throw error;
+          limitedPacketHashes.add(packetHash);
+        }
+      }
+      return { results, limitedPacketHashes };
+    },
+  };
+}
@@ -7,14 +7,19 @@ import {
  upsertPathHistoryCache,
  type PathHistorySegmentRow,
 } from '../db/index.js';
-import { resolveMultiObserverBetaPath, type BetaResolvedPayload } from '../path-beta/resolver.js';
-import { runBoundedItems } from '../analysis/boundedRun.js';
+import {
+  createMultiObserverPathBatchResolver,
+  type BetaResolvedPayload,
+} from '../path-beta/resolver.js';
 import { BoundedSegmentCounter } from '../analysis/boundedSegmentCounter.js';
 import {
+  AnalysisRunDeadlineExceededError,
  AnalysisRunAlreadyActiveError,
  analysisGeneration,
  beginAnalysisRun,
+  checkpointAnalysisRun,
  finishAnalysisRun,
+  getLatestTimedOutAnalysisCheckpoint,
  startAnalysisRunHeartbeat,
 } from '../analysis/runState.js';
 import { observeWorkerOutcome } from '../metrics.js';
@@ -23,6 +28,13 @@ import {
  pathHistoryNextDelayMs,
  pathHistoryRetryIntervalMs,
 } from './pathHistorySchedule.js';
+import {
+  PATH_HISTORY_MODEL_GENERATION,
+  pathHistoryBatches,
+  pathHistorySelectionIdentity,
+  resumablePathHistoryCheckpoint,
+  type PathHistoryCheckpoint,
+} from './pathHistoryCheckpoint.js';

 const RETRY_INTERVAL_MS = pathHistoryRetryIntervalMs(process.env['PATH_HISTORY_RETRY_INTERVAL_MS']);
 const WINDOW_HOURS = 168;
@@ -41,9 +53,9 @@ const RUN_DEADLINE_MS = Math.max(
  60_000,
  Number(process.env['PATH_HISTORY_RUN_DEADLINE_MS'] ?? 120 * 60_000) || 120 * 60_000,
 );
-const CONCURRENCY = Math.max(
-  1,
-  Math.min(16, Math.trunc(Number(process.env['PATH_HISTORY_CONCURRENCY'] ?? 2) || 2)),
+const BATCH_SIZE = Math.max(
+  8,
+  Math.min(512, Math.trunc(Number(process.env['PATH_HISTORY_BATCH_SIZE'] ?? 64) || 64)),
 );
 const SCOPES = ['ukmesh', 'test'] as const;

@@ -100,20 +112,103 @@ function collectPurpleSegments(result: BetaResolvedPayload, sink: Set<string>):
  }
 }

-async function refreshScope(scope: ScopeName): Promise<'finished' | 'active-run'> {
+async function refreshScope(scope: ScopeName): Promise<'finished' | 'active-run' | 'retry'> {
  const visibilityGeneration = await getPublicVisibilityGeneration();
-  const packetHashes = await getRecentPathHistoryPacketHashes(
-    WINDOW_HOURS,
+  const latestTimedOut = await getLatestTimedOutAnalysisCheckpoint('path-history', scope);
+  const parsedCheckpointCandidate = resumablePathHistoryCheckpoint(latestTimedOut?.metadata, {
+    scope,
+    modelGeneration: PATH_HISTORY_MODEL_GENERATION,
+    privacyGeneration: visibilityGeneration,
+  });
+  const checkpointCandidate = parsedCheckpointCandidate
+    && latestTimedOut
+    && Date.parse(latestTimedOut?.windowStart ?? '') === Date.parse(parsedCheckpointCandidate.windowStart)
+    && Date.parse(latestTimedOut?.windowEnd ?? '') === Date.parse(parsedCheckpointCandidate.windowEnd)
+    && latestTimedOut.modelGeneration === parsedCheckpointCandidate.modelGeneration
+    && latestTimedOut.privacyGeneration === parsedCheckpointCandidate.privacyGeneration
+    && latestTimedOut.checkpoint === parsedCheckpointCandidate.nextIndex
+    ? parsedCheckpointCandidate
+    : null;
+
+  let windowEnd = checkpointCandidate
+    ? new Date(checkpointCandidate.windowEnd)
+    : new Date();
+  let windowStart = checkpointCandidate
+    ? new Date(checkpointCandidate.windowStart)
+    : new Date(windowEnd.getTime() - WINDOW_HOURS * 60 * 60 * 1000);
+  let packetHashes = await getRecentPathHistoryPacketHashes(
+    windowStart,
+    windowEnd,
    scope,
    MAX_PACKET_HASHES,
    MIN_HISTORY_PATH_HASH_BYTES,
  );
-  const counts = new BoundedSegmentCounter(SEGMENT_COUNTER_CAPACITY);
-  let resolvedPacketCount = 0;
-  let skippedPacketCount = 0;
+  let selectionIdentity = pathHistorySelectionIdentity({
+    scope,
+    windowStart,
+    windowEnd,
+    modelGeneration: PATH_HISTORY_MODEL_GENERATION,
+    privacyGeneration: visibilityGeneration,
+    packetHashes,
+  });
+  let resume = resumablePathHistoryCheckpoint(latestTimedOut?.metadata, {
+    scope,
+    modelGeneration: PATH_HISTORY_MODEL_GENERATION,
+    privacyGeneration: visibilityGeneration,
+    selectionIdentity,
+    packetCount: packetHashes.length,
+  });
+  let counts: BoundedSegmentCounter;
+  try {
+    counts = resume
+      ? BoundedSegmentCounter.fromSnapshot(SEGMENT_COUNTER_CAPACITY, resume.segmentCounter)
+      : new BoundedSegmentCounter(SEGMENT_COUNTER_CAPACITY);
+  } catch {
+    resume = null;
+    counts = new BoundedSegmentCounter(SEGMENT_COUNTER_CAPACITY);
+  }

-  const windowEnd = new Date();
-  const windowStart = new Date(windowEnd.getTime() - WINDOW_HOURS * 60 * 60 * 1000);
+  if (checkpointCandidate && !resume) {
+    // Late-arriving rows, a capacity change, or damaged metadata invalidates the
+    // old fixed selection. Start a fresh current window rather than mixing it
+    // with partial state from another generation.
+    windowEnd = new Date();
+    windowStart = new Date(windowEnd.getTime() - WINDOW_HOURS * 60 * 60 * 1000);
+    packetHashes = await getRecentPathHistoryPacketHashes(
+      windowStart,
+      windowEnd,
+      scope,
+      MAX_PACKET_HASHES,
+      MIN_HISTORY_PATH_HASH_BYTES,
+    );
+    selectionIdentity = pathHistorySelectionIdentity({
+      scope,
+      windowStart,
+      windowEnd,
+      modelGeneration: PATH_HISTORY_MODEL_GENERATION,
+      privacyGeneration: visibilityGeneration,
+      packetHashes,
+    });
+    counts = new BoundedSegmentCounter(SEGMENT_COUNTER_CAPACITY);
+  }
+  let nextIndex = resume?.nextIndex ?? 0;
+  let resolvedPacketCount = resume?.resolvedPacketCount ?? 0;
+  let skippedPacketCount = resume?.skippedPacketCount ?? 0;
+
+  const makeCheckpoint = (): PathHistoryCheckpoint => ({
+    version: 1,
+    scope,
+    windowStart: windowStart.toISOString(),
+    windowEnd: windowEnd.toISOString(),
+    modelGeneration: PATH_HISTORY_MODEL_GENERATION,
+    privacyGeneration: visibilityGeneration,
+    selectionIdentity,
+    packetCount: packetHashes.length,
+    nextIndex,
+    resolvedPacketCount,
+    skippedPacketCount,
+    segmentCounter: counts.snapshot(),
+  });
  let run;
  try {
    run = await beginAnalysisRun({
@@ -124,7 +219,7 @@ async function refreshScope(scope: ScopeName): Promise<'finished' | 'active-run'
      totalItems: packetHashes.length,
      deadlineMs: RUN_DEADLINE_MS,
      privacyGeneration: visibilityGeneration,
-      modelGeneration: 'path-history-v2',
+      modelGeneration: PATH_HISTORY_MODEL_GENERATION,
    });
  } catch (error) {
    if (error instanceof AnalysisRunAlreadyActiveError || (error as { code?: string }).code === '55P03') {
@@ -151,20 +246,51 @@ async function refreshScope(scope: ScopeName): Promise<'finished' | 'active-run'
      console.warn(`[path-history] scope=${scope} selected no packets; preserving last complete snapshot`);
      return 'finished';
    }
-    const outcome = await runBoundedItems(packetHashes, async (packetHash, _index, signal) => {
+    const resolver = await createMultiObserverPathBatchResolver(
+      scope,
+      visibilityGeneration,
+      stopHeartbeat.signal,
+    );
+    if (resume) {
+      console.log(
+        `[path-history] scope=${scope} resuming checkpoint=${nextIndex}/${packetHashes.length} `
+          + `window=${windowStart.toISOString()}..${windowEnd.toISOString()}`,
+      );
+    }
+    for (const batch of pathHistoryBatches(packetHashes, nextIndex, BATCH_SIZE)) {
      stopHeartbeat.assertOwned();
-      signal.throwIfAborted();
-      try {
-        const resolved = await resolveMultiObserverBetaPath(packetHash, scope, undefined, undefined, {
-          touchPredictedOnline: false,
-          log: false,
-          pinContextForBatch: true,
-          requiredVisibilityGeneration: visibilityGeneration,
-          signal,
+      stopHeartbeat.signal.throwIfAborted();
+      // Exactly one visibility read fences each observation batch. The context
+      // was preloaded against the same generation and publication rechecks it.
+      const batchVisibilityGeneration = await getPublicVisibilityGeneration(stopHeartbeat.signal);
+      if (batchVisibilityGeneration !== visibilityGeneration) {
+        await finish({
+          status: 'stale',
+          checkpoint: nextIndex,
+          error: 'public visibility changed during generation',
+          metadata: { pathHistoryCheckpoint: makeCheckpoint() },
        });
+        console.warn(
+          `[path-history] scope=${scope} visibility changed during run; preserving the current snapshot`,
+        );
+        return 'finished';
+      }
+      const resolvedBatch = await resolver.resolveBatch(
+        batch.items,
+        windowStart,
+        windowEnd,
+        stopHeartbeat.signal,
+      );
+      stopHeartbeat.assertOwned();
+      for (const packetHash of batch.items) {
+        if (resolvedBatch.limitedPacketHashes.has(packetHash)) {
+          skippedPacketCount += 1;
+          continue;
+        }
+        const resolved = resolvedBatch.results.get(packetHash) ?? null;
        stopHeartbeat.assertOwned();
        if (!resolved?.ok || resolved.results.length < 1) {
-          return;
+          continue;
        }
        const packetSegments = new Set<string>();
        for (const result of resolved.results) collectPurpleSegments(result, packetSegments);
@@ -172,39 +298,14 @@ async function refreshScope(scope: ScopeName): Promise<'finished' | 'active-run'
          resolvedPacketCount += 1;
          for (const key of packetSegments) counts.observe(key);
        }
-        return;
-      } catch (error) {
-        if ((error as Error).message === 'PATH_HISTORY_LIMIT') {
-          skippedPacketCount += 1;
-          return;
-        }
-        throw error;
      }
-    }, {
-      windowStart,
-      windowEnd,
-      deadlineMs: RUN_DEADLINE_MS,
-      concurrency: CONCURRENCY,
-      collectResults: false,
-      maxErrors: 100,
-      runId: run.runId,
-      signal: stopHeartbeat.signal,
-    });
-    if (outcome.status !== 'complete') {
-      await finish({
-        status: outcome.status,
-        checkpoint: outcome.checkpoint,
-        error: outcome.errors[0]?.message,
-        metadata: { errors: outcome.errors.slice(0, 20) },
-      });
-      console.error('[path-history] incomplete generation; preserving last complete snapshot', {
-        scope,
-        runId: outcome.runId,
-        status: outcome.status,
-        checkpoint: outcome.checkpoint,
-        errors: outcome.errors.slice(0, 5),
-      });
-      return 'finished';
+      nextIndex = batch.endIndex;
+      await checkpointAnalysisRun(
+        run,
+        nextIndex,
+        { pathHistoryCheckpoint: makeCheckpoint() },
+        stopHeartbeat.signal,
+      );
    }

    const segmentCounts: SegmentCount[] = counts.candidates(MIN_SEGMENT_COUNT)
@@ -214,6 +315,20 @@ async function refreshScope(scope: ScopeName): Promise<'finished' | 'active-run'
      }))
      .slice(0, MAX_SEGMENTS);

+    const publicationVisibilityGeneration = await getPublicVisibilityGeneration();
+    if (publicationVisibilityGeneration !== visibilityGeneration) {
+      await finish({
+        status: 'stale',
+        checkpoint: nextIndex,
+        error: 'public visibility changed before publication',
+        metadata: { pathHistoryCheckpoint: makeCheckpoint() },
+      });
+      console.warn(
+        `[path-history] scope=${scope} visibility changed before publication; preserving the current snapshot`,
+      );
+      return 'finished';
+    }
+
    const published = await upsertPathHistoryCache({
      scope,
      windowStart,
@@ -226,7 +341,7 @@ async function refreshScope(scope: ScopeName): Promise<'finished' | 'active-run'
    if (!published) {
      await finish({
        status: 'stale',
-        checkpoint: outcome.checkpoint,
+        checkpoint: nextIndex,
        error: 'public visibility changed during generation',
        metadata: { visibilityGeneration },
      });
@@ -242,7 +357,7 @@ async function refreshScope(scope: ScopeName): Promise<'finished' | 'active-run'
    });
    await finish({
      status: 'complete',
-      checkpoint: outcome.checkpoint,
+      checkpoint: nextIndex,
      generation,
      metadata: {
        skippedPacketCount,
@@ -254,19 +369,26 @@ async function refreshScope(scope: ScopeName): Promise<'finished' | 'active-run'
    });

    console.log(
-      `[path-history] scope=${scope} run=${outcome.runId} packets=${packetHashes.length} skipped=${skippedPacketCount} resolved=${resolvedPacketCount} segments=${segmentCounts.length}`,
+      `[path-history] scope=${scope} run=${run.runId} packets=${packetHashes.length} skipped=${skippedPacketCount} resolved=${resolvedPacketCount} segments=${segmentCounts.length}`,
    );
    return 'finished';
  } catch (error) {
    try {
-      const timedOut = Date.now() >= run.deadlineAt.getTime();
-      if (timedOut) await stopHeartbeat.stopForTerminal();
-      else await stopHeartbeat();
+      const timedOut = Date.now() >= run.deadlineAt.getTime()
+        || error instanceof AnalysisRunDeadlineExceededError
+        || (error as Error).message === 'analysis run deadline exceeded';
      await finish({
        status: timedOut ? 'timed_out' : 'failed',
-        checkpoint: 0,
+        checkpoint: nextIndex,
        error: error instanceof Error ? error.message : String(error),
+        metadata: { pathHistoryCheckpoint: makeCheckpoint() },
      });
+      if (timedOut) {
+        console.warn(
+          `[path-history] scope=${scope} timed out at checkpoint=${nextIndex}/${packetHashes.length}; retry will resume`,
+        );
+        return 'retry';
+      }
    } catch (finishError) {
      console.error('[path-history] could not record failed run', (finishError as Error).message);
    }
@@ -299,7 +421,8 @@ async function refreshAll(tag: 'initial' | 'scheduled'): Promise<boolean> {
    }
    let retrySoon = false;
    for (const scope of SCOPES) {
-      if (await refreshScope(scope) === 'active-run') retrySoon = true;
+      const result = await refreshScope(scope);
+      if (result === 'active-run' || result === 'retry') retrySoon = true;
    }
    observeWorkerOutcome('path_history', 'refresh', 'success');
    return retrySoon;
@@ -315,10 +438,12 @@ async function refreshAll(tag: 'initial' | 'scheduled'): Promise<boolean> {
 async function main() {
  startWorkerMetrics();
  await initDb();
+  let consecutiveRetries = 0;
  const scheduleNext = (retrySoon: boolean) => {
+    consecutiveRetries = retrySoon ? consecutiveRetries + 1 : 0;
    setTimeout(() => {
      void refreshAll('scheduled').then(scheduleNext);
-    }, pathHistoryNextDelayMs(retrySoon, RETRY_INTERVAL_MS));
+    }, pathHistoryNextDelayMs(retrySoon, RETRY_INTERVAL_MS, consecutiveRetries));
  };
  scheduleNext(await refreshAll('initial'));
 }
@@ -0,0 +1,76 @@
+import assert from 'node:assert/strict';
+import test from 'node:test';
+import { BoundedSegmentCounter } from '../analysis/boundedSegmentCounter.js';
+import {
+  PATH_HISTORY_MODEL_GENERATION,
+  pathHistoryBatches,
+  pathHistorySelectionIdentity,
+  resumablePathHistoryCheckpoint,
+  type PathHistoryCheckpoint,
+} from './pathHistoryCheckpoint.js';
+
+test('path history batching preserves selection order and resumes at a batch boundary', () => {
+  const hashes = ['h0', 'h1', 'h2', 'h3', 'h4', 'h5', 'h6'];
+  assert.deepEqual(pathHistoryBatches(hashes, 0, 3), [
+    { startIndex: 0, endIndex: 3, items: ['h0', 'h1', 'h2'] },
+    { startIndex: 3, endIndex: 6, items: ['h3', 'h4', 'h5'] },
+    { startIndex: 6, endIndex: 7, items: ['h6'] },
+  ]);
+  assert.deepEqual(pathHistoryBatches(hashes, 3, 3).flatMap((batch) => batch.items), hashes.slice(3));
+});
+
+test('path history checkpoint resumes only for the identical fixed selection', () => {
+  const windowStart = new Date('2026-08-01T00:00:00.000Z');
+  const windowEnd = new Date('2026-08-08T00:00:00.000Z');
+  const packetHashes = ['b', 'a', 'c'];
+  const selectionIdentity = pathHistorySelectionIdentity({
+    scope: 'ukmesh',
+    windowStart,
+    windowEnd,
+    modelGeneration: PATH_HISTORY_MODEL_GENERATION,
+    privacyGeneration: 7,
+    packetHashes,
+  });
+  const counter = new BoundedSegmentCounter(8);
+  counter.observe('1,2|3,4');
+  const checkpoint: PathHistoryCheckpoint = {
+    version: 1,
+    scope: 'ukmesh',
+    windowStart: windowStart.toISOString(),
+    windowEnd: windowEnd.toISOString(),
+    modelGeneration: PATH_HISTORY_MODEL_GENERATION,
+    privacyGeneration: 7,
+    selectionIdentity,
+    packetCount: 3,
+    nextIndex: 2,
+    resolvedPacketCount: 1,
+    skippedPacketCount: 0,
+    segmentCounter: counter.snapshot(),
+  };
+  const metadata = { pathHistoryCheckpoint: checkpoint };
+  assert.deepEqual(resumablePathHistoryCheckpoint(metadata, {
+    scope: 'ukmesh',
+    modelGeneration: PATH_HISTORY_MODEL_GENERATION,
+    privacyGeneration: 7,
+    selectionIdentity,
+    packetCount: 3,
+  }), checkpoint);
+  assert.equal(resumablePathHistoryCheckpoint(metadata, {
+    scope: 'ukmesh',
+    modelGeneration: PATH_HISTORY_MODEL_GENERATION,
+    privacyGeneration: 8,
+  }), null);
+  assert.equal(resumablePathHistoryCheckpoint(metadata, {
+    scope: 'ukmesh',
+    modelGeneration: PATH_HISTORY_MODEL_GENERATION,
+    privacyGeneration: 7,
+    selectionIdentity: pathHistorySelectionIdentity({
+      scope: 'ukmesh',
+      windowStart,
+      windowEnd,
+      modelGeneration: PATH_HISTORY_MODEL_GENERATION,
+      privacyGeneration: 7,
+      packetHashes: ['a', 'b', 'c'],
+    }),
+  }), null);
+});
@@ -0,0 +1,104 @@
+import { createHash } from 'node:crypto';
+import type { BoundedSegmentCounterSnapshot } from '../analysis/boundedSegmentCounter.js';
+
+export const PATH_HISTORY_MODEL_GENERATION = 'path-history-v2';
+
+export type PathHistoryCheckpoint = {
+  version: 1;
+  scope: string;
+  windowStart: string;
+  windowEnd: string;
+  modelGeneration: string;
+  privacyGeneration: number;
+  selectionIdentity: string;
+  packetCount: number;
+  nextIndex: number;
+  resolvedPacketCount: number;
+  skippedPacketCount: number;
+  segmentCounter: BoundedSegmentCounterSnapshot;
+};
+
+export function pathHistorySelectionIdentity(input: {
+  scope: string;
+  windowStart: Date;
+  windowEnd: Date;
+  modelGeneration: string;
+  privacyGeneration: number;
+  packetHashes: readonly string[];
+}): string {
+  return createHash('sha256').update(JSON.stringify({
+    scope: input.scope,
+    windowStart: input.windowStart.toISOString(),
+    windowEnd: input.windowEnd.toISOString(),
+    modelGeneration: input.modelGeneration,
+    privacyGeneration: input.privacyGeneration,
+    packetHashes: input.packetHashes,
+  })).digest('hex');
+}
+
+function integerAtLeast(value: unknown, minimum: number): value is number {
+  return Number.isSafeInteger(value) && Number(value) >= minimum;
+}
+
+export function resumablePathHistoryCheckpoint(
+  metadata: unknown,
+  expected: {
+    scope: string;
+    modelGeneration: string;
+    privacyGeneration: number;
+    selectionIdentity?: string;
+    packetCount?: number;
+  },
+): PathHistoryCheckpoint | null {
+  if (!metadata || typeof metadata !== 'object') return null;
+  const value = (metadata as { pathHistoryCheckpoint?: unknown }).pathHistoryCheckpoint;
+  if (!value || typeof value !== 'object') return null;
+  const checkpoint = value as Partial<PathHistoryCheckpoint>;
+  if (
+    checkpoint.version !== 1
+    || checkpoint.scope !== expected.scope
+    || checkpoint.modelGeneration !== expected.modelGeneration
+    || checkpoint.privacyGeneration !== expected.privacyGeneration
+    || typeof checkpoint.windowStart !== 'string'
+    || !Number.isFinite(Date.parse(checkpoint.windowStart))
+    || typeof checkpoint.windowEnd !== 'string'
+    || !Number.isFinite(Date.parse(checkpoint.windowEnd))
+    || Date.parse(checkpoint.windowStart) > Date.parse(checkpoint.windowEnd)
+    || typeof checkpoint.selectionIdentity !== 'string'
+    || !/^[a-f0-9]{64}$/.test(checkpoint.selectionIdentity)
+    || !integerAtLeast(checkpoint.packetCount, 1)
+    || !integerAtLeast(checkpoint.nextIndex, 0)
+    || checkpoint.nextIndex > checkpoint.packetCount
+    || !integerAtLeast(checkpoint.resolvedPacketCount, 0)
+    || checkpoint.resolvedPacketCount > checkpoint.nextIndex
+    || !integerAtLeast(checkpoint.skippedPacketCount, 0)
+    || checkpoint.skippedPacketCount > checkpoint.nextIndex
+    || !checkpoint.segmentCounter
+    || typeof checkpoint.segmentCounter !== 'object'
+  ) return null;
+  if (
+    expected.selectionIdentity !== undefined
+    && checkpoint.selectionIdentity !== expected.selectionIdentity
+  ) return null;
+  if (expected.packetCount !== undefined && checkpoint.packetCount !== expected.packetCount) return null;
+  return checkpoint as PathHistoryCheckpoint;
+}
+
+export function pathHistoryBatches<T>(
+  items: readonly T[],
+  startIndex: number,
+  batchSize: number,
+): Array<{ startIndex: number; endIndex: number; items: T[] }> {
+  if (!Number.isSafeInteger(startIndex) || startIndex < 0 || startIndex > items.length) {
+    throw new Error('INVALID_PATH_HISTORY_BATCH_START');
+  }
+  if (!Number.isSafeInteger(batchSize) || batchSize < 1) {
+    throw new Error('INVALID_PATH_HISTORY_BATCH_SIZE');
+  }
+  const batches: Array<{ startIndex: number; endIndex: number; items: T[] }> = [];
+  for (let index = startIndex; index < items.length; index += batchSize) {
+    const endIndex = Math.min(items.length, index + batchSize);
+    batches.push({ startIndex: index, endIndex, items: items.slice(index, endIndex) });
+  }
+  return batches;
+}
@@ -2,14 +2,20 @@ import assert from 'node:assert/strict';
 import test from 'node:test';
 import {
  PATH_HISTORY_REFRESH_INTERVAL_MS,
+  pathHistoryRetryDelayMs,
  pathHistoryNextDelayMs,
  pathHistoryRetryIntervalMs,
 } from './pathHistorySchedule.js';

-test('path history retries active leases promptly with bounded configuration', () => {
-  assert.equal(pathHistoryRetryIntervalMs(undefined), 60_000);
-  assert.equal(pathHistoryRetryIntervalMs('1000'), 10_000);
-  assert.equal(pathHistoryRetryIntervalMs('900000'), 300_000);
-  assert.equal(pathHistoryNextDelayMs(true, 45_000), 45_000);
-  assert.equal(pathHistoryNextDelayMs(false, 45_000), PATH_HISTORY_REFRESH_INTERVAL_MS);
+test('path history retries with bounded exponential backoff', () => {
+  assert.equal(pathHistoryRetryIntervalMs(undefined), 300_000);
+  assert.equal(pathHistoryRetryIntervalMs('1000'), 300_000);
+  assert.equal(pathHistoryRetryIntervalMs('900000'), 900_000);
+  assert.equal(pathHistoryRetryDelayMs(1, 300_000), 300_000);
+  assert.equal(pathHistoryRetryDelayMs(2, 300_000), 600_000);
+  assert.equal(pathHistoryRetryDelayMs(3, 300_000), 1_200_000);
+  assert.equal(pathHistoryRetryDelayMs(4, 300_000), 1_800_000);
+  assert.equal(pathHistoryRetryDelayMs(10, 300_000), 1_800_000);
+  assert.equal(pathHistoryNextDelayMs(true, 300_000, 2), 600_000);
+  assert.equal(pathHistoryNextDelayMs(false, 300_000, 2), PATH_HISTORY_REFRESH_INTERVAL_MS);
 });
@@ -1,10 +1,22 @@
 export const PATH_HISTORY_REFRESH_INTERVAL_MS = 60 * 60 * 1_000;
+export const PATH_HISTORY_MAX_RETRY_INTERVAL_MS = 30 * 60 * 1_000;

 export function pathHistoryRetryIntervalMs(rawValue: string | undefined): number {
-  const parsed = Number(rawValue ?? 60_000) || 60_000;
-  return Math.min(5 * 60_000, Math.max(10_000, parsed));
+  const parsed = Number(rawValue ?? 5 * 60_000) || 5 * 60_000;
+  return Math.min(PATH_HISTORY_MAX_RETRY_INTERVAL_MS, Math.max(5 * 60_000, parsed));
 }

-export function pathHistoryNextDelayMs(retrySoon: boolean, retryIntervalMs: number): number {
-  return retrySoon ? retryIntervalMs : PATH_HISTORY_REFRESH_INTERVAL_MS;
+export function pathHistoryRetryDelayMs(retryAttempt: number, retryIntervalMs: number): number {
+  const exponent = Math.max(0, Math.min(10, Math.trunc(retryAttempt) - 1));
+  return Math.min(PATH_HISTORY_MAX_RETRY_INTERVAL_MS, retryIntervalMs * (2 ** exponent));
+}
+
+export function pathHistoryNextDelayMs(
+  retrySoon: boolean,
+  retryIntervalMs: number,
+  retryAttempt = 1,
+): number {
+  return retrySoon
+    ? pathHistoryRetryDelayMs(retryAttempt, retryIntervalMs)
+    : PATH_HISTORY_REFRESH_INTERVAL_MS;
 }