fix: resolve advisory_packages join in postgres, skip 1.5TB purl scan (CM-1362)

scripts/services/blast-radius-worker.yaml

@@ -5,7 +5,7 @@ x-env-args: &env-args
   NODE_ENV: docker
   SERVICE: blast-radius-worker
   CROWD_TEMPORAL_TASKQUEUE: blast-radius-worker
-  CROWD_TEMPORAL_NAMESPACE: ${CROWD_PACKAGES_TEMPORAL_NAMESPACE}
+  CROWD_TEMPORAL_NAMESPACE: ${CROWD_PACKAGES_TEMPORAL_NAMESPACE:-default}
   SHELL: /bin/sh
   SUPPRESS_NO_CONFIG_WARNING: 'true'
 

scripts/services/bq-dataset-ingest.yaml

@@ -5,7 +5,7 @@ x-env-args: &env-args
   NODE_ENV: docker
   SERVICE: bq-dataset-ingest
   CROWD_TEMPORAL_TASKQUEUE: bq-dataset-ingest
-  CROWD_TEMPORAL_NAMESPACE: ${CROWD_PACKAGES_TEMPORAL_NAMESPACE}
+  CROWD_TEMPORAL_NAMESPACE: ${CROWD_PACKAGES_TEMPORAL_NAMESPACE:-default}
   SHELL: /bin/sh
   SUPPRESS_NO_CONFIG_WARNING: 'true'
 

scripts/services/maven-worker.yaml

@@ -5,7 +5,7 @@ x-env-args: &env-args
   NODE_ENV: docker
   SERVICE: maven-worker
   CROWD_TEMPORAL_TASKQUEUE: packages-worker
-  CROWD_TEMPORAL_NAMESPACE: ${CROWD_PACKAGES_TEMPORAL_NAMESPACE}
+  CROWD_TEMPORAL_NAMESPACE: ${CROWD_PACKAGES_TEMPORAL_NAMESPACE:-default}
   SHELL: /bin/sh
   SUPPRESS_NO_CONFIG_WARNING: 'true'
 

scripts/services/npm-worker.yaml

@@ -5,7 +5,7 @@ x-env-args: &env-args
   NODE_ENV: docker
   SERVICE: npm-worker
   CROWD_TEMPORAL_TASKQUEUE: npm-worker
-  CROWD_TEMPORAL_NAMESPACE: ${CROWD_PACKAGES_TEMPORAL_NAMESPACE}
+  CROWD_TEMPORAL_NAMESPACE: ${CROWD_PACKAGES_TEMPORAL_NAMESPACE:-default}
   SHELL: /bin/sh
   SUPPRESS_NO_CONFIG_WARNING: 'true'
 

scripts/services/osv-worker.yaml

@@ -5,7 +5,7 @@ x-env-args: &env-args
   NODE_ENV: docker
   SERVICE: osv-worker
   CROWD_TEMPORAL_TASKQUEUE: osv-worker
-  CROWD_TEMPORAL_NAMESPACE: ${CROWD_PACKAGES_TEMPORAL_NAMESPACE}
+  CROWD_TEMPORAL_NAMESPACE: ${CROWD_PACKAGES_TEMPORAL_NAMESPACE:-default}
   SHELL: /bin/sh
   SUPPRESS_NO_CONFIG_WARNING: 'true'
 

scripts/services/packagist-worker.yaml

@@ -5,7 +5,7 @@ x-env-args: &env-args
   NODE_ENV: docker
   SERVICE: packagist-worker
   CROWD_TEMPORAL_TASKQUEUE: packagist-worker
-  CROWD_TEMPORAL_NAMESPACE: ${CROWD_PACKAGES_TEMPORAL_NAMESPACE}
+  CROWD_TEMPORAL_NAMESPACE: ${CROWD_PACKAGES_TEMPORAL_NAMESPACE:-default}
   SHELL: /bin/sh
   SUPPRESS_NO_CONFIG_WARNING: 'true'
 

scripts/services/pypi-worker.yaml

@@ -5,7 +5,7 @@ x-env-args: &env-args
   NODE_ENV: docker
   SERVICE: pypi-worker
   CROWD_TEMPORAL_TASKQUEUE: pypi-worker
-  CROWD_TEMPORAL_NAMESPACE: ${CROWD_PACKAGES_TEMPORAL_NAMESPACE}
+  CROWD_TEMPORAL_NAMESPACE: ${CROWD_PACKAGES_TEMPORAL_NAMESPACE:-default}
   SHELL: /bin/sh
   SUPPRESS_NO_CONFIG_WARNING: 'true'
 

scripts/services/security-contacts-worker.yaml

@@ -7,7 +7,7 @@ x-env-args: &env-args
   SHELL: /bin/sh
   SUPPRESS_NO_CONFIG_WARNING: 'true'
   CROWD_TEMPORAL_TASKQUEUE: security-contacts-worker
-  CROWD_TEMPORAL_NAMESPACE: ${CROWD_PACKAGES_TEMPORAL_NAMESPACE}
+  CROWD_TEMPORAL_NAMESPACE: ${CROWD_PACKAGES_TEMPORAL_NAMESPACE:-default}
 
 services:
   security-contacts-worker:

services/apps/packages_worker/src/deps-dev/README.md

@@ -56,8 +56,8 @@ The mode-specific key takes precedence over the generic key. Value must be a pos
 | `BQ_DATASET_INGEST_PACKAGE_DEPENDENCIES_MAX_BQ_GB`      | 25000 full / 10000 incr | `package_dependencies`      | Full scans `*Latest`. Incremental is a snapshot edge-diff (today vs watermark partitions of `DependencyGraphEdges` + `GoRequirements` + `NuGetRequirements`), matched on `(root, to_name)` excluding the resolved `to_version` to drop re-resolution churn (~4.1TB, Option A). Mode-specific `…_FULL_…` / `…_INCREMENTAL_…` keys take precedence. |
 | `BQ_DATASET_INGEST_REPOS_MAX_BQ_GB`                     |                    2000 | `repos`                     |                                                                                                                                                                                                                                                                                                                                                   |
 | `BQ_DATASET_INGEST_PACKAGE_REPOS_MAX_BQ_GB`             |                    2000 | `package_repos`             |                                                                                                                                                                                                                                                                                                                                                   |
-| `BQ_DATASET_INGEST_ADVISORIES_MAX_BQ_GB`                |                      10 | `advisories`                |                                                                                                                                                                                                                                                                                                                                                   |
-| `BQ_DATASET_INGEST_ADVISORY_PACKAGES_MAX_BQ_GB`         |                    1500 | `advisory_packages`         |                                                                                                                                                                                                                                                                                                                                                   |
+| `BQ_DATASET_INGEST_ADVISORIES_MAX_BQ_GB`                |                      20 | `advisories`                | `AdvisoriesLatest` only — small table.                                                                                                                                                                                                                                                                                                            |
+| `BQ_DATASET_INGEST_ADVISORY_PACKAGES_MAX_BQ_GB`         |                      50 | `advisory_packages`         | `AdvisoriesLatest` + `UNNEST`, no BQ-side purl lookup — package_id is resolved in Postgres against `packages` instead (CM-1362). Was 1500 (a `PackageVersionsLatest` scan for a purl join key); measured actual is ~1.4 GB.                                                                                                                       |
 | `BQ_DATASET_INGEST_DEPENDENT_COUNTS_MAX_BQ_GB`          |                    2000 | `dependent_counts`          | Edges only (NPM/MAVEN/PYPI/CARGO) from the `Dependents` reverse index. GO/NUGET/RUBYGEMS are absent from `Dependents` and run as separate kinds below.                                                                                                                                                                                            |
 | `BQ_DATASET_INGEST_DEPENDENT_COUNTS_GO_MAX_BQ_GB`       |                    5000 | `dependent_counts_go`       | GO exact reverse transitive closure over `GoRequirementsLatest` (script mode). All 3 count columns. Ceiling is a `maximumBytesBilled` runaway cap above the validated full-pipeline spend (2.31 TB incl. repos aggregation), not a dry-run gate.                                                                                                  |
 | `BQ_DATASET_INGEST_DEPENDENT_COUNTS_NUGET_MAX_BQ_GB`    |                     200 | `dependent_counts_nuget`    | NUGET exact reverse transitive closure over `NuGetRequirementsLatest` (script mode). All 3 count columns. `maximumBytesBilled` runaway cap above the measured ~32 GB.                                                                                                                                                                             |

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

@@ -1,3 +1,5 @@
+import { ApplicationFailure } from '@temporalio/client'
+
 import {
   OsspckgsJobKind,
   OsspckgsSyncMode,
@@ -224,8 +226,11 @@ export async function bqExportToGcs(input: BqExportToGcsInput): Promise<BqExport
       'BQ dry-run complete',
     )
     if (dryRunBytes > ceiling) {
-      throw new Error(
+      // Non-retryable: the dry-run byte count is deterministic for a given query, so retrying
+      // (the caller's default maximumAttempts: 3) just repeats the same failed dry-run 3 times.
+      throw ApplicationFailure.nonRetryable(
         `BQ dry-run for ${jobKind} reports ${dryRunBytes} bytes > ceiling ${ceiling} — aborting`,
+        'BQ_CEILING_EXCEEDED',
       )
     }
   }

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

@@ -8,5 +8,6 @@ export * from './getLastSnapshot'
 export * from './getResumeExport'
 export * from './checkDependentCountsGuard'
 export * from './checkEdgeSnapshotQuality'
+export * from './notifyBqCeilingSkip'
 export * from './probePartitionExists'
 export * from './resolveSnapshotDate'

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

@@ -0,0 +1,33 @@
+import { SlackChannel, SlackPersona, sendSlackNotification } from '@crowd/slack'
+
+export interface NotifyBqCeilingSkipInput {
+  jobKind: string
+  message: string
+}
+
+// The advisories ceiling breach is caught one level up in bootstrapOsspckgs instead of inside a
+// guard activity, so it needs its own alert call to keep repeated skips from going unnoticed
+// (CM-1362 review). Uses CDP_AKRITES_ALERTS — the packages_worker team's own channel (matches
+// service.ts / blast-radius-worker.ts), per reviewer request — rather than CDP_CRITICAL_ALERTS.
+export async function notifyBqCeilingSkip(input: NotifyBqCeilingSkipInput): Promise<void> {
+  // ingestAdvisories merges `advisories` to Postgres before exporting `advisory_packages`
+  // (bootstrapOsspckgs.ts), so a breach on the latter still leaves this run's new advisory rows
+  // committed with no package links — "existing data untouched" would misstate that (bugbot
+  // review on CM-1362). A breach on `advisories` itself precedes any write this run, so that
+  // claim stays accurate there.
+  const impact =
+    input.jobKind === 'advisory_packages'
+      ? "This run's advisories were already merged without package links — they'll link up on the next successful run."
+      : 'Ingest skipped for this run so scorecard/ranking still complete. Existing data untouched.'
+  sendSlackNotification(
+    SlackChannel.CDP_AKRITES_ALERTS,
+    SlackPersona.CRITICAL_ALERTER,
+    `:warning: ${input.jobKind} BQ byte ceiling exceeded — soft-failed`,
+    [
+      {
+        title: 'Action',
+        text: `${impact} ${input.message}`,
+      },
+    ],
+  )
+}

services/apps/packages_worker/src/deps-dev/queries/advisoriesSql.ts

@@ -13,26 +13,20 @@ SELECT
 FROM \`bigquery-public-data.deps_dev_v1.AdvisoriesLatest\`
 `
 
+// No purl here — package_id is resolved in Postgres against the already-ingested `packages`
+// table (see ADVISORY_PACKAGES_MERGE_SQL in workflows/ingestAdvisories.ts). Pulling purl from
+// BigQuery required scanning all of PackageVersionsLatest (~1.5 TB) just for a join key we
+// already have locally (CM-1362).
 export function buildAdvisoryPackagesSql(systems: string): string {
   return `
-WITH purl_map AS (
-  SELECT System, Name, ANY_VALUE(REGEXP_REPLACE(Purl, r'@[^@]*$', '')) AS purl
-  FROM \`bigquery-public-data.deps_dev_v1.PackageVersionsLatest\`
-  WHERE System IN (${systems})
-    AND Purl IS NOT NULL
-    AND Name NOT LIKE '%>%'
-  GROUP BY System, Name
-)
 SELECT
   a.SourceID             AS osv_id,
   LOWER(pkg.System)      AS ecosystem,
   pkg.Name               AS package_name,
-  pm.purl                AS purl,
   pkg.AffectedVersions   AS range_raw,
   pkg.UnaffectedVersions AS unaffected_raw
 FROM \`bigquery-public-data.deps_dev_v1.AdvisoriesLatest\` a,
 UNNEST(a.Packages) AS pkg
-LEFT JOIN purl_map pm ON pm.System = pkg.System AND pm.Name = pkg.Name
 WHERE pkg.System IN (${systems})
   AND pkg.Name NOT LIKE '%>%'
 `

services/apps/packages_worker/src/deps-dev/queries/pgIdentity.ts

@@ -0,0 +1,22 @@
+// The (namespace, name) split ingestPackages.ts writes into `packages` (see MERGE_SQL there).
+// Any consumer that needs to join back onto that identity by name must derive it identically,
+// so both sides share this one definition instead of drifting apart.
+export function packageNameSplitSql(
+  alias: string,
+  nameCol: string,
+): { namespace: string; name: string } {
+  const ecosystem = `${alias}.ecosystem`
+  const rawName = `${alias}.${nameCol}`
+  return {
+    namespace: `CASE
+      WHEN ${ecosystem} = 'maven' THEN SPLIT_PART(${rawName}, ':', 1)
+      WHEN ${rawName} LIKE '@%/%' THEN SPLIT_PART(${rawName}, '/', 1)
+      ELSE NULL
+    END`,
+    name: `CASE
+      WHEN ${ecosystem} = 'maven' THEN SPLIT_PART(${rawName}, ':', 2)
+      WHEN ${rawName} LIKE '@%/%' THEN SPLIT_PART(${rawName}, '/', 2)
+      ELSE ${rawName}
+    END`,
+  }
+}

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

@@ -24,6 +24,11 @@ const { getLastSnapshot, probePartitionExists, resolveSnapshotDate } = proxyActi
   retry: { maximumAttempts: 3 },
 })
 
+const { notifyBqCeilingSkip } = proxyActivities<typeof depsDevActivities>({
+  startToCloseTimeout: '1 minute',
+  retry: { maximumAttempts: 3 },
+})
+
 type JobKind =
   | 'packages'
   | 'repos'
@@ -300,19 +305,37 @@ export async function bootstrapOsspckgs(opts: {
     }
   }
   if (runs('advisories') || runs('advisory_packages')) {
-    await executeChild(ingestAdvisories, {
-      args: [
-        {
-          runId,
-          syncMode: opts.mode,
-          today,
-          watermark: wm('advisories'),
-          ecosystems: opts.ecosystems,
-          reuseExports: opts.reuseExports,
-          exportName: opts.exportName,
-        },
-      ],
-    })
+    try {
+      await executeChild(ingestAdvisories, {
+        args: [
+          {
+            runId,
+            syncMode: opts.mode,
+            today,
+            watermark: wm('advisories'),
+            ecosystems: opts.ecosystems,
+            reuseExports: opts.reuseExports,
+            exportName: opts.exportName,
+          },
+        ],
+      })
+    } catch (err) {
+      // Only soft-fail on the BQ byte-ceiling guard (CM-1362), mirroring the dependent_counts /
+      // package_dependencies handling above. advisories is the last data kind — letting a ceiling
+      // breach here propagate unhandled used to strand scorecard + ranking below for the whole
+      // run. All other errors (BQ timeout, DB failure, etc.) still propagate.
+      const cause = err instanceof ChildWorkflowFailure ? err.cause : err
+      if (!(cause instanceof ApplicationFailure) || cause.type !== 'BQ_CEILING_EXCEEDED') {
+        throw err
+      }
+      // Unlike checkDependentCountsGuard/checkEdgeSnapshotQuality, this failure happens before
+      // any ingest-job row is created, so there's no failed-job row for an operator to notice —
+      // alert explicitly or repeated skips go unnoticed (review comment on CM-1362).
+      // ingestAdvisories carries the failing export's jobKind as the failure detail ('advisories'
+      // or 'advisory_packages') so the alert names the export that actually breached.
+      const jobKind = typeof cause.details?.[0] === 'string' ? cause.details[0] : 'advisories'
+      await notifyBqCeilingSkip({ jobKind, message: cause.message })
+    }
   }
   if (runs('scorecard')) {
     await executeChild(ingestScorecard, {

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

@@ -1,9 +1,15 @@
-import { proxyActivities } from '@temporalio/workflow'
+import { ActivityFailure, ApplicationFailure, proxyActivities } from '@temporalio/workflow'
 
 import type * as depsDevActivities from '../activities'
 import { ADVISORIES_SQL, buildAdvisoryPackagesSql } from '../queries/advisoriesSql'
+import { packageNameSplitSql } from '../queries/pgIdentity'
 import { toSystemsFilter } from '../queries/systems'
 
+const { namespace: NAMESPACE_SPLIT_SQL, name: NAME_SPLIT_SQL } = packageNameSplitSql(
+  'r',
+  'package_name',
+)
+
 const { bqExportToGcs } = proxyActivities<typeof depsDevActivities>({
   startToCloseTimeout: '1 hour',
   retry: { maximumAttempts: 3, initialInterval: '1 minute', backoffCoefficient: 2 },
@@ -25,6 +31,25 @@ const { mergeStagingToTable } = proxyActivities<typeof depsDevActivities>({
   retry: { maximumAttempts: 1 },
 })
 
+// bqExportToGcs throws ApplicationFailure.nonRetryable('BQ_CEILING_EXCEEDED') directly from
+// activity code, which the SDK surfaces here as ActivityFailure.cause. Rethrow as a workflow-level
+// ApplicationFailure carrying the job kind as a detail so bootstrapOsspckgs's unwrap (err.cause on
+// the resulting ChildWorkflowFailure) can match it, soft-fail, and alert on whichever export
+// actually breached — mirroring the DEPENDENT_COUNTS_GUARD / EDGE_SNAPSHOT_GUARD pattern there.
+// Applies to both exports below (review comment on CM-1362): a ceiling breach on either one must
+// be recognized, not just advisory_packages.
+async function exportWithCeilingGuard(input: Parameters<typeof bqExportToGcs>[0]) {
+  try {
+    return await bqExportToGcs(input)
+  } catch (err) {
+    const cause = err instanceof ActivityFailure ? err.cause : err
+    if (cause instanceof ApplicationFailure && cause.type === 'BQ_CEILING_EXCEEDED') {
+      throw ApplicationFailure.nonRetryable(cause.message, 'BQ_CEILING_EXCEEDED', input.jobKind)
+    }
+    throw err
+  }
+}
+
 const ADVISORIES_STAGING_TABLE = 'staging.osspckgs_advisories_raw'
 const ADVISORY_PACKAGES_STAGING_TABLE = 'staging.osspckgs_advisory_packages_raw'
 
@@ -42,16 +67,18 @@ CREATE UNLOGGED TABLE IF NOT EXISTS staging.osspckgs_advisories_raw (
 )
 `
 
-const ADVISORY_PACKAGES_STAGING_DDL = `
-CREATE UNLOGGED TABLE IF NOT EXISTS staging.osspckgs_advisory_packages_raw (
+// Two-statement DDL: DROP before CREATE so a deployed table with the old `purl` column doesn't
+// silently stick around under `CREATE ... IF NOT EXISTS`. Staging is TRUNCATED/recreated every run.
+const ADVISORY_PACKAGES_STAGING_DDL = [
+  `DROP TABLE IF EXISTS staging.osspckgs_advisory_packages_raw`,
+  `CREATE UNLOGGED TABLE staging.osspckgs_advisory_packages_raw (
   osv_id         text,
   ecosystem      text,
   package_name   text,
-  purl           text,
   range_raw      text,
   unaffected_raw text
-)
-`
+)`,
+]
 
 const ADVISORIES_MERGE_SQL = `
 INSERT INTO advisories (osv_id, source, source_url, summary, details, cvss, severity, aliases, published_at, created_at, updated_at)
@@ -63,17 +90,32 @@ FROM staging.osspckgs_advisories_raw
 ON CONFLICT (osv_id) DO NOTHING
 `
 
+// package_id is resolved here by reconstructing the same (ecosystem, namespace, name) identity
+// ingestPackages.ts writes into `packages` (packageNameSplitSql, shared by both), rather than by
+// a BQ-sourced purl (CM-1362 — the purl_map scan cost ~1.5 TB per run for a join key we already
+// had locally). COALESCE(p.namespace,'') mirrors the unique index expression verbatim, so this
+// stays an index lookup. Still a LEFT JOIN: package_id stays nullable, resolveMissingPackageIds
+// keeps its catch-up role for anything unresolved here.
 const ADVISORY_PACKAGES_MERGE_SQL = `
+WITH s AS (
+  SELECT r.osv_id, r.ecosystem, r.package_name,
+    ${NAMESPACE_SPLIT_SQL} AS namespace,
+    ${NAME_SPLIT_SQL} AS name
+  FROM staging.osspckgs_advisory_packages_raw r
+)
 INSERT INTO advisory_packages (advisory_id, package_id, ecosystem, package_name, created_at, updated_at)
 SELECT
   adv.id,
   p.id,
   s.ecosystem,
   s.package_name,
   NOW(), NOW()
-FROM staging.osspckgs_advisory_packages_raw s
+FROM s
 JOIN advisories adv ON adv.osv_id = s.osv_id
-LEFT JOIN packages p ON p.purl = s.purl
+LEFT JOIN packages p
+  ON p.ecosystem = s.ecosystem
+ AND COALESCE(p.namespace, '') = COALESCE(s.namespace, '')
+ AND p.name = s.name
 ON CONFLICT (advisory_id, ecosystem, package_name) DO NOTHING
 `
 
@@ -160,7 +202,6 @@ const ADVISORY_PACKAGES_PG_COLUMNS = [
   'osv_id',
   'ecosystem',
   'package_name',
-  'purl',
   'range_raw',
   'unaffected_raw',
 ]
@@ -179,7 +220,7 @@ export async function ingestAdvisories(opts: {
   const systems = toSystemsFilter(opts.ecosystems)
 
   // Step 1: advisories header rows
-  const advisoriesExport = await bqExportToGcs({
+  const advisoriesExport = await exportWithCeilingGuard({
     jobKind: 'advisories',
     sql: ADVISORIES_SQL,
     runId: opts.runId,
@@ -253,13 +294,15 @@ export async function ingestAdvisories(opts: {
   }
 
   // Step 2: advisory_packages + affected ranges (FK → advisories must exist first)
-  const pkgsExport = await bqExportToGcs({
+  const pkgsExport = await exportWithCeilingGuard({
     jobKind: 'advisory_packages',
     sql: buildAdvisoryPackagesSql(systems),
     runId: opts.runId,
     syncMode: opts.syncMode,
     snapshotAt: opts.today,
-    maxBytesGb: 1500,
+    // No purl_map (CM-1362): measured actual scan is ~1.4 GB against AdvisoriesLatest; 50 GB
+    // keeps this a real regression gate instead of a ceiling that periodically needs raising.
+    maxBytesGb: 50,
     reuseExports: opts.reuseExports,
     exportName: opts.exportName,
     ecosystems: opts.ecosystems,