diff --git a/docs/adr/0004-go-nuget-transitive-dependent-counts.md b/docs/adr/0004-go-nuget-transitive-dependent-counts.md new file mode 100644 index 0000000000..faa9c26557 --- /dev/null +++ b/docs/adr/0004-go-nuget-transitive-dependent-counts.md @@ -0,0 +1,100 @@ +# ADR-0004: Compute GO/NUGET transitive dependent counts via exact reverse closure (over HLL approximation) + +**Date**: 2026-06-23 +**Status**: accepted +**Deciders**: Uroš Marolt + +## Context +deps.dev publishes its reverse-dependent index (`Dependents`) and its resolved/transitive +dependency graph (`Dependencies`, `DependencyGraphEdges`) for only four ecosystems — +NPM, MAVEN, PYPI, CARGO (verified via BQ 2026-06-23: each `GROUP BY System` returns exactly +those four). GO and NUGET have only raw direct manifests (`GoRequirementsLatest`, +`NuGetRequirementsLatest`). The `dependent_counts` ingest was therefore split three ways +(edges / go / nuget — see the dependent-counts split). Direct `dependent_count` for GO/NUGET +is a cheap manifest invert, but `transitive_dependent_count` (MinimumDepth>1) and the all-depth +`dependent_repos_count` require the reverse **transitive closure**, which deps.dev does not +provide for these two ecosystems — so we must compute it ourselves from the direct edges. + +## Decision +Compute the **exact** reverse transitive closure for GO/NUGET — a semi-naive fixpoint over +module-level edges (names hashed to INT64 via `FARM_FINGERPRINT`, persisted + clustered) iterated +to convergence — so GO/NUGET transitive counts are true integers matching the edge-system +methodology. We chose exact over an HLL-sketch approximation, accepting the higher cost for now. + +## Alternatives Considered + +### Alternative 1: HLL sketch propagation (approximate) +- **Pros**: ~cents/run, seconds, tiny tables (one HyperLogLog sketch per subject, ~188K rows for + GO, merged along edges to a fixpoint — never materializes the billions of pairs). Full-depth. +- **Cons**: probabilistic cardinality estimate with bounded relative error ≈ `1.04/sqrt(2^p)` + (~1.6% at precision 12, ~0.4% at 15). Methodology differs from the exact edge-system counts, so + GO/NUGET would be ~1% approximate while the other four ecosystems are exact. +- **Why not**: the decision is to have identical, exact integer methodology across all six + ecosystems. The measured exact cost (below) is acceptable for now. HLL remains the documented + fallback if weekly cost becomes prohibitive. + +### Alternative 2: Bounded-depth closure (cap at K hops) +- **Pros**: K deterministic joins in a single statement, dry-runnable, cost-bounded, no scripting. +- **Cons**: undercounts. The GO closure needs **31 hops** to converge; new pairs peak at hop 8 + (~507M) and only decay to zero by hop 31. Capping at a small K drops hundreds of millions of + genuine transitive pairs. +- **Why not**: a low cap is materially wrong; a cap high enough to be correct is no cheaper than + full convergence. + +## Consequences + +### Positive +- Exact integer counts, full depth, identical methodology to NPM/MAVEN/PYPI/CARGO. +- Deterministic and reproducible; no estimator error to explain to downstream consumers. +- Termination is guaranteed by construction: each iteration's frontier is anti-joined against the + accumulated `reach`, so only brand-new pairs survive and the finite pair space converges (proven: + GO converged at hop 31 with `new_pairs → 0`). + +### Negative +- This is by far the heaviest job in the packages pipeline. Measured exact runs (2026-06-23, + INT64-fingerprint-optimized; raw string names would be ~5×). The build-`reach` figures are from the + closure probe; the full-pipeline figures (incl. the all-depth repos aggregation + per-subject + counts → `_export_data`) are from the end-to-end validation run of the production script: + - **GO**: 5.81B closure pairs, 31 iterations. Build reach ≈ 1.86 TB; **full pipeline = 2.31 TB + billed (~$14.5 at $6.25/TiB), 99 slot-hours, 16 min wall, 132 child statements** (validated). + - **NUGET**: 114M closure pairs, 19 iterations, ~4 min wall, ~32 GB billed (~$0.19); full pipeline + validated → 191,762 subject rows. + - GO dominates (its graph is ~50× larger and deeper). Combined ≈ **~$15/run, ~20 min, weekly**. + For comparison the edge `dependent_counts` job scans ~310 GB (~$2). +- Requires a semi-naive BQ scripting job (not a single exportable SELECT), so it does not fit the + existing one-query→GCS pipeline shape as-is (integrated via an `isScript` mode — see below). + +### Risks +- Cost/runtime grow with the GO/NUGET graphs. Mitigations: INT64 fingerprints (~5× cheaper), + clustering on the join keys, and a documented escape hatch to switch GO/NUGET to HLL (Alternative + 1) if cost becomes prohibitive — counts in the hundred-thousands make ~1% error invisible for + popularity/ranking. +- `FARM_FINGERPRINT` collisions are negligible at ~1–2M nodes (~1e-6) but non-zero; acceptable for + these metrics. Switch to a dictionary-encoded INT id if exactness must be collision-proof. + +## Pipeline integration (decided 2026-06-23) +- **No persistent scratch dataset.** The closure runs as one multi-statement scripting job using + session-scoped `CREATE TEMP TABLE`s (edges → INT64-fingerprinted edges → `reach` fixpoint → + fingerprint→name lookup → `_export_data`). Temp tables are clustered on the join keys and are + auto-dropped when the script's session ends — on success *and* failure — so there is no lingering + storage to bill or clean up. (The interactive probe used regular tables in + `lfx-insights:scratch_closure_probe`; that is leftover, dropped separately — not the pipeline.) +- **Reuses the existing job path, not a new kind.** `bqExportToGcs` gains an `isScript` mode: when + set, `sql` is the full script (ending in `CREATE TEMP TABLE _export_data AS …`) and the activity + appends only the `EXPORT DATA … AS SELECT * FROM _export_data` statement rather than wrapping + `sql` in a subquery. Everything downstream (listParquetFiles → guard → chunked + gcsParquetToStaging → mergeStagingToTable) is unchanged, and the `dependent_counts_go` / + `dependent_counts_nuget` kinds, guard baselines, and monitor entries stay continuous. +- **Cost guard.** A dry-run cannot price a `WHILE` loop, so script mode skips it and enforces the + byte ceiling server-side via `maximumBytesBilled` (= the per-variant `maxBytesGb`). The script + also carries an iteration cap as the deterministic runaway guard (convergence already proven: + GO hop 31, NUGET hop 19, both `new_pairs → 0`). +- **Final output → counts.** `_export_data` emits `(purl, dependent_count, + transitive_dependent_count, dependent_repos_count)`: fingerprints in `reach` are mapped back to + names via a `names` temp table, joined to `purl_map` (name→purl). `transitive_dependent_count = + COUNT(DISTINCT dep over reach) − direct_count`; `dependent_repos_count` = distinct + `PackageVersionToProject` SOURCE_REPO_TYPE repos over the all-depth dependent set. +- **Weekly trigger.** No new schedule — the `go`/`nuget` variants are already child workflows of + `bootstrapOsspckgs`, which the existing `osspckgs-bootstrap-weekly` schedule runs incrementally. + +Related: ADR-0003 (deps BQ table selection — same root cause: GO/NUGET absent from the resolved-graph tables). diff --git a/docs/adr/README.md b/docs/adr/README.md index 9bb2045dca..e96d02f8b3 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -10,6 +10,7 @@ Use the `/adr` skill in Claude Code to record new ADRs or query past decisions. | --------------------------------------------------- | ---------------------------------------- | ------ | ---------- | | [ADR-0001](./0001-oss-packages-design-decisions.md) | OSS packages — design decisions (living) | living | 2026-05-27 | | [ADR-0003](./0003-deps-bq-table-selection.md) | Use DependencyGraphEdgesLatest for deps ingestion; defer DependenciesLatest until NUGET or GO needed | accepted | 2026-05-29 | +| [ADR-0004](./0004-go-nuget-transitive-dependent-counts.md) | Compute GO/NUGET transitive dependent counts via exact reverse closure (over HLL approximation) | accepted | 2026-06-23 | ## Why ADRs? diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 7541586a7a..ec3949f97c 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -1373,9 +1373,6 @@ importers: '@types/node': specifier: ^20.8.2 version: 20.12.7 - '@types/pg-copy-streams': - specifier: ^1.2.5 - version: 1.2.5 '@types/semver': specifier: ^7.5.8 version: 7.5.8 @@ -4890,12 +4887,6 @@ packages: '@types/node@22.19.10': resolution: {integrity: sha512-tF5VOugLS/EuDlTBijk0MqABfP8UxgYazTLo3uIn3b4yJgg26QRbVYJYsDtHrjdDUIRfP70+VfhTTc+CE1yskw==} - '@types/pg-copy-streams@1.2.5': - resolution: {integrity: sha512-7D6/GYW2uHIaVU6S/5omI+6RZnwlZBpLQDZAH83xX1rjxAOK0f6/deKyyUTewxqts145VIGn6XWYz1YGf50G5g==} - - '@types/pg@8.20.0': - resolution: {integrity: sha512-bEPFOaMAHTEP1EzpvHTbmwR8UsFyHSKsRisLIHVMXnpNefSbGA1bD6CVy+qKjGSqmZqNqBDV2azOBo8TgkcVow==} - '@types/q@1.5.8': resolution: {integrity: sha512-hroOstUScF6zhIi+5+x0dzqrHA1EJi+Irri6b1fxolMTqqHIV/Cg77EtnQcZqZCu8hR3mX2BzIxN4/GzI68Kfw==} @@ -14044,17 +14035,6 @@ snapshots: dependencies: undici-types: 6.21.0 - '@types/pg-copy-streams@1.2.5': - dependencies: - '@types/node': 20.12.7 - '@types/pg': 8.20.0 - - '@types/pg@8.20.0': - dependencies: - '@types/node': 20.12.7 - pg-protocol: 1.6.1 - pg-types: 2.2.0 - '@types/q@1.5.8': {} '@types/qs@6.9.15': {} diff --git a/services/apps/packages_worker/package.json b/services/apps/packages_worker/package.json index d372b3465c..88351e70ec 100644 --- a/services/apps/packages_worker/package.json +++ b/services/apps/packages_worker/package.json @@ -85,7 +85,6 @@ "devDependencies": { "@types/jsonwebtoken": "^9.0.0", "@types/node": "^20.8.2", - "@types/pg-copy-streams": "^1.2.5", "@types/semver": "^7.5.8", "@types/unzipper": "^0.10.10", "nodemon": "^3.0.1", diff --git a/services/apps/packages_worker/src/bin/bq-dataset-ingest.ts b/services/apps/packages_worker/src/bin/bq-dataset-ingest.ts index a81c14914a..141973a626 100644 --- a/services/apps/packages_worker/src/bin/bq-dataset-ingest.ts +++ b/services/apps/packages_worker/src/bin/bq-dataset-ingest.ts @@ -1,10 +1,8 @@ import { scheduleOsspckgsBootstrap } from '../deps-dev/schedules/bootstrap' -import { scheduleOsspckgsCleanup } from '../schedules/cleanup' import { svc } from '../service' setImmediate(async () => { await svc.init() await scheduleOsspckgsBootstrap() - await scheduleOsspckgsCleanup() await svc.start() }) diff --git a/services/apps/packages_worker/src/cargo/loadDump.ts b/services/apps/packages_worker/src/cargo/loadDump.ts index 12e52999cb..d3884d4cb4 100644 --- a/services/apps/packages_worker/src/cargo/loadDump.ts +++ b/services/apps/packages_worker/src/cargo/loadDump.ts @@ -97,7 +97,9 @@ async function copyCsv( ): Promise { const con = await db.connect() try { - // @types/pg-copy-streams Submittable doesn't match @types/pg's query overload. + // pg-promise's bundled IClient.query types only return Promise — they lack node-pg's + // `query(qs: T): T` stream overload. At runtime con.client IS a node-pg + // Client and returns the CopyStreamQuery, so cast through to recover it. // eslint-disable-next-line @typescript-eslint/no-explicit-any const stream: CopyStreamQuery = (con.client as any).query( copyFrom(`COPY ${STAGING_SCHEMA}.${table} FROM STDIN CSV HEADER`), diff --git a/services/apps/packages_worker/src/deps-dev/README.md b/services/apps/packages_worker/src/deps-dev/README.md index 8e283a9e4e..9701055aab 100644 --- a/services/apps/packages_worker/src/deps-dev/README.md +++ b/services/apps/packages_worker/src/deps-dev/README.md @@ -30,7 +30,10 @@ See `src/scripts/triggerBootstrap.ts --help` for full options. Each job kind has a hardcoded `maxBytesGb` ceiling checked via a BQ dry-run before the real export fires. If the dry-run exceeds the ceiling the activity -fails immediately (no BQ cost incurred). +fails immediately (no BQ cost incurred). Script-mode kinds (the GO/NUGET reverse +transitive closure — a multi-statement `WHILE` loop a dry-run cannot price) skip +the dry-run and instead enforce the ceiling server-side via `maximumBytesBilled`; +there the ceiling is a runaway cap set above expected spend, not a tight gate. Every ceiling can be overridden at runtime via an env variable — useful when a table grows past the default ceiling without requiring a code deploy: @@ -45,31 +48,32 @@ The mode-specific key takes precedence over the generic key. Value must be a pos **When adding a new job kind, add a row to this table.** -| Env var override | Default (GB) | Job kind | Notes | -| -------------------------------------------------- | -----------: | ---------------------- | --------------------------------------------------------- | -| `BQ_DATASET_INGEST_PACKAGES_FULL_MAX_BQ_GB` | 6000 | `packages` | Full only (set in `ingestPackages.ts`) | -| `BQ_DATASET_INGEST_PACKAGES_INCREMENTAL_MAX_BQ_GB` | 400 | `packages` | Incremental only (set in `ingestPackages.ts`) | -| `BQ_DATASET_INGEST_VERSIONS_MAX_BQ_GB` | 400 | `versions` | | -| `BQ_DATASET_INGEST_PACKAGE_DEPENDENCIES_MAX_BQ_GB` | 10000 | `package_dependencies` | Incremental always scans ~2 full-day partitions (~3.85TB) | -| `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_DEPENDENT_COUNTS_MAX_BQ_GB` | 2000 | `dependent_counts` | | -| `BQ_DATASET_INGEST_SCORECARD_REPOS_MAX_BQ_GB` | 50 | `scorecard_repos` | | -| `BQ_DATASET_INGEST_SCORECARD_CHECKS_MAX_BQ_GB` | 500 | `scorecard_checks` | | +| Env var override | Default (GB) | Job kind | Notes | +| ---------------------------------------------------- | ----------------------: | ------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `BQ_DATASET_INGEST_PACKAGES_FULL_MAX_BQ_GB` | 6000 | `packages` | Full only (set in `ingestPackages.ts`) | +| `BQ_DATASET_INGEST_PACKAGES_INCREMENTAL_MAX_BQ_GB` | 400 | `packages` | Incremental only (set in `ingestPackages.ts`) | +| `BQ_DATASET_INGEST_VERSIONS_MAX_BQ_GB` | 400 | `versions` | | +| `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_DEPENDENT_COUNTS_MAX_BQ_GB` | 2000 | `dependent_counts` | Edges only (NPM/MAVEN/PYPI/CARGO) from the `Dependents` reverse index. GO/NUGET 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. | +| `BQ_DATASET_INGEST_SCORECARD_REPOS_MAX_BQ_GB` | 50 | `scorecard_repos` | | +| `BQ_DATASET_INGEST_SCORECARD_CHECKS_MAX_BQ_GB` | 500 | `scorecard_checks` | | The override logic lives in `src/deps-dev/activities/bqExportToGcs.ts`. ## Environment variables -| Variable | Required | Purpose | -| ------------------------------ | --------- | -------------------------------------------------------------------------------------------------- | -| `OSSPCKGS_GCP_PROJECT` | yes | GCP project ID for BQ and GCS | -| `OSSPCKGS_GCS_BUCKET` | yes | GCS bucket for Parquet exports | -| `OSSPCKGS_GCP_CREDENTIALS_B64` | yes | Base64-encoded GCP service account JSON | -| `OSSPCKGS_DEPS_TABLE` | no | Set to `B` to use `DependenciesLatest` (ADR-0003 Option B) instead of `DependencyGraphEdgesLatest` | -| `CROWD_TEMPORAL_SERVER_URL` | yes | Temporal server address | -| `CROWD_TEMPORAL_NAMESPACE` | yes | Temporal namespace (overrides `backend-config` default) | -| `CROWD_TEMPORAL_CERTIFICATE` | prod only | Base64-encoded mTLS client certificate | -| `CROWD_TEMPORAL_PRIVATE_KEY` | prod only | Base64-encoded mTLS private key | +| Variable | Required | Purpose | +| ------------------------------ | --------- | ------------------------------------------------------- | +| `OSSPCKGS_GCP_PROJECT` | yes | GCP project ID for BQ and GCS | +| `OSSPCKGS_GCS_BUCKET` | yes | GCS bucket for Parquet exports | +| `OSSPCKGS_GCP_CREDENTIALS_B64` | yes | Base64-encoded GCP service account JSON | +| `CROWD_TEMPORAL_SERVER_URL` | yes | Temporal server address | +| `CROWD_TEMPORAL_NAMESPACE` | yes | Temporal namespace (overrides `backend-config` default) | +| `CROWD_TEMPORAL_CERTIFICATE` | prod only | Base64-encoded mTLS client certificate | +| `CROWD_TEMPORAL_PRIVATE_KEY` | prod only | Base64-encoded mTLS private key | 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 a594295af0..ebdc531bea 100644 --- a/services/apps/packages_worker/src/deps-dev/activities/bqExportToGcs.ts +++ b/services/apps/packages_worker/src/deps-dev/activities/bqExportToGcs.ts @@ -25,6 +25,13 @@ export interface BqExportToGcsInput { reuseExports?: boolean exportName?: string ecosystems?: string[] + // Script mode (GO/NUGET reverse transitive closure). When true, `sql` is a full multi-statement + // BQ script (semi-naive fixpoint over TEMP tables) that ends by creating `TEMP TABLE _export_data` + // holding the final result set. The activity appends only the EXPORT DATA statement instead of + // wrapping `sql` in a subquery (a script cannot be a subquery). The up-front dry-run ceiling check + // 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 } export interface BqExportToGcsOutput { @@ -45,6 +52,7 @@ export async function bqExportToGcs(input: BqExportToGcsInput): Promise ceiling) { - throw new Error( - `BQ dry-run for ${jobKind} reports ${dryRunBytes} bytes > ceiling ${ceiling} — aborting`, + + // Single-SELECT exports: dry-run up-front, abort if the scan exceeds the ceiling. + // Script exports (isScript): a dry-run only validates the first statement and cannot price the + // semi-naive WHILE loop, so the dry-run is skipped here; the ceiling is instead enforced + // server-side via maximumBytesBilled on the real job below (see ADR-0004). + if (!isScript) { + // 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 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`, + ) + } } const provisionalDate = snapshotAt ? new Date(snapshotAt) : null @@ -196,12 +217,18 @@ export async function bqExportToGcs(input: BqExportToGcsInput): Promise { const qx = await getPackagesDb() - const prevRowCount = await dalGetLastCompletedJobRowCount(qx, 'dependent_counts') + const jobKind = input.jobKind ?? 'dependent_counts' + const prevRowCount = await dalGetLastCompletedJobRowCount(qx, jobKind) if (prevRowCount === null || prevRowCount <= 0) { return { ok: true, prevRowCount, dropPct: null } @@ -32,7 +37,7 @@ export async function checkDependentCountsGuard( sendSlackNotification( SlackChannel.CDP_CRITICAL_ALERTS, SlackPersona.CRITICAL_ALERTER, - ':warning: dependent_counts row count anomaly detected', + `:warning: ${jobKind} row count anomaly detected`, [ { title: 'Snapshot', diff --git a/services/apps/packages_worker/src/deps-dev/activities/checkEdgeSnapshotQuality.ts b/services/apps/packages_worker/src/deps-dev/activities/checkEdgeSnapshotQuality.ts new file mode 100644 index 0000000000..955e91d273 --- /dev/null +++ b/services/apps/packages_worker/src/deps-dev/activities/checkEdgeSnapshotQuality.ts @@ -0,0 +1,221 @@ +import { SlackChannel, SlackPersona, sendSlackNotification } from '@crowd/slack' + +import { bigquery } from '../config' +import { assertSnapshotDate } from '../queries/depsSql' + +// deps.dev ships weekly full snapshots of the resolved dependency graph in +// DependencyGraphEdges (NPM/MAVEN/PYPI/CARGO). On 2026-06-11 and 2026-06-15 that +// pipeline shipped corrupt: every (package, version) collapsed to 1–2 arbitrary edges, +// each duplicated exactly ~100×. An incremental package_dependencies run against a +// corrupt snapshot exports ~550M garbage rows and burns ~5h before PG's +// ON CONFLICT DO NOTHING discards 99.5% of them — wasteful, though not data-destroying. +// +// This guard runs BEFORE the BQ export. It probes a small set of high-fanout canary packages +// over the today partition and rejects the snapshot when the duplication ratio +// (rows / distinct edges) is far above the healthy baseline of ~1.0 (corrupt ≈ 100). +// +// Cost: the filter is written as `(System = x AND Name IN (...))` OR-groups so it prunes on the +// table's clustering (System, Name, Version) — measured ~3.8GB billed (≈ $0.02), vs ~1.76TB if +// the predicate is a `(System, Name) IN UNNEST(structs)` tuple (which defeats cluster pruning). +// +// GO/NUGET are sourced from manifest tables (GoRequirements/NuGetRequirements) which are not +// produced by the resolution pipeline and were unaffected, so they are not probed here. + +export interface CheckEdgeSnapshotQualityInput { + 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 { + system: string + name: string + rows: number + edges: number + ratio: number | null // rows / distinct edges; ~1.0 healthy, ~100 when corrupt + present: boolean +} + +export interface CheckEdgeSnapshotQualityOutput { + ok: boolean + reason?: string + canaries: CanaryStat[] +} + +// Systems whose direct deps come from the resolved graph (DependencyGraphEdges). +// Mirrors EDGE_SYSTEMS in queries/depsSql.ts — GO/NUGET are manifest-sourced and excluded. +const EDGE_SYSTEMS = new Set(['NPM', 'MAVEN', 'PYPI', 'CARGO']) + +// High-fanout packages that every healthy snapshot resolves with MANY distinct direct edges. +// All verified present with ratio ≈ 1.0 on the healthy 2026-06-01 snapshot. Deliberately avoid +// dependency-free packages (e.g. lodash, urllib3) — they have zero direct edges even when healthy, +// so they'd masquerade as "missing". Keep several per system so a single yank can't trip the guard. +const CANARIES: ReadonlyArray<{ system: string; name: string }> = [ + { system: 'NPM', name: 'express' }, + { system: 'NPM', name: 'webpack' }, + { system: 'NPM', name: 'react' }, + { system: 'NPM', name: 'eslint' }, + { system: 'NPM', name: '@babel/core' }, + { system: 'MAVEN', name: 'com.google.guava:guava' }, + { system: 'MAVEN', name: 'org.springframework:spring-context' }, + { system: 'MAVEN', name: 'org.apache.httpcomponents:httpclient' }, + { system: 'PYPI', name: 'requests' }, + { system: 'PYPI', name: 'flask' }, + { system: 'PYPI', name: 'django' }, + { system: 'PYPI', name: 'pandas' }, + { system: 'CARGO', name: 'serde' }, + { system: 'CARGO', name: 'tokio' }, + { system: 'CARGO', name: 'clap' }, + { system: 'CARGO', name: 'reqwest' }, +] + +// Healthy ratio is ~1.0 (max observed ≈ 1.8 for serde); the corrupt snapshots ran ~100×. +// 5 leaves wide margin for benign multiplicity while catching the ×100 collapse unambiguously. +const DUP_RATIO_REJECT = 5 + +interface CanaryRow { + system: string + name: string + row_count: number | string | { value: string } + edge_count: number | string | { value: string } +} + +// BigQuery returns INT64 columns as number | string | { value }; coerce defensively. +function toNum(v: number | string | { value: string }): number { + if (typeof v === 'number') return v + if (typeof v === 'string') return Number(v) + return Number(v.value) +} + +export async function checkEdgeSnapshotQuality( + input: CheckEdgeSnapshotQualityInput, +): Promise { + const inScope = new Set( + input.ecosystems.map((e) => e.toUpperCase()).filter((e) => EDGE_SYSTEMS.has(e)), + ) + const activeCanaries = CANARIES.filter((c) => inScope.has(c.system)) + + // No resolved-graph ecosystems requested (e.g. GO/NUGET only) → nothing to probe. + if (activeCanaries.length === 0) { + return { ok: true, canaries: [] } + } + + // 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() + for (const c of activeCanaries) { + const names = bySystem.get(c.system) ?? [] + names.push(c.name) + bySystem.set(c.system, names) + } + const systemPredicates = [...bySystem.entries()] + .map( + ([system, names]) => + `(e.System = '${system}' AND e.Name IN (${names.map((n) => `'${n}'`).join(', ')}))`, + ) + .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 = ` + SELECT + e.System AS system, + 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 \`${edgesTable}\` e + WHERE ${snapshotFilter}e.From.Name = e.Name AND e.From.Version = e.Version + AND ( + ${systemPredicates} + ) + GROUP BY e.System, e.Name + ` + + const [job] = await bigquery.createQueryJob({ query, location: 'US' }) + const [rows] = await job.getQueryResults() + const resultRows = rows as CanaryRow[] + + const canaries: CanaryStat[] = activeCanaries.map((c) => { + const r = resultRows.find((row) => row.system === c.system && row.name === c.name) + if (!r) { + return { system: c.system, name: c.name, rows: 0, edges: 0, ratio: null, present: false } + } + const rowCount = toNum(r.row_count) + const edgeCount = toNum(r.edge_count) + return { + system: c.system, + name: c.name, + rows: rowCount, + edges: edgeCount, + ratio: edgeCount > 0 ? rowCount / edgeCount : null, + present: true, + } + }) + + // Corruption shows up as a ×100 duplication ratio on the canaries that DO resolve, and/or as + // canaries vanishing entirely. Base the verdict on the ratio of PRESENT canaries (missingness + // alone is noisy — packages get yanked), with a floor against a mass collapse. + const present = canaries.filter((c) => c.present) + const overDuplicated = present.filter((c) => c.ratio !== null && c.ratio > DUP_RATIO_REJECT) + const minPresent = Math.ceil(activeCanaries.length / 2) + const massCollapse = present.length < minPresent + // Reject at HALF-OR-MORE (not strict majority): if half the high-fanout canaries show ×100 + // duplication the snapshot is corrupt enough to abort. Bias toward rejecting — a false reject + // just skips one cycle (existing rows preserved), a false accept burns ~5h on ~550M garbage rows. + const halfOrMoreOverDuplicated = overDuplicated.length >= Math.ceil(present.length / 2) + const ok = !massCollapse && !halfOrMoreOverDuplicated + + if (ok) { + return { ok: true, canaries } + } + + const dupDetail = overDuplicated + .map((c) => `${c.system}:${c.name} ratio=${(c.ratio ?? 0).toFixed(1)} (${c.rows}/${c.edges})`) + .join(', ') + const reason = massCollapse + ? `only ${present.length}/${activeCanaries.length} canaries resolved (min ${minPresent}) — snapshot looks collapsed` + : `${overDuplicated.length}/${present.length} present canaries over-duplicated (ratio > ${DUP_RATIO_REJECT}; healthy ≈ 1.0): ${dupDetail}` + + sendSlackNotification( + SlackChannel.CDP_CRITICAL_ALERTS, + SlackPersona.CRITICAL_ALERTER, + ':warning: deps.dev edge snapshot quality anomaly detected', + [ + { + title: 'Snapshot', + // Full/fill probe *Latest (no partition filter), so the literal date isn't what was scanned. + text: input.fullScan ? `${input.snapshotDate} (probed *Latest)` : input.snapshotDate, + }, + { + title: 'Canaries', + text: reason, + }, + { + title: 'Action', + text: 'package_dependencies ingest aborted before export — existing rows preserved. The deps.dev resolved-graph snapshot looks corrupt; re-run once a healthy snapshot is published.', + }, + ], + ) + + return { ok: false, reason, canaries } +} diff --git a/services/apps/packages_worker/src/deps-dev/activities/gcsParquetToStaging.ts b/services/apps/packages_worker/src/deps-dev/activities/gcsParquetToStaging.ts index e4efbdf4de..8ec9923b0e 100644 --- a/services/apps/packages_worker/src/deps-dev/activities/gcsParquetToStaging.ts +++ b/services/apps/packages_worker/src/deps-dev/activities/gcsParquetToStaging.ts @@ -114,66 +114,76 @@ export async function gcsParquetToStaging(input: GcsToStagingInput): Promise f.name.endsWith('.parquet')).map((f) => f.name) - } else { - throw new Error('gcsParquetToStaging: must provide either fileNames or gcsPrefix') - } - - const totalFiles = input.totalFiles ?? parquetFileNames.length + // Row is 'loading' from here; a staging-load failure must flip it to 'failed' with the reason + // rather than leave it stuck 'loading'. Mirrors the rankPackages pattern (criticality/activities). + try { + if (stagingDdl) { + const stmts = Array.isArray(stagingDdl) ? stagingDdl : [stagingDdl] + for (const stmt of stmts) await qx.result(stmt) + } + await qx.result(`TRUNCATE ${stagingTable}`) + // Reset progress at the start of a fresh run (first chunk). This handles job ID reuse + // via --export-name: a previous run's 193/193 would otherwise persist due to GREATEST. + if (filesOffset === 0 && input.totalFiles != null) { + await updateLoadingProgress(qx, jobId, 0, input.totalFiles, true) + } - log.info( - { jobId, stagingTable, fileCount: parquetFileNames.length, filesOffset, totalFiles }, - 'Loading parquet files into staging', - ) + let parquetFileNames: string[] + if (input.fileNames) { + parquetFileNames = input.fileNames + } else if (input.gcsPrefix) { + const objectPrefix = gcsPrefixToObjectPrefix(input.gcsPrefix) + const [files] = await bucket.getFiles({ prefix: objectPrefix }) + parquetFileNames = files.filter((f) => f.name.endsWith('.parquet')).map((f) => f.name) + } else { + throw new Error('gcsParquetToStaging: must provide either fileNames or gcsPrefix') + } - let totalLoaded = 0 + const totalFiles = input.totalFiles ?? parquetFileNames.length - for (let i = 0; i < parquetFileNames.length; i += MAX_CONCURRENT) { - const chunk = parquetFileNames.slice(i, i + MAX_CONCURRENT) - const counts = await Promise.all( - chunk.map((name) => loadParquetFile(qx, stagingTable, pgColumns, name, tsCols, decCols)), - ) - totalLoaded += counts.reduce((a, b) => a + b, 0) - const doneInBatch = i + chunk.length - const doneGlobal = filesOffset + doneInBatch log.info( - { - jobId, - totalLoaded, - progress: `${doneGlobal}/${totalFiles} (${Math.round((doneGlobal / totalFiles) * 100)}%)`, - }, - 'Staging load progress', + { jobId, stagingTable, fileCount: parquetFileNames.length, filesOffset, totalFiles }, + 'Loading parquet files into staging', ) - Context.current().heartbeat({ done: doneGlobal, total: totalFiles }) - await updateLoadingProgress(qx, jobId, doneGlobal, totalFiles) - } - const cumulativeStagingRows = (input.priorStagingRows ?? 0) + totalLoaded - await markJobStatus(qx, jobId, 'loading', { - rowCountStaging: cumulativeStagingRows, - tableRowCounts: { [`staging:${stagingTable}`]: totalLoaded }, - }) + let totalLoaded = 0 + + for (let i = 0; i < parquetFileNames.length; i += MAX_CONCURRENT) { + const chunk = parquetFileNames.slice(i, i + MAX_CONCURRENT) + const counts = await Promise.all( + chunk.map((name) => loadParquetFile(qx, stagingTable, pgColumns, name, tsCols, decCols)), + ) + totalLoaded += counts.reduce((a, b) => a + b, 0) + const doneInBatch = i + chunk.length + const doneGlobal = filesOffset + doneInBatch + log.info( + { + jobId, + totalLoaded, + progress: `${doneGlobal}/${totalFiles} (${Math.round((doneGlobal / totalFiles) * 100)}%)`, + }, + 'Staging load progress', + ) + Context.current().heartbeat({ done: doneGlobal, total: totalFiles }) + await updateLoadingProgress(qx, jobId, doneGlobal, totalFiles) + } + + const cumulativeStagingRows = (input.priorStagingRows ?? 0) + totalLoaded + await markJobStatus(qx, jobId, 'loading', { + rowCountStaging: cumulativeStagingRows, + tableRowCounts: { [`staging:${stagingTable}`]: totalLoaded }, + }) - await qx.result(`ANALYZE ${stagingTable}`) + await qx.result(`ANALYZE ${stagingTable}`) - log.info({ jobId, stagingTable, totalLoaded }, 'Staging load complete') + log.info({ jobId, stagingTable, totalLoaded }, 'Staging load complete') - return { rowsLoaded: totalLoaded } + return { rowsLoaded: totalLoaded } + } catch (err) { + await markJobStatus(qx, jobId, 'failed', { + errorMessage: err instanceof Error ? err.message : String(err), + finishedAt: new Date(), + }) + throw err + } } 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 11e1ed49c8..3de67ac1d2 100644 --- a/services/apps/packages_worker/src/deps-dev/activities/index.ts +++ b/services/apps/packages_worker/src/deps-dev/activities/index.ts @@ -10,5 +10,6 @@ export * from './gcsParquetToStaging' export * from './mergeStagingToTable' export * from './getLastSnapshot' export * from './checkDependentCountsGuard' +export * from './checkEdgeSnapshotQuality' export * from './probePartitionExists' export * from './resolveSnapshotDate' diff --git a/services/apps/packages_worker/src/deps-dev/activities/mergeStagingToTable.ts b/services/apps/packages_worker/src/deps-dev/activities/mergeStagingToTable.ts index fdbb481d5c..4a144a8c85 100644 --- a/services/apps/packages_worker/src/deps-dev/activities/mergeStagingToTable.ts +++ b/services/apps/packages_worker/src/deps-dev/activities/mergeStagingToTable.ts @@ -50,51 +50,60 @@ export async function mergeStagingToTable(input: MergeStagingInput): Promise = {} - let rowsAffected = 0 - const tableRowCounts: Record = {} + await qx.tx(async (tx) => { + for (const sql of prepareStatements) { + await tx.result(sql) + } + for (let i = 0; i < statements.length; i++) { + const count = await tx.result(statements[i]) + rowsAffected += count + const table = names[i] ?? `table_${i}` + tableRowCounts[table] = (tableRowCounts[table] ?? 0) + count + } + }) - await qx.tx(async (tx) => { - for (const sql of prepareStatements) { - await tx.result(sql) - } - for (let i = 0; i < statements.length; i++) { - const count = await tx.result(statements[i]) - rowsAffected += count - const table = names[i] ?? `table_${i}` - tableRowCounts[table] = (tableRowCounts[table] ?? 0) + count + if (chunkInfo) { + log.info( + { jobId, rowsAffected, tableRowCounts, chunk: `${chunkInfo.index + 1}/${chunkInfo.total}` }, + 'Chunk merge complete', + ) } - }) - if (chunkInfo) { - log.info( - { jobId, rowsAffected, tableRowCounts, chunk: `${chunkInfo.index + 1}/${chunkInfo.total}` }, - 'Chunk merge complete', - ) - } - - if (!isFinal) { - await markJobStatus(qx, jobId, 'merging', { - rowCountPg: priorRowsAffected + rowsAffected, - }) - } + if (!isFinal) { + await markJobStatus(qx, jobId, 'merging', { + rowCountPg: priorRowsAffected + rowsAffected, + }) + } - if (isFinal) { - const totalRowsAffected = priorRowsAffected + rowsAffected - const totalTableRowCounts: Record = { ...priorTableRowCounts } - for (const [table, count] of Object.entries(tableRowCounts)) { - totalTableRowCounts[table] = (totalTableRowCounts[table] ?? 0) + count + if (isFinal) { + const totalRowsAffected = priorRowsAffected + rowsAffected + const totalTableRowCounts: Record = { ...priorTableRowCounts } + for (const [table, count] of Object.entries(tableRowCounts)) { + totalTableRowCounts[table] = (totalTableRowCounts[table] ?? 0) + count + } + await markJobStatus(qx, jobId, 'done', { + finishedAt: new Date(), + rowCountPg: totalRowsAffected, + tableRowCounts: totalTableRowCounts, + }) + log.info( + { jobId, rowsAffected: totalRowsAffected, tableRowCounts: totalTableRowCounts }, + 'Merge complete', + ) } - await markJobStatus(qx, jobId, 'done', { + + return { rowsAffected, tableRowCounts } + } catch (err) { + await markJobStatus(qx, jobId, 'failed', { + errorMessage: err instanceof Error ? err.message : String(err), finishedAt: new Date(), - rowCountPg: totalRowsAffected, - tableRowCounts: totalTableRowCounts, }) - log.info( - { jobId, rowsAffected: totalRowsAffected, tableRowCounts: totalTableRowCounts }, - 'Merge complete', - ) + throw err } - - return { rowsAffected, tableRowCounts } } diff --git a/services/apps/packages_worker/src/deps-dev/config.ts b/services/apps/packages_worker/src/deps-dev/config.ts index c9565ebc94..b934e9c499 100644 --- a/services/apps/packages_worker/src/deps-dev/config.ts +++ b/services/apps/packages_worker/src/deps-dev/config.ts @@ -12,10 +12,6 @@ export const GCS_BUCKET = requireEnv('OSSPCKGS_GCS_BUCKET') export const DEPS_DEV_DATASET = 'bigquery-public-data.deps_dev_v1' export const SCORECARD_DATASET = 'openssf.scorecardcron' -// ADR-0003: Option A = DependencyGraphEdgesLatest (prod default, has version_constraint). -// Set OSSPCKGS_DEPS_TABLE=B locally to use DependenciesLatest (cheaper, no version_constraint). -export const DEPS_TABLE_OPTION: 'A' | 'B' = process.env.OSSPCKGS_DEPS_TABLE === 'B' ? 'B' : 'A' - const credentials = JSON.parse( Buffer.from(requireEnv('OSSPCKGS_GCP_CREDENTIALS_B64'), 'base64').toString('utf8'), ) diff --git a/services/apps/packages_worker/src/deps-dev/queries/dependentCountsSql.ts b/services/apps/packages_worker/src/deps-dev/queries/dependentCountsSql.ts index 9a949cc4a0..f34d4c9b39 100644 --- a/services/apps/packages_worker/src/deps-dev/queries/dependentCountsSql.ts +++ b/services/apps/packages_worker/src/deps-dev/queries/dependentCountsSql.ts @@ -1,13 +1,24 @@ +import { assertSnapshotDate } from './depsSql' + +// The deps.dev `Dependents` reverse index only covers the resolved-graph ecosystems +// {NPM, MAVEN, PYPI, CARGO} — GO and NUGET are entirely absent from the table (verified via BQ +// 2026-06-17). GO/NUGET dependent counts are produced separately by inverting their manifest tables +// (see buildGoDependentCountsSql / buildNugetDependentCountsSql). This builder is therefore scoped +// to the four edge systems; the job kind stays `dependent_counts` so existing job history, the +// monitor, and the row-count guard baseline remain continuous. +// // DependentsLatest scans all historical snapshots (~77 TB, ~$303). Use one partition instead. -// No ecosystem filter on Dependents — counts dependents from all ecosystems pointing at our packages. // MinimumDepth=1 = direct dependents (package declares an explicit dependency edge). // MinimumDepth>1 = transitive dependents (only reachable via intermediaries). +const EDGE_DEPENDENT_SYSTEMS = `('NPM', 'MAVEN', 'PYPI', 'CARGO')` + export function buildDependentCountsSql(snapshotDate: string): string { + assertSnapshotDate(snapshotDate) 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 ('NPM', 'GO', 'MAVEN', 'PYPI', 'NUGET', 'CARGO') + WHERE System IN ${EDGE_DEPENDENT_SYSTEMS} AND Purl IS NOT NULL AND Name NOT LIKE '%>%' GROUP BY System, Name @@ -29,9 +40,157 @@ LEFT JOIN ( ) pvp ON pvp.System = d.Dependent.System AND pvp.Name = d.Dependent.Name WHERE d.SnapshotAt >= TIMESTAMP('${snapshotDate}') AND d.SnapshotAt < TIMESTAMP(DATE_ADD(DATE '${snapshotDate}', INTERVAL 1 DAY)) - AND d.System IN ('NPM', 'GO', 'MAVEN', 'PYPI', 'NUGET', 'CARGO') + AND d.System IN ${EDGE_DEPENDENT_SYSTEMS} AND d.MinimumDepth >= 1 AND d.DependentIsHighestReleaseWithResolution = TRUE GROUP BY pm.purl ` } + +// GO/NUGET reverse-dependent counts. deps.dev publishes NO reverse-dependent or transitive graph for +// these two ecosystems (absent from Dependents/Dependencies/DependencyGraphEdges — see ADR-0004), so +// we compute the EXACT reverse transitive closure ourselves from the direct manifests it DOES publish +// (GoRequirementsLatest / NuGetRequirementsLatest). The "Latest" tables hold the highest release per +// package, matching the DependentIsHighestReleaseWithResolution semantics of the edge query. +// +// Unlike the edge query (a single SELECT), this is a multi-statement BQ SCRIPT: a semi-naive fixpoint +// over session-scoped TEMP tables, run via bqExportToGcs's isScript mode. It ends by creating +// `_export_data` with all three count columns, which the activity exports. Names are hashed to INT64 +// via FARM_FINGERPRINT for the loop (~5× cheaper scans; collisions ~1e-6), then mapped back to purls. +// +// Edge orientation matches the v1 direct-only builder: dep = requirer (g/n.Name), subj = depended-upon +// (d/dep.Name), so dependent_count(subj) = distinct requirers — purl keys line up with the merge. + +// GO needs 31 hops to converge, NUGET 19 (measured 2026-06-23). The cap is a deterministic runaway +// guard (e.g. a corrupt snapshot with a version cycle that never saturates), set well above both. +const MAX_CLOSURE_ITERATIONS = 60 + +// Builds the full closure script for one ecosystem. `edgesSql` must SELECT DISTINCT a `dep` (requirer) +// and `subj` (depended-upon) name column from that ecosystem's manifest table. +function buildClosureScript( + system: 'GO' | 'NUGET', + edgesSql: string, + snapshotDate: string, +): string { + assertSnapshotDate(snapshotDate) + return ` +DECLARE new_pairs INT64 DEFAULT 1; +DECLARE iter INT64 DEFAULT 0; + +-- 1. module-level direct edges (dep depends on subj); versions collapsed, self-loops + malformed names dropped. +CREATE TEMP TABLE edges_raw CLUSTER BY subj AS +${edgesSql}; + +-- 2. fingerprint names -> INT64 (≈5× cheaper scans; collisions ~1e-6, see ADR-0004). +CREATE TEMP TABLE fe CLUSTER BY subj, dep AS +SELECT DISTINCT FARM_FINGERPRINT(dep) AS dep, FARM_FINGERPRINT(subj) AS subj FROM edges_raw; + +-- 3. fingerprint -> name lookup (both endpoints) for mapping results back to purls/repos. +CREATE TEMP TABLE names AS +SELECT DISTINCT FARM_FINGERPRINT(name) AS fp, name FROM ( + SELECT dep AS name FROM edges_raw UNION DISTINCT SELECT subj AS name FROM edges_raw +); + +-- 4. seed reach (full closure, incl. direct) and frontier with the direct edges. +CREATE TEMP TABLE reach CLUSTER BY subj, dep AS SELECT dep, subj FROM fe; +CREATE TEMP TABLE frontier CLUSTER BY subj AS SELECT dep, subj FROM fe; + +-- 5. semi-naive fixpoint: extend the frontier one base edge deeper, keep only pairs not already in reach. +-- Terminates when no new pairs survive (finite pair space). iter cap = runaway guard. +WHILE new_pairs > 0 AND iter < ${MAX_CLOSURE_ITERATIONS} DO + CREATE OR REPLACE TEMP TABLE next_frontier CLUSTER BY subj AS + SELECT DISTINCT f.dep AS dep, e.subj AS subj + FROM frontier f + JOIN fe e ON e.dep = f.subj + LEFT JOIN reach r ON r.dep = f.dep AND r.subj = e.subj + WHERE r.dep IS NULL; + SET new_pairs = (SELECT COUNT(*) FROM next_frontier); + INSERT INTO reach SELECT dep, subj FROM next_frontier; + CREATE OR REPLACE TEMP TABLE frontier CLUSTER BY subj AS SELECT dep, subj FROM next_frontier; + SET iter = iter + 1; +END WHILE; + +-- 5b. The loop exits on convergence (new_pairs = 0) OR the iteration cap. If it stopped on the cap +-- with pairs still pending, the closure is INCOMPLETE: transitive_dependent_count would be +-- silently undercounted and the row-count guard wouldn't catch it (row volume stays ~flat). +-- Fail the job loudly instead of exporting partial counts. Raising the cap is the fix only if +-- the graph legitimately deepened past it. +IF new_pairs > 0 THEN + RAISE USING MESSAGE = FORMAT( + '${system} reverse-dependent closure did not converge within ${MAX_CLOSURE_ITERATIONS} iterations (new_pairs=%d still pending) - aborting to avoid undercounting transitive dependents', + new_pairs); +END IF; + +-- 6. repo mapping for the all-depth dependent_repos_count: latest source repo per package within a +-- 60-day window anchored on the run date. Literal date bounds prune partitions (a subquery-derived +-- SnapshotAt = (SELECT MAX ...) would NOT prune and scans the whole table). +CREATE TEMP TABLE dep_repos AS +SELECT Name, repo FROM ( + SELECT Name, ProjectName AS repo, + ROW_NUMBER() OVER (PARTITION BY Name ORDER BY SnapshotAt DESC) AS rn + FROM \`bigquery-public-data.deps_dev_v1.PackageVersionToProject\` + WHERE System = '${system}' AND RelationType = 'SOURCE_REPO_TYPE' + AND SnapshotAt >= TIMESTAMP(DATE_SUB(DATE '${snapshotDate}', INTERVAL 60 DAY)) + AND SnapshotAt < TIMESTAMP(DATE_ADD(DATE '${snapshotDate}', INTERVAL 1 DAY)) +) +WHERE rn = 1; + +-- collapse fingerprint -> repo once (~millions) so the multi-billion-row reach join hits one small table. +CREATE TEMP TABLE fp_repo CLUSTER BY fp AS +SELECT n.fp AS fp, dr.repo AS repo +FROM names n JOIN dep_repos dr ON dr.Name = n.name; + +-- 7. per-subject counts -> purl. Self-pairs (cycles) excluded so a package is never its own dependent. +CREATE TEMP TABLE _export_data AS +WITH +purl_map AS ( + SELECT Name, ANY_VALUE(REGEXP_REPLACE(Purl, r'@[^@]+$', '')) AS purl + FROM \`bigquery-public-data.deps_dev_v1.PackageVersionsLatest\` + WHERE System = '${system}' AND Purl IS NOT NULL AND Name NOT LIKE '%>%' + GROUP BY Name +), +direct AS ( + SELECT subj, COUNT(DISTINCT dep) AS dependent_count + FROM fe WHERE dep != subj GROUP BY subj +), +total AS ( + SELECT subj, COUNT(DISTINCT dep) AS reach_count + FROM reach WHERE dep != subj GROUP BY subj +), +repos AS ( + SELECT r.subj, COUNT(DISTINCT fr.repo) AS dependent_repos_count + FROM reach r JOIN fp_repo fr ON fr.fp = r.dep + WHERE r.dep != r.subj + GROUP BY r.subj +) +SELECT + pm.purl AS purl, + d.dependent_count AS dependent_count, + t.reach_count - d.dependent_count AS transitive_dependent_count, + COALESCE(rp.dependent_repos_count, 0) AS dependent_repos_count +FROM total t +JOIN names nm ON nm.fp = t.subj +JOIN purl_map pm ON pm.Name = nm.name +JOIN direct d ON d.subj = t.subj +LEFT JOIN repos rp ON rp.subj = t.subj; +` +} + +export function buildGoDependentCountsSql(snapshotDate: string): string { + const edgesSql = ` +SELECT DISTINCT g.Name AS dep, d.Name AS subj +FROM \`bigquery-public-data.deps_dev_v1.GoRequirementsLatest\` g, +UNNEST(g.DirectDependencies) AS d +WHERE d.Name NOT LIKE '%>%' AND g.Name NOT LIKE '%>%' AND g.Name != d.Name` + return buildClosureScript('GO', edgesSql, snapshotDate) +} + +export function buildNugetDependentCountsSql(snapshotDate: string): string { + const edgesSql = ` +SELECT DISTINCT n.Name AS dep, dp.Name AS subj +FROM \`bigquery-public-data.deps_dev_v1.NuGetRequirementsLatest\` n, +UNNEST(n.DependencyGroups) AS grp, +UNNEST(grp.Dependencies) AS dp +WHERE dp.Name NOT LIKE '%>%' AND n.Name NOT LIKE '%>%' AND n.Name != dp.Name` + return buildClosureScript('NUGET', edgesSql, snapshotDate) +} 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 b713ed343c..1acc249b2f 100644 --- a/services/apps/packages_worker/src/deps-dev/queries/depsSql.ts +++ b/services/apps/packages_worker/src/deps-dev/queries/depsSql.ts @@ -90,9 +90,123 @@ WHERE d.System IN (${filter}) return parts.join('\nUNION ALL\n') } -// --- Incremental SQL helpers --- -// Uses base tables (GoRequirements, NuGetRequirements, DependencyGraphEdges) with SnapshotAt filter. -// All CTEs combined in one WITH clause so UNION ALL can reference them freely. +// --- Incremental SQL helpers (snapshot edge-diff) --- +// +// Diff today's direct edges against the watermark (last successfully ingested) snapshot, matched on +// the edge identity (ecosystem, root_name, root_version, to_name) — deliberately EXCLUDING the +// resolved dependency version (to_version) and the requirement string from the match key. +// +// Why exclude to_version: a version's declared direct dependencies are immutable (manifest), but +// deps.dev RE-RESOLVES each dependency's concrete version every snapshot (a range like "^4.17.0" +// resolves to whatever patch is newest that week). Including to_version in the match makes every +// stable edge look "new" whenever any dependency ships a patch — the re-resolution churn that +// produced the ~555M-row exports. Matching on (root, to_name) only: +// - drops that churn, AND +// - still catches genuinely-new edges: brand-new versions' edges AND edges deps.dev resolved +// LATE for already-published versions. (A version lands in PackageVersions on publish, but its +// resolved graph can appear a snapshot or more later — measured ~14% of edge-bearing versions. +// A version-level diff misses those; an edge-level diff catches them, because the edge is +// simply new vs the watermark.) +// +// Consistent with how PG stores the data: the unique key is (version_id, depends_on_id, +// dependency_kind) — depends_on_version_id is NOT in it and the merge is ON CONFLICT DO NOTHING, so +// depends_on_version_id is never updated. to_version / version_constraint are still SELECTed (to +// populate the columns on first insert) but are not part of the diff key; the today CTEs collapse +// to one row per edge identity via GROUP BY + MAX (also defuses ×100 duplication if a corrupt +// snapshot ever slips past the edge-quality guard). +// +// All CTEs go into one WITH clause so each branch's UNION ALL SELECT can reference its CTE pair. + +const DEPS_DEV = 'bigquery-public-data.deps_dev_v1' + +// Snapshot dates always arrive as YYYY-MM-DD (resolveSnapshotDate slices the timestamp; +// bootstrap uses toISOString().slice(0,10)). Validate before embedding in SQL so a malformed or +// operator-supplied value fails loudly instead of producing broken — or injectable — SQL. +export function assertSnapshotDate(date: string): void { + if (!/^\d{4}-\d{2}-\d{2}$/.test(date)) { + throw new Error(`Invalid snapshot date '${date}' — expected YYYY-MM-DD`) + } +} + +// SnapshotAt is not exactly midnight, so filter by a [date, date+1day) range, never `= TIMESTAMP(date)`. +function snapshotRange(col: string, date: string): string { + assertSnapshotDate(date) + return `${col} >= TIMESTAMP('${date}') + AND ${col} < TIMESTAMP(DATE_ADD(DATE '${date}', INTERVAL 1 DAY))` +} + +// Anti-join: today edges not present in the watermark snapshot, matched on the edge identity. +function antiJoinSelect(todayCte: string, watermarkCte: string): string { + return `SELECT t.ecosystem, t.root_name, t.root_version, t.to_name, t.to_version, t.version_constraint +FROM ${todayCte} t +LEFT JOIN ${watermarkCte} l + ON l.ecosystem = t.ecosystem AND l.root_name = t.root_name + AND l.root_version = t.root_version AND l.to_name = t.to_name +WHERE l.to_name IS NULL` +} + +// GO + NUGET come from manifest tables (GoRequirements / NuGetRequirements) regardless of Option +// A/B, so their incremental branches are shared. Each returns its today + watermark CTEs + select. +function goIncrementalBranch(today: string, watermark: string): { ctes: string[]; select: string } { + return { + ctes: [ + `today_go AS ( + SELECT + 'go' AS ecosystem, + g.Name AS root_name, + g.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}.GoRequirements\` g, + UNNEST(g.DirectDependencies) AS d + WHERE ${snapshotRange('g.SnapshotAt', today)} + GROUP BY 1, 2, 3, 4 +)`, + `watermark_go AS ( + SELECT 'go' AS ecosystem, g.Name AS root_name, g.Version AS root_version, d.Name AS to_name + FROM \`${DEPS_DEV}.GoRequirements\` g, + UNNEST(g.DirectDependencies) AS d + WHERE ${snapshotRange('g.SnapshotAt', watermark)} + GROUP BY 1, 2, 3, 4 +)`, + ], + select: antiJoinSelect('today_go', 'watermark_go'), + } +} + +function nugetIncrementalBranch( + today: string, + watermark: string, +): { ctes: string[]; select: string } { + return { + ctes: [ + `today_nuget AS ( + SELECT + 'nuget' AS ecosystem, + n.Name AS root_name, + n.Version AS root_version, + dep.Name AS to_name, + CAST(NULL AS STRING) AS to_version, + MAX(dep.Requirement) AS version_constraint + FROM \`${DEPS_DEV}.NuGetRequirements\` n, + UNNEST(n.DependencyGroups) AS grp, + UNNEST(grp.Dependencies) AS dep + WHERE ${snapshotRange('n.SnapshotAt', today)} + GROUP BY 1, 2, 3, 4 +)`, + `watermark_nuget AS ( + SELECT 'nuget' AS ecosystem, n.Name AS root_name, n.Version AS root_version, dep.Name AS to_name + FROM \`${DEPS_DEV}.NuGetRequirements\` n, + UNNEST(n.DependencyGroups) AS grp, + UNNEST(grp.Dependencies) AS dep + WHERE ${snapshotRange('n.SnapshotAt', watermark)} + GROUP BY 1, 2, 3, 4 +)`, + ], + select: antiJoinSelect('today_nuget', 'watermark_nuget'), + } +} export function buildDepsIncrementalSqlA( today: string, @@ -111,106 +225,40 @@ export function buildDepsIncrementalSqlA( ctes.push( `today_edges AS ( SELECT - LOWER(e.System) AS ecosystem, - e.Name AS root_name, - e.Version AS root_version, - e.To.Name AS to_name, - e.To.Version AS to_version, - e.Requirement AS version_constraint - FROM \`bigquery-public-data.deps_dev_v1.DependencyGraphEdges\` e - WHERE e.SnapshotAt >= TIMESTAMP('${today}') - AND e.SnapshotAt < TIMESTAMP(DATE_ADD(DATE '${today}', INTERVAL 1 DAY)) + LOWER(e.System) AS ecosystem, + e.Name AS root_name, + e.Version AS root_version, + e.To.Name AS to_name, + MAX(e.To.Version) AS to_version, + MAX(e.Requirement) AS version_constraint + FROM \`${DEPS_DEV}.DependencyGraphEdges\` e + WHERE ${snapshotRange('e.SnapshotAt', today)} AND e.System IN (${filter}) AND e.From.Name = e.Name AND e.From.Version = e.Version + GROUP BY 1, 2, 3, 4 )`, `watermark_edges AS ( - SELECT e.System, e.Name, e.Version, e.To.Name AS to_name, e.To.Version AS to_version - FROM \`bigquery-public-data.deps_dev_v1.DependencyGraphEdges\` e - WHERE e.SnapshotAt >= TIMESTAMP('${watermark}') - AND e.SnapshotAt < TIMESTAMP(DATE_ADD(DATE '${watermark}', INTERVAL 1 DAY)) + SELECT LOWER(e.System) AS ecosystem, e.Name AS root_name, e.Version AS root_version, e.To.Name AS to_name + FROM \`${DEPS_DEV}.DependencyGraphEdges\` e + WHERE ${snapshotRange('e.SnapshotAt', watermark)} AND e.System IN (${filter}) AND e.From.Name = e.Name AND e.From.Version = e.Version - GROUP BY e.System, e.Name, e.Version, e.To.Name, e.To.Version + GROUP BY 1, 2, 3, 4 )`, ) - selects.push( - `SELECT t.* -FROM today_edges t -LEFT JOIN watermark_edges l - ON LOWER(l.System) = t.ecosystem AND l.Name = t.root_name AND l.Version = t.root_version - AND l.to_name = t.to_name AND l.to_version = t.to_version -WHERE l.to_name IS NULL`, - ) + selects.push(antiJoinSelect('today_edges', 'watermark_edges')) } if (includeGo) { - ctes.push( - `today_go AS ( - SELECT - 'go' AS ecosystem, - g.Name AS root_name, - g.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.GoRequirements\` g, - UNNEST(g.DirectDependencies) AS d - WHERE g.SnapshotAt >= TIMESTAMP('${today}') - AND g.SnapshotAt < TIMESTAMP(DATE_ADD(DATE '${today}', INTERVAL 1 DAY)) -)`, - `watermark_go AS ( - SELECT g.Name, g.Version, d.Name AS to_name, d.Requirement - FROM \`bigquery-public-data.deps_dev_v1.GoRequirements\` g, - UNNEST(g.DirectDependencies) AS d - WHERE g.SnapshotAt >= TIMESTAMP('${watermark}') - AND g.SnapshotAt < TIMESTAMP(DATE_ADD(DATE '${watermark}', INTERVAL 1 DAY)) - GROUP BY g.Name, g.Version, d.Name, d.Requirement -)`, - ) - selects.push( - `SELECT t.* -FROM today_go t -LEFT JOIN watermark_go l - ON l.Name = t.root_name AND l.Version = t.root_version AND l.to_name = t.to_name - AND l.Requirement = t.version_constraint -WHERE l.to_name IS NULL`, - ) + const { ctes: goCtes, select } = goIncrementalBranch(today, watermark) + ctes.push(...goCtes) + selects.push(select) } if (includeNuget) { - ctes.push( - `today_nuget AS ( - SELECT - 'nuget' AS ecosystem, - n.Name AS root_name, - n.Version AS root_version, - dep.Name AS to_name, - CAST(NULL AS STRING) AS to_version, - dep.Requirement AS version_constraint - FROM \`bigquery-public-data.deps_dev_v1.NuGetRequirements\` n, - UNNEST(n.DependencyGroups) AS grp, - UNNEST(grp.Dependencies) AS dep - WHERE n.SnapshotAt >= TIMESTAMP('${today}') - AND n.SnapshotAt < TIMESTAMP(DATE_ADD(DATE '${today}', INTERVAL 1 DAY)) -)`, - `watermark_nuget AS ( - SELECT n.Name, n.Version, dep.Name AS to_name, dep.Requirement - FROM \`bigquery-public-data.deps_dev_v1.NuGetRequirements\` n, - UNNEST(n.DependencyGroups) AS grp, - UNNEST(grp.Dependencies) AS dep - WHERE n.SnapshotAt >= TIMESTAMP('${watermark}') - AND n.SnapshotAt < TIMESTAMP(DATE_ADD(DATE '${watermark}', INTERVAL 1 DAY)) - GROUP BY n.Name, n.Version, dep.Name, dep.Requirement -)`, - ) - selects.push( - `SELECT t.* -FROM today_nuget t -LEFT JOIN watermark_nuget l - ON l.Name = t.root_name AND l.Version = t.root_version AND l.to_name = t.to_name - AND l.Requirement = t.version_constraint -WHERE l.to_name IS NULL`, - ) + const { ctes: nugetCtes, select } = nugetIncrementalBranch(today, watermark) + ctes.push(...nugetCtes) + selects.push(select) } return `WITH\n${ctes.join(',\n')}\n${selects.join('\nUNION ALL\n')}` @@ -237,102 +285,36 @@ export function buildDepsIncrementalSqlB( d.Name AS root_name, d.Version AS root_version, d.Dependency.Name AS to_name, - d.Dependency.Version AS to_version, + MAX(d.Dependency.Version) AS to_version, CAST(NULL AS STRING) AS version_constraint - FROM \`bigquery-public-data.deps_dev_v1.Dependencies\` d - WHERE d.SnapshotAt >= TIMESTAMP('${today}') - AND d.SnapshotAt < TIMESTAMP(DATE_ADD(DATE '${today}', INTERVAL 1 DAY)) + FROM \`${DEPS_DEV}.Dependencies\` d + WHERE ${snapshotRange('d.SnapshotAt', today)} AND d.System IN (${filter}) AND d.MinimumDepth = 1 + GROUP BY 1, 2, 3, 4 )`, `watermark_deps AS ( - SELECT d.System, d.Name, d.Version, d.Dependency.Name AS to_name, d.Dependency.Version AS to_version - FROM \`bigquery-public-data.deps_dev_v1.Dependencies\` d - WHERE d.SnapshotAt >= TIMESTAMP('${watermark}') - AND d.SnapshotAt < TIMESTAMP(DATE_ADD(DATE '${watermark}', INTERVAL 1 DAY)) + SELECT LOWER(d.System) AS ecosystem, d.Name AS root_name, d.Version AS root_version, d.Dependency.Name AS to_name + FROM \`${DEPS_DEV}.Dependencies\` d + WHERE ${snapshotRange('d.SnapshotAt', watermark)} AND d.System IN (${filter}) AND d.MinimumDepth = 1 - GROUP BY d.System, d.Name, d.Version, d.Dependency.Name, d.Dependency.Version + GROUP BY 1, 2, 3, 4 )`, ) - selects.push( - `SELECT t.* -FROM today_deps t -LEFT JOIN watermark_deps l - ON LOWER(l.System) = t.ecosystem AND l.Name = t.root_name AND l.Version = t.root_version - AND l.to_name = t.to_name AND l.to_version = t.to_version -WHERE l.to_name IS NULL`, - ) + selects.push(antiJoinSelect('today_deps', 'watermark_deps')) } if (includeGo) { - ctes.push( - `today_go AS ( - SELECT - 'go' AS ecosystem, - g.Name AS root_name, - g.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.GoRequirements\` g, - UNNEST(g.DirectDependencies) AS d - WHERE g.SnapshotAt >= TIMESTAMP('${today}') - AND g.SnapshotAt < TIMESTAMP(DATE_ADD(DATE '${today}', INTERVAL 1 DAY)) -)`, - `watermark_go AS ( - SELECT g.Name, g.Version, d.Name AS to_name, d.Requirement - FROM \`bigquery-public-data.deps_dev_v1.GoRequirements\` g, - UNNEST(g.DirectDependencies) AS d - WHERE g.SnapshotAt >= TIMESTAMP('${watermark}') - AND g.SnapshotAt < TIMESTAMP(DATE_ADD(DATE '${watermark}', INTERVAL 1 DAY)) - GROUP BY g.Name, g.Version, d.Name, d.Requirement -)`, - ) - selects.push( - `SELECT t.* -FROM today_go t -LEFT JOIN watermark_go l - ON l.Name = t.root_name AND l.Version = t.root_version AND l.to_name = t.to_name - AND l.Requirement = t.version_constraint -WHERE l.to_name IS NULL`, - ) + const { ctes: goCtes, select } = goIncrementalBranch(today, watermark) + ctes.push(...goCtes) + selects.push(select) } if (includeNuget) { - ctes.push( - `today_nuget AS ( - SELECT - 'nuget' AS ecosystem, - n.Name AS root_name, - n.Version AS root_version, - dep.Name AS to_name, - CAST(NULL AS STRING) AS to_version, - dep.Requirement AS version_constraint - FROM \`bigquery-public-data.deps_dev_v1.NuGetRequirements\` n, - UNNEST(n.DependencyGroups) AS grp, - UNNEST(grp.Dependencies) AS dep - WHERE n.SnapshotAt >= TIMESTAMP('${today}') - AND n.SnapshotAt < TIMESTAMP(DATE_ADD(DATE '${today}', INTERVAL 1 DAY)) -)`, - `watermark_nuget AS ( - SELECT n.Name, n.Version, dep.Name AS to_name, dep.Requirement - FROM \`bigquery-public-data.deps_dev_v1.NuGetRequirements\` n, - UNNEST(n.DependencyGroups) AS grp, - UNNEST(grp.Dependencies) AS dep - WHERE n.SnapshotAt >= TIMESTAMP('${watermark}') - AND n.SnapshotAt < TIMESTAMP(DATE_ADD(DATE '${watermark}', INTERVAL 1 DAY)) - GROUP BY n.Name, n.Version, dep.Name, dep.Requirement -)`, - ) - selects.push( - `SELECT t.* -FROM today_nuget t -LEFT JOIN watermark_nuget l - ON l.Name = t.root_name AND l.Version = t.root_version AND l.to_name = t.to_name - AND l.Requirement = t.version_constraint -WHERE l.to_name IS NULL`, - ) + const { ctes: nugetCtes, select } = nugetIncrementalBranch(today, watermark) + ctes.push(...nugetCtes) + 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/schedules/bootstrap.ts b/services/apps/packages_worker/src/deps-dev/schedules/bootstrap.ts index 7add6203a2..cd3d03a462 100644 --- a/services/apps/packages_worker/src/deps-dev/schedules/bootstrap.ts +++ b/services/apps/packages_worker/src/deps-dev/schedules/bootstrap.ts @@ -21,7 +21,12 @@ export async function scheduleOsspckgsBootstrap(): Promise { type: 'startWorkflow', workflowType: bootstrapOsspckgs, taskQueue: 'bq-dataset-ingest', - workflowExecutionTimeout: '12 hours', + // Per-attempt cap, not whole-tree. workflowExecutionTimeout spans the entire retry chain + + // continue-as-new, so a 12h execution timeout left retries with zero budget after attempt 1 + // ate it all — and 12h was too short for the package_dependencies merge loop anyway + // (~500 chunks × ~1min + BQ exports + preceding kinds). workflowRunTimeout bounds each + // attempt independently, so the 3 retries each get a fresh 24h. + workflowRunTimeout: '24 hours', retry: { initialInterval: '1 minute', backoffCoefficient: 2, 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 0258fa40e7..c12edceed8 100644 --- a/services/apps/packages_worker/src/deps-dev/workflows/bootstrapOsspckgs.ts +++ b/services/apps/packages_worker/src/deps-dev/workflows/bootstrapOsspckgs.ts @@ -33,9 +33,12 @@ type JobKind = | 'advisories' | 'advisory_packages' | 'dependent_counts' + | 'dependent_counts_go' + | 'dependent_counts_nuget' // deps.dev retains weekly snapshots for ~3 years; 1095 days (3 years) gives comfortable headroom. // advisories/advisory_packages use AdvisoriesLatest (no partition history) → effectively unlimited. +// dependent_counts_go/_nuget read *RequirementsLatest (no partition history) → effectively unlimited. const RETENTION_DAYS_BY_KIND: Record = { packages: 1095, repos: 1095, @@ -45,6 +48,8 @@ const RETENTION_DAYS_BY_KIND: Record = { advisories: 999_999, advisory_packages: 999_999, dependent_counts: 1095, + dependent_counts_go: 999_999, + dependent_counts_nuget: 999_999, } // Kinds whose incremental diff is driven by a BQ partition snapshot date. @@ -182,6 +187,39 @@ export async function bootstrapOsspckgs(opts: { } } } + // GO/NUGET reverse-dependent counts: separate kinds, manifest-sourced (GoRequirementsLatest / + // NuGetRequirementsLatest), computed via the exact reverse transitive closure script. The manifests + // are *Latest views (no resolution needed); `today` is the snapshot_at stamp AND the anchor for the + // dependent_repos partition window (latest PackageVersionToProject snapshot within 60 days). Each + // guards against its own history and merges a disjoint purl space, so an edge-snapshot corruption + // that aborts `dependent_counts` never blocks these. + for (const variant of ['go', 'nuget'] as const) { + const kind = `dependent_counts_${variant}` as const + if (!runs(kind)) continue + try { + await executeChild(ingestDependentCounts, { + args: [ + { + runId, + // Honor --snapshot-date like the partition kinds do (snap()). The closure reads *Latest + // manifests regardless, but anchors the dependent_repos 60-day window on this date, so a + // recovery run with an override must use it for a consistent window. + snapshotDate: opts.snapshotDate ?? today, + variant, + reuseExports: opts.reuseExports, + exportName: opts.exportName, + }, + ], + workflowId: `${runId}-${kind}`, + }) + } catch (err) { + // Soft-fail only on the row-count guard, mirroring the edge dependent_counts handling above. + const cause = err instanceof ChildWorkflowFailure ? err.cause : err + if (!(cause instanceof ApplicationFailure) || cause.type !== 'DEPENDENT_COUNTS_GUARD') { + throw err + } + } + } if (runs('repos') || runs('package_repos')) { await executeChild(ingestRepos, { args: [ @@ -211,21 +249,33 @@ export async function bootstrapOsspckgs(opts: { }) } if (runs('package_dependencies')) { - await executeChild(ingestDependencies, { - args: [ - { - runId, - syncMode: opts.mode, - today: snap('package_dependencies'), - watermark: wm('package_dependencies'), - ecosystems: opts.ecosystems, - reuseExports: opts.reuseExports, - depsTableOption: opts.depsTableOption, - exportName: opts.exportName, - fillConstraints: opts.fillConstraints, - }, - ], - }) + try { + await executeChild(ingestDependencies, { + args: [ + { + runId, + syncMode: opts.mode, + today: snap('package_dependencies'), + watermark: wm('package_dependencies'), + ecosystems: opts.ecosystems, + reuseExports: opts.reuseExports, + depsTableOption: opts.depsTableOption, + exportName: opts.exportName, + fillConstraints: opts.fillConstraints, + }, + ], + }) + } catch (err) { + // Only soft-fail on the edge-snapshot quality guard (corrupt deps.dev resolved-graph + // snapshot). Skipping leaves existing package_dependencies untouched and lets the rest + // of the bootstrap proceed; the next healthy snapshot ingests naturally. Mirror the + // dependent_counts guard: unwrap the ChildWorkflowFailure to inspect the cause; all + // other errors propagate. + const cause = err instanceof ChildWorkflowFailure ? err.cause : err + if (!(cause instanceof ApplicationFailure) || cause.type !== 'EDGE_SNAPSHOT_GUARD') { + throw err + } + } } if (runs('advisories') || runs('advisory_packages')) { await executeChild(ingestAdvisories, { diff --git a/services/apps/packages_worker/src/deps-dev/workflows/cleanupOsspckgs.ts b/services/apps/packages_worker/src/deps-dev/workflows/cleanupOsspckgs.ts deleted file mode 100644 index 3ef98d74b1..0000000000 --- a/services/apps/packages_worker/src/deps-dev/workflows/cleanupOsspckgs.ts +++ /dev/null @@ -1,6 +0,0 @@ -// GCS objects are covered by a 7-day lifecycle rule on the bucket — no explicit cleanup needed. -// This workflow is reserved for staging-table GC (orphaned tables older than 24h) once -// versioned staging tables are implemented (B4). -export async function cleanupOsspckgs(): Promise { - // no-op until versioned staging table GC is wired -} diff --git a/services/apps/packages_worker/src/deps-dev/workflows/index.ts b/services/apps/packages_worker/src/deps-dev/workflows/index.ts index 8ef97c9e3f..642653ca2c 100644 --- a/services/apps/packages_worker/src/deps-dev/workflows/index.ts +++ b/services/apps/packages_worker/src/deps-dev/workflows/index.ts @@ -1,5 +1,4 @@ export * from './bootstrapOsspckgs' -export * from './cleanupOsspckgs' export * from './ingestAdvisories' export * from './ingestDependentCounts' export * from './ingestDependencies' 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 0309c079bb..111f77534f 100644 --- a/services/apps/packages_worker/src/deps-dev/workflows/ingestDependencies.ts +++ b/services/apps/packages_worker/src/deps-dev/workflows/ingestDependencies.ts @@ -1,4 +1,4 @@ -import { proxyActivities } from '@temporalio/workflow' +import { ApplicationFailure, proxyActivities } from '@temporalio/workflow' import type * as depsDevActivities from '../activities' import { @@ -54,6 +54,11 @@ const { setJobStep } = proxyActivities({ retry: { maximumAttempts: 3 }, }) +const { checkEdgeSnapshotQuality } = proxyActivities({ + startToCloseTimeout: '5 minutes', + retry: { maximumAttempts: 3 }, +}) + const STAGING_TABLE = 'staging.osspckgs_deps_raw' const STAGING_DDL = ` @@ -219,19 +224,52 @@ 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. + // + // 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 = - 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', sql, runId: opts.runId, - syncMode: opts.syncMode, + // 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, - maxBytesGb: opts.syncMode === 'full' || isFill ? 25000 : 10000, + // 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, diff --git a/services/apps/packages_worker/src/deps-dev/workflows/ingestDependentCounts.ts b/services/apps/packages_worker/src/deps-dev/workflows/ingestDependentCounts.ts index e9df9b9ad8..aa50eb1b81 100644 --- a/services/apps/packages_worker/src/deps-dev/workflows/ingestDependentCounts.ts +++ b/services/apps/packages_worker/src/deps-dev/workflows/ingestDependentCounts.ts @@ -1,7 +1,13 @@ import { ApplicationFailure, proxyActivities } from '@temporalio/workflow' +import type { OsspckgsJobKind } from '@crowd/data-access-layer' + import type * as depsDevActivities from '../activities' -import { buildDependentCountsSql } from '../queries/dependentCountsSql' +import { + buildDependentCountsSql, + buildGoDependentCountsSql, + buildNugetDependentCountsSql, +} from '../queries/dependentCountsSql' const { bqExportToGcs } = proxyActivities({ startToCloseTimeout: '1 hour', @@ -34,14 +40,53 @@ const { setJobStep } = proxyActivities({ retry: { maximumAttempts: 3 }, }) -const STAGING_TABLE = 'staging.osspckgs_dependent_counts_raw' +// The dependent-counts ingest runs three independent ways, one per source of reverse-dependent data: +// - 'edges' : NPM/MAVEN/PYPI/CARGO from the deps.dev `Dependents` reverse index (single SELECT) +// - 'go' : GO from the GoRequirementsLatest exact reverse transitive closure (BQ script) +// - 'nuget' : NUGET from the NuGetRequirementsLatest exact reverse transitive closure (BQ script) +// All three produce the same three columns (dependent_count, transitive_dependent_count, +// dependent_repos_count) — deps.dev gives the edge systems their transitive graph directly, while +// GO/NUGET compute it via the semi-naive closure script (isScript, see ADR-0004 + dependentCountsSql). +// Each way is its own job kind, staging table, guard baseline, and purl-disjoint merge — so a corrupt +// edge snapshot aborts only the edge way and the GO/NUGET counts (manifest-sourced, unaffected) still +// update. 'edges' keeps the original `dependent_counts` kind so existing job history / monitor / guard +// baseline stay continuous. +export type DependentCountsVariant = 'edges' | 'go' | 'nuget' + +interface VariantConfig { + jobKind: OsspckgsJobKind + stagingTable: string + stagingDdl: string[] + mergeSql: string + pgColumns: string[] + maxBytesGb: number + buildSql: (snapshotDate: string) => string + // 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. Unset (single-SELECT) for the edges variant; set for GO/NUGET. + isScript?: boolean +} + +// Two-statement DDL: DROP before CREATE so an existing table with stale columns doesn't block the +// new schema. Staging table is TRUNCATED/recreated on every run anyway. +const EDGES_STAGING = 'staging.osspckgs_dependent_counts_raw' +const GO_STAGING = 'staging.osspckgs_dependent_counts_go_raw' +const NUGET_STAGING = 'staging.osspckgs_dependent_counts_nuget_raw' -// Two-statement DDL: DROP before CREATE so an existing table with the old dependent_packages_count -// column doesn't block the new column names. Staging table is TRUNCATED on every chunk anyway, -// so DROP+CREATE is semantically equivalent to IF NOT EXISTS for normal runs. -const STAGING_DDL = [ - `DROP TABLE IF EXISTS staging.osspckgs_dependent_counts_raw`, - `CREATE UNLOGGED TABLE staging.osspckgs_dependent_counts_raw ( +// All three ways now produce the same shape: purl + the three count columns. Each variant has its own +// staging table (parallel-safe) but identical DDL/merge; the merges touch disjoint purl spaces (one +// per ecosystem), so they never collide. +const PG_COLUMNS = [ + 'purl', + 'dependent_count', + 'transitive_dependent_count', + 'dependent_repos_count', +] + +const stagingDdl = (table: string): string[] => [ + `DROP TABLE IF EXISTS ${table}`, + `CREATE UNLOGGED TABLE ${table} ( purl text, dependent_count bigint, transitive_dependent_count bigint, @@ -49,24 +94,54 @@ const STAGING_DDL = [ )`, ] -// Strip @version from both sides — BQ ANY_VALUE(Purl) is non-deterministic across separate +// Strip @version from the staging purl — BQ ANY_VALUE(Purl) is non-deterministic across separate // query executions and may include or omit the version suffix. -const MERGE_SQL = ` +const dependentCountsMerge = (staging: string): string => ` UPDATE packages SET dependent_count = s.dependent_count, transitive_dependent_count = s.transitive_dependent_count, dependent_repos_count = s.dependent_repos_count, last_synced_at = NOW() -FROM staging.osspckgs_dependent_counts_raw s +FROM ${staging} s WHERE packages.purl = REGEXP_REPLACE(s.purl, '@[^@]+$', '') ` -const PG_COLUMNS = [ - 'purl', - 'dependent_count', - 'transitive_dependent_count', - 'dependent_repos_count', -] +const VARIANTS: Record = { + edges: { + jobKind: 'dependent_counts', + stagingTable: EDGES_STAGING, + stagingDdl: stagingDdl(EDGES_STAGING), + mergeSql: dependentCountsMerge(EDGES_STAGING), + pgColumns: PG_COLUMNS, + maxBytesGb: 2000, + buildSql: (snapshotDate) => buildDependentCountsSql(snapshotDate), + }, + // GO/NUGET run the exact reverse transitive closure script (isScript). maxBytesGb is the + // server-side maximumBytesBilled runaway cap, set well above the validated full-pipeline spend + // (GO 2.31 TB incl. the all-depth repos aggregation; NUGET ~32 GB) so a normal week never trips + // it — the iteration cap inside the script is the deterministic guard. Env-overridable via + // BQ_DATASET_INGEST_DEPENDENT_COUNTS_GO_MAX_BQ_GB / _NUGET_. + go: { + jobKind: 'dependent_counts_go', + stagingTable: GO_STAGING, + stagingDdl: stagingDdl(GO_STAGING), + mergeSql: dependentCountsMerge(GO_STAGING), + pgColumns: PG_COLUMNS, + maxBytesGb: 5000, + buildSql: (snapshotDate) => buildGoDependentCountsSql(snapshotDate), + isScript: true, + }, + nuget: { + jobKind: 'dependent_counts_nuget', + stagingTable: NUGET_STAGING, + stagingDdl: stagingDdl(NUGET_STAGING), + mergeSql: dependentCountsMerge(NUGET_STAGING), + pgColumns: PG_COLUMNS, + maxBytesGb: 200, + buildSql: (snapshotDate) => buildNugetDependentCountsSql(snapshotDate), + isScript: true, + }, +} const ROWS_PER_CHUNK = 1_000_000 @@ -75,16 +150,20 @@ export async function ingestDependentCounts(opts: { snapshotDate: string reuseExports?: boolean exportName?: string + variant?: DependentCountsVariant // defaults to 'edges' for backward compatibility }): Promise { + const cfg = VARIANTS[opts.variant ?? 'edges'] + const exportResult = await bqExportToGcs({ - jobKind: 'dependent_counts', - sql: buildDependentCountsSql(opts.snapshotDate), + jobKind: cfg.jobKind, + sql: cfg.buildSql(opts.snapshotDate), runId: opts.runId, syncMode: 'full', snapshotAt: opts.snapshotDate, - maxBytesGb: 2000, + maxBytesGb: cfg.maxBytesGb, reuseExports: opts.reuseExports, exportName: opts.exportName, + isScript: cfg.isScript, }) const { fileNames, rowCounts } = await listParquetFiles({ gcsPrefix: exportResult.gcsPrefix }) @@ -95,10 +174,11 @@ export async function ingestDependentCounts(opts: { const guard = await checkDependentCountsGuard({ currentRowCount: totalRows, snapshotDate: opts.snapshotDate, + jobKind: cfg.jobKind, }) if (!guard.ok) { throw ApplicationFailure.nonRetryable( - `dependent_counts guard failed: ${String(totalRows)} rows vs prev max ${String(guard.prevRowCount)} ` + + `${cfg.jobKind} guard failed: ${String(totalRows)} rows vs prev max ${String(guard.prevRowCount)} ` + `(${((guard.dropPct ?? 0) * 100).toFixed(1)}% drop). Slack alert sent. Aborting to preserve existing data.`, 'DEPENDENT_COUNTS_GUARD', ) @@ -130,9 +210,9 @@ export async function ingestDependentCounts(opts: { const { rowsLoaded } = await gcsParquetToStaging({ jobId: exportResult.jobId, - stagingTable: STAGING_TABLE, - stagingDdl: STAGING_DDL, - pgColumns: PG_COLUMNS, + stagingTable: cfg.stagingTable, + stagingDdl: cfg.stagingDdl, + pgColumns: cfg.pgColumns, fileNames: chunk, filesOffset: start, totalFiles, @@ -142,7 +222,7 @@ export async function ingestDependentCounts(opts: { const { rowsAffected, tableRowCounts } = await mergeStagingToTable({ jobId: exportResult.jobId, - mergeSql: MERGE_SQL, + mergeSql: cfg.mergeSql, tableNames: 'packages', isFinal, priorRowsAffected, diff --git a/services/apps/packages_worker/src/schedules/cleanup.ts b/services/apps/packages_worker/src/schedules/cleanup.ts deleted file mode 100644 index b4e165b118..0000000000 --- a/services/apps/packages_worker/src/schedules/cleanup.ts +++ /dev/null @@ -1,40 +0,0 @@ -import { ScheduleAlreadyRunning, ScheduleOverlapPolicy } from '@temporalio/client' - -import { cleanupOsspckgs } from '../deps-dev/workflows' -import { svc } from '../service' - -export async function scheduleOsspckgsCleanup(): Promise { - const { temporal } = svc - if (!temporal) throw new Error('Temporal client not initialized') - - try { - await temporal.schedule.create({ - scheduleId: 'osspckgs-cleanup-daily', - spec: { - cronExpressions: ['30 3 * * *'], - }, - policies: { - overlap: ScheduleOverlapPolicy.SKIP, - catchupWindow: '30 minutes', - }, - action: { - type: 'startWorkflow', - workflowType: cleanupOsspckgs, - taskQueue: 'bq-dataset-ingest', - workflowExecutionTimeout: '1 hour', - retry: { - initialInterval: '1 minute', - backoffCoefficient: 2, - maximumAttempts: 3, - }, - args: [], - }, - }) - } catch (err) { - if (err instanceof ScheduleAlreadyRunning) { - svc.log.info('Schedule osspckgs-cleanup-daily already registered.') - } else { - throw err - } - } -} diff --git a/services/apps/packages_worker/src/scripts/monitorOsspckgs.ts b/services/apps/packages_worker/src/scripts/monitorOsspckgs.ts index e2e325daaf..00643f778d 100644 --- a/services/apps/packages_worker/src/scripts/monitorOsspckgs.ts +++ b/services/apps/packages_worker/src/scripts/monitorOsspckgs.ts @@ -234,6 +234,8 @@ const KIND_TABLES: Record = { advisories: ['advisories'], advisory_packages: ['advisory_packages', 'advisory_affected_ranges'], dependent_counts: ['packages'], + dependent_counts_go: ['packages'], + dependent_counts_nuget: ['packages'], ranking: ['packages'], } diff --git a/services/apps/packages_worker/src/scripts/triggerBootstrap.ts b/services/apps/packages_worker/src/scripts/triggerBootstrap.ts index e688cb42a5..a9891cb948 100644 --- a/services/apps/packages_worker/src/scripts/triggerBootstrap.ts +++ b/services/apps/packages_worker/src/scripts/triggerBootstrap.ts @@ -11,6 +11,8 @@ const VALID_KINDS = [ 'advisories', 'advisory_packages', 'dependent_counts', + 'dependent_counts_go', + 'dependent_counts_nuget', 'scorecard', ] as const @@ -27,6 +29,8 @@ Options: ${VALID_KINDS.join(', ')} "repos" and "package_repos" always run together (one workflow). "advisories" and "advisory_packages" always run together (one workflow). + "dependent_counts" = edges (NPM/MAVEN/PYPI/CARGO) only; GO and NUGET + reverse counts run as "dependent_counts_go" / "dependent_counts_nuget". --snapshot-date DATE Override BQ snapshot resolution for all partition-filtered kinds. Use YYYY-MM-DD. Skips resolveSnapshotDate BQ call and uses this date directly. Useful for re-running with a known-good older snapshot. diff --git a/services/apps/packages_worker/src/types/pg-copy-streams.d.ts b/services/apps/packages_worker/src/types/pg-copy-streams.d.ts new file mode 100644 index 0000000000..0d79d2dc9d --- /dev/null +++ b/services/apps/packages_worker/src/types/pg-copy-streams.d.ts @@ -0,0 +1,39 @@ +// Local type declarations for pg-copy-streams v7. +// +// The DefinitelyTyped package (@types/pg-copy-streams) is abandoned at v1.2.5 and never tracked +// the v7 API: it advertises a stale major and lacks `both`. v7's runtime shape is identical for +// `from`/`to` (Writable/Readable streams carrying `text`/`rowCount`) and adds `both` (a Duplex). +// We own the declaration here instead of depending on a mismatched @types major. +// +// Kept self-contained (no `import ... from 'pg'`): the streams are pg "Submittable" objects — node-pg +// calls submit(connection) internally — but our code never references pg's Submittable/Connection +// types directly (loadDump.ts only reads `rowCount` and pipes the stream), so we avoid coupling this +// package to @types/pg. `submit` is typed loosely for the same reason. +declare module 'pg-copy-streams' { + import { Duplex, Readable, ReadableOptions, Writable, WritableOptions } from 'stream' + + export class CopyStreamQuery extends Writable { + text: string + rowCount: number + submit(connection: unknown): void + } + + export class CopyToStreamQuery extends Readable { + text: string + rowCount: number + submit(connection: unknown): void + } + + export class CopyBothQueryStream extends Duplex { + text: string + rowCount: number + submit(connection: unknown): void + } + + export function from(txt: string, options?: WritableOptions): CopyStreamQuery + export function to(txt: string, options?: ReadableOptions): CopyToStreamQuery + export function both( + txt: string, + options?: ReadableOptions & WritableOptions, + ): CopyBothQueryStream +} diff --git a/services/apps/packages_worker/src/workflows/index.ts b/services/apps/packages_worker/src/workflows/index.ts index 219aa742fc..09416fdaba 100644 --- a/services/apps/packages_worker/src/workflows/index.ts +++ b/services/apps/packages_worker/src/workflows/index.ts @@ -6,7 +6,6 @@ export { } from '../npm/workflows' export { bootstrapOsspckgs, - cleanupOsspckgs, ingestPackages, ingestVersions, ingestRepos, diff --git a/services/libs/data-access-layer/src/osspckgs/ingestJobs.ts b/services/libs/data-access-layer/src/osspckgs/ingestJobs.ts index 9606035e23..71c536d3d8 100644 --- a/services/libs/data-access-layer/src/osspckgs/ingestJobs.ts +++ b/services/libs/data-access-layer/src/osspckgs/ingestJobs.ts @@ -9,6 +9,8 @@ export type OsspckgsJobKind = | 'advisories' | 'advisory_packages' | 'dependent_counts' + | 'dependent_counts_go' + | 'dependent_counts_nuget' | 'scorecard_repos' | 'scorecard_checks' | 'ranking'