diff --git a/backend/src/analysis/boundedSegmentCounter.test.ts b/backend/src/analysis/boundedSegmentCounter.test.ts index da56ed9..7f021b1 100644 --- a/backend/src/analysis/boundedSegmentCounter.test.ts +++ b/backend/src/analysis/boundedSegmentCounter.test.ts @@ -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/, + ); +}); diff --git a/backend/src/analysis/boundedSegmentCounter.ts b/backend/src/analysis/boundedSegmentCounter.ts index 13bd6a3..d167156 100644 --- a/backend/src/analysis/boundedSegmentCounter.ts +++ b/backend/src/analysis/boundedSegmentCounter.ts @@ -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(); + 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); } diff --git a/backend/src/analysis/runState.ts b/backend/src/analysis/runState.ts index a706c83..391a4b8 100644 --- a/backend/src/analysis/runState.ts +++ b/backend/src/analysis/runState.ts @@ -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 { + 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) & { signal: AbortSignal; assertOwned: () => void; diff --git a/backend/src/db/index.ts b/backend/src/db/index.ts index 19cb33e..5397c71 100644 --- a/backend/src/db/index.ts +++ b/backend/src/db/index.ts @@ -92,7 +92,7 @@ function observerRegionFromTopic(topic: string): string | null { async function queryPool( targetPool: pg.Pool, poolName: 'oltp' | 'analytics', - text: string, + querySpec: string | pg.QueryConfig, params?: unknown[], signal?: AbortSignal, ): Promise> { @@ -100,7 +100,12 @@ async function queryPool( let outcome = 'success'; try { updateDbPoolMetrics(poolName, targetPool); - if (!signal) return await targetPool.query(text, params); + const execute = (target: pg.Pool | pg.PoolClient) => ( + typeof querySpec === 'string' + ? target.query(querySpec, params) + : target.query(querySpec) + ); + if (!signal) return await execute(targetPool); signal.throwIfAborted(); const client = await targetPool.connect(); let destroyed = false; @@ -125,7 +130,7 @@ async function queryPool( try { signal.throwIfAborted(); const result = await Promise.race([ - client.query(text, params), + execute(client), aborted, ]); signal.throwIfAborted(); @@ -153,6 +158,16 @@ export async function query( return queryPool(pool, 'oltp', text, params, signal); } +export async function namedQuery( + name: string, + text: string, + params?: unknown[], + signal?: AbortSignal, +): Promise> { + if (!/^[a-z0-9_-]{1,63}$/.test(name)) throw new Error('INVALID_PREPARED_STATEMENT_NAME'); + return queryPool(pool, 'oltp', { name, text, values: params }, undefined, signal); +} + export async function analyticsQuery( 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 { 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); } diff --git a/backend/src/path-beta/resolver.ts b/backend/src/path-beta/resolver.ts index facb0b8..1c5286d 100644 --- a/backend/src/path-beta/resolver.ts +++ b/backend/src/path-beta/resolver.ts @@ -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 { 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, stickyAgeFraction?: number, options?: PathResolutionOptions, ): Promise { - throwIfResolutionAborted(options); - const allResult = await query( - `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(); - 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, + stickyAgeFraction?: number, + options?: PathResolutionOptions, +): Promise { + throwIfResolutionAborted(options); + const allResult = await query( + `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; + limitedPacketHashes: Set; +}; + +export async function createMultiObserverPathBatchResolver( + network: string, + visibilityGeneration: number, + signal?: AbortSignal, +): Promise<{ + resolveBatch: ( + packetHashes: readonly string[], + windowStart: Date, + windowEnd: Date, + signal?: AbortSignal, + ) => Promise; +}> { + 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(); + const limitedPacketHashes = new Set(); + if (packetHashes.length === 0) return { results, limitedPacketHashes }; + + const packetResult = await namedQuery( + '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(); + 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 }; + }, + }; +} diff --git a/backend/src/workers/path-history.ts b/backend/src/workers/path-history.ts index 48dfac6..331f110 100644 --- a/backend/src/workers/path-history.ts +++ b/backend/src/workers/path-history.ts @@ -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): } } -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(); 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 { } 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 { 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')); } diff --git a/backend/src/workers/pathHistoryCheckpoint.test.ts b/backend/src/workers/pathHistoryCheckpoint.test.ts new file mode 100644 index 0000000..17990f6 --- /dev/null +++ b/backend/src/workers/pathHistoryCheckpoint.test.ts @@ -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); +}); diff --git a/backend/src/workers/pathHistoryCheckpoint.ts b/backend/src/workers/pathHistoryCheckpoint.ts new file mode 100644 index 0000000..d2061ca --- /dev/null +++ b/backend/src/workers/pathHistoryCheckpoint.ts @@ -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; + 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( + 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; +} diff --git a/backend/src/workers/pathHistorySchedule.test.ts b/backend/src/workers/pathHistorySchedule.test.ts index 0ac3a61..f418312 100644 --- a/backend/src/workers/pathHistorySchedule.test.ts +++ b/backend/src/workers/pathHistorySchedule.test.ts @@ -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); }); diff --git a/backend/src/workers/pathHistorySchedule.ts b/backend/src/workers/pathHistorySchedule.ts index 66f9977..10152a9 100644 --- a/backend/src/workers/pathHistorySchedule.ts +++ b/backend/src/workers/pathHistorySchedule.ts @@ -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; }