From 6406cdc70d43bfb55a8d5d5d3869ed6a2020ccec Mon Sep 17 00:00:00 2001 From: Florian <45694132+flo-bit@users.noreply.github.com> Date: Wed, 5 Aug 2026 02:30:23 +0200 Subject: [PATCH] Pipeline backfill projection checkpoints --- .changeset/backfill-throughput.md | 2 +- apps/benchmark/README.md | 3 + .../calendar-records-only-comparison.json | 142 ++++++ .../configs/calendar-records-only.config.json | 15 + apps/benchmark/src/bench.ts | 21 + packages/contrail/src/adapters/sqlite.ts | 33 +- packages/contrail/src/core/backfill.ts | 460 +++++++++++++----- packages/contrail/src/core/db/records.ts | 7 +- packages/contrail/src/core/ingest.ts | 11 +- packages/contrail/tests/dialect.test.ts | 37 ++ packages/contrail/tests/ingest.test.ts | 31 ++ 11 files changed, 633 insertions(+), 129 deletions(-) create mode 100644 apps/benchmark/baselines/calendar-records-only-comparison.json create mode 100644 apps/benchmark/configs/calendar-records-only.config.json diff --git a/.changeset/backfill-throughput.md b/.changeset/backfill-throughput.md index 2aa8788..c918ece 100644 --- a/.changeset/backfill-throughput.md +++ b/.changeset/backfill-throughput.md @@ -2,4 +2,4 @@ "@atmo-dev/contrail": patch --- -Group historical fetches by PDS with bounded per-host concurrency, cancel timed-out requests, defer failed initial accounts to scheduled retries, and rebuild FTS and relation counts with set-based SQL after canonical bulk loading. +Group historical fetches by PDS with bounded per-host concurrency, stream resolved identities directly into host workers, cancel timed-out requests, defer failed initial accounts to scheduled retries, atomically commit canonical pages with cursor checkpoints, and rebuild FTS and relation counts with set-based SQL after canonical bulk loading. SQLite batches now use synchronous transactions so concurrent backfill work cannot overlap or partially commit. diff --git a/apps/benchmark/README.md b/apps/benchmark/README.md index f3bfcd5..0bcac0c 100644 --- a/apps/benchmark/README.md +++ b/apps/benchmark/README.md @@ -10,6 +10,8 @@ pnpm bench --config calendar.config.json `calendar.config.json` mirrors the retained indexing shape from `11-atproto/02-atmo-rsvp`: calendar events, RSVPs, profiles, follows/feeds, query indexes, relation counts, and event full-text search. Removed product modules and read-only pipeline handlers are intentionally absent because they do not participate in indexing. External sinks are also omitted so the benchmark measures Contrail and D1 rather than Meilisearch latency. +For a narrow source/storage comparison, `calendar-records-only.config.json` indexes only calendar events and RSVPs. It explicitly disables profiles, follows, feeds, FTS, relation counts, field-query indexes, and Constellation. The retained [`calendar-records-only-comparison.json`](baselines/calendar-records-only-comparison.json) compares this workload with HappyView using matched 100-resolution, 10-PDS, and 3-DID limits; it also records Contrail's validated 20-PDS D1 setting. + ## Comparing concurrency Each command runs in a new Node process and starts from a fresh local D1. Identity resolution, active PDS hosts, and accounts per PDS are separate controls: @@ -19,6 +21,7 @@ pnpm bench --config calendar.config.json --concurrency 100 pnpm bench --config calendar.config.json --pds-concurrency 5 --dids-per-pds 3 pnpm bench --config calendar.config.json --pds-concurrency 10 --dids-per-pds 3 pnpm bench --config calendar.config.json --pds-concurrency 20 --dids-per-pds 3 +pnpm bench --config calendar-records-only.config.json --pds-concurrency 10 --dids-per-pds 3 ``` The current defaults are 100 concurrent identity resolutions, 20 active PDS hosts, 3 accounts per PDS, and one immediate attempt. Failures retain their cursor and move to scheduled cron retries instead of slowing the initial pass. The checked-in 774.49-second baseline records the older global-concurrency/5-attempt behavior and remains the historical comparison point. diff --git a/apps/benchmark/baselines/calendar-records-only-comparison.json b/apps/benchmark/baselines/calendar-records-only-comparison.json new file mode 100644 index 0000000..60fc0f1 --- /dev/null +++ b/apps/benchmark/baselines/calendar-records-only-comparison.json @@ -0,0 +1,142 @@ +{ + "format": "contrail.comparison-benchmark", + "version": 1, + "observed_at": "2026-08-05T00:09:00Z", + "workload": { + "collections": [ + "community.lexicon.calendar.event", + "community.lexicon.calendar.rsvp" + ], + "profiles": false, + "follows": false, + "feeds": false, + "full_text_search": false, + "relation_counts": false, + "field_query_indexes": false, + "live_ingestion_during_measurement": false, + "discovered_accounts": 1633 + }, + "happyview": { + "version": "2.11.8", + "commit": "6035c557e3ff2939eb0923bb1161663106c97223", + "source_patch": "unbounded per-host scheduler queue deadlock fix", + "backend": "native-sqlite", + "options": { + "identity_concurrency": 100, + "pds_concurrency": 10, + "dids_per_pds": 3 + }, + "discovery_and_backfill_ms": 45332.32, + "harness_wall_ms": 45474, + "indexed_records": 20869, + "records": { + "community.lexicon.calendar.event": 14608, + "community.lexicon.calendar.rsvp": 6261 + }, + "indexed_records_per_second": 460.4, + "repos": { + "discovered": 1633, + "resolved": 1629, + "processed": 1629, + "pending": 4 + } + }, + "contrail_native_sqlite": { + "backend": "native-sqlite", + "config": "configs/calendar-records-only.config.json", + "options": { + "identity_concurrency": 100, + "pds_concurrency": 10, + "dids_per_pds": 3, + "max_immediate_attempts": 1 + }, + "discovery_and_backfill_ms": 43880.78, + "full_harness_ms": 43923.33, + "indexed_records": 20882, + "records": { + "community.lexicon.calendar.event": 14610, + "community.lexicon.calendar.rsvp": 6272 + }, + "indexed_records_per_second": 475.9, + "accounts": { + "complete": 1591, + "retrying": 42, + "failed": 0 + } + }, + "contrail_native_sqlite_validated_limits": { + "backend": "native-sqlite", + "config": "configs/calendar-records-only.config.json", + "options": { + "identity_concurrency": 100, + "pds_concurrency": 20, + "dids_per_pds": 3, + "max_immediate_attempts": 1 + }, + "discovery_and_backfill_ms": 34490.29, + "full_harness_ms": 34520.07, + "indexed_records": 20882, + "records": { + "community.lexicon.calendar.event": 14610, + "community.lexicon.calendar.rsvp": 6272 + }, + "indexed_records_per_second": 605.4, + "accounts": { + "complete": 1590, + "retrying": 43, + "failed": 0 + } + }, + "contrail_matched_limits": { + "backend": "wrangler-local-d1", + "config": "configs/calendar-records-only.config.json", + "options": { + "identity_concurrency": 100, + "pds_concurrency": 10, + "dids_per_pds": 3, + "max_immediate_attempts": 1 + }, + "discovery_and_backfill_ms": 57269.02, + "full_harness_ms": 57720.11, + "indexed_records": 20882, + "records": { + "community.lexicon.calendar.event": 14610, + "community.lexicon.calendar.rsvp": 6272 + }, + "indexed_records_per_second": 364.6, + "accounts": { + "complete": 1591, + "retrying": 42, + "failed": 0 + } + }, + "contrail_validated_limits": { + "backend": "wrangler-local-d1", + "config": "configs/calendar-records-only.config.json", + "options": { + "identity_concurrency": 100, + "pds_concurrency": 20, + "dids_per_pds": 3, + "max_immediate_attempts": 1 + }, + "discovery_and_backfill_ms": 49754.94, + "full_harness_ms": 50281.28, + "indexed_records": 20878, + "records": { + "community.lexicon.calendar.event": 14607, + "community.lexicon.calendar.rsvp": 6271 + }, + "indexed_records_per_second": 419.6, + "accounts": { + "complete": 1588, + "retrying": 45, + "failed": 0 + } + }, + "notes": [ + "The live network changed slightly between sequential runs; record totals differ by at most 13 (0.06%).", + "The native-SQLite 10x3 row matches both backend and HappyView scheduling; the native-SQLite 20x3 row uses Contrail's validated host limit. The D1 rows show Contrail's primary deployment target.", + "HappyView can count a repository processed despite a collection fetch failure; Contrail keeps any unfinished collection durable and retrying.", + "Contrail phase metrics reported effectively zero derived-rebuild time because this fixture has no FTS or relation counts." + ] +} diff --git a/apps/benchmark/configs/calendar-records-only.config.json b/apps/benchmark/configs/calendar-records-only.config.json new file mode 100644 index 0000000..0c95249 --- /dev/null +++ b/apps/benchmark/configs/calendar-records-only.config.json @@ -0,0 +1,15 @@ +{ + "namespace": "rsvp.atmo", + "profiles": [], + "constellation": false, + "collections": { + "event": { + "collection": "community.lexicon.calendar.event", + "searchable": false + }, + "rsvp": { + "collection": "community.lexicon.calendar.rsvp", + "searchable": false + } + } +} diff --git a/apps/benchmark/src/bench.ts b/apps/benchmark/src/bench.ts index 3ea6b7b..b196791 100644 --- a/apps/benchmark/src/bench.ts +++ b/apps/benchmark/src/bench.ts @@ -215,6 +215,7 @@ async function main(): Promise { let backfillMs = 0; let discovered = 0; let acceptedRecords = 0; + let backfillMetrics: any; let overview: any; let fetchInstrumentation: ReturnType | undefined; @@ -254,6 +255,9 @@ async function main(): Promise { pdsConcurrency: options.pdsConcurrency, didsPerPds: options.didsPerPds, maxAttempts: options.maxAttempts, + onMetrics(metrics) { + backfillMetrics = metrics; + }, onProgress(progress) { const now = Date.now(); if (now - lastProgressAt < 2_000) return; @@ -325,6 +329,7 @@ async function main(): Promise { accepted_records: acceptedRecords, indexed_records: indexedRecords, peak_rss_kib: process.resourceUsage().maxRSS, + phases: backfillMetrics, network: { max_concurrent: fetchInstrumentation?.maxActive() ?? 0, requests: network, @@ -354,6 +359,22 @@ async function main(): Promise { `${overview.backfill.accounts.failed} failed`, ); console.log(`network max: ${result.network.max_concurrent} concurrent requests`); + if (backfillMetrics) { + console.log( + `phases: resolution ${(backfillMetrics.resolution_ms / 1000).toFixed(2)}s, ` + + `derived ${(backfillMetrics.derived_rebuild_ms / 1000).toFixed(2)}s`, + ); + for (const [collection, metric] of Object.entries( + backfillMetrics.collections, + ) as Array<[string, any]>) { + console.log( + `collection: ${collection} — ${metric.fetched_records} fetched, ` + + `${metric.accepted_records} accepted, ${metric.requests} requests, ` + + `${(metric.fetch_ms / 1000).toFixed(2)}s fetch, ` + + `${(metric.projection_and_checkpoint_ms / 1000).toFixed(2)}s project/checkpoint`, + ); + } + } for (const [key, metric] of Object.entries(network)) { console.log( `network: ${key} — ${metric.requests} requests, ` + diff --git a/packages/contrail/src/adapters/sqlite.ts b/packages/contrail/src/adapters/sqlite.ts index 2f46dce..34b2511 100644 --- a/packages/contrail/src/adapters/sqlite.ts +++ b/packages/contrail/src/adapters/sqlite.ts @@ -2,18 +2,27 @@ import { DatabaseSync } from "node:sqlite"; import type { Database, Statement } from "../core/types"; import { sqliteDialect } from "../core/dialect"; +interface SqliteStatement extends Statement { + _runSync(): unknown; +} + export function createSqliteDatabase(path: string): Database { const raw = new DatabaseSync(path); raw.exec("PRAGMA journal_mode = WAL"); - function wrapStatement(sql: string, boundValues: any[] = []): Statement { + function wrapStatement( + sql: string, + boundValues: any[] = [] + ): SqliteStatement { + const runSync = () => raw.prepare(sql).run(...boundValues); return { - bind(...values: any[]): Statement { + bind(...values: any[]): SqliteStatement { return wrapStatement(sql, values); }, async run() { - return raw.prepare(sql).run(...boundValues); + return runSync(); }, + _runSync: runSync, async all() { return { results: raw.prepare(sql).all(...boundValues) as T[] }; }, @@ -28,11 +37,21 @@ export function createSqliteDatabase(path: string): Database { return wrapStatement(sql); }, async batch(stmts: Statement[]): Promise { - const results: any[] = []; - for (const stmt of stmts) { - results.push(await stmt.run()); + raw.exec("BEGIN IMMEDIATE"); + try { + const results: unknown[] = []; + // Keep the whole transaction synchronous. Yielding between statements + // allows another concurrent batch to start a nested transaction on this + // single SQLite connection. + for (const stmt of stmts) { + results.push((stmt as SqliteStatement)._runSync()); + } + raw.exec("COMMIT"); + return results; + } catch (error) { + raw.exec("ROLLBACK"); + throw error; } - return results; }, dialect: sqliteDialect, }; diff --git a/packages/contrail/src/core/backfill.ts b/packages/contrail/src/core/backfill.ts index cde5bbb..21ee665 100644 --- a/packages/contrail/src/core/backfill.ts +++ b/packages/contrail/src/core/backfill.ts @@ -58,6 +58,69 @@ const BACKFILL_RETRY_MAX_MS = 48 * 60 * 60_000; const DEFAULT_SCHEDULED_MAX_ATTEMPTS = 10; const DERIVED_PROJECTIONS_DIRTY_KEY = "backfill_derived_projections_dirty"; +export interface BackfillCollectionMetrics { + requests: number; + pages: number; + fetched_records: number; + accepted_records: number; + record_bytes: number; + fetch_ms: number; + projection_and_checkpoint_ms: number; +} + +export interface BackfillRunMetrics { + resolution_ms: number; + derived_rebuild_ms: number; + collections: Record; +} + +type BackfillMetricsAccumulator = BackfillRunMetrics; + +function emptyBackfillMetrics(): BackfillMetricsAccumulator { + return { + resolution_ms: 0, + derived_rebuild_ms: 0, + collections: {}, + }; +} + +function collectionMetrics( + metrics: BackfillMetricsAccumulator, + collection: string +): BackfillCollectionMetrics { + return (metrics.collections[collection] ??= { + requests: 0, + pages: 0, + fetched_records: 0, + accepted_records: 0, + record_bytes: 0, + fetch_ms: 0, + projection_and_checkpoint_ms: 0, + }); +} + +function roundedMetrics( + metrics: BackfillMetricsAccumulator +): BackfillRunMetrics { + const round = (value: number) => Math.round(value * 100) / 100; + return { + resolution_ms: round(metrics.resolution_ms), + derived_rebuild_ms: round(metrics.derived_rebuild_ms), + collections: Object.fromEntries( + Object.entries(metrics.collections).map(([collection, values]) => [ + collection, + { + ...values, + fetch_ms: round(values.fetch_ms), + projection_and_checkpoint_ms: round( + values.projection_and_checkpoint_ms + ), + }, + ]) + ), + }; +} + function positiveInteger(value: number | undefined, fallback: number): number { return typeof value === "number" && Number.isFinite(value) && value > 0 ? Math.floor(value) @@ -160,6 +223,10 @@ export interface BackfillOptions { maxPages?: number; /** Defer FTS and relation counts until the bulk pass finishes. */ skipDerivedProjections?: boolean; + /** @internal Aggregate benchmark instrumentation owned by the bulk run. */ + metrics?: BackfillMetricsAccumulator; + /** @internal Cursor already loaded by the bulk scheduler. */ + resumeState?: { cursor: string | null }; } interface BackfillUserAttempt { @@ -180,12 +247,22 @@ async function backfillUserAttempt( return { records: 0, completed: false, failed: false }; } - const status = await db - .prepare( - "SELECT completed, pds_cursor, retries FROM backfills WHERE did = ? AND collection = ?" - ) - .bind(did, collection) - .first<{ completed: number; pds_cursor: string | null; retries: number }>(); + const status = options?.resumeState + ? { + completed: 0, + pds_cursor: options.resumeState.cursor, + retries: 0, + } + : await db + .prepare( + "SELECT completed, pds_cursor, retries FROM backfills WHERE did = ? AND collection = ?" + ) + .bind(did, collection) + .first<{ + completed: number; + pds_cursor: string | null; + retries: number; + }>(); if (status?.completed) { return { records: 0, completed: true, failed: false }; @@ -250,20 +327,30 @@ async function backfillUserAttempt( let totalInserted = 0; let done = false; let pages = 0; + const metrics = options?.metrics + ? collectionMetrics(options.metrics, collection) + : undefined; try { while (Date.now() < deadline) { const response = await withRetry( - (signal) => - client!.get("com.atproto.repo.listRecords", { - params: { - repo: did as Did, - collection, - limit: PAGE_SIZE, - cursor: currentCursor, - }, - signal, - }), + async (signal) => { + const started = performance.now(); + if (metrics) metrics.requests++; + try { + return await client!.get("com.atproto.repo.listRecords", { + params: { + repo: did as Did, + collection, + limit: PAGE_SIZE, + cursor: currentCursor, + }, + signal, + }); + } finally { + if (metrics) metrics.fetch_ms += performance.now() - started; + } + }, `listRecords(${did}/${collection})`, retries, timeout @@ -282,14 +369,17 @@ async function backfillUserAttempt( return { records: totalInserted, completed: false, failed: true }; } - if (response.data.records.length === 0) { - done = true; - break; + const records = response.data.records; + if (metrics) { + metrics.pages++; + metrics.fetched_records += records.length; } const now = Date.now(); const nowUs = now * 1000; - const events: IngestEvent[] = response.data.records.map((record) => + const nextCursor = response.data.cursor ?? undefined; + const pageDone = records.length === 0 || !nextCursor; + const events: IngestEvent[] = records.map((record) => createIngestEvent({ uri: record.uri, did, @@ -303,29 +393,41 @@ async function backfillUserAttempt( }), ); - if (events.length > 0) { - const result = await ingestRecords(db, events, config, { - skipReplayDetection: options?.skipReplayDetection, - skipFeedFanout: true, - knownDids: options?.knownDids, - skipDerivedProjections: options?.skipDerivedProjections, - // Let sinks bulk-flush differently from live ingestion. - phase: "backfill", - }); - totalInserted += result.accepted.length; + if (metrics) { + metrics.record_bytes += events.reduce( + (bytes, event) => bytes + (event.record?.length ?? 0), + 0 + ); } - - currentCursor = response.data.cursor ?? undefined; - - await db + const projectionStarted = performance.now(); + const checkpoint = db .prepare( - "UPDATE backfills SET pds_cursor = ?, retries = 0, scheduled_retries = 0, last_error = NULL, last_attempt_at = ?, next_retry_at = NULL, retry_exhausted = 0 WHERE did = ? AND collection = ?" + "UPDATE backfills SET pds_cursor = ?, completed = ?, retries = 0, scheduled_retries = 0, last_error = NULL, last_attempt_at = ?, next_retry_at = NULL, retry_exhausted = 0 WHERE did = ? AND collection = ?" ) - .bind(currentCursor ?? null, now, did, collection) - .run(); + .bind(nextCursor ?? null, pageDone ? 1 : 0, now, did, collection); + const result = await ingestRecords(db, events, config, { + skipReplayDetection: options?.skipReplayDetection, + skipFeedFanout: true, + knownDids: options?.knownDids, + skipDerivedProjections: options?.skipDerivedProjections, + // Canonical projection and cursor acknowledgement commit atomically. + trailingStatements: [checkpoint], + // Let sinks bulk-flush differently from live ingestion. + phase: "backfill", + }); + totalInserted += result.accepted.length; + if (metrics) { + metrics.accepted_records += result.accepted.length; + metrics.projection_and_checkpoint_ms += + performance.now() - projectionStarted; + } + currentCursor = nextCursor; + if (options?.resumeState) { + options.resumeState.cursor = currentCursor ?? null; + } pages++; - if (!currentCursor) { + if (pageDone) { done = true; break; } @@ -342,15 +444,6 @@ async function backfillUserAttempt( return { records: totalInserted, completed: false, failed: true }; } - if (done) { - await db - .prepare( - "UPDATE backfills SET completed = 1, retries = 0, scheduled_retries = 0, last_error = NULL, last_attempt_at = ?, next_retry_at = NULL, retry_exhausted = 0 WHERE did = ? AND collection = ?" - ) - .bind(Date.now(), did, collection) - .run(); - } - return { records: totalInserted, completed: done, failed: false }; } @@ -395,6 +488,8 @@ export interface BackfillAllOptions { * later attempts. Values above 1 are retained for explicit manual recovery. */ maxAttempts?: number; onProgress?: (progress: BackfillProgress) => void; + /** Receives aggregate source/projection timings after a complete invocation. */ + onMetrics?: (metrics: BackfillRunMetrics) => void; } /** Keep a fixed number of jobs active without batch barriers. Returning true @@ -440,6 +535,115 @@ async function drainQueue( }); } +function createStreamingHostScheduler( + hostConcurrency: number, + didsPerHost: number, + run: (pds: string, did: string) => Promise, + consume: (result: TResult) => boolean | void +): { + add(pds: string, did: string): void; + finish(): Promise; +} { + type HostState = { + pending: string[]; + active: number; + queued: boolean; + running: boolean; + }; + const hosts = new Map(); + const waiting: string[] = []; + let activeHosts = 0; + let producerDone = false; + let settled = false; + let resolveFinished!: () => void; + let rejectFinished!: (error: unknown) => void; + const finished = new Promise((resolve, reject) => { + resolveFinished = resolve; + rejectFinished = reject; + }); + + const fail = (error: unknown) => { + if (settled) return; + settled = true; + rejectFinished(error); + }; + + const maybeFinish = () => { + if (settled || !producerDone) return; + if (activeHosts === 0 && waiting.length === 0) { + settled = true; + resolveFinished(); + } + }; + + const pumpHosts = () => { + if (settled) return; + while (activeHosts < hostConcurrency && waiting.length > 0) { + const pds = waiting.shift()!; + const state = hosts.get(pds)!; + state.queued = false; + if (state.running || state.pending.length === 0) continue; + state.running = true; + activeHosts++; + pumpDids(pds, state); + } + maybeFinish(); + }; + + const finishHost = (state: HostState) => { + if (!state.running) return; + state.running = false; + activeHosts--; + pumpHosts(); + }; + + function pumpDids(pds: string, state: HostState): void { + if (settled) return; + while (state.active < didsPerHost && state.pending.length > 0) { + const did = state.pending.shift()!; + state.active++; + run(pds, did).then( + (result) => { + state.active--; + if (consume(result) === true) state.pending.push(did); + pumpDids(pds, state); + }, + fail + ); + } + if (state.active === 0 && state.pending.length === 0) { + finishHost(state); + } + } + + return { + add(pds, did) { + if (producerDone) throw new Error("Cannot add work after scheduler finish"); + let state = hosts.get(pds); + if (!state) { + state = { pending: [], active: 0, queued: false, running: false }; + hosts.set(pds, state); + } + state.pending.push(did); + if (state.running) { + pumpDids(pds, state); + } else if (!state.queued) { + state.queued = true; + waiting.push(pds); + pumpHosts(); + } + }, + finish() { + producerDone = true; + for (const [pds, state] of hosts) { + if (state.running) pumpDids(pds, state); + } + pumpHosts(); + return finished; + }, + }; +} + async function loadKnownBackfillDids(db: Database): Promise> { const rows = await db .prepare("SELECT DISTINCT did FROM backfills") @@ -472,6 +676,7 @@ async function backfillPendingWork( ); let totalBackfilled = 0; const knownDids = await loadKnownBackfillDids(db); + const metrics = emptyBackfillMetrics(); // Anchor the jetstream cursor to now if it hasn't been set yet, so records // emitted during backfill are replayed once jetstream starts. @@ -496,39 +701,110 @@ async function backfillPendingWork( while (true) { const pending = await db .prepare( - "SELECT did, collection FROM backfills WHERE completed = 0 AND retries < ? ORDER BY did" + "SELECT did, collection, pds_cursor FROM backfills WHERE completed = 0 AND retries < ? ORDER BY did" ) .bind(maxAttempts) - .all<{ did: string; collection: string }>(); + .all<{ did: string; collection: string; pds_cursor: string | null }>(); const rows = pending.results ?? []; if (rows.length === 0) break; - const byDid = new Map(); + type CollectionWork = { + collection: string; + cursor: string | null; + completed: boolean; + failed: boolean; + }; + const byDid = new Map(); for (const row of rows) { const collections = byDid.get(row.did) ?? []; - collections.push(row.collection); + collections.push({ + collection: row.collection, + cursor: row.pds_cursor, + completed: false, + failed: false, + }); byDid.set(row.did, collections); } const dids = [...byDid.keys()]; - const byPds = new Map(); let roundBackfilled = 0; let usersComplete = 0; let usersFailed = 0; - const emitProgress = () => + let lastHeartbeat = Date.now(); + const emitProgress = () => { options?.onProgress?.({ records: totalBackfilled + roundBackfilled, usersComplete, usersTotal: dids.length, usersFailed, }); + const now = Date.now(); + if (now - lastHeartbeat >= 30_000) { + lastHeartbeat = now; + void heartbeatBackfillRun(db, runId).catch((error) => + (config.logger ?? console).warn("Backfill heartbeat failed", error) + ); + } + }; await heartbeatBackfillRun(db, runId); - // Resolve first, then schedule by host. This deliberately separates cheap - // identity fan-out from PDS traffic and lets every host share one client. + // Start host work as soon as each identity resolves. This removes the + // all-identities barrier while retaining both host and per-host limits. + const clients = new Map(); + const hostScheduler = createStreamingHostScheduler( + pdsConcurrency, + didsPerPds, + async (pds, did) => { + let client = clients.get(pds); + if (!client) { + client = createPdsClient(pds); + clients.set(pds, client); + } + const works = byDid.get(did)!; + let records = 0; + for (const work of works) { + if (work.completed || work.failed) continue; + const attempt = await backfillUserAttempt( + db, + did, + work.collection, + Infinity, + config, + { + client, + knownDids, + skipReplayDetection: true, + maxRetries: 0, + requestTimeout, + maxPages: 1, + skipDerivedProjections: true, + metrics, + resumeState: work, + } + ); + records += attempt.records; + work.completed = attempt.completed; + work.failed = attempt.failed; + } + return { + records, + completed: works.every((work) => work.completed), + failed: works.some((work) => work.failed), + }; + }, + (result) => { + roundBackfilled += result.records; + if (result.completed) usersComplete++; + else if (result.failed) usersFailed++; + emitProgress(); + return !result.completed && !result.failed; + } + ); + + const resolutionStarted = performance.now(); await drainQueue( dids, resolutionConcurrency, @@ -543,80 +819,23 @@ async function backfillPendingWork( if (!pds) throw new Error(`PDS not found for ${did}`); return { did, pds: pds.replace(/\/+$/, ""), failed: false }; } catch (error) { - for (const collection of byDid.get(did)!) { - await markFailed(db, did, collection, error); + for (const work of byDid.get(did)!) { + await markFailed(db, did, work.collection, error); + work.failed = true; } return { did, pds: null, failed: true }; } }, (result) => { - if (result.pds) { - const hostDids = byPds.get(result.pds) ?? []; - hostDids.push(result.did); - byPds.set(result.pds, hostDids); - } else if (result.failed) { + if (result.pds) hostScheduler.add(result.pds, result.did); + else if (result.failed) { usersFailed++; emitProgress(); } } ); - - // Keep only a small number of PDS hosts active. Within each host, process a - // few accounts and one collection page at a time. This preserves connection - // reuse, prevents request storms, and keeps large repositories from blocking - // unrelated hosts or accounts on the same host. - await drainQueue( - [...byPds.entries()], - pdsConcurrency, - async ([pds, hostDids]) => { - const client = createPdsClient(pds); - await drainQueue( - hostDids, - didsPerPds, - async (did) => { - const attempts: BackfillUserAttempt[] = []; - for (const collection of byDid.get(did)!) { - attempts.push( - await backfillUserAttempt( - db, - did, - collection, - Infinity, - config, - { - client, - knownDids, - skipReplayDetection: true, - maxRetries: 0, - requestTimeout, - maxPages: 1, - skipDerivedProjections: true, - } - ) - ); - } - return { - records: attempts.reduce( - (sum, attempt) => sum + attempt.records, - 0 - ), - completed: attempts.every((attempt) => attempt.completed), - failed: attempts.some((attempt) => attempt.failed), - }; - }, - (result) => { - roundBackfilled += result.records; - if (result.completed) usersComplete++; - else if (result.failed) usersFailed++; - emitProgress(); - return !result.completed && !result.failed; - } - ); - await heartbeatBackfillRun(db, runId); - return undefined; - }, - () => false - ); + metrics.resolution_ms += performance.now() - resolutionStarted; + await hostScheduler.finish(); totalBackfilled += roundBackfilled; await heartbeatBackfillRun(db, runId); @@ -628,9 +847,12 @@ async function backfillPendingWork( // Canonical records and cursors are durable now. Rebuild expensive derived // projections once with set-based SQL instead of hundreds of statements per // network page. This also repairs a prior interrupted bulk pass. + const derivedStarted = performance.now(); await rebuildDerivedProjections(db, config); + metrics.derived_rebuild_ms += performance.now() - derivedStarted; await setMeta(db, DERIVED_PROJECTIONS_DIRTY_KEY, "0"); await heartbeatBackfillRun(db, runId); + options?.onMetrics?.(roundedMetrics(metrics)); return totalBackfilled; } diff --git a/packages/contrail/src/core/db/records.ts b/packages/contrail/src/core/db/records.ts index 2b555b6..4c9afd7 100644 --- a/packages/contrail/src/core/db/records.ts +++ b/packages/contrail/src/core/db/records.ts @@ -675,6 +675,8 @@ export async function projectEvents( /** Ingest phase forwarded to `config.sinks`. `"live"` for jetstream / * persistent ingest (default), `"backfill"` for replay / rebuild. */ phase?: "live" | "backfill"; + /** Statements committed after projection in the same database batch. */ + trailingStatements?: Statement[]; } ): Promise { if (events.length === 0) return; @@ -751,8 +753,11 @@ export async function projectEvents( // batch observe the final records. batch.unshift(...buildRecordMutationStatements(db, storageMutations.values())); - // Build deduplicated count statements — one UPDATE per unique target + // Build deduplicated count statements — one UPDATE per unique target. batch.push(...buildBatchCountStatements(db, config, countTargets)); + if (options?.trailingStatements?.length) { + batch.push(...options.trailingStatements); + } await db.batch(batch); diff --git a/packages/contrail/src/core/ingest.ts b/packages/contrail/src/core/ingest.ts index bcc4244..2a3ca5a 100644 --- a/packages/contrail/src/core/ingest.ts +++ b/packages/contrail/src/core/ingest.ts @@ -1,5 +1,10 @@ import { isDid } from "@atcute/lexicons/syntax"; -import type { ContrailConfig, Database, IngestEvent } from "./types"; +import type { + ContrailConfig, + Database, + IngestEvent, + Statement, +} from "./types"; import { getNestedValue, resolveCollectionKey, @@ -51,6 +56,8 @@ export interface IngestRecordsOptions { phase?: "live" | "backfill"; /** Known actors used to filter dependent records without another DB read. */ knownDids?: ReadonlySet; + /** Statements committed after projection in the same database batch. */ + trailingStatements?: Statement[]; } export interface IngestDropCounts { @@ -167,6 +174,8 @@ export async function ingestRecords( if (subjectFiltered.length > 0) { await projectEvents(db, subjectFiltered, config, options); + } else if (options.trailingStatements?.length) { + await db.batch(options.trailingStatements); } return { accepted: subjectFiltered, dropped, discoveredDids }; diff --git a/packages/contrail/tests/dialect.test.ts b/packages/contrail/tests/dialect.test.ts index cc5a372..209d728 100644 --- a/packages/contrail/tests/dialect.test.ts +++ b/packages/contrail/tests/dialect.test.ts @@ -44,6 +44,43 @@ describe("Database.dialect", () => { expect(db.dialect).toBeDefined(); expect(db.dialect.recordColumnType).toBe("TEXT"); }); + + it("serializes concurrent SQLite batches without nested transactions", async () => { + const db = createSqliteDatabase(":memory:"); + await db.prepare("CREATE TABLE concurrent_test (value INTEGER UNIQUE)").run(); + + await Promise.all( + Array.from({ length: 50 }, (_, value) => + db.batch([ + db + .prepare("INSERT INTO concurrent_test (value) VALUES (?)") + .bind(value), + ]) + ) + ); + + const row = await db + .prepare("SELECT COUNT(*) AS count FROM concurrent_test") + .first<{ count: number }>(); + expect(row?.count).toBe(50); + }); + + it("rolls back a failed SQLite batch", async () => { + const db = createSqliteDatabase(":memory:"); + await db.prepare("CREATE TABLE values_test (value TEXT UNIQUE)").run(); + + await expect( + db.batch([ + db.prepare("INSERT INTO values_test (value) VALUES (?)").bind("one"), + db.prepare("INSERT INTO values_test (value) VALUES (?)").bind("one"), + ]) + ).rejects.toThrow(); + + const row = await db + .prepare("SELECT COUNT(*) AS count FROM values_test") + .first<{ count: number }>(); + expect(row?.count).toBe(0); + }); }); describe("postgresDialect", () => { diff --git a/packages/contrail/tests/ingest.test.ts b/packages/contrail/tests/ingest.test.ts index 38dafea..2da7cae 100644 --- a/packages/contrail/tests/ingest.test.ts +++ b/packages/contrail/tests/ingest.test.ts @@ -106,6 +106,37 @@ describe("ingestRecords", () => { expect((await queryRecords(db, config, { collection: "event" })).records).toHaveLength(30); }); + it("commits trailing checkpoint statements atomically with projection", async () => { + await db.prepare("CREATE TABLE checkpoint_test (value TEXT UNIQUE)").run(); + await db + .prepare("INSERT INTO checkpoint_test (value) VALUES (?)") + .bind("duplicate") + .run(); + const event = createIngestEvent({ + did: "did:plc:alice", + collection: "com.example.event", + rkey: "atomic", + operation: "create", + cid: "cid-atomic", + value: { keep: true }, + timeUs: 1, + }); + + await expect( + ingestRecords(db, [event], config, { + trailingStatements: [ + db + .prepare("INSERT INTO checkpoint_test (value) VALUES (?)") + .bind("duplicate"), + ], + }) + ).rejects.toThrow(); + + expect( + (await queryRecords(db, config, { collection: "event" })).records + ).toHaveLength(0); + }); + it("drops malformed records and untracked collections before projection", async () => { const malformed = createIngestEvent({ did: "did:plc:alice", -- 2.51.2