From 2bec760dd61b12556c27bc5bcb2aee0c2bd8b10d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Uro=C5=A1=20Marolt?= Date: Tue, 30 Jun 2026 17:01:19 +0200 Subject: [PATCH 01/15] fix: drive osspckgs monitor total ETA off file progress MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Uroš Marolt --- .../src/scripts/monitorOsspckgs.ts | 36 +++++++++---------- 1 file changed, 16 insertions(+), 20 deletions(-) diff --git a/services/apps/packages_worker/src/scripts/monitorOsspckgs.ts b/services/apps/packages_worker/src/scripts/monitorOsspckgs.ts index 00643f778d..78a416f822 100644 --- a/services/apps/packages_worker/src/scripts/monitorOsspckgs.ts +++ b/services/apps/packages_worker/src/scripts/monitorOsspckgs.ts @@ -755,34 +755,30 @@ function computeChunkEta(job: any) { } // ETA for the entire job (all remaining files + merges). -// Uses job.started_at so rate includes both loading and merging time naturally. +// Driven by file progress, not pgRows-vs-bqRows: pgRows is the merged delta and +// converges to staging size (after dedup), never to the raw BQ scan count — so a +// bqRows target overstates remaining work, and a pgRows/total-elapsed rate is diluted +// by the export+download+staging phases where pgRows is still 0. Files are the unit +// that actually advances, and elapsed already amortizes every per-file phase. // eslint-disable-next-line @typescript-eslint/no-explicit-any function computeTotalEta(job: any) { if (!['loading', 'merging'].includes(job.status)) return null - const pgRows = Number(job.row_count_pg) || 0 const trc = job.table_row_counts ?? {} const filesDone = Number(trc['progress:done']) || 0 const filesTotal = Number(trc['progress:total']) || 0 - const stagingRows = Number(job.row_count_staging) || 0 - if (!filesDone || !filesTotal || !pgRows || !job.started_at) return null - const bqRows = Number(job.row_count_bq) || 0 - const rowsPerFile = - bqRows > 0 && filesTotal > 0 - ? bqRows / filesTotal - : stagingRows > 0 - ? stagingRows / filesDone - : 0 - if (!rowsPerFile) return null - const estimatedTotal = bqRows > 0 ? bqRows : rowsPerFile * filesTotal - if (pgRows >= estimatedTotal) return null + if (!filesDone || !filesTotal || !job.started_at) return null + if (filesDone >= filesTotal) return null const elapsedMs = Date.now() - new Date(job.started_at).getTime() - if (elapsedMs < 10000 || pgRows < 1000) return null - const ratePerMs = pgRows / elapsedMs - if (ratePerMs <= 0) return null + if (elapsedMs < 10000) return null + const msPerFile = elapsedMs / filesDone + const remainingMs = msPerFile * (filesTotal - filesDone) + if (remainingMs <= 0) return null + // Throughput display: staging rows landed per minute over the whole job. + const stagingRows = Number(job.row_count_staging) || 0 return { - ms: (estimatedTotal - pgRows) / ratePerMs, - ratio: Math.min(pgRows / estimatedTotal, 1), - ratePerMin: ratePerMs * 60000, + ms: remainingMs, + ratio: filesDone / filesTotal, + ratePerMin: stagingRows > 0 ? (stagingRows / elapsedMs) * 60000 : 0, } } From 7c452337b045bdb85ac78fa1c48754c4127992cc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Uro=C5=A1=20Marolt?= Date: Tue, 30 Jun 2026 17:03:59 +0200 Subject: [PATCH 02/15] fix: drop diluted-rate fallback in osspckgs monitor merge ETA MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Uroš Marolt --- .../src/scripts/monitorOsspckgs.ts | 41 ++++++++----------- 1 file changed, 17 insertions(+), 24 deletions(-) diff --git a/services/apps/packages_worker/src/scripts/monitorOsspckgs.ts b/services/apps/packages_worker/src/scripts/monitorOsspckgs.ts index 78a416f822..a975c33328 100644 --- a/services/apps/packages_worker/src/scripts/monitorOsspckgs.ts +++ b/services/apps/packages_worker/src/scripts/monitorOsspckgs.ts @@ -710,8 +710,10 @@ function updateChunkHistory(job: any, now: number) { } // ETA to finish merging the current staging chunk. -// Uses historical merge rate from completed chunks when available; falls back to overall job -// throughput (includes loading time, so slightly conservative) when none observed yet. +// Uses historical merge rate from completed chunks only. With no chunk history yet +// (e.g. monitor started mid-merge) there is no trustworthy throughput signal — pgRows/elapsed +// is diluted by the export+download phases where pgRows is ~0 and mis-projects by ~30x — so we +// show "—" until a chunk boundary is observed rather than print a wildly wrong number. // eslint-disable-next-line @typescript-eslint/no-explicit-any function computeChunkEta(job: any) { if (job.status !== 'merging') return null @@ -719,29 +721,20 @@ function computeChunkEta(job: any) { if (!stagingRows) return null const hist = chunkMergeHistory.get(job.id) if (!hist || hist.mergeStart == null) return null - - let rateRowsPerMs: number - if (hist.completedChunks.length > 0) { - // Exponential recency weighting: chunk i gets weight 2^i (oldest=0, newest=n-1). - // Most recent chunk contributes ~50% of the rate; history stabilises it. - const chunks = hist.completedChunks - let weightedRate = 0, - totalWeight = 0 - for (let i = 0; i < chunks.length; i++) { - const w = Math.pow(2, i) - weightedRate += (chunks[i].stagingRows / chunks[i].durationMs) * w - totalWeight += w - } - if (totalWeight === 0) return null - rateRowsPerMs = weightedRate / totalWeight - } else { - // No completed chunks observed — derive rate from overall job progress (conservative: includes loading) - const pgRows = Number(job.row_count_pg) || 0 - if (!pgRows || !job.started_at) return null - const jobElapsedMs = Date.now() - new Date(job.started_at).getTime() - if (jobElapsedMs < 10000) return null - rateRowsPerMs = pgRows / jobElapsedMs + if (hist.completedChunks.length === 0) return null + + // Exponential recency weighting: chunk i gets weight 2^i (oldest=0, newest=n-1). + // Most recent chunk contributes ~50% of the rate; history stabilises it. + const chunks = hist.completedChunks + let weightedRate = 0, + totalWeight = 0 + for (let i = 0; i < chunks.length; i++) { + const w = Math.pow(2, i) + weightedRate += (chunks[i].stagingRows / chunks[i].durationMs) * w + totalWeight += w } + if (totalWeight === 0) return null + const rateRowsPerMs = weightedRate / totalWeight const estimatedDurationMs = stagingRows / rateRowsPerMs const elapsedMs = Date.now() - hist.mergeStart From 66c89b70ae387f88c4aa718b4cdb40364af14bea Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Uro=C5=A1=20Marolt?= Date: Thu, 2 Jul 2026 20:15:12 +0200 Subject: [PATCH 03/15] feat: resume partially-merged package_dependencies ingest by job id MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Uroš Marolt --- .../deps-dev/activities/getResumeExport.ts | 42 +++++ .../src/deps-dev/activities/index.ts | 1 + .../deps-dev/workflows/bootstrapOsspckgs.ts | 8 +- .../deps-dev/workflows/ingestDependencies.ts | 178 +++++++++++++----- .../src/scripts/triggerBootstrap.ts | 50 ++++- .../src/osspckgs/ingestJobs.ts | 41 ++++ 6 files changed, 272 insertions(+), 48 deletions(-) create mode 100644 services/apps/packages_worker/src/deps-dev/activities/getResumeExport.ts diff --git a/services/apps/packages_worker/src/deps-dev/activities/getResumeExport.ts b/services/apps/packages_worker/src/deps-dev/activities/getResumeExport.ts new file mode 100644 index 0000000000..8561988d66 --- /dev/null +++ b/services/apps/packages_worker/src/deps-dev/activities/getResumeExport.ts @@ -0,0 +1,42 @@ +import { getIngestJobForResume } from '@crowd/data-access-layer' + +import { getPackagesDb } from '../../db' + +export interface GetResumeExportInput { + jobId: number +} + +export interface GetResumeExportOutput { + jobId: number + jobKind: string + status: string + syncMode: string + gcsPrefix: string | null + progressDone: number + progressTotal: number + rowCountPg: number +} + +// Pure fetch of a prior job's resume-relevant fields (export path, status, file-load progress, +// rows merged). Returns null if the job id does not exist. All validation — kind, status, presence +// of an export, and that the parquet files still exist — is done by the caller (ingestDependencies) +// so it can fail fast with non-retryable errors instead of retrying a bad-input activity. +export async function getResumeExport( + input: GetResumeExportInput, +): Promise { + const qx = await getPackagesDb() + const job = await getIngestJobForResume(qx, input.jobId) + if (!job) { + return null + } + return { + jobId: job.id, + jobKind: job.jobKind, + status: job.status, + syncMode: job.syncMode, + gcsPrefix: job.gcsPrefix, + progressDone: job.progressDone, + progressTotal: job.progressTotal, + rowCountPg: job.rowCountPg, + } +} diff --git a/services/apps/packages_worker/src/deps-dev/activities/index.ts b/services/apps/packages_worker/src/deps-dev/activities/index.ts index 3de67ac1d2..1fd3a15b0d 100644 --- a/services/apps/packages_worker/src/deps-dev/activities/index.ts +++ b/services/apps/packages_worker/src/deps-dev/activities/index.ts @@ -9,6 +9,7 @@ export * from './listParquetFiles' export * from './gcsParquetToStaging' export * from './mergeStagingToTable' export * from './getLastSnapshot' +export * from './getResumeExport' export * from './checkDependentCountsGuard' export * from './checkEdgeSnapshotQuality' export * from './probePartitionExists' diff --git a/services/apps/packages_worker/src/deps-dev/workflows/bootstrapOsspckgs.ts b/services/apps/packages_worker/src/deps-dev/workflows/bootstrapOsspckgs.ts index c12edceed8..92554c5f5f 100644 --- a/services/apps/packages_worker/src/deps-dev/workflows/bootstrapOsspckgs.ts +++ b/services/apps/packages_worker/src/deps-dev/workflows/bootstrapOsspckgs.ts @@ -69,12 +69,17 @@ export async function bootstrapOsspckgs(opts: { exportName?: string snapshotDate?: string // YYYY-MM-DD — override BQ snapshot resolution for all partition-filtered kinds fillConstraints?: boolean // re-export full deps BQ data, upsert version_constraint where NULL + resumeJobId?: number // resume a partially-merged package_dependencies job by id (skips its BQ export) }): Promise { // B3: deterministic timestamps — workflowInfo().startTime is replay-stable; new Date() is not. const start = workflowInfo().startTime const runId = start.toISOString().replace(/[:.]/g, '-') const today = start.toISOString().slice(0, 10) + // Resume mode reuses a prior job's export, so there is no fresh BQ export to validate. Skip the + // incremental watermark/partition checks below — the resumed partition may not match `today`. + const resume = opts.resumeJobId != null + // Recovery: each child workflow updates osspckgs_ingest_jobs independently. // If a child fails mid-bootstrap, re-run with the SAME mode. // Already-done kinds skip naturally (today→today diff = 0 rows). @@ -119,7 +124,7 @@ export async function bootstrapOsspckgs(opts: { // Validate all watermarks up-front before touching BQ (fail fast, not mid-run) for (const jobKind of jobKinds) { - if (opts.mode === 'incremental') { + if (opts.mode === 'incremental' && !resume) { const { snapshotAt } = await getLastSnapshot({ jobKind }) if (!snapshotAt) { throw new ApplicationFailure(`No watermark for ${jobKind} — run full bootstrap first`) @@ -262,6 +267,7 @@ export async function bootstrapOsspckgs(opts: { depsTableOption: opts.depsTableOption, exportName: opts.exportName, fillConstraints: opts.fillConstraints, + resumeJobId: opts.resumeJobId, }, ], }) diff --git a/services/apps/packages_worker/src/deps-dev/workflows/ingestDependencies.ts b/services/apps/packages_worker/src/deps-dev/workflows/ingestDependencies.ts index 111f77534f..26072caec4 100644 --- a/services/apps/packages_worker/src/deps-dev/workflows/ingestDependencies.ts +++ b/services/apps/packages_worker/src/deps-dev/workflows/ingestDependencies.ts @@ -24,10 +24,17 @@ const { gcsParquetToStaging } = proxyActivities({ }) const { mergeStagingToTable } = proxyActivities({ - startToCloseTimeout: '2 hours', + // A single ~1M-row chunk merge into the live-index/live-FK package_dependencies table (incremental + // does not drop indexes) can run long; 4h gives headroom over the previously-observed 2h overrun. + startToCloseTimeout: '4 hours', retry: { maximumAttempts: 1 }, }) +const { getResumeExport } = proxyActivities({ + startToCloseTimeout: '1 minute', + retry: { maximumAttempts: 3 }, +}) + const { createVersionsLookup } = proxyActivities({ startToCloseTimeout: '2 hours', retry: { maximumAttempts: 2, initialInterval: '30 seconds' }, @@ -219,6 +226,9 @@ export async function ingestDependencies(opts: { depsTableOption?: 'A' | 'B' exportName?: string fillConstraints?: boolean // re-export full BQ data, upsert version_constraint only where NULL + // Resume a partially-merged prior job: skip the BQ export, reuse that job's exact parquet files, + // and restart the chunk loop where its staging load left off. Idempotent modes only (see gate). + resumeJobId?: number }): Promise<{ rowCountBq: number }> { const ecosystems = opts.ecosystems ?? DEPS_DEFAULT_ECOSYSTEMS const isFill = opts.fillConstraints === true @@ -228,57 +238,128 @@ export async function ingestDependencies(opts: { // source below and which source the guard probes — they must match. const fullScan = opts.syncMode === 'full' || isFill - // Guard against corrupt deps.dev resolved-graph snapshots BEFORE the (multi-hour) export. - // Skip when reusing a prior export — we're re-importing already-validated parquet, not - // scanning the live snapshot. Both full (*Latest = newest snapshot) and incremental can hit - // a bad snapshot, so the guard runs for both. Probes only resolved-graph ecosystems; a clean - // GO/NUGET-only run finds no canaries and passes through. - // - // Option A only: the guard's canary ratios are DependencyGraphEdges-schema-specific. Option B - // ingests the separate Dependencies/DependenciesLatest table, which the 2026-06 corruption was - // never observed in and has no calibrated baseline — probing Edges there would "validate" a table - // we don't ingest (false confidence) and could abort a healthy Option B run when only Edges is bad. - // Option B is a manual, non-scheduled cost-experiment path (--deps-table-b); leave it unguarded by - // design rather than invent a guard for an unproven threat. Option A is the production default. - if (!opts.reuseExports && tableOption === 'A') { - const guard = await checkEdgeSnapshotQuality({ snapshotDate: opts.today, ecosystems, fullScan }) - if (!guard.ok) { + const resume = opts.resumeJobId != null + + // Resume is only safe for idempotent merges: incremental uses ON CONFLICT DO NOTHING and fill uses + // ON CONFLICT DO UPDATE, so re-running an overlapping chunk no-ops. A full (non-fill) load uses a + // plain INSERT with the UNIQUE constraint dropped, so a re-merged chunk would duplicate rows. + if (resume && fullScan && !isFill) { + throw ApplicationFailure.nonRetryable( + `resume (resumeJobId) is not supported for full loads — plain INSERT would duplicate rows. ` + + `Use incremental or --fill-constraints.`, + 'RESUME_UNSUPPORTED', + ) + } + + let exportResult: { jobId: number; gcsPrefix: string; rowCount: number } + // Files already loaded into staging by the prior (resumed) job, and rows it had already merged. + let resumeProgressDone = 0 + let resumeRowCountPg = 0 + if (opts.resumeJobId != null) { + // Reuse the prior job's exact export by id — bypasses reuse-by-kind (which excludes 'failed' + // jobs and could grab a different, older export) and skips the multi-hour BQ scan entirely. + // Validate hard here (non-retryable) so a bad id fails immediately instead of corrupting state. + const prior = await getResumeExport({ jobId: opts.resumeJobId }) + if (!prior) { + throw ApplicationFailure.nonRetryable( + `resume job ${opts.resumeJobId} not found`, + 'RESUME_INVALID', + ) + } + if (prior.jobKind !== 'package_dependencies') { + throw ApplicationFailure.nonRetryable( + `resume job ${opts.resumeJobId} is kind '${prior.jobKind}', expected 'package_dependencies'`, + 'RESUME_INVALID', + ) + } + // Resumable = has an export to reuse with unfinished merge work: exported (no chunk merged yet), + // loading/merging (interrupted mid-run), failed (the common case). done = nothing to do; + // cleaned = parquet already deleted; pending/exporting = no export produced yet. + const RESUMABLE_STATUSES = ['exported', 'loading', 'merging', 'failed'] + if (!RESUMABLE_STATUSES.includes(prior.status)) { + throw ApplicationFailure.nonRetryable( + `resume job ${opts.resumeJobId} has status '${prior.status}' — not resumable ` + + `(expected one of: ${RESUMABLE_STATUSES.join(', ')})`, + 'RESUME_INVALID', + ) + } + if (!prior.gcsPrefix) { throw ApplicationFailure.nonRetryable( - `edge snapshot quality guard failed for ${opts.today}: ${guard.reason}. ` + - `Slack alert sent. Aborting before export to preserve existing package_dependencies and compute.`, - 'EDGE_SNAPSHOT_GUARD', + `resume job ${opts.resumeJobId} has no gcs_prefix — nothing was exported to resume from`, + 'RESUME_INVALID', ) } + exportResult = { jobId: prior.jobId, gcsPrefix: prior.gcsPrefix, rowCount: 0 } + resumeProgressDone = prior.progressDone + resumeRowCountPg = prior.rowCountPg + } else { + // Guard against corrupt deps.dev resolved-graph snapshots BEFORE the (multi-hour) export. + // Skip when reusing a prior export — we're re-importing already-validated parquet, not + // scanning the live snapshot. Both full (*Latest = newest snapshot) and incremental can hit + // a bad snapshot, so the guard runs for both. Probes only resolved-graph ecosystems; a clean + // GO/NUGET-only run finds no canaries and passes through. + // + // Option A only: the guard's canary ratios are DependencyGraphEdges-schema-specific. Option B + // ingests the separate Dependencies/DependenciesLatest table, which the 2026-06 corruption was + // never observed in and has no calibrated baseline — probing Edges there would "validate" a table + // we don't ingest (false confidence) and could abort a healthy Option B run when only Edges is bad. + // Option B is a manual, non-scheduled cost-experiment path (--deps-table-b); leave it unguarded by + // design rather than invent a guard for an unproven threat. Option A is the production default. + if (!opts.reuseExports && tableOption === 'A') { + const guard = await checkEdgeSnapshotQuality({ + snapshotDate: opts.today, + ecosystems, + fullScan, + }) + if (!guard.ok) { + throw ApplicationFailure.nonRetryable( + `edge snapshot quality guard failed for ${opts.today}: ${guard.reason}. ` + + `Slack alert sent. Aborting before export to preserve existing package_dependencies and compute.`, + 'EDGE_SNAPSHOT_GUARD', + ) + } + } + // Fill mode always uses full SQL — needs all rows to find which have NULL version_constraint in DB. + const sql = fullScan + ? buildDepsFullSql(ecosystems, tableOption) + : buildDepsIncrementalSql(opts.today, opts.watermark ?? '', ecosystems, tableOption) + + exportResult = await bqExportToGcs({ + jobKind: 'package_dependencies', + sql, + runId: opts.runId, + // Report the PHYSICAL scan mode, not the requested one. A fill run (fillConstraints) forces a + // full *Latest scan even when opts.syncMode is 'incremental'; passing the raw mode would record + // the job as incremental and make bqExportToGcs pick the INCREMENTAL byte-ceiling env override + // for a query that's actually full. fullScan already gates the SQL + maxBytesGb below. + syncMode: fullScan ? 'full' : opts.syncMode, + snapshotAt: opts.today, + // Full/fill scan the *Latest views (everything) → 25000. Incremental is a snapshot edge-diff + // (today vs watermark partitions of DependencyGraphEdges + GoRequirements + NuGetRequirements); + // measured ~4.1TB for Option A. 10000 leaves ~2.4x headroom and still trips a runaway full-table + // scan. Overridable via BQ_DATASET_INGEST_PACKAGE_DEPENDENCIES[_INCREMENTAL]_MAX_BQ_GB (see README). + maxBytesGb: fullScan ? 25000 : 10000, + reuseExports: opts.reuseExports, + exportName: opts.exportName, + ecosystems, + }) } - // Fill mode always uses full SQL — needs all rows to find which have NULL version_constraint in DB. - const sql = fullScan - ? buildDepsFullSql(ecosystems, tableOption) - : buildDepsIncrementalSql(opts.today, opts.watermark ?? '', ecosystems, tableOption) - - const exportResult = await bqExportToGcs({ - jobKind: 'package_dependencies', - sql, - runId: opts.runId, - // Report the PHYSICAL scan mode, not the requested one. A fill run (fillConstraints) forces a - // full *Latest scan even when opts.syncMode is 'incremental'; passing the raw mode would record - // the job as incremental and make bqExportToGcs pick the INCREMENTAL byte-ceiling env override - // for a query that's actually full. fullScan already gates the SQL + maxBytesGb below. - syncMode: fullScan ? 'full' : opts.syncMode, - snapshotAt: opts.today, - // Full/fill scan the *Latest views (everything) → 25000. Incremental is a snapshot edge-diff - // (today vs watermark partitions of DependencyGraphEdges + GoRequirements + NuGetRequirements); - // measured ~4.1TB for Option A. 10000 leaves ~2.4x headroom and still trips a runaway full-table - // scan. Overridable via BQ_DATASET_INGEST_PACKAGE_DEPENDENCIES[_INCREMENTAL]_MAX_BQ_GB (see README). - maxBytesGb: fullScan ? 25000 : 10000, - reuseExports: opts.reuseExports, - exportName: opts.exportName, - ecosystems, - }) const { fileNames, rowCounts } = await listParquetFiles({ gcsPrefix: exportResult.gcsPrefix }) const totalFiles = fileNames.length const totalRows = rowCounts.reduce((a, b) => a + b, 0) + // Resume must find the exact parquet it recorded. Empty means GCS expired/deleted the files — + // fail loudly rather than fall through to the 0-row early-return below, which would mark the + // job 'done' and silently abandon the un-merged tail. + if (opts.resumeJobId != null && (totalFiles === 0 || totalRows === 0)) { + throw ApplicationFailure.nonRetryable( + `resume job ${opts.resumeJobId}: no parquet files at ${exportResult.gcsPrefix} ` + + `(export expired or deleted) — run a fresh export instead of --resume-job`, + 'RESUME_INVALID', + ) + } + if (totalFiles === 0 || totalRows === 0) { await mergeStagingToTable({ jobId: exportResult.jobId, @@ -305,11 +386,18 @@ export async function ingestDependencies(opts: { ? Math.max(1, Math.round((ROWS_PER_CHUNK * fileNames.length) / totalRows)) : Math.min(fileNames.length, 2) const totalChunks = Math.ceil(fileNames.length / filesPerChunk) - let priorRowsAffected = 0 + // Resume: the prior job recorded files loaded (progress:done) but not a per-merge-chunk index, + // and the merge lags the load by up to one chunk. Restart one chunk before where the load + // stopped and let ON CONFLICT no-op the overlap. Clamp to [0, totalChunks). Seed priorRowsAffected + // with the prior job's merged count so the final 'done' row_count_pg stays roughly accurate. + const startChunk = resume + ? Math.max(0, Math.min(Math.floor(resumeProgressDone / filesPerChunk) - 1, totalChunks - 1)) + : 0 + let priorRowsAffected = resume ? resumeRowCountPg : 0 let priorStagingRows = 0 const priorTableRowCounts: Record = {} - for (let chunkIndex = 0; chunkIndex < totalChunks; chunkIndex++) { + for (let chunkIndex = startChunk; chunkIndex < totalChunks; chunkIndex++) { const start = chunkIndex * filesPerChunk const chunk = fileNames.slice(start, start + filesPerChunk) const isLastChunk = chunkIndex === totalChunks - 1 diff --git a/services/apps/packages_worker/src/scripts/triggerBootstrap.ts b/services/apps/packages_worker/src/scripts/triggerBootstrap.ts index a9891cb948..c7aade9a98 100644 --- a/services/apps/packages_worker/src/scripts/triggerBootstrap.ts +++ b/services/apps/packages_worker/src/scripts/triggerBootstrap.ts @@ -45,6 +45,10 @@ Options: rows where it is currently NULL. Use after a --deps-table-b run to backfill missing version_constraint values without re-inserting rows. Only affects package_dependencies. + --resume-job Resume a partially-merged package_dependencies job by id. Skips BQ, + reuses that job's exact parquet export, and restarts the chunk loop + where its staging load left off (idempotent ON CONFLICT overlap). + Requires: --kinds package_dependencies AND (incremental | --fill-constraints). --help Show this help Examples: @@ -59,6 +63,7 @@ Examples: pnpm trigger-bootstrap:local full --kinds dependent_counts pnpm trigger-bootstrap:local full --kinds packages,versions pnpm trigger-bootstrap:local full --kinds dependent_counts --snapshot-date 2026-06-04 + pnpm trigger-bootstrap:local incremental --kinds package_dependencies --resume-job 194 ` async function main(): Promise { @@ -97,6 +102,23 @@ async function main(): Promise { process.exit(1) } + const resumeJobIdx = args.indexOf('--resume-job') + if ( + resumeJobIdx !== -1 && + (resumeJobIdx + 1 >= args.length || args[resumeJobIdx + 1].startsWith('--')) + ) { + console.error('--resume-job requires a value (numeric job id)') + process.exit(1) + } + let resumeJobId: number | undefined + if (resumeJobIdx !== -1) { + resumeJobId = Number(args[resumeJobIdx + 1]) + if (!Number.isInteger(resumeJobId) || resumeJobId <= 0) { + console.error(`--resume-job must be a positive integer, got: ${args[resumeJobIdx + 1]}`) + process.exit(1) + } + } + const kindsIdx = args.indexOf('--kinds') if (kindsIdx !== -1 && (kindsIdx + 1 >= args.length || args[kindsIdx + 1].startsWith('--'))) { console.error('--kinds requires a value') @@ -115,7 +137,13 @@ async function main(): Promise { const positional: string[] = [] for (let i = 0; i < args.length; i++) { if (args[i].startsWith('--')) { - if (args[i] === '--export-name' || args[i] === '--kinds' || args[i] === '--snapshot-date') i++ // skip value + if ( + args[i] === '--export-name' || + args[i] === '--kinds' || + args[i] === '--snapshot-date' || + args[i] === '--resume-job' + ) + i++ // skip value continue } positional.push(args[i]) @@ -142,6 +170,21 @@ async function main(): Promise { } } + // Resume is deps-only and merge-idempotent-only: it re-runs already-committed chunks, safe under + // incremental (ON CONFLICT DO NOTHING) or --fill-constraints (ON CONFLICT DO UPDATE), not a full load. + if (resumeJobId !== undefined) { + if (!kinds || kinds.length !== 1 || kinds[0] !== 'package_dependencies') { + console.error('--resume-job requires: --kinds package_dependencies (resume is deps-only)') + process.exit(1) + } + if (mode !== 'incremental' && !fillConstraints) { + console.error( + '--resume-job requires incremental mode or --fill-constraints (idempotent merges only)', + ) + process.exit(1) + } + } + const cfg = TEMPORAL_CONFIG() if (!cfg.serverUrl || !cfg.namespace) { console.error('Missing CROWD_TEMPORAL_SERVER_URL or CROWD_TEMPORAL_NAMESPACE') @@ -153,7 +196,8 @@ async function main(): Promise { const ecosystemSuffix = ecosystems ? `-${ecosystems.join('-').toLowerCase()}` : '' const reuseSuffix = reuseExports ? '-reuse' : '' const tableSuffix = depsTableOption === 'B' ? '-depsB' : '' - const workflowId = `bootstrap-osspckgs-${mode}${ecosystemSuffix}${reuseSuffix}${tableSuffix}-${Date.now()}` + const resumeSuffix = resumeJobId !== undefined ? `-resume${resumeJobId}` : '' + const workflowId = `bootstrap-osspckgs-${mode}${ecosystemSuffix}${reuseSuffix}${tableSuffix}${resumeSuffix}-${Date.now()}` const handle = await client.workflow.start(bootstrapOsspckgs, { taskQueue: 'bq-dataset-ingest', workflowId, @@ -167,6 +211,7 @@ async function main(): Promise { exportName, snapshotDate, fillConstraints, + resumeJobId, }, ], }) @@ -178,6 +223,7 @@ async function main(): Promise { depsTableOption === 'B' ? '--deps-table-b' : '', exportName ? `--export-name ${exportName}` : '', fillConstraints ? '--fill-constraints' : '', + resumeJobId !== undefined ? `--resume-job ${resumeJobId}` : '', ].filter(Boolean) console.log( `Started workflow ${handle.workflowId}${ecosystems ? ` (ecosystems: ${ecosystems.join(', ')})` : ''}${flags.length ? ` [${flags.join(' ')}]` : ''}`, diff --git a/services/libs/data-access-layer/src/osspckgs/ingestJobs.ts b/services/libs/data-access-layer/src/osspckgs/ingestJobs.ts index 71c536d3d8..24254a1638 100644 --- a/services/libs/data-access-layer/src/osspckgs/ingestJobs.ts +++ b/services/libs/data-access-layer/src/osspckgs/ingestJobs.ts @@ -86,6 +86,47 @@ export async function findLatestExportedJobByKind( // Returns an existing exported job for the given GCS prefix so callers can // skip re-running the BQ export when retrying a failed workflow. +// Loads the fields needed to resume a partially-merged chunked job by id: the exact GCS export +// path (so the same parquet files — and therefore the same chunk boundaries — are re-listed) plus +// the file-level load progress and rows-merged-so-far. progressDone/progressTotal come from the +// table_row_counts JSONB written by updateLoadingProgress. Returns null if the job id is unknown. +export async function getIngestJobForResume( + qx: QueryExecutor, + id: number, +): Promise<{ + id: number + jobKind: string + status: string + syncMode: string + gcsPrefix: string | null + progressDone: number + progressTotal: number + rowCountPg: number +} | null> { + const row = await qx.selectOneOrNone( + ` + SELECT id, job_kind, status, sync_mode, gcs_prefix, row_count_pg, + COALESCE((table_row_counts->>'progress:done')::int, 0) AS progress_done, + COALESCE((table_row_counts->>'progress:total')::int, 0) AS progress_total + FROM osspckgs_ingest_jobs + WHERE id = $(id) + `, + { id }, + ) + return row + ? { + id: Number(row.id), + jobKind: row.job_kind as string, + status: row.status as string, + syncMode: row.sync_mode as string, + gcsPrefix: (row.gcs_prefix as string | null) ?? null, + progressDone: Number(row.progress_done ?? 0), + progressTotal: Number(row.progress_total ?? 0), + rowCountPg: Number(row.row_count_pg ?? 0), + } + : null +} + export async function findExportedJobByGcsPrefix( qx: QueryExecutor, gcsPrefix: string, From 92bef2c8374c185eae1d019f58a365eb4f74fa4a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Uro=C5=A1=20Marolt?= Date: Fri, 3 Jul 2026 10:22:41 +0200 Subject: [PATCH 04/15] fix: show 'merging' phase step during osspckgs merge loop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Uroš Marolt --- .../src/deps-dev/workflows/ingestDependencies.ts | 5 +++++ services/apps/packages_worker/src/scripts/monitorOsspckgs.ts | 4 +++- 2 files changed, 8 insertions(+), 1 deletion(-) diff --git a/services/apps/packages_worker/src/deps-dev/workflows/ingestDependencies.ts b/services/apps/packages_worker/src/deps-dev/workflows/ingestDependencies.ts index 26072caec4..ecb3b05a3f 100644 --- a/services/apps/packages_worker/src/deps-dev/workflows/ingestDependencies.ts +++ b/services/apps/packages_worker/src/deps-dev/workflows/ingestDependencies.ts @@ -397,6 +397,11 @@ export async function ingestDependencies(opts: { let priorStagingRows = 0 const priorTableRowCounts: Record = {} + // Advance the phase label to 'merging' before the loop. The merge itself sets status='merging' + // but never a step, so without this the monitor keeps showing the last phase (creating_lookup for + // fill/incremental, drop_indexes for full) alongside the 'merging' status. Set once, not per chunk. + await setJobStep({ jobId: exportResult.jobId, step: 'merging' }) + for (let chunkIndex = startChunk; chunkIndex < totalChunks; chunkIndex++) { const start = chunkIndex * filesPerChunk const chunk = fileNames.slice(start, start + filesPerChunk) diff --git a/services/apps/packages_worker/src/scripts/monitorOsspckgs.ts b/services/apps/packages_worker/src/scripts/monitorOsspckgs.ts index d194644604..f88afd7559 100644 --- a/services/apps/packages_worker/src/scripts/monitorOsspckgs.ts +++ b/services/apps/packages_worker/src/scripts/monitorOsspckgs.ts @@ -114,7 +114,9 @@ function statusStr(status: string, step?: string | null, stuck?: boolean) { const c = stuck ? A.yellow : (STATUS_COLOR[status as keyof typeof STATUS_COLOR] ?? '') const i = stuck ? '⚠' : (STATUS_ICON[status as keyof typeof STATUS_ICON] ?? '?') const label = - step && !['done', 'failed', 'cleaned'].includes(status) ? `${status}·${step}` : status + step && step !== status && !['done', 'failed', 'cleaned'].includes(status) + ? `${status}·${step}` + : status return `${c}${i} ${label}${A.reset}` } From 1fb19328e5a74dd34361aee3b35f1dedf3bd5a2c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Uro=C5=A1=20Marolt?= Date: Fri, 3 Jul 2026 10:41:25 +0200 Subject: [PATCH 05/15] fix: eliminate flicker in osspckgs monitor with buffered redraw MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Uroš Marolt --- .../src/scripts/monitorOsspckgs.ts | 28 ++++++++++++++++--- 1 file changed, 24 insertions(+), 4 deletions(-) diff --git a/services/apps/packages_worker/src/scripts/monitorOsspckgs.ts b/services/apps/packages_worker/src/scripts/monitorOsspckgs.ts index f88afd7559..e816f16d77 100644 --- a/services/apps/packages_worker/src/scripts/monitorOsspckgs.ts +++ b/services/apps/packages_worker/src/scripts/monitorOsspckgs.ts @@ -22,6 +22,8 @@ const A = { altOn: '\x1b[?1049h', altOff: '\x1b[?1049l', clear: '\x1b[2J\x1b[H', + home: '\x1b[H', // cursor to top-left without clearing — overwrite in place (no flicker) + eos: '\x1b[0J', // erase from cursor to end of screen — clears leftover rows from a taller frame // fg black: '\x1b[30m', red: '\x1b[31m', @@ -44,10 +46,18 @@ const move = (row: number, col: number) => `${ESC}${row};${col}H` const clearLine = () => `${ESC}2K` const w = process.stdout +// When frameBuf is non-null a render() is in progress: accumulate the whole frame into one string +// and flush it with a single w.write() so the terminal paints it as one frame (no partial-frame +// flicker). Outside render(), write() flushes immediately as before. +let frameBuf: string | null = null +let maxRow = 0 + function write(s: string) { - w.write(s) + if (frameBuf !== null) frameBuf += s + else w.write(s) } function writeln(row: number, col: number, s: string) { + if (row > maxRow) maxRow = row write(move(row, col) + clearLine() + s) } @@ -786,7 +796,17 @@ function computeTotalEta(job: any) { function render() { const { rows, columns } = process.stdout - write(A.reset + A.hide + A.clear) + // Buffer the whole frame, then flush once. Home (not clear) + overwrite-in-place kills the flicker; + // eos at the end erases any rows a shorter frame left behind (e.g. detail panel closed, resize). + frameBuf = '' + maxRow = 0 + write(A.reset + A.hide + A.home) + const flush = () => { + write(move(maxRow + 1, 1) + A.eos + A.show) + const out = frameBuf ?? '' + frameBuf = null + w.write(out) + } const refreshStr = lastRefresh ? `last refresh: ${lastRefresh}` : 'loading…' const title = ` ${A.bold}${A.cyan}OSSPCKGS Monitor${A.reset} ${A.dim}${refreshStr}${A.reset}` @@ -796,7 +816,7 @@ function render() { if (error) { writeln(3, 1, `${A.red}DB Error: ${error}${A.reset}`) - write(A.show) + flush() return } @@ -865,7 +885,7 @@ function render() { } } - write(A.show) + flush() } // ── Data ─────────────────────────────────────────────────────────────────────── From f1192cb68707cf064cdb30daff1d17c886114dcf Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Uro=C5=A1=20Marolt?= Date: Fri, 3 Jul 2026 15:26:21 +0200 Subject: [PATCH 06/15] refactor: remove index drop/recreate from osspckgs ingest (CM-1296) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Uroš Marolt --- .../src/deps-dev/activities/index.ts | 4 - .../managePackageDepsConstraints.ts | 93 ------- .../activities/managePackageDepsIndexes.ts | 183 -------------- .../activities/manageVersionsConstraints.ts | 76 ------ .../activities/manageVersionsIndexes.ts | 148 ------------ .../src/deps-dev/queries/depsSql.ts | 2 +- .../deps-dev/workflows/ingestDependencies.ts | 227 +++++------------- .../src/deps-dev/workflows/ingestVersions.ts | 165 ++++--------- 8 files changed, 105 insertions(+), 793 deletions(-) delete mode 100644 services/apps/packages_worker/src/deps-dev/activities/managePackageDepsConstraints.ts delete mode 100644 services/apps/packages_worker/src/deps-dev/activities/managePackageDepsIndexes.ts delete mode 100644 services/apps/packages_worker/src/deps-dev/activities/manageVersionsConstraints.ts delete mode 100644 services/apps/packages_worker/src/deps-dev/activities/manageVersionsIndexes.ts diff --git a/services/apps/packages_worker/src/deps-dev/activities/index.ts b/services/apps/packages_worker/src/deps-dev/activities/index.ts index 1fd3a15b0d..1e578ae4ee 100644 --- a/services/apps/packages_worker/src/deps-dev/activities/index.ts +++ b/services/apps/packages_worker/src/deps-dev/activities/index.ts @@ -1,10 +1,6 @@ export * from './bqExportToGcs' export * from './setJobStep' export * from './createVersionsLookup' -export * from './managePackageDepsConstraints' -export * from './managePackageDepsIndexes' -export * from './manageVersionsConstraints' -export * from './manageVersionsIndexes' export * from './listParquetFiles' export * from './gcsParquetToStaging' export * from './mergeStagingToTable' diff --git a/services/apps/packages_worker/src/deps-dev/activities/managePackageDepsConstraints.ts b/services/apps/packages_worker/src/deps-dev/activities/managePackageDepsConstraints.ts deleted file mode 100644 index 22369a5a01..0000000000 --- a/services/apps/packages_worker/src/deps-dev/activities/managePackageDepsConstraints.ts +++ /dev/null @@ -1,93 +0,0 @@ -import { getServiceChildLogger } from '@crowd/logging' - -import { getPackagesDb } from '../../db' - -const log = getServiceChildLogger('managePackageDepsConstraints') - -// FK constraints on package_dependencies — dropped before full-load INSERT and rebuilt after. -// Dropping FKs eliminates the per-row trigger checks (referential integrity lookups against -// packages/versions) on every inserted row. Safe to drop because the INSERT...SELECT JOINs -// against packages/versions guarantee only valid rows are inserted. -// On Oracle Cloud managed PostgreSQL, session_replication_role=replica is blocked even with -// the REPLICATION role granted, so this is the only way to skip FK trigger overhead. -// Only fetch FKs referencing parent tables (not partition children). -// Oracle Cloud PG creates per-partition FK constraints automatically when a FK references a -// partitioned table — one on the parent (e.g. → versions) plus one per partition (→ versions_p0 -// … versions_p31). Dropping the parent FK cascades and removes all partition-level FKs. -// Filtering out partition-referencing FKs (those whose confrelid has a parent in pg_inherits) -// avoids trying to drop already-cascaded constraints and hitting "does not exist" errors. -const FK_CONSTRAINTS_SQL = ` - SELECT conname, - pg_get_constraintdef(oid) AS condef - FROM pg_constraint - WHERE conrelid = 'package_dependencies'::regclass - AND contype = 'f' - AND NOT EXISTS ( - SELECT 1 FROM pg_inherits WHERE inhrelid = confrelid - ) -` - -// Known FK definitions — used to recreate constraints after full load. -// Matches the migration V1779710880__initial_schema.sql. -const FK_DEFINITIONS = [ - { - columns: 'depends_on_id', - sql: `ALTER TABLE package_dependencies ADD FOREIGN KEY (depends_on_id) REFERENCES packages (id)`, - }, - { - columns: 'version_id, package_id', - sql: `ALTER TABLE package_dependencies ADD FOREIGN KEY (version_id, package_id) REFERENCES versions (id, package_id)`, - }, - { - columns: 'depends_on_version_id, depends_on_id', - sql: `ALTER TABLE package_dependencies ADD FOREIGN KEY (depends_on_version_id, depends_on_id) REFERENCES versions (id, package_id)`, - }, -] - -export async function dropPackageDepsConstraints(): Promise<{ dropped: string[] }> { - const qx = await getPackagesDb() - - const rows: Array<{ conname: string; condef: string }> = await qx.select(FK_CONSTRAINTS_SQL) - - if (rows.length === 0) { - log.info('No FK constraints found on package_dependencies — already dropped or never created') - return { dropped: [] } - } - - log.info( - { constraints: rows.map((r) => r.conname) }, - 'Dropping FK constraints on package_dependencies', - ) - - for (const row of rows) { - await qx.result(`ALTER TABLE package_dependencies DROP CONSTRAINT IF EXISTS "${row.conname}"`) - } - - log.info({ dropped: rows.map((r) => r.conname) }, 'FK constraints dropped') - return { dropped: rows.map((r) => r.conname) } -} - -export async function rebuildPackageDepsConstraints(): Promise<{ rebuilt: string[] }> { - const qx = await getPackagesDb() - - const existing: Array<{ conname: string; condef: string }> = await qx.select(FK_CONSTRAINTS_SQL) - - const rebuilt: string[] = [] - - for (const def of FK_DEFINITIONS) { - const alreadyExists = existing.some((r) => r.condef.includes(`FOREIGN KEY (${def.columns})`)) - if (alreadyExists) { - log.info({ columns: def.columns }, 'FK constraint already exists, skipping') - continue - } - log.info( - { columns: def.columns }, - 'Rebuilding FK constraint on package_dependencies (validates all rows)', - ) - await qx.result(def.sql) - rebuilt.push(def.columns) - } - - log.info({ rebuilt }, 'FK constraints rebuilt') - return { rebuilt } -} diff --git a/services/apps/packages_worker/src/deps-dev/activities/managePackageDepsIndexes.ts b/services/apps/packages_worker/src/deps-dev/activities/managePackageDepsIndexes.ts deleted file mode 100644 index 12ffac79fe..0000000000 --- a/services/apps/packages_worker/src/deps-dev/activities/managePackageDepsIndexes.ts +++ /dev/null @@ -1,183 +0,0 @@ -import { getServiceChildLogger } from '@crowd/logging' - -import { getPackagesDb } from '../../db' - -const log = getServiceChildLogger('managePackageDepsIndexes') - -// Secondary indexes on package_dependencies — safe to drop before full-load INSERT and rebuild after. -// The PRIMARY KEY (id, depends_on_id) stays. The UNIQUE (version_id, depends_on_id, dependency_kind) -// is also dropped/rebuilt for full loads — dropping it removes the per-row B-tree probe on INSERT, -// which is the dominant cost when loading 1B+ rows. The full-load INSERT uses plain INSERT (no ON -// CONFLICT) while the constraint is absent. If BQ data has duplicates the ADD CONSTRAINT will fail, -// which tells us we need BQ-level dedup first. -const UNIQUE_CONSTRAINT_SQL = ` - SELECT conname - FROM pg_constraint c - JOIN pg_class t ON t.oid = c.conrelid - WHERE t.relname = 'package_dependencies' - AND c.contype = 'u' - LIMIT 1 -` - -// Only secondary (non-constraint) indexes — PK excluded, UNIQUE excluded (handled separately above). -const NON_CONSTRAINT_INDEXES_SQL = ` - SELECT i.indexname, i.indexdef - FROM pg_indexes i - WHERE i.tablename = 'package_dependencies' - AND NOT EXISTS ( - SELECT 1 FROM pg_constraint c - JOIN pg_class t ON t.oid = c.conrelid AND t.relname = 'package_dependencies' - JOIN pg_class ix ON ix.oid = c.conindid AND ix.relname = i.indexname - WHERE c.contype IN ('p', 'u') - ) -` - -const SECONDARY_INDEXES: Array<{ columns: string; createSql: string }> = [ - { - columns: 'depends_on_id, depends_on_version_id', - createSql: `CREATE INDEX ON package_dependencies (depends_on_id, depends_on_version_id)`, - }, - { - columns: 'version_id', - createSql: `CREATE INDEX ON package_dependencies (version_id)`, - }, -] - -// Drops secondary (non-constraint) indexes on package_dependencies. -// Idempotent: DROP INDEX IF EXISTS on each found index. -// For a partitioned table, dropping the parent index cascades to all 64 partitions. -export async function dropPackageDepsIndexes(): Promise<{ - dropped: string[] - droppedConstraint: string | null -}> { - const qx = await getPackagesDb() - - const rows: Array<{ indexname: string }> = await qx.select(NON_CONSTRAINT_INDEXES_SQL) - const names = rows.map((r: { indexname: string }) => r.indexname) - - if (names.length > 0) { - log.info({ indexes: names }, 'Dropping secondary indexes on package_dependencies') - for (const name of names) { - await qx.result(`DROP INDEX IF EXISTS ${name}`) - } - } else { - log.info('No secondary indexes found — already dropped or never created') - } - - // Also drop the UNIQUE constraint so full-load INSERTs skip the per-row B-tree probe. - // Plain INSERT (no ON CONFLICT) is used while the constraint is absent. - const constraintRow = await qx.selectOneOrNone(UNIQUE_CONSTRAINT_SQL) - let droppedConstraint: string | null = null - if (constraintRow) { - log.info( - { constraint: constraintRow.conname }, - 'Dropping UNIQUE constraint on package_dependencies', - ) - await qx.result(`ALTER TABLE package_dependencies DROP CONSTRAINT ${constraintRow.conname}`) - droppedConstraint = constraintRow.conname - } else { - log.info('UNIQUE constraint not found — already dropped') - } - - log.info({ indexes: names, droppedConstraint }, 'Drop phase complete') - return { dropped: names, droppedConstraint } -} - -// Rebuilds secondary indexes on package_dependencies. -// Idempotent: existence check uses exact column-list match — "(version_id)" — to avoid -// false positives from substring matches (e.g. "depends_on_version_id" contains "version_id"). -// Only non-constraint indexes are checked so the UNIQUE constraint cannot cause a false positive. -export async function rebuildPackageDepsIndexes(): Promise<{ - rebuilt: string[] - rebuiltConstraint: boolean -}> { - const qx = await getPackagesDb() - - const existing: Array<{ indexdef: string }> = await qx.select(NON_CONSTRAINT_INDEXES_SQL) - const existingDefs = existing.map((r: { indexdef: string }) => r.indexdef.toLowerCase()) - - const toRebuild = SECONDARY_INDEXES.filter( - (idx) => !existingDefs.some((def: string) => def.includes(`(${idx.columns})`)), - ) - const rebuilt: string[] = [] - - // Build indexes in parallel — each on its own connection so they run concurrently. - await Promise.all( - toRebuild.map(async (idx) => { - const conn = await getPackagesDb() - log.info({ columns: idx.columns }, 'Creating index on package_dependencies') - await conn.result(idx.createSql) - rebuilt.push(idx.columns) - }), - ) - - // Remove cross-chunk duplicates before rebuilding the UNIQUE constraint. - // DISTINCT ON deduplicates within a single chunk, but the same (root, dep) pair can appear in - // multiple BQ parquet files (different chunks), each inserting its own row. The DELETE here is - // a one-time scan at the end of the full load — cheaper than failing the constraint rebuild. - // depends_on_id is included in the USING join so PG routes each delete to the correct partition. - // - // Run per partition (depends_on_id % 64 = p) so each iteration prunes to one of the 64 - // partitions instead of scanning all 1.15B rows at once. Same total rows read; each pass - // fits in work_mem and avoids cross-partition sort. - // Run in parallel batches of DEDUP_CONCURRENCY to cut wall-clock from ~10h to ~1-2h. - const NUM_PARTITIONS = 64 - const DEDUP_CONCURRENCY = 8 - let totalDedupDeleted = 0 - for (let batch = 0; batch < NUM_PARTITIONS; batch += DEDUP_CONCURRENCY) { - const partitions = Array.from( - { length: Math.min(DEDUP_CONCURRENCY, NUM_PARTITIONS - batch) }, - (_, i) => batch + i, - ) - const counts = await Promise.all( - partitions.map(async (p) => { - const conn = await getPackagesDb() - // work_mem (not maintenance_work_mem) controls sort memory for window functions. - // 2GB avoids disk spill during the ROW_NUMBER() sort on each partition. - return conn.tx(async (tx) => { - await tx.result(`SET LOCAL work_mem = '2GB'`) - return tx.result(` - DELETE FROM package_dependencies pd - USING ( - SELECT id, depends_on_id - FROM ( - SELECT id, depends_on_id, - ROW_NUMBER() OVER ( - PARTITION BY version_id, depends_on_id, dependency_kind - ORDER BY id - ) AS rn - FROM package_dependencies - WHERE depends_on_id % ${NUM_PARTITIONS} = ${p} - ) sub - WHERE rn > 1 - ) dupes - WHERE pd.id = dupes.id AND pd.depends_on_id = dupes.depends_on_id - `) - }) - }), - ) - totalDedupDeleted += counts.reduce((sum, n) => sum + n, 0) - } - log.info( - { rowsDeleted: totalDedupDeleted }, - 'Cross-chunk duplicate rows removed from package_dependencies', - ) - - const constraintRow = await qx.selectOneOrNone(UNIQUE_CONSTRAINT_SQL) - let rebuiltConstraint = false - if (!constraintRow) { - log.info( - 'Rebuilding UNIQUE constraint on package_dependencies (this may take a while on large tables)', - ) - await qx.result( - `ALTER TABLE package_dependencies ADD UNIQUE (version_id, depends_on_id, dependency_kind)`, - ) - rebuiltConstraint = true - log.info('UNIQUE constraint rebuilt') - } else { - log.info({ constraint: constraintRow.conname }, 'UNIQUE constraint already exists, skipping') - } - - log.info({ rebuilt, rebuiltConstraint }, 'Rebuild phase complete') - return { rebuilt, rebuiltConstraint } -} diff --git a/services/apps/packages_worker/src/deps-dev/activities/manageVersionsConstraints.ts b/services/apps/packages_worker/src/deps-dev/activities/manageVersionsConstraints.ts deleted file mode 100644 index 7740770f43..0000000000 --- a/services/apps/packages_worker/src/deps-dev/activities/manageVersionsConstraints.ts +++ /dev/null @@ -1,76 +0,0 @@ -import { getServiceChildLogger } from '@crowd/logging' - -import { getPackagesDb } from '../../db' - -const log = getServiceChildLogger('manageVersionsConstraints') - -// FK constraints on versions — dropped before full-load INSERT and rebuilt after. -// Dropping FKs eliminates the per-row trigger checks (referential integrity lookups against -// packages) on every inserted row. Safe to drop because the INSERT...SELECT JOINs against -// packages guarantee only valid rows are inserted. -// On Oracle Cloud managed PostgreSQL, session_replication_role=replica is blocked even with -// the REPLICATION role granted, so this is the only way to skip FK trigger overhead. -// Only fetch FKs referencing parent tables (not partition children) — see managePackageDepsConstraints.ts -// for the full explanation. versions only has one FK (→ packages, not partitioned) so this filter -// is a no-op here, but kept for consistency and future safety. -const FK_CONSTRAINTS_SQL = ` - SELECT conname, - pg_get_constraintdef(oid) AS condef - FROM pg_constraint - WHERE conrelid = 'versions'::regclass - AND contype = 'f' - AND NOT EXISTS ( - SELECT 1 FROM pg_inherits WHERE inhrelid = confrelid - ) -` - -// Known FK definitions — used to recreate constraints after full load. -// Matches the migration V1779710880__initial_schema.sql. -const FK_DEFINITIONS = [ - { - columns: 'package_id', - sql: `ALTER TABLE versions ADD FOREIGN KEY (package_id) REFERENCES packages (id)`, - }, -] - -export async function dropVersionsConstraints(): Promise<{ dropped: string[] }> { - const qx = await getPackagesDb() - - const rows: Array<{ conname: string; condef: string }> = await qx.select(FK_CONSTRAINTS_SQL) - - if (rows.length === 0) { - log.info('No FK constraints found on versions — already dropped or never created') - return { dropped: [] } - } - - log.info({ constraints: rows.map((r) => r.conname) }, 'Dropping FK constraints on versions') - - for (const row of rows) { - await qx.result(`ALTER TABLE versions DROP CONSTRAINT IF EXISTS "${row.conname}"`) - } - - log.info({ dropped: rows.map((r) => r.conname) }, 'FK constraints dropped') - return { dropped: rows.map((r) => r.conname) } -} - -export async function rebuildVersionsConstraints(): Promise<{ rebuilt: string[] }> { - const qx = await getPackagesDb() - - const existing: Array<{ conname: string; condef: string }> = await qx.select(FK_CONSTRAINTS_SQL) - - const rebuilt: string[] = [] - - for (const def of FK_DEFINITIONS) { - const alreadyExists = existing.some((r) => r.condef.includes(`FOREIGN KEY (${def.columns})`)) - if (alreadyExists) { - log.info({ columns: def.columns }, 'FK constraint already exists, skipping') - continue - } - log.info({ columns: def.columns }, 'Rebuilding FK constraint on versions (validates all rows)') - await qx.result(def.sql) - rebuilt.push(def.columns) - } - - log.info({ rebuilt }, 'FK constraints rebuilt') - return { rebuilt } -} diff --git a/services/apps/packages_worker/src/deps-dev/activities/manageVersionsIndexes.ts b/services/apps/packages_worker/src/deps-dev/activities/manageVersionsIndexes.ts deleted file mode 100644 index 239385e609..0000000000 --- a/services/apps/packages_worker/src/deps-dev/activities/manageVersionsIndexes.ts +++ /dev/null @@ -1,148 +0,0 @@ -import { getServiceChildLogger } from '@crowd/logging' - -import { getPackagesDb } from '../../db' - -const log = getServiceChildLogger('manageVersionsIndexes') - -const UNIQUE_CONSTRAINT_SQL = ` - SELECT conname - FROM pg_constraint c - JOIN pg_class t ON t.oid = c.conrelid - WHERE t.relname = 'versions' - AND c.contype = 'u' - LIMIT 1 -` - -// Only secondary (non-constraint) indexes on the parent versions table. -// DROP INDEX on the parent cascades to all 32 partitions. -const NON_CONSTRAINT_INDEXES_SQL = ` - SELECT i.indexname, i.indexdef - FROM pg_indexes i - WHERE i.tablename = 'versions' - AND NOT EXISTS ( - SELECT 1 FROM pg_constraint c - JOIN pg_class t ON t.oid = c.conrelid AND t.relname = 'versions' - JOIN pg_class ix ON ix.oid = c.conindid AND ix.relname = i.indexname - WHERE c.contype IN ('p', 'u') - ) -` - -const SECONDARY_INDEXES: Array<{ columns: string; createSql: string; matchString?: string }> = [ - { - columns: 'published_at desc', - createSql: `CREATE INDEX ON versions (published_at DESC)`, - }, - { - columns: 'package_id', - createSql: `CREATE INDEX ON versions (package_id) WHERE is_latest`, - // Include WHERE clause so the check doesn't match a hypothetical plain (package_id) index. - matchString: '(package_id) where is_latest', - }, - { - columns: "ecosystem, coalesce(namespace, ''), name, number", - createSql: `CREATE INDEX ON versions (ecosystem, COALESCE(namespace, ''), name, number)`, - matchString: "(ecosystem, coalesce(namespace, ''), name, number)", - }, -] - -export async function dropVersionsIndexes(): Promise<{ - dropped: string[] - droppedConstraint: string | null -}> { - const qx = await getPackagesDb() - - const rows: Array<{ indexname: string }> = await qx.select(NON_CONSTRAINT_INDEXES_SQL) - const names = rows.map((r) => r.indexname) - - if (names.length > 0) { - log.info({ indexes: names }, 'Dropping secondary indexes on versions') - for (const name of names) { - await qx.result(`DROP INDEX IF EXISTS ${name}`) - } - } else { - log.info('No secondary indexes found — already dropped or never created') - } - - const constraintRow = await qx.selectOneOrNone(UNIQUE_CONSTRAINT_SQL) - let droppedConstraint: string | null = null - if (constraintRow) { - log.info({ constraint: constraintRow.conname }, 'Dropping UNIQUE constraint on versions') - await qx.result(`ALTER TABLE versions DROP CONSTRAINT ${constraintRow.conname}`) - droppedConstraint = constraintRow.conname - } else { - log.info('UNIQUE constraint not found — already dropped') - } - - log.info({ indexes: names, droppedConstraint }, 'Drop phase complete') - return { dropped: names, droppedConstraint } -} - -export async function rebuildVersionsIndexes(): Promise<{ - rebuilt: string[] - rebuiltConstraint: boolean -}> { - const qx = await getPackagesDb() - - const existing: Array<{ indexdef: string }> = await qx.select(NON_CONSTRAINT_INDEXES_SQL) - const existingDefs = existing.map((r) => r.indexdef.toLowerCase()) - - const toRebuild = SECONDARY_INDEXES.filter( - (idx) => !existingDefs.some((def) => def.includes(idx.matchString ?? `(${idx.columns})`)), - ) - const rebuilt: string[] = [] - - // Build indexes in parallel — each on its own connection so they run concurrently. - // maintenance_work_mem per connection: with 32 partitions and default 64MB, PG spills to - // disk on every partition; 2GB lets the sort fit in RAM and cuts build time dramatically. - await Promise.all( - toRebuild.map(async (idx) => { - const conn = await getPackagesDb() - await conn.tx(async (t) => { - await t.result(`SET LOCAL maintenance_work_mem = '2GB'`) - log.info({ columns: idx.columns }, 'Creating index on versions') - await t.result(idx.createSql) - }) - rebuilt.push(idx.columns) - }), - ) - - // Remove cross-chunk duplicates before rebuilding the UNIQUE constraint. - // versions is HASH-partitioned by package_id (32 partitions). Loop over each partition table - // directly so PG scans one partition at a time rather than all 32 at once. - const NUM_PARTITIONS = 32 - let totalDedupDeleted = 0 - for (let p = 0; p < NUM_PARTITIONS; p++) { - const table = `versions_p${p}` - const deleted = await qx.result(` - DELETE FROM ${table} v - USING ( - SELECT id - FROM ( - SELECT id, - ROW_NUMBER() OVER (PARTITION BY package_id, number ORDER BY id) AS rn - FROM ${table} - ) sub - WHERE rn > 1 - ) dupes - WHERE v.id = dupes.id - `) - totalDedupDeleted += deleted - } - if (totalDedupDeleted > 0) { - log.info({ rowsDeleted: totalDedupDeleted }, 'Cross-chunk duplicate rows removed from versions') - } - - const constraintRow = await qx.selectOneOrNone(UNIQUE_CONSTRAINT_SQL) - let rebuiltConstraint = false - if (!constraintRow) { - log.info('Rebuilding UNIQUE constraint on versions (this may take a while on large tables)') - await qx.result(`ALTER TABLE versions ADD UNIQUE (package_id, number)`) - rebuiltConstraint = true - log.info('UNIQUE constraint rebuilt') - } else { - log.info({ constraint: constraintRow.conname }, 'UNIQUE constraint already exists, skipping') - } - - log.info({ rebuilt, rebuiltConstraint }, 'Rebuild phase complete') - return { rebuilt, rebuiltConstraint } -} diff --git a/services/apps/packages_worker/src/deps-dev/queries/depsSql.ts b/services/apps/packages_worker/src/deps-dev/queries/depsSql.ts index 1acc249b2f..25c6d882d2 100644 --- a/services/apps/packages_worker/src/deps-dev/queries/depsSql.ts +++ b/services/apps/packages_worker/src/deps-dev/queries/depsSql.ts @@ -22,7 +22,7 @@ FROM \`bigquery-public-data.deps_dev_v1.GoRequirementsLatest\` g, UNNEST(g.DirectDependencies) AS d` // NuGet groups deps by TargetFramework — flatten all groups, dedup handled downstream -// by DISTINCT ON in MERGE_SQL_FULL and ON CONFLICT in MERGE_SQL. +// by ON CONFLICT in MERGE_SQL (and DISTINCT ON in the fill-constraints variant). const NUGET_FULL_PART = ` SELECT 'nuget' AS ecosystem, diff --git a/services/apps/packages_worker/src/deps-dev/workflows/ingestDependencies.ts b/services/apps/packages_worker/src/deps-dev/workflows/ingestDependencies.ts index ecb3b05a3f..71c5619a98 100644 --- a/services/apps/packages_worker/src/deps-dev/workflows/ingestDependencies.ts +++ b/services/apps/packages_worker/src/deps-dev/workflows/ingestDependencies.ts @@ -24,8 +24,8 @@ const { gcsParquetToStaging } = proxyActivities({ }) const { mergeStagingToTable } = proxyActivities({ - // A single ~1M-row chunk merge into the live-index/live-FK package_dependencies table (incremental - // does not drop indexes) can run long; 4h gives headroom over the previously-observed 2h overrun. + // A single ~1M-row chunk merge into the live-index/live-FK package_dependencies table can run + // long; 4h gives headroom over the previously-observed 2h overrun. startToCloseTimeout: '4 hours', retry: { maximumAttempts: 1 }, }) @@ -40,22 +40,6 @@ const { createVersionsLookup } = proxyActivities({ retry: { maximumAttempts: 2, initialInterval: '30 seconds' }, }) -const { dropPackageDepsIndexes, rebuildPackageDepsIndexes } = proxyActivities< - typeof depsDevActivities ->({ - // Index builds + parallel dedup on 1B+ rows — 24h covers worst-case sequential retry. - startToCloseTimeout: '24 hours', - retry: { maximumAttempts: 2, initialInterval: '1 minute' }, -}) - -const { dropPackageDepsConstraints, rebuildPackageDepsConstraints } = proxyActivities< - typeof depsDevActivities ->({ - // FK validation on 1B+ rows can take hours. - startToCloseTimeout: '24 hours', - retry: { maximumAttempts: 2, initialInterval: '1 minute' }, -}) - const { setJobStep } = proxyActivities({ startToCloseTimeout: '30 seconds', retry: { maximumAttempts: 3 }, @@ -119,46 +103,10 @@ LEFT JOIN versions dv ON dv.package_id = pd.id AND dv.number = sp.to_version ON CONFLICT (version_id, depends_on_id, dependency_kind) DO NOTHING ` -// Full-load variant: UNIQUE constraint is dropped before the chunk loop so plain INSERT is safe. -// DISTINCT ON (pv.id, pd.id) deduplicates before INSERT — BQ can emit multiple rows for the same -// (root_name, root_version, to_name) with different to_version values; both resolve to the same -// (version_id, depends_on_id) pair, which violates the UNIQUE constraint on rebuild. -const MERGE_SQL_FULL = ` -INSERT INTO package_dependencies ( - package_id, version_id, depends_on_id, depends_on_version_id, - version_constraint, dependency_kind, is_optional, created_at, updated_at -) -SELECT DISTINCT ON (pv.id, pd.id) - pv.package_id, pv.id, pd.id, dv.id, - sp.version_constraint, 'direct', FALSE, NOW(), NOW() -FROM staging.osspckgs_deps_raw sp -JOIN staging.osspckgs_versions_lookup pv ON pv.ecosystem = sp.ecosystem - AND pv.ns = CASE - WHEN sp.ecosystem = 'maven' THEN SPLIT_PART(sp.root_name, ':', 1) - WHEN sp.root_name LIKE '@%/%' THEN SPLIT_PART(sp.root_name, '/', 1) - ELSE '' END - AND pv.name = CASE - WHEN sp.ecosystem = 'maven' THEN SPLIT_PART(sp.root_name, ':', 2) - WHEN sp.root_name LIKE '@%/%' THEN SPLIT_PART(sp.root_name, '/', 2) - ELSE sp.root_name END - AND pv.number = sp.root_version -JOIN packages pd ON pd.ecosystem = sp.ecosystem - AND COALESCE(pd.namespace, '') = CASE - WHEN sp.ecosystem = 'maven' THEN SPLIT_PART(sp.to_name, ':', 1) - WHEN sp.to_name LIKE '@%/%' THEN SPLIT_PART(sp.to_name, '/', 1) - ELSE '' END - AND pd.name = CASE - WHEN sp.ecosystem = 'maven' THEN SPLIT_PART(sp.to_name, ':', 2) - WHEN sp.to_name LIKE '@%/%' THEN SPLIT_PART(sp.to_name, '/', 2) - ELSE sp.to_name END -LEFT JOIN versions dv ON dv.package_id = pd.id AND dv.number = sp.to_version -ORDER BY pv.id, pd.id, sp.to_version DESC NULLS LAST -` - -// Fill-constraints variant: UNIQUE constraint stays in place (not dropped), so ON CONFLICT is valid. -// Upserts version_constraint only for rows where it is currently NULL — safe to run against a table -// already populated by --deps-table-b (which sets version_constraint = NULL for all rows). -// DISTINCT ON matches MERGE_SQL_FULL to resolve duplicate (root, dep) pairs from BQ before the upsert. +// Fill-constraints variant: UNIQUE constraint stays in place, so ON CONFLICT is valid. Upserts +// version_constraint only for rows where it is currently NULL — safe to run against a table already +// populated by --deps-table-b (which sets version_constraint = NULL for all rows). DISTINCT ON +// resolves duplicate (root, dep) pairs from BQ (same root/dep, different to_version) before the upsert. const MERGE_SQL_FILL_CONSTRAINTS = ` INSERT INTO package_dependencies ( package_id, version_id, depends_on_id, depends_on_version_id, @@ -195,7 +143,7 @@ ON CONFLICT (version_id, depends_on_id, dependency_kind) DO UPDATE ` // SET LOCAL scopes settings to this transaction only. -// synchronous_commit=off skips WAL flush wait — safe for plain INSERT on full loads. +// synchronous_commit=off skips WAL flush wait — safe for these re-runnable ON CONFLICT merges. // max_parallel_workers_per_gather parallelises the SELECT side of INSERT...SELECT. // session_replication_role=replica would skip FK trigger checks but requires superuser — // blocked on Oracle Cloud managed PostgreSQL regardless of REPLICATION role. @@ -240,17 +188,6 @@ export async function ingestDependencies(opts: { const resume = opts.resumeJobId != null - // Resume is only safe for idempotent merges: incremental uses ON CONFLICT DO NOTHING and fill uses - // ON CONFLICT DO UPDATE, so re-running an overlapping chunk no-ops. A full (non-fill) load uses a - // plain INSERT with the UNIQUE constraint dropped, so a re-merged chunk would duplicate rows. - if (resume && fullScan && !isFill) { - throw ApplicationFailure.nonRetryable( - `resume (resumeJobId) is not supported for full loads — plain INSERT would duplicate rows. ` + - `Use incremental or --fill-constraints.`, - 'RESUME_UNSUPPORTED', - ) - } - let exportResult: { jobId: number; gcsPrefix: string; rowCount: number } // Files already loaded into staging by the prior (resumed) job, and rows it had already merged. let resumeProgressDone = 0 @@ -373,108 +310,62 @@ export async function ingestDependencies(opts: { await setJobStep({ jobId: exportResult.jobId, step: 'creating_lookup' }) await createVersionsLookup({ ecosystems }) - if (opts.syncMode === 'full' && !isFill) { - await setJobStep({ jobId: exportResult.jobId, step: 'drop_constraints' }) - await dropPackageDepsConstraints() - await setJobStep({ jobId: exportResult.jobId, step: 'drop_indexes' }) - await dropPackageDepsIndexes() - } - - try { - const filesPerChunk = - totalRows > 0 - ? Math.max(1, Math.round((ROWS_PER_CHUNK * fileNames.length) / totalRows)) - : Math.min(fileNames.length, 2) - const totalChunks = Math.ceil(fileNames.length / filesPerChunk) - // Resume: the prior job recorded files loaded (progress:done) but not a per-merge-chunk index, - // and the merge lags the load by up to one chunk. Restart one chunk before where the load - // stopped and let ON CONFLICT no-op the overlap. Clamp to [0, totalChunks). Seed priorRowsAffected - // with the prior job's merged count so the final 'done' row_count_pg stays roughly accurate. - const startChunk = resume - ? Math.max(0, Math.min(Math.floor(resumeProgressDone / filesPerChunk) - 1, totalChunks - 1)) - : 0 - let priorRowsAffected = resume ? resumeRowCountPg : 0 - let priorStagingRows = 0 - const priorTableRowCounts: Record = {} - - // Advance the phase label to 'merging' before the loop. The merge itself sets status='merging' - // but never a step, so without this the monitor keeps showing the last phase (creating_lookup for - // fill/incremental, drop_indexes for full) alongside the 'merging' status. Set once, not per chunk. - await setJobStep({ jobId: exportResult.jobId, step: 'merging' }) - - for (let chunkIndex = startChunk; chunkIndex < totalChunks; chunkIndex++) { - const start = chunkIndex * filesPerChunk - const chunk = fileNames.slice(start, start + filesPerChunk) - const isLastChunk = chunkIndex === totalChunks - 1 - // For full-load (non-fill), don't mark done here — rebuild runs after the chunk loop - const isFinal = isLastChunk && !(opts.syncMode === 'full' && !isFill) - - const { rowsLoaded } = await gcsParquetToStaging({ - jobId: exportResult.jobId, - stagingTable: STAGING_TABLE, - stagingDdl: STAGING_DDL, - pgColumns: PG_COLUMNS, - fileNames: chunk, - filesOffset: start, - totalFiles, - priorStagingRows, - }) - priorStagingRows += rowsLoaded - - const { rowsAffected, tableRowCounts } = await mergeStagingToTable({ - jobId: exportResult.jobId, - prepareSql: MERGE_PREPARE_SQL, - mergeSql: isFill - ? MERGE_SQL_FILL_CONSTRAINTS - : opts.syncMode === 'full' - ? MERGE_SQL_FULL - : MERGE_SQL, - tableNames: 'package_dependencies', - isFinal, - priorRowsAffected, - priorTableRowCounts, - chunkInfo: { index: chunkIndex, total: totalChunks }, - }) + const filesPerChunk = + totalRows > 0 + ? Math.max(1, Math.round((ROWS_PER_CHUNK * fileNames.length) / totalRows)) + : Math.min(fileNames.length, 2) + const totalChunks = Math.ceil(fileNames.length / filesPerChunk) + // Resume: the prior job recorded files loaded (progress:done) but not a per-merge-chunk index, + // and the merge lags the load by up to one chunk. Restart one chunk before where the load + // stopped and let ON CONFLICT no-op the overlap. Clamp to [0, totalChunks). Seed priorRowsAffected + // with the prior job's merged count so the final 'done' row_count_pg stays roughly accurate. + const startChunk = resume + ? Math.max(0, Math.min(Math.floor(resumeProgressDone / filesPerChunk) - 1, totalChunks - 1)) + : 0 + let priorRowsAffected = resume ? resumeRowCountPg : 0 + let priorStagingRows = 0 + const priorTableRowCounts: Record = {} + + // Advance the phase label to 'merging' before the loop. The merge itself sets status='merging' + // but never a step, so without this the monitor keeps showing the last phase (creating_lookup) + // alongside the 'merging' status. Set once, not per chunk. + await setJobStep({ jobId: exportResult.jobId, step: 'merging' }) + + for (let chunkIndex = startChunk; chunkIndex < totalChunks; chunkIndex++) { + const start = chunkIndex * filesPerChunk + const chunk = fileNames.slice(start, start + filesPerChunk) + const isFinal = chunkIndex === totalChunks - 1 + + const { rowsLoaded } = await gcsParquetToStaging({ + jobId: exportResult.jobId, + stagingTable: STAGING_TABLE, + stagingDdl: STAGING_DDL, + pgColumns: PG_COLUMNS, + fileNames: chunk, + filesOffset: start, + totalFiles, + priorStagingRows, + }) + priorStagingRows += rowsLoaded - priorRowsAffected += rowsAffected - if (!isFinal) { - for (const [k, v] of Object.entries(tableRowCounts)) { - priorTableRowCounts[k] = (priorTableRowCounts[k] ?? 0) + v - } - } - } + const { rowsAffected, tableRowCounts } = await mergeStagingToTable({ + jobId: exportResult.jobId, + prepareSql: MERGE_PREPARE_SQL, + mergeSql: isFill ? MERGE_SQL_FILL_CONSTRAINTS : MERGE_SQL, + tableNames: 'package_dependencies', + isFinal, + priorRowsAffected, + priorTableRowCounts, + chunkInfo: { index: chunkIndex, total: totalChunks }, + }) - if (opts.syncMode === 'full' && !isFill) { - await setJobStep({ jobId: exportResult.jobId, step: 'rebuild_indexes' }) - await rebuildPackageDepsIndexes() - await setJobStep({ jobId: exportResult.jobId, step: 'rebuild_constraints' }) - await rebuildPackageDepsConstraints() - // Finalize after rebuild — marks done with correct finishedAt - await mergeStagingToTable({ - jobId: exportResult.jobId, - mergeSql: [], - tableNames: [], - isFinal: true, - priorRowsAffected, - priorTableRowCounts, - }) - } - } catch (err) { - if (opts.syncMode === 'full' && !isFill) { - try { - await setJobStep({ jobId: exportResult.jobId, step: 'rebuild_indexes' }) - await rebuildPackageDepsIndexes() - } catch (_) { - /* best-effort */ - } - try { - await setJobStep({ jobId: exportResult.jobId, step: 'rebuild_constraints' }) - await rebuildPackageDepsConstraints() - } catch (_) { - /* best-effort */ + priorRowsAffected += rowsAffected + if (!isFinal) { + for (const [k, v] of Object.entries(tableRowCounts)) { + priorTableRowCounts[k] = (priorTableRowCounts[k] ?? 0) + v } } - throw err } + return { rowCountBq: exportResult.rowCount } } diff --git a/services/apps/packages_worker/src/deps-dev/workflows/ingestVersions.ts b/services/apps/packages_worker/src/deps-dev/workflows/ingestVersions.ts index af0cd716cb..c54c75241a 100644 --- a/services/apps/packages_worker/src/deps-dev/workflows/ingestVersions.ts +++ b/services/apps/packages_worker/src/deps-dev/workflows/ingestVersions.ts @@ -25,25 +25,6 @@ const { mergeStagingToTable } = proxyActivities({ retry: { maximumAttempts: 1 }, }) -const { dropVersionsIndexes, rebuildVersionsIndexes } = proxyActivities({ - // Index builds on large partitioned tables can take hours. - startToCloseTimeout: '6 hours', - retry: { maximumAttempts: 2, initialInterval: '1 minute' }, -}) - -const { dropVersionsConstraints, rebuildVersionsConstraints } = proxyActivities< - typeof depsDevActivities ->({ - // FK validation on large partitioned tables can take hours. - startToCloseTimeout: '6 hours', - retry: { maximumAttempts: 2, initialInterval: '1 minute' }, -}) - -const { setJobStep } = proxyActivities({ - startToCloseTimeout: '30 seconds', - retry: { maximumAttempts: 3 }, -}) - const STAGING_TABLE = 'staging.osspckgs_versions_raw' const STAGING_DDL = ` @@ -58,6 +39,8 @@ CREATE UNLOGGED TABLE IF NOT EXISTS staging.osspckgs_versions_raw ( ) ` +// Merge against the live UNIQUE index — ON CONFLICT DO NOTHING makes every chunk idempotent, so the +// table's indexes/keys/constraints are never dropped. Both full and incremental use this path. const MERGE_SQL = ` INSERT INTO versions ( package_id, ecosystem, namespace, name, number, published_at, is_prerelease, licenses, last_synced_at, @@ -71,24 +54,8 @@ JOIN packages p ON p.purl = s.purl ON CONFLICT (package_id, number) DO NOTHING ` -// Full-load variant: UNIQUE constraint is dropped before the chunk loop so plain INSERT is safe. -// DISTINCT ON (p.id, s.number) deduplicates before INSERT — guards against duplicate (purl, number) -// rows in BQ data that would cause the UNIQUE constraint rebuild to fail. -const MERGE_SQL_FULL = ` -INSERT INTO versions ( - package_id, ecosystem, namespace, name, number, published_at, is_prerelease, licenses, last_synced_at, - created_at -) -SELECT DISTINCT ON (p.id, s.number) - p.id, s.ecosystem, p.namespace, p.name, s.number, s.published_at, s.is_prerelease, s.licenses, NOW(), - NOW() -FROM staging.osspckgs_versions_raw s -JOIN packages p ON p.purl = s.purl -ORDER BY p.id, s.number, s.published_at DESC NULLS LAST -` - // SET LOCAL scopes settings to this transaction only. -// synchronous_commit=off skips WAL flush wait — safe for plain INSERT on full loads. +// synchronous_commit=off skips WAL flush wait — safe for these re-runnable ON CONFLICT merges. // max_parallel_workers_per_gather parallelises the SELECT side of INSERT...SELECT. // session_replication_role=replica would skip FK trigger checks but requires superuser — // blocked on Oracle Cloud managed PostgreSQL regardless of REPLICATION role. @@ -150,94 +117,52 @@ export async function ingestVersions(opts: { return { rowCountBq: exportResult.rowCount } } - if (opts.syncMode === 'full') { - await setJobStep({ jobId: exportResult.jobId, step: 'drop_constraints' }) - await dropVersionsConstraints() - await setJobStep({ jobId: exportResult.jobId, step: 'drop_indexes' }) - await dropVersionsIndexes() - } + const totalRows = rowCounts.reduce((a, b) => a + b, 0) + const filesPerChunk = + totalRows > 0 + ? Math.max(1, Math.round((ROWS_PER_CHUNK * fileNames.length) / totalRows)) + : Math.min(fileNames.length, 2) + const totalChunks = Math.ceil(fileNames.length / filesPerChunk) + let priorRowsAffected = 0 + let priorStagingRows = 0 + const priorTableRowCounts: Record = {} + + for (let chunkIndex = 0; chunkIndex < totalChunks; chunkIndex++) { + const start = chunkIndex * filesPerChunk + const chunk = fileNames.slice(start, start + filesPerChunk) + const isFinal = chunkIndex === totalChunks - 1 + + const { rowsLoaded } = await gcsParquetToStaging({ + jobId: exportResult.jobId, + stagingTable: STAGING_TABLE, + stagingDdl: STAGING_DDL, + pgColumns: PG_COLUMNS, + timestampColumns: ['published_at'], + fileNames: chunk, + filesOffset: start, + totalFiles, + priorStagingRows, + }) + priorStagingRows += rowsLoaded - try { - const totalRows = rowCounts.reduce((a, b) => a + b, 0) - const filesPerChunk = - totalRows > 0 - ? Math.max(1, Math.round((ROWS_PER_CHUNK * fileNames.length) / totalRows)) - : Math.min(fileNames.length, 2) - const totalChunks = Math.ceil(fileNames.length / filesPerChunk) - let priorRowsAffected = 0 - let priorStagingRows = 0 - const priorTableRowCounts: Record = {} - - for (let chunkIndex = 0; chunkIndex < totalChunks; chunkIndex++) { - const start = chunkIndex * filesPerChunk - const chunk = fileNames.slice(start, start + filesPerChunk) - const isLastChunk = chunkIndex === totalChunks - 1 - // For full-load, don't mark done here — rebuild runs after the chunk loop - const isFinal = isLastChunk && opts.syncMode !== 'full' - - const { rowsLoaded } = await gcsParquetToStaging({ - jobId: exportResult.jobId, - stagingTable: STAGING_TABLE, - stagingDdl: STAGING_DDL, - pgColumns: PG_COLUMNS, - timestampColumns: ['published_at'], - fileNames: chunk, - filesOffset: start, - totalFiles, - priorStagingRows, - }) - priorStagingRows += rowsLoaded - - const { rowsAffected, tableRowCounts } = await mergeStagingToTable({ - jobId: exportResult.jobId, - prepareSql: MERGE_PREPARE_SQL, - mergeSql: opts.syncMode === 'full' ? MERGE_SQL_FULL : MERGE_SQL, - tableNames: 'versions', - isFinal, - priorRowsAffected, - priorTableRowCounts, - chunkInfo: { index: chunkIndex, total: totalChunks }, - }) - - priorRowsAffected += rowsAffected - if (!isFinal) { - for (const [k, v] of Object.entries(tableRowCounts)) { - priorTableRowCounts[k] = (priorTableRowCounts[k] ?? 0) + v - } - } - } + const { rowsAffected, tableRowCounts } = await mergeStagingToTable({ + jobId: exportResult.jobId, + prepareSql: MERGE_PREPARE_SQL, + mergeSql: MERGE_SQL, + tableNames: 'versions', + isFinal, + priorRowsAffected, + priorTableRowCounts, + chunkInfo: { index: chunkIndex, total: totalChunks }, + }) - if (opts.syncMode === 'full') { - await setJobStep({ jobId: exportResult.jobId, step: 'rebuild_indexes' }) - await rebuildVersionsIndexes() - await setJobStep({ jobId: exportResult.jobId, step: 'rebuild_constraints' }) - await rebuildVersionsConstraints() - // Finalize after rebuild — marks done with correct finishedAt - await mergeStagingToTable({ - jobId: exportResult.jobId, - mergeSql: [], - tableNames: [], - isFinal: true, - priorRowsAffected, - priorTableRowCounts, - }) - } - } catch (err) { - if (opts.syncMode === 'full') { - try { - await setJobStep({ jobId: exportResult.jobId, step: 'rebuild_indexes' }) - await rebuildVersionsIndexes() - } catch (_) { - /* best-effort */ - } - try { - await setJobStep({ jobId: exportResult.jobId, step: 'rebuild_constraints' }) - await rebuildVersionsConstraints() - } catch (_) { - /* best-effort */ + priorRowsAffected += rowsAffected + if (!isFinal) { + for (const [k, v] of Object.entries(tableRowCounts)) { + priorTableRowCounts[k] = (priorTableRowCounts[k] ?? 0) + v } } - throw err } + return { rowCountBq: exportResult.rowCount } } From 631143824092b4849329dabb278046f8ff2345b6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Uro=C5=A1=20Marolt?= Date: Fri, 3 Jul 2026 15:37:33 +0200 Subject: [PATCH 07/15] feat: add rubygems ecosystem to bq-dataset-ingest (CM-1296) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Uroš Marolt --- .../activities/createVersionsLookup.ts | 2 +- .../src/deps-dev/queries/depsSql.ts | 67 ++++++++++++++++++- .../src/deps-dev/queries/systems.ts | 4 +- .../src/scripts/monitorOsspckgs.ts | 2 +- .../src/scripts/triggerBootstrap.ts | 6 +- 5 files changed, 71 insertions(+), 10 deletions(-) diff --git a/services/apps/packages_worker/src/deps-dev/activities/createVersionsLookup.ts b/services/apps/packages_worker/src/deps-dev/activities/createVersionsLookup.ts index 89228c4da0..826dbb6d35 100644 --- a/services/apps/packages_worker/src/deps-dev/activities/createVersionsLookup.ts +++ b/services/apps/packages_worker/src/deps-dev/activities/createVersionsLookup.ts @@ -4,7 +4,7 @@ import { getPackagesDb } from '../../db' const log = getServiceChildLogger('createVersionsLookup') -const VALID_ECOSYSTEMS = new Set(['npm', 'go', 'maven', 'pypi', 'nuget', 'cargo']) +const VALID_ECOSYSTEMS = new Set(['npm', 'go', 'maven', 'pypi', 'nuget', 'cargo', 'rubygems']) // Builds a persistent UNLOGGED lookup table in the staging schema for the root-version JOIN in // ingestDependencies. Using a temp table per chunk would rebuild 4GB for every chunk on npm diff --git a/services/apps/packages_worker/src/deps-dev/queries/depsSql.ts b/services/apps/packages_worker/src/deps-dev/queries/depsSql.ts index 25c6d882d2..b8e23d5fbd 100644 --- a/services/apps/packages_worker/src/deps-dev/queries/depsSql.ts +++ b/services/apps/packages_worker/src/deps-dev/queries/depsSql.ts @@ -1,12 +1,13 @@ // NPM, MAVEN, PYPI, CARGO are in DependencyGraphEdgesLatest / DependenciesLatest. // GO uses GoRequirementsLatest (DirectDependencies, no resolved to_version). // NUGET uses NuGetRequirementsLatest (DependencyGroups → Dependencies, no resolved to_version). -// Confirmed via BQ query 2026-06-17: DependencyGraphEdgesLatest and DependenciesLatest -// both contain exactly {NPM, MAVEN, PYPI, CARGO} — GO and NUGET absent from both. +// RUBYGEMS uses RubyGemsRequirementsLatest (RuntimeDependencies, no resolved to_version). +// Confirmed via BQ query: DependencyGraphEdgesLatest and DependenciesLatest contain exactly +// {NPM, MAVEN, PYPI, CARGO} — GO, NUGET and RUBYGEMS absent from both (rubygems verified 2026-07-03). const EDGE_SYSTEMS = new Set(['NPM', 'MAVEN', 'PYPI', 'CARGO']) -export const DEPS_DEFAULT_ECOSYSTEMS = ['NPM', 'GO', 'MAVEN', 'PYPI', 'NUGET', 'CARGO'] +export const DEPS_DEFAULT_ECOSYSTEMS = ['NPM', 'GO', 'MAVEN', 'PYPI', 'NUGET', 'CARGO', 'RUBYGEMS'] // --- Full SQL helpers --- @@ -35,6 +36,18 @@ FROM \`bigquery-public-data.deps_dev_v1.NuGetRequirementsLatest\` n, UNNEST(n.DependencyGroups) AS grp, UNNEST(grp.Dependencies) AS dep` +// RubyGems: RuntimeDependencies only (the runtime graph — dev deps excluded, matching GO/NUGET). +const RUBYGEMS_FULL_PART = ` +SELECT + 'rubygems' AS ecosystem, + r.Name AS root_name, + r.Version AS root_version, + d.Name AS to_name, + CAST(NULL AS STRING) AS to_version, + d.Requirement AS version_constraint +FROM \`bigquery-public-data.deps_dev_v1.RubyGemsRequirementsLatest\` r, +UNNEST(r.RuntimeDependencies) AS d` + // ADR-0003 Option A: DependencyGraphEdgesLatest for NPM/MAVEN/PYPI/CARGO — has version_constraint. // GO + NUGET always come from their ecosystem-specific tables regardless of option. export function buildDepsFullSqlA(ecosystems: string[]): string { @@ -59,6 +72,7 @@ WHERE e.System IN (${filter}) if (ecosystems.includes('GO')) parts.push(GO_FULL_PART) if (ecosystems.includes('NUGET')) parts.push(NUGET_FULL_PART) + if (ecosystems.includes('RUBYGEMS')) parts.push(RUBYGEMS_FULL_PART) return parts.join('\nUNION ALL\n') } @@ -86,6 +100,7 @@ WHERE d.System IN (${filter}) if (ecosystems.includes('GO')) parts.push(GO_FULL_PART) if (ecosystems.includes('NUGET')) parts.push(NUGET_FULL_PART) + if (ecosystems.includes('RUBYGEMS')) parts.push(RUBYGEMS_FULL_PART) return parts.join('\nUNION ALL\n') } @@ -208,6 +223,38 @@ function nugetIncrementalBranch( } } +// RubyGems manifest — RuntimeDependencies only (runtime graph; dev deps excluded, matching GO/NUGET). +function rubygemsIncrementalBranch( + today: string, + watermark: string, +): { ctes: string[]; select: string } { + return { + ctes: [ + `today_rubygems AS ( + SELECT + 'rubygems' AS ecosystem, + r.Name AS root_name, + r.Version AS root_version, + d.Name AS to_name, + CAST(NULL AS STRING) AS to_version, + MAX(d.Requirement) AS version_constraint + FROM \`${DEPS_DEV}.RubyGemsRequirements\` r, + UNNEST(r.RuntimeDependencies) AS d + WHERE ${snapshotRange('r.SnapshotAt', today)} + GROUP BY 1, 2, 3, 4 +)`, + `watermark_rubygems AS ( + SELECT 'rubygems' AS ecosystem, r.Name AS root_name, r.Version AS root_version, d.Name AS to_name + FROM \`${DEPS_DEV}.RubyGemsRequirements\` r, + UNNEST(r.RuntimeDependencies) AS d + WHERE ${snapshotRange('r.SnapshotAt', watermark)} + GROUP BY 1, 2, 3, 4 +)`, + ], + select: antiJoinSelect('today_rubygems', 'watermark_rubygems'), + } +} + export function buildDepsIncrementalSqlA( today: string, watermark: string, @@ -216,6 +263,7 @@ export function buildDepsIncrementalSqlA( const edgeSystems = ecosystems.filter((s) => EDGE_SYSTEMS.has(s)) const includeGo = ecosystems.includes('GO') const includeNuget = ecosystems.includes('NUGET') + const includeRubygems = ecosystems.includes('RUBYGEMS') const ctes: string[] = [] const selects: string[] = [] @@ -261,6 +309,12 @@ export function buildDepsIncrementalSqlA( selects.push(select) } + if (includeRubygems) { + const { ctes: rubygemsCtes, select } = rubygemsIncrementalBranch(today, watermark) + ctes.push(...rubygemsCtes) + selects.push(select) + } + return `WITH\n${ctes.join(',\n')}\n${selects.join('\nUNION ALL\n')}` } @@ -272,6 +326,7 @@ export function buildDepsIncrementalSqlB( const depsSystems = ecosystems.filter((s) => EDGE_SYSTEMS.has(s)) const includeGo = ecosystems.includes('GO') const includeNuget = ecosystems.includes('NUGET') + const includeRubygems = ecosystems.includes('RUBYGEMS') const ctes: string[] = [] const selects: string[] = [] @@ -317,6 +372,12 @@ export function buildDepsIncrementalSqlB( selects.push(select) } + if (includeRubygems) { + const { ctes: rubygemsCtes, select } = rubygemsIncrementalBranch(today, watermark) + ctes.push(...rubygemsCtes) + selects.push(select) + } + return `WITH\n${ctes.join(',\n')}\n${selects.join('\nUNION ALL\n')}` } diff --git a/services/apps/packages_worker/src/deps-dev/queries/systems.ts b/services/apps/packages_worker/src/deps-dev/queries/systems.ts index c00cf8b601..c68d73c7d3 100644 --- a/services/apps/packages_worker/src/deps-dev/queries/systems.ts +++ b/services/apps/packages_worker/src/deps-dev/queries/systems.ts @@ -1,6 +1,6 @@ -export const DEFAULT_SYSTEMS = `'NPM', 'GO', 'MAVEN', 'PYPI', 'NUGET', 'CARGO'` +export const DEFAULT_SYSTEMS = `'NPM', 'GO', 'MAVEN', 'PYPI', 'NUGET', 'CARGO', 'RUBYGEMS'` -const VALID_SYSTEMS = new Set(['NPM', 'GO', 'MAVEN', 'PYPI', 'NUGET', 'CARGO']) +const VALID_SYSTEMS = new Set(['NPM', 'GO', 'MAVEN', 'PYPI', 'NUGET', 'CARGO', 'RUBYGEMS']) export function toSystemsFilter(ecosystems?: string[]): string { if (!ecosystems || ecosystems.length === 0) return DEFAULT_SYSTEMS diff --git a/services/apps/packages_worker/src/scripts/monitorOsspckgs.ts b/services/apps/packages_worker/src/scripts/monitorOsspckgs.ts index e816f16d77..b060256ecc 100644 --- a/services/apps/packages_worker/src/scripts/monitorOsspckgs.ts +++ b/services/apps/packages_worker/src/scripts/monitorOsspckgs.ts @@ -285,7 +285,7 @@ async function fetchTableCounts(): Promise> { // ── Ecosystem extraction ─────────────────────────────────────────────────────── -const KNOWN_ECOSYSTEMS = ['npm', 'go', 'maven', 'pypi', 'nuget', 'cargo'] +const KNOWN_ECOSYSTEMS = ['npm', 'go', 'maven', 'pypi', 'nuget', 'cargo', 'rubygems'] // Reads ecosystems from table_row_counts['meta:ecosystems'] (new jobs) or falls back to // parsing gcs_prefix/export_name for jobs created before the meta key was added. diff --git a/services/apps/packages_worker/src/scripts/triggerBootstrap.ts b/services/apps/packages_worker/src/scripts/triggerBootstrap.ts index c7aade9a98..1e87b988e3 100644 --- a/services/apps/packages_worker/src/scripts/triggerBootstrap.ts +++ b/services/apps/packages_worker/src/scripts/triggerBootstrap.ts @@ -21,8 +21,8 @@ Usage: trigger-bootstrap [full|incremental] [ECOSYSTEMS] [options] Arguments: full | incremental Sync mode (default: full) - ECOSYSTEMS Comma-separated list: CARGO,NPM,MAVEN,GO,PYPI,NUGET - Omit for all 6 ecosystems + ECOSYSTEMS Comma-separated list: CARGO,NPM,MAVEN,GO,PYPI,NUGET,RUBYGEMS + Omit for all 7 ecosystems Options: --kinds Run only specific job kinds (default: all). Comma-separated: @@ -159,7 +159,7 @@ async function main(): Promise { ? positional[1].split(',').map((e) => e.trim().toUpperCase()) : undefined - const VALID_ECOSYSTEMS = new Set(['NPM', 'GO', 'MAVEN', 'PYPI', 'NUGET', 'CARGO']) + const VALID_ECOSYSTEMS = new Set(['NPM', 'GO', 'MAVEN', 'PYPI', 'NUGET', 'CARGO', 'RUBYGEMS']) if (ecosystems) { const invalid = ecosystems.filter((e) => !VALID_ECOSYSTEMS.has(e)) if (invalid.length > 0) { From dfafcbeb95ce4754e2c23ffbdc48e874053d8ea3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Uro=C5=A1=20Marolt?= Date: Fri, 3 Jul 2026 18:09:35 +0200 Subject: [PATCH 08/15] fix: restore merge dedup and relax deps resume guards (CM-1296) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Uroš Marolt --- .../src/deps-dev/workflows/bootstrapOsspckgs.ts | 12 ++++++++++++ .../src/deps-dev/workflows/ingestDependencies.ts | 7 ++++++- .../src/deps-dev/workflows/ingestVersions.ts | 15 ++++++++++++++- .../src/scripts/triggerBootstrap.ts | 14 +++++--------- 4 files changed, 37 insertions(+), 11 deletions(-) diff --git a/services/apps/packages_worker/src/deps-dev/workflows/bootstrapOsspckgs.ts b/services/apps/packages_worker/src/deps-dev/workflows/bootstrapOsspckgs.ts index 92554c5f5f..c48056bfc9 100644 --- a/services/apps/packages_worker/src/deps-dev/workflows/bootstrapOsspckgs.ts +++ b/services/apps/packages_worker/src/deps-dev/workflows/bootstrapOsspckgs.ts @@ -93,6 +93,18 @@ export async function bootstrapOsspckgs(opts: { const activeKinds = opts.kinds ? new Set(opts.kinds) : null const runs = (kind: string) => !activeKinds || activeKinds.has(kind) + // Resume reuses a prior package_dependencies export and skips watermark/partition validation. + // Hard-enforce it targets ONLY package_dependencies so a stray resumeJobId can't silently run other + // kinds without their safety checks. The CLI validates this too; this is the fail-fast backstop. + if ( + resume && + !(activeKinds && activeKinds.size === 1 && activeKinds.has('package_dependencies')) + ) { + throw new ApplicationFailure( + 'resumeJobId is only valid with kinds=[package_dependencies] — refusing to skip validation for other kinds', + ) + } + const jobKinds: JobKind[] = ( ['packages', 'versions', 'package_dependencies', 'advisories', 'advisory_packages'] as JobKind[] ).filter((k) => runs(k)) diff --git a/services/apps/packages_worker/src/deps-dev/workflows/ingestDependencies.ts b/services/apps/packages_worker/src/deps-dev/workflows/ingestDependencies.ts index 71c5619a98..1fd4546754 100644 --- a/services/apps/packages_worker/src/deps-dev/workflows/ingestDependencies.ts +++ b/services/apps/packages_worker/src/deps-dev/workflows/ingestDependencies.ts @@ -76,7 +76,7 @@ INSERT INTO package_dependencies ( package_id, version_id, depends_on_id, depends_on_version_id, version_constraint, dependency_kind, is_optional, created_at, updated_at ) -SELECT +SELECT DISTINCT ON (pv.id, pd.id) pv.package_id, pv.id, pd.id, dv.id, sp.version_constraint, 'direct', FALSE, NOW(), NOW() FROM staging.osspckgs_deps_raw sp @@ -100,6 +100,11 @@ JOIN packages pd ON pd.ecosystem = sp.ecosystem WHEN sp.to_name LIKE '@%/%' THEN SPLIT_PART(sp.to_name, '/', 2) ELSE sp.to_name END LEFT JOIN versions dv ON dv.package_id = pd.id AND dv.number = sp.to_version +-- DISTINCT ON collapses duplicate (root, dep) pairs BQ emits with different resolved to_version; +-- ORDER BY keeps the highest to_version (best-resolvable depends_on_version_id), matching the fill +-- variant. ON CONFLICT then handles cross-chunk / pre-existing rows. Full lost this when MERGE_SQL_FULL +-- (which had the same DISTINCT ON) was removed with the drop/rebuild-index step. +ORDER BY pv.id, pd.id, sp.to_version DESC NULLS LAST ON CONFLICT (version_id, depends_on_id, dependency_kind) DO NOTHING ` diff --git a/services/apps/packages_worker/src/deps-dev/workflows/ingestVersions.ts b/services/apps/packages_worker/src/deps-dev/workflows/ingestVersions.ts index c54c75241a..96fae1e89e 100644 --- a/services/apps/packages_worker/src/deps-dev/workflows/ingestVersions.ts +++ b/services/apps/packages_worker/src/deps-dev/workflows/ingestVersions.ts @@ -25,6 +25,11 @@ const { mergeStagingToTable } = proxyActivities({ retry: { maximumAttempts: 1 }, }) +const { setJobStep } = proxyActivities({ + startToCloseTimeout: '30 seconds', + retry: { maximumAttempts: 3 }, +}) + const STAGING_TABLE = 'staging.osspckgs_versions_raw' const STAGING_DDL = ` @@ -46,11 +51,15 @@ INSERT INTO versions ( package_id, ecosystem, namespace, name, number, published_at, is_prerelease, licenses, last_synced_at, created_at ) -SELECT +SELECT DISTINCT ON (p.id, s.number) p.id, s.ecosystem, p.namespace, p.name, s.number, s.published_at, s.is_prerelease, s.licenses, NOW(), NOW() FROM staging.osspckgs_versions_raw s JOIN packages p ON p.purl = s.purl +-- DISTINCT ON + ORDER BY keeps the most recently published row when BQ emits duplicate +-- (purl, number) rows with differing published_at/licenses, so ON CONFLICT DO NOTHING no longer +-- retains an arbitrary insert-order row. +ORDER BY p.id, s.number, s.published_at DESC NULLS LAST ON CONFLICT (package_id, number) DO NOTHING ` @@ -127,6 +136,10 @@ export async function ingestVersions(opts: { let priorStagingRows = 0 const priorTableRowCounts: Record = {} + // Mark the phase before the loop so the monitor shows 'merging' instead of the stale prior step + // (e.g. 'loading') during the merge — mirrors ingestDependencies. + await setJobStep({ jobId: exportResult.jobId, step: 'merging' }) + for (let chunkIndex = 0; chunkIndex < totalChunks; chunkIndex++) { const start = chunkIndex * filesPerChunk const chunk = fileNames.slice(start, start + filesPerChunk) diff --git a/services/apps/packages_worker/src/scripts/triggerBootstrap.ts b/services/apps/packages_worker/src/scripts/triggerBootstrap.ts index 1e87b988e3..e6fde65c6b 100644 --- a/services/apps/packages_worker/src/scripts/triggerBootstrap.ts +++ b/services/apps/packages_worker/src/scripts/triggerBootstrap.ts @@ -48,7 +48,8 @@ Options: --resume-job Resume a partially-merged package_dependencies job by id. Skips BQ, reuses that job's exact parquet export, and restarts the chunk loop where its staging load left off (idempotent ON CONFLICT overlap). - Requires: --kinds package_dependencies AND (incremental | --fill-constraints). + Requires: --kinds package_dependencies. Works in any mode (all deps + merges are idempotent). --help Show this help Examples: @@ -170,19 +171,14 @@ async function main(): Promise { } } - // Resume is deps-only and merge-idempotent-only: it re-runs already-committed chunks, safe under - // incremental (ON CONFLICT DO NOTHING) or --fill-constraints (ON CONFLICT DO UPDATE), not a full load. + // Resume is deps-only: it re-runs already-committed chunks. Safe in every mode because all deps + // merges are now idempotent — full and incremental use ON CONFLICT DO NOTHING, --fill-constraints + // uses ON CONFLICT DO UPDATE. (Full became idempotent when the drop/rebuild-index step was removed.) if (resumeJobId !== undefined) { if (!kinds || kinds.length !== 1 || kinds[0] !== 'package_dependencies') { console.error('--resume-job requires: --kinds package_dependencies (resume is deps-only)') process.exit(1) } - if (mode !== 'incremental' && !fillConstraints) { - console.error( - '--resume-job requires incremental mode or --fill-constraints (idempotent merges only)', - ) - process.exit(1) - } } const cfg = TEMPORAL_CONFIG() From a103b06fb6f75a4c95a2d5378b56f37d99d5c746 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Uro=C5=A1=20Marolt?= Date: Fri, 3 Jul 2026 19:22:01 +0200 Subject: [PATCH 09/15] fix: resume deps ingest with original export settings (CM-1296) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Uroš Marolt --- .../src/deps-dev/activities/bqExportToGcs.ts | 16 ++++++++++++++-- .../src/deps-dev/activities/getResumeExport.ts | 4 ++++ .../src/deps-dev/workflows/ingestDependencies.ts | 14 ++++++++++++-- .../data-access-layer/src/osspckgs/ingestJobs.ts | 12 +++++++++++- 4 files changed, 41 insertions(+), 5 deletions(-) diff --git a/services/apps/packages_worker/src/deps-dev/activities/bqExportToGcs.ts b/services/apps/packages_worker/src/deps-dev/activities/bqExportToGcs.ts index ebdc531bea..3e377988c7 100644 --- a/services/apps/packages_worker/src/deps-dev/activities/bqExportToGcs.ts +++ b/services/apps/packages_worker/src/deps-dev/activities/bqExportToGcs.ts @@ -32,6 +32,12 @@ export interface BqExportToGcsInput { // is replaced by a server-side maximumBytesBilled cap, since a dry-run only validates the first // statement and cannot predict the WHILE loop's total scan. See ADR-0004. isScript?: boolean + // Fill-constraints run (package_dependencies only): the export is a full Option-A scan but the + // downstream merge upserts version_constraint (ON CONFLICT DO UPDATE) instead of DO NOTHING. Full + // and fill produce identical parquet, so sync_mode alone can't tell them apart — persist it in the + // job meta so a --resume-job run reprocesses the export with the correct merge instead of silently + // reverting to DO NOTHING (which skips the backfill). + isFill?: boolean } export interface BqExportToGcsOutput { @@ -53,6 +59,7 @@ export async function bqExportToGcs(input: BqExportToGcsInput): Promise = {} + if (ecosystems) exportMeta['meta:ecosystems'] = ecosystems + if (isFill) exportMeta['meta:fill'] = 1 await markJobStatus(qx, jobId, 'exporting', { - ...(ecosystems ? { tableRowCounts: { 'meta:ecosystems': ecosystems } } : {}), + ...(Object.keys(exportMeta).length > 0 ? { tableRowCounts: exportMeta } : {}), }) // From here the row is 'exporting'; any BQ failure (incl. script-mode maximumBytesBilled aborts) diff --git a/services/apps/packages_worker/src/deps-dev/activities/getResumeExport.ts b/services/apps/packages_worker/src/deps-dev/activities/getResumeExport.ts index 8561988d66..793f2c9f95 100644 --- a/services/apps/packages_worker/src/deps-dev/activities/getResumeExport.ts +++ b/services/apps/packages_worker/src/deps-dev/activities/getResumeExport.ts @@ -15,6 +15,8 @@ export interface GetResumeExportOutput { progressDone: number progressTotal: number rowCountPg: number + ecosystems: string[] | null + fill: boolean } // Pure fetch of a prior job's resume-relevant fields (export path, status, file-load progress, @@ -38,5 +40,7 @@ export async function getResumeExport( progressDone: job.progressDone, progressTotal: job.progressTotal, rowCountPg: job.rowCountPg, + ecosystems: job.ecosystems, + fill: job.fill, } } diff --git a/services/apps/packages_worker/src/deps-dev/workflows/ingestDependencies.ts b/services/apps/packages_worker/src/deps-dev/workflows/ingestDependencies.ts index 1fd4546754..678520b052 100644 --- a/services/apps/packages_worker/src/deps-dev/workflows/ingestDependencies.ts +++ b/services/apps/packages_worker/src/deps-dev/workflows/ingestDependencies.ts @@ -183,8 +183,9 @@ export async function ingestDependencies(opts: { // and restart the chunk loop where its staging load left off. Idempotent modes only (see gate). resumeJobId?: number }): Promise<{ rowCountBq: number }> { - const ecosystems = opts.ecosystems ?? DEPS_DEFAULT_ECOSYSTEMS - const isFill = opts.fillConstraints === true + // let: on resume these are overridden with the prior export's stored settings (see resume block). + let ecosystems = opts.ecosystems ?? DEPS_DEFAULT_ECOSYSTEMS + let isFill = opts.fillConstraints === true // Fill mode forces Option A — Option B selects NULL for version_constraint, making the fill a no-op. const tableOption = isFill ? 'A' : (opts.depsTableOption ?? 'A') // Full/fill scan the *Latest views; incremental reads the `today` partition. Drives both the SQL @@ -234,6 +235,13 @@ export async function ingestDependencies(opts: { exportResult = { jobId: prior.jobId, gcsPrefix: prior.gcsPrefix, rowCount: 0 } resumeProgressDone = prior.progressDone resumeRowCountPg = prior.rowCountPg + // Reprocess the export with its ORIGINAL settings, not the current CLI opts. A narrower + // ecosystems list would drop versions-lookup rows (deps silently unmerged); the wrong fill flag + // would pick ON CONFLICT DO NOTHING and skip the version_constraint backfill. Pre-meta jobs (no + // stored ecosystems) fall back to all ecosystems — a superset lookup never drops rows. fill is + // OR'd with the current flag so an explicit --fill-constraints still works for pre-meta jobs. + ecosystems = prior.ecosystems ?? DEPS_DEFAULT_ECOSYSTEMS + isFill = prior.fill || isFill } else { // Guard against corrupt deps.dev resolved-graph snapshots BEFORE the (multi-hour) export. // Skip when reusing a prior export — we're re-importing already-validated parquet, not @@ -284,6 +292,8 @@ export async function ingestDependencies(opts: { reuseExports: opts.reuseExports, exportName: opts.exportName, ecosystems, + // Persist fill in the job meta so a later --resume-job restores the upsert merge (see resume block). + isFill, }) } diff --git a/services/libs/data-access-layer/src/osspckgs/ingestJobs.ts b/services/libs/data-access-layer/src/osspckgs/ingestJobs.ts index e101707246..f63769ab52 100644 --- a/services/libs/data-access-layer/src/osspckgs/ingestJobs.ts +++ b/services/libs/data-access-layer/src/osspckgs/ingestJobs.ts @@ -104,12 +104,20 @@ export async function getIngestJobForResume( progressDone: number progressTotal: number rowCountPg: number + // Original export parameters, so resume reprocesses the export with the SAME settings rather than + // whatever the current CLI opts happen to be: ecosystems (drives the versions-lookup filter) and + // fill (drives the merge SQL — fill upserts version_constraint, non-fill is ON CONFLICT DO NOTHING). + // ecosystems is null for jobs exported before the meta key existed; fill defaults to false. + ecosystems: string[] | null + fill: boolean } | null> { const row = await qx.selectOneOrNone( ` SELECT id, job_kind, status, sync_mode, gcs_prefix, row_count_pg, COALESCE((table_row_counts->>'progress:done')::int, 0) AS progress_done, - COALESCE((table_row_counts->>'progress:total')::int, 0) AS progress_total + COALESCE((table_row_counts->>'progress:total')::int, 0) AS progress_total, + table_row_counts->'meta:ecosystems' AS ecosystems, + COALESCE((table_row_counts->>'meta:fill')::bool, false) AS fill FROM osspckgs_ingest_jobs WHERE id = $(id) `, @@ -125,6 +133,8 @@ export async function getIngestJobForResume( progressDone: Number(row.progress_done ?? 0), progressTotal: Number(row.progress_total ?? 0), rowCountPg: Number(row.row_count_pg ?? 0), + ecosystems: (row.ecosystems as string[] | null) ?? null, + fill: Boolean(row.fill), } : null } From ac7eee92a55d77060c47fa4b8b0bb920698609bb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Uro=C5=A1=20Marolt?= Date: Fri, 3 Jul 2026 19:41:09 +0200 Subject: [PATCH 10/15] docs: fix stale comments after merge dedup + resume changes (CM-1296) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Uroš Marolt --- services/apps/packages_worker/src/deps-dev/queries/depsSql.ts | 4 ++-- services/libs/data-access-layer/src/osspckgs/ingestJobs.ts | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/services/apps/packages_worker/src/deps-dev/queries/depsSql.ts b/services/apps/packages_worker/src/deps-dev/queries/depsSql.ts index b8e23d5fbd..098623975e 100644 --- a/services/apps/packages_worker/src/deps-dev/queries/depsSql.ts +++ b/services/apps/packages_worker/src/deps-dev/queries/depsSql.ts @@ -22,8 +22,8 @@ SELECT FROM \`bigquery-public-data.deps_dev_v1.GoRequirementsLatest\` g, UNNEST(g.DirectDependencies) AS d` -// NuGet groups deps by TargetFramework — flatten all groups, dedup handled downstream -// by ON CONFLICT in MERGE_SQL (and DISTINCT ON in the fill-constraints variant). +// NuGet groups deps by TargetFramework — flatten all groups, duplicate (root, dep) pairs are +// deduped downstream by DISTINCT ON + ON CONFLICT in MERGE_SQL (and the fill-constraints variant). const NUGET_FULL_PART = ` SELECT 'nuget' AS ecosystem, diff --git a/services/libs/data-access-layer/src/osspckgs/ingestJobs.ts b/services/libs/data-access-layer/src/osspckgs/ingestJobs.ts index f63769ab52..3cef453ef1 100644 --- a/services/libs/data-access-layer/src/osspckgs/ingestJobs.ts +++ b/services/libs/data-access-layer/src/osspckgs/ingestJobs.ts @@ -86,8 +86,6 @@ export async function findLatestExportedJobByKind( : null } -// Returns an existing exported job for the given GCS prefix so callers can -// skip re-running the BQ export when retrying a failed workflow. // Loads the fields needed to resume a partially-merged chunked job by id: the exact GCS export // path (so the same parquet files — and therefore the same chunk boundaries — are re-listed) plus // the file-level load progress and rows-merged-so-far. progressDone/progressTotal come from the @@ -139,6 +137,8 @@ export async function getIngestJobForResume( : null } +// Returns an existing exported job for the given GCS prefix so callers can +// skip re-running the BQ export when retrying a failed workflow. export async function findExportedJobByGcsPrefix( qx: QueryExecutor, gcsPrefix: string, From ec0e676f0d1b1c32959c084cea4064b806317cf9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Uro=C5=A1=20Marolt?= Date: Sat, 4 Jul 2026 08:08:43 +0200 Subject: [PATCH 11/15] fix: stamp meta:fill on deps export reuse (CM-1296) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Uroš Marolt --- .../src/deps-dev/activities/bqExportToGcs.ts | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/services/apps/packages_worker/src/deps-dev/activities/bqExportToGcs.ts b/services/apps/packages_worker/src/deps-dev/activities/bqExportToGcs.ts index 3e377988c7..dfb980d009 100644 --- a/services/apps/packages_worker/src/deps-dev/activities/bqExportToGcs.ts +++ b/services/apps/packages_worker/src/deps-dev/activities/bqExportToGcs.ts @@ -6,6 +6,7 @@ import { findExportedJobByGcsPrefix, findLatestExportedJobByKind, markJobStatus, + mergeJobTableRowCounts, } from '@crowd/data-access-layer' import { getServiceChildLogger } from '@crowd/logging' @@ -86,6 +87,10 @@ export async function bqExportToGcs(input: BqExportToGcsInput): Promise Date: Sat, 4 Jul 2026 20:31:26 +0200 Subject: [PATCH 12/15] docs: sync deps incremental byte-ceiling note for rubygems (CM-1296) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Uroš Marolt --- .../src/deps-dev/workflows/ingestDependencies.ts | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/services/apps/packages_worker/src/deps-dev/workflows/ingestDependencies.ts b/services/apps/packages_worker/src/deps-dev/workflows/ingestDependencies.ts index 678520b052..69b057be05 100644 --- a/services/apps/packages_worker/src/deps-dev/workflows/ingestDependencies.ts +++ b/services/apps/packages_worker/src/deps-dev/workflows/ingestDependencies.ts @@ -285,9 +285,10 @@ export async function ingestDependencies(opts: { syncMode: fullScan ? 'full' : opts.syncMode, snapshotAt: opts.today, // Full/fill scan the *Latest views (everything) → 25000. Incremental is a snapshot edge-diff - // (today vs watermark partitions of DependencyGraphEdges + GoRequirements + NuGetRequirements); - // measured ~4.1TB for Option A. 10000 leaves ~2.4x headroom and still trips a runaway full-table - // scan. Overridable via BQ_DATASET_INGEST_PACKAGE_DEPENDENCIES[_INCREMENTAL]_MAX_BQ_GB (see README). + // (today vs watermark partitions of DependencyGraphEdges + GoRequirements + NuGetRequirements + + // RubyGemsRequirements); measured ~4.1TB for Option A (pre-RubyGems, which adds a small increment). + // 10000 leaves ~2.4x headroom and still trips a runaway full-table scan. Overridable via + // BQ_DATASET_INGEST_PACKAGE_DEPENDENCIES[_INCREMENTAL]_MAX_BQ_GB (see README). maxBytesGb: fullScan ? 25000 : 10000, reuseExports: opts.reuseExports, exportName: opts.exportName, From 9065aa6a5e44ca56c015280cb59e4c7fc348e372 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Uro=C5=A1=20Marolt?= Date: Sat, 4 Jul 2026 20:49:29 +0200 Subject: [PATCH 13/15] fix: clear stale meta:fill on non-fill deps export reuse (CM-1296) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Uroš Marolt --- .../src/deps-dev/activities/bqExportToGcs.ts | 22 ++++++++++--------- 1 file changed, 12 insertions(+), 10 deletions(-) diff --git a/services/apps/packages_worker/src/deps-dev/activities/bqExportToGcs.ts b/services/apps/packages_worker/src/deps-dev/activities/bqExportToGcs.ts index dfb980d009..28438c9adf 100644 --- a/services/apps/packages_worker/src/deps-dev/activities/bqExportToGcs.ts +++ b/services/apps/packages_worker/src/deps-dev/activities/bqExportToGcs.ts @@ -87,10 +87,12 @@ export async function bqExportToGcs(input: BqExportToGcsInput): Promise Date: Sat, 4 Jul 2026 21:02:07 +0200 Subject: [PATCH 14/15] fix: fail deps resume when parquet file count changed (CM-1296) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Uroš Marolt --- .../src/deps-dev/workflows/ingestDependencies.ts | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/services/apps/packages_worker/src/deps-dev/workflows/ingestDependencies.ts b/services/apps/packages_worker/src/deps-dev/workflows/ingestDependencies.ts index 69b057be05..68c9792ea3 100644 --- a/services/apps/packages_worker/src/deps-dev/workflows/ingestDependencies.ts +++ b/services/apps/packages_worker/src/deps-dev/workflows/ingestDependencies.ts @@ -197,6 +197,7 @@ export async function ingestDependencies(opts: { let exportResult: { jobId: number; gcsPrefix: string; rowCount: number } // Files already loaded into staging by the prior (resumed) job, and rows it had already merged. let resumeProgressDone = 0 + let resumeProgressTotal = 0 let resumeRowCountPg = 0 if (opts.resumeJobId != null) { // Reuse the prior job's exact export by id — bypasses reuse-by-kind (which excludes 'failed' @@ -234,6 +235,7 @@ export async function ingestDependencies(opts: { } exportResult = { jobId: prior.jobId, gcsPrefix: prior.gcsPrefix, rowCount: 0 } resumeProgressDone = prior.progressDone + resumeProgressTotal = prior.progressTotal resumeRowCountPg = prior.rowCountPg // Reprocess the export with its ORIGINAL settings, not the current CLI opts. A narrower // ecosystems list would drop versions-lookup rows (deps silently unmerged); the wrong fill flag @@ -313,6 +315,20 @@ export async function ingestDependencies(opts: { ) } + // The stored progress:done maps to chunk boundaries computed from the file count the prior run + // saw (progress:total). If the re-listed count differs (objects deleted, prefix re-exported), the + // boundaries no longer align: startChunk can clamp past the un-merged files and the loop marks the + // job 'done' having skipped them. Fail loudly instead. Only when the prior run actually recorded + // progress (>0) — a fresh 'exported' resume has progress:total 0 and restarts from chunk 0 safely. + if (resume && resumeProgressTotal > 0 && totalFiles !== resumeProgressTotal) { + throw ApplicationFailure.nonRetryable( + `resume job ${opts.resumeJobId}: parquet file count changed since the prior run ` + + `(recorded ${resumeProgressTotal}, now ${totalFiles} at ${exportResult.gcsPrefix}) — ` + + `the export was modified or re-exported; run a fresh export instead of --resume-job`, + 'RESUME_INVALID', + ) + } + if (totalFiles === 0 || totalRows === 0) { await mergeStagingToTable({ jobId: exportResult.jobId, From 359a4cbd9d0fcab4199511b6f80df447e3733313 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Uro=C5=A1=20Marolt?= Date: Sun, 5 Jul 2026 19:22:39 +0200 Subject: [PATCH 15/15] fix: raise versions merge activity timeout to 4h (CM-1296) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Uroš Marolt --- .../packages_worker/src/deps-dev/workflows/ingestVersions.ts | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/services/apps/packages_worker/src/deps-dev/workflows/ingestVersions.ts b/services/apps/packages_worker/src/deps-dev/workflows/ingestVersions.ts index 96fae1e89e..53796e92fb 100644 --- a/services/apps/packages_worker/src/deps-dev/workflows/ingestVersions.ts +++ b/services/apps/packages_worker/src/deps-dev/workflows/ingestVersions.ts @@ -21,7 +21,10 @@ const { gcsParquetToStaging } = proxyActivities({ }) const { mergeStagingToTable } = proxyActivities({ - startToCloseTimeout: '1 hour', + // A single ~1M-row chunk merge into the live-index/live-constraint versions table can run long; + // 4h matches ingestDependencies and gives headroom over the previously-observed 2h overrun. The + // merge is non-retryable (maximumAttempts: 1), so too short a timeout fails the whole ingest. + startToCloseTimeout: '4 hours', retry: { maximumAttempts: 1 }, })