fix: comments

Signed-off-by: Uroš Marolt <uros@marolt.me>

Author: themarolt

services/apps/packages_worker/src/deps-dev/activities/bqExportToGcs.ts

@@ -196,7 +196,9 @@ export async function bqExportToGcs(input: BqExportToGcsInput): Promise<BqExport
     // M4: explicit location to avoid cross-region error when account default != US
     const [dryRunJob] = await bigquery.createQueryJob({ query: sql, dryRun: true, location: 'US' })
     const dryRunBytes = Number(dryRunJob.metadata.statistics.totalBytesProcessed ?? 0)
-    log.info({ jobKind, dryRunBytes, maxBytesGb }, 'BQ dry-run complete')
+    // Log the effective ceiling (env override may differ from the default maxBytesGb) and the
+    // computed byte ceiling, so ops can see what the abort decision is actually compared against.
+    log.info({ jobKind, dryRunBytes, maxBytesGb, effectiveMaxBytesGb, ceiling }, 'BQ dry-run complete')
     if (dryRunBytes > ceiling) {
       throw new Error(
         `BQ dry-run for ${jobKind} reports ${dryRunBytes} bytes > ceiling ${ceiling} — aborting`,

services/apps/packages_worker/src/deps-dev/activities/checkEdgeSnapshotQuality.ts

@@ -22,8 +22,13 @@ import { assertSnapshotDate } from '../queries/depsSql'
 // produced by the resolution pipeline and were unaffected, so they are not probed here.
 
 export interface CheckEdgeSnapshotQualityInput {
-  snapshotDate: string // YYYY-MM-DD — the resolved BQ partition date being ingested
+  snapshotDate: string // YYYY-MM-DD — the BQ partition the incremental diff reads (ignored when fullScan)
   ecosystems: string[]
+  // Probe the SAME source the ingest reads, so the guard validates the snapshot that actually gets
+  // ingested. Full/fill scan the *Latest views (newest snapshot, no date filter) → probe *Latest.
+  // Incremental reads the `snapshotDate` partition → probe that partition. Without this, a full run
+  // with an older --snapshot-date would validate a stale partition while ingesting a corrupt *Latest.
+  fullScan: boolean
 }
 
 export interface CanaryStat {
@@ -99,8 +104,6 @@ export async function checkEdgeSnapshotQuality(
     return { ok: true, canaries: [] }
   }
 
-  assertSnapshotDate(input.snapshotDate)
-
   // Group canaries by system into `(System = x AND Name IN (...))` predicates so the filter prunes
   // on the (System, Name, Version) clustering — keeps the probe in the single-GB / pennies range.
   const bySystem = new Map<string, string[]>()
@@ -116,6 +119,22 @@ export async function checkEdgeSnapshotQuality(
     )
     .join('\n        OR ')
 
+  // Probe the same source the ingest reads (see fullScan): full/fill scan the *Latest view (newest
+  // snapshot, no date filter); incremental reads the `snapshotDate` partition. The date is only
+  // interpolated (and validated) on the partition path; *Latest needs no date.
+  let edgesTable: string
+  let snapshotFilter: string
+  if (input.fullScan) {
+    edgesTable = 'bigquery-public-data.deps_dev_v1.DependencyGraphEdgesLatest'
+    snapshotFilter = ''
+  } else {
+    assertSnapshotDate(input.snapshotDate)
+    edgesTable = 'bigquery-public-data.deps_dev_v1.DependencyGraphEdges'
+    snapshotFilter = `e.SnapshotAt >= TIMESTAMP('${input.snapshotDate}')
+      AND e.SnapshotAt <  TIMESTAMP(DATE_ADD(DATE '${input.snapshotDate}', INTERVAL 1 DAY))
+      AND `
+  }
+
   // Direct edges only (From = graph root), matching what the ingest reads. %T formats NULLs
   // as the literal "NULL" so COUNT(DISTINCT ...) isn't nulled out by unresolved To.Version.
   const query = `
@@ -124,10 +143,8 @@ export async function checkEdgeSnapshotQuality(
       e.Name   AS name,
       COUNT(*) AS row_count,
       COUNT(DISTINCT FORMAT('%T|%T|%T', e.From.Version, e.To.Name, e.To.Version)) AS edge_count
-    FROM \`bigquery-public-data.deps_dev_v1.DependencyGraphEdges\` e
-    WHERE e.SnapshotAt >= TIMESTAMP('${input.snapshotDate}')
-      AND e.SnapshotAt <  TIMESTAMP(DATE_ADD(DATE '${input.snapshotDate}', INTERVAL 1 DAY))
-      AND e.From.Name = e.Name AND e.From.Version = e.Version
+    FROM \`${edgesTable}\` e
+    WHERE ${snapshotFilter}e.From.Name = e.Name AND e.From.Version = e.Version
       AND (
         ${systemPredicates}
       )

services/apps/packages_worker/src/deps-dev/workflows/ingestDependencies.ts

@@ -224,14 +224,17 @@ export async function ingestDependencies(opts: {
   const 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
+  // 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.
   if (!opts.reuseExports) {
-    const guard = await checkEdgeSnapshotQuality({ snapshotDate: opts.today, ecosystems })
+    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}. ` +
@@ -241,10 +244,9 @@ export async function ingestDependencies(opts: {
     }
   }
   // Fill mode always uses full SQL — needs all rows to find which have NULL version_constraint in DB.
-  const sql =
-    opts.syncMode === 'full' || isFill
-      ? buildDepsFullSql(ecosystems, tableOption)
-      : buildDepsIncrementalSql(opts.today, opts.watermark ?? '', ecosystems, tableOption)
+  const sql = fullScan
+    ? buildDepsFullSql(ecosystems, tableOption)
+    : buildDepsIncrementalSql(opts.today, opts.watermark ?? '', ecosystems, tableOption)
 
   const exportResult = await bqExportToGcs({
     jobKind: 'package_dependencies',

services/apps/packages_worker/src/deps-dev/workflows/ingestDependentCounts.ts

@@ -64,7 +64,7 @@ interface VariantConfig {
   // When true, buildSql returns a multi-statement BQ script (the GO/NUGET exact reverse transitive
   // closure) that ends by creating TEMP TABLE _export_data, rather than a single SELECT. The export
   // activity then appends only EXPORT DATA and enforces the byte ceiling via maximumBytesBilled
-  // instead of a dry-run. See ADR-0004. Left unset (single-SELECT) until the closure SQL lands.
+  // instead of a dry-run. See ADR-0004. Unset (single-SELECT) for the edges variant; set for GO/NUGET.
   isScript?: boolean
 }