Merge remote-tracking branch 'origin/main' into fix/advisory-packages-bq-ceiling-CM-1362

Author: themarolt

CLAUDE.md

@@ -69,4 +69,16 @@ Millions of rows. Every query matters.
 - Define types properly — extend and reuse existing types. Don't sprinkle `any`.
 - Don't touch working code outside the scope of the current task.
 - Prefer doing less over introducing risk. Weigh trade-offs before acting.
-- Default to no comments. When one is warranted (non-obvious WHY), keep it to 2 lines max.
+- No comments. Code must be self-explanatory: name things well and extract
+  functions instead of explaining them.
+- A comment is allowed ONLY if one of these is true, and the reason cannot be
+  expressed in code:
+  - workaround for an external bug/API quirk (link it)
+  - non-obvious invariant or ordering the caller must respect
+  - performance/concurrency constraint that looks removable but isn't
+  - genuinely convoluted logic (e.g. inherited/legacy complexity) that can't
+    be simplified or extracted right now — prefer refactoring over commenting
+  - `TODO(CM-XXX):` with a ticket
+- Never comment: what the code does, section headers, JSDoc on obvious
+  functions, notes about the change you just made.
+- When allowed: 2 lines max.

backend/src/osspckgs/migrations/V1785740540__packagist_transitive_runs.sql

@@ -0,0 +1,14 @@
+-- Run ledger for the packagist transitive-dependents lane: one row per run;
+-- prepare retries reuse 'pending', and 'merging' rows end 'done' or 'failed'.
+CREATE TABLE packagist_transitive_runs (
+  id                        serial      PRIMARY KEY,
+  status                    text        NOT NULL DEFAULT 'pending'
+                            CHECK (status IN ('pending', 'merging', 'done', 'failed')),
+  edge_count                bigint,     -- distinct package-level direct edges snapshotted
+  packages_with_dependents  bigint,     -- rows in the closure output
+  processed_rows            bigint,     -- packagist packages visited by the merge drain
+  changed_rows              bigint,     -- rows whose transitive_dependent_count actually changed
+  error_message             text,
+  started_at                timestamptz NOT NULL DEFAULT now(),
+  finished_at               timestamptz
+);

docs/adr/0009-packagist-worker-design-decisions.md

@@ -27,6 +27,7 @@ like the one deps.dev exposes in BigQuery.
 | Dependency model | decided |
 | Metadata enrichment scope | decided |
 | Lane architecture & cadence | decided |
+| Transitive dependent counts | decided |
 
 ---
 
@@ -187,6 +188,78 @@ that week's enrichment (recoverable via the manual trigger).
 
 ---
 
+### Transitive dependent counts: weekly materialized reverse closure over our own edges
+
+A fifth lane, `computePackagistTransitiveDependents`, populates
+`packages.transitive_dependent_count` for packagist weekly: snapshot the direct edges to
+package-level pairs (`staging.packagist_transitive_edges`), compute the **exact** reverse
+transitive closure in one recursive statement (`staging.packagist_transitive_counts`), then
+keyset-merge into `packages` with zero-fill and `IS DISTINCT FROM` churn protection. It is
+chained off the metadata drain's natural completion (event, not clock — the same idiom as
+seed → metadata) and tracked in its own run ledger, `packagist_transitive_runs`
+(`pending → merging → done | failed`, one row per run with graph sizes and merge totals):
+per-purl `packagist_package_state` exists for lanes whose HTTP fetches fail per package —
+a whole-ecosystem batch has no per-purl outcomes — and `osspckgs_ingest_jobs` is the
+BQ-ingest ledger (gcs/bq columns, ingest-shaped kinds), so overloading it was rejected. The
+run aborts loudly on an empty edge snapshot, the merge side refuses to run against an empty
+counts table (a crash-truncated UNLOGGED staging table can never be zero-filled over good
+data), and a permanently failed merge marks the run `failed` instead of stranding it in
+`merging`. A weekly **ledger-gated backstop cron** (Monday, after the Sunday chain) covers
+broken chains: it no-ops when the ledger shows a `done` run within 6 days or while the metadata drain is
+still running (whose completion chains the closure), else chain-starts the fixed workflow
+id — clock as safety net, event chain as primary.
+
+**Provenance.** deps.dev has no Packagist coverage, and Packagist's own API reports only
+direct dependents — so this is the only ecosystem where the transitive signal must come from
+our own stored graph. `rank_packages()` already carries `transitive_dependents` as a coverage
+signal but silently drops it while the ecosystem total is zero; measurement showed the closure
+surfaces hidden infrastructure with tiny direct counts (symfony/polyfill-php80 ≈ 196K
+transitive dependents vs 471 registry-reported direct) and closes most of the critical-set gap
+vs ecosyste.ms. `is_critical` is a BOOL_OR across signals, so enabling this can only add
+critical packages — the same measurement-first precedent as the sonatype signal.
+
+**This does not reopen the "no resolved graph at ingest" decision** (see *Dependency model*
+above). No versions are resolved and no per-edge resolved targets are stored; the closure is a
+package-level derived aggregate — the same count columns deps.dev hands us pre-computed for
+npm/maven/pypi/cargo, and the same exact-closure computation the deps-dev worker already runs
+in BigQuery for GO/NUGET/RUBYGEMS. Packagist's edges live in Postgres instead of BigQuery, so
+the closure runs there.
+
+**Semantics** match the deps.dev convention (`MinimumDepth > 1`): dependents reachable only at
+depth ≥ 2, direct dependents excluded, computed from the same edge set (never mixed with the
+registry-reported `dependent_count`, which stays its own signal); dev-deps excluded (Composer
+does not install them transitively); cycles terminate and a package is never its own
+dependent; `0` means "computed, none" vs NULL "never computed". Count updates are **not
+audited** — bulk derived analytics, matching the deps.dev dependent-counts merges rather than
+this worker's registry-fact auditing.
+
+**Costs, measured.** The snapshot is the only step touching the full `package_dependencies`
+table (~1.5B rows, no index on `package_id`): a deliberate weekly parallel seq scan,
+~10–20 min projected in prod — the same accepted access pattern as the criticality PageRank
+edge loader. The closure itself runs on the ~919K-row package-level snapshot: ~29.3M reachable
+pairs in ~51 s (validated end-to-end locally on the full real dataset: 88 s wall clock,
+454,455 rows merged, +95 packages newly critical from the signal alone).
+
+**Consequences.**
+
+_Positive:_ the criticality gap for packagist closes without any ranking change; the lane is
+idempotent and resumable (continueAsNew keyset drain, pending-run reuse); Tinybird sees only
+real changes.
+
+_Negative / trade-offs:_ a weekly full scan of the shared 1.5B-row table; counts refresh at
+most weekly (with the edges), so they lag registry reality by up to a week.
+
+_Risks / escape hatch:_ if the weekly scan becomes a problem, the metadata lane already holds
+each package's edges in memory at write time and could co-maintain the package-level edge
+table incrementally, eliminating the scan — the snapshot step is isolated in one activity so
+the source can be swapped without touching the closure or merge. Undercount caveats: edges
+exist only for dependency targets that resolved to known package rows, and edge-less packages
+count as leaves — both bias counts down, so criticality conclusions stay conservative.
+
+**Decided**: 2026-07-27
+
+---
+
 ## Changelog
 
 - **2026-07-13** — ADR created. First entry: _Dependency model: direct edges + declared constraints,
@@ -198,3 +271,9 @@ that week's enrichment (recoverable via the manual trigger).
 - **2026-07-17** — Corrected stale "critical slice only" wording left over in the _Dependency
   model_ entry from before the _Metadata enrichment scope_ decision widened it to all packages;
   set `Status` to `accepted` (matching ADR-0005's precedent for a living/consolidated doc).
+- **2026-07-27** — Added _Transitive dependent counts: weekly materialized reverse closure over
+  our own edges_ (the fifth lane; resolves the `transitive_dependent_count` gap flagged in the
+  original risks).
+- **2026-08-04** — Revised _Transitive dependent counts_: run state moved to the dedicated
+  `packagist_transitive_runs` ledger, and a ledger-gated weekly backstop cron added for weeks
+  where the seed→metadata chain breaks.

services/apps/packages_worker/src/activities.ts

@@ -47,6 +47,12 @@ export {
   getCriticalPackagistCount,
   packagistCurrentTimestamp,
   packagistStopAfterFirstPage,
+  preparePackagistTransitiveCounts,
+  mergePackagistTransitiveBatch,
+  finishPackagistTransitiveRun,
+  failPackagistTransitiveRun,
+  packagistTransitiveRanRecently,
+  packagistMetadataDrainRunning,
 } from './packagist/activities'
 export { processRubyGemsCoreBatch, processRubyGemsCriticalBatch } from './rubygems/activities'
 export {

services/apps/packages_worker/src/criticality/activities.ts

@@ -1,6 +1,6 @@
 import { Context } from '@temporalio/activity'
 
-import { createIngestJob, markJobStatus } from '@crowd/data-access-layer'
+import { createIngestJob, findPendingJobByKind, markJobStatus } from '@crowd/data-access-layer'
 import { getServiceChildLogger } from '@crowd/logging'
 
 import { getPackagesDb } from '../db'
@@ -70,13 +70,9 @@ export async function rankPackages(): Promise<{ scoredRows: number; rankedRows:
 
   // On retry, a pending row from the prior attempt may already exist — reuse it.
   // Do NOT reuse a done row: it belongs to a previous bootstrap run and ranking must re-execute.
-  const existing = await qx.selectOneOrNone(
-    `SELECT id FROM osspckgs_ingest_jobs
-     WHERE job_kind = 'ranking' AND status = 'pending'
-     ORDER BY id DESC LIMIT 1`,
-  )
-
-  const jobId = existing?.id ?? (await createIngestJob(qx, 'ranking', 'ranking', null))
+  const jobId =
+    (await findPendingJobByKind(qx, 'ranking')) ??
+    (await createIngestJob(qx, 'ranking', 'ranking', null))
   try {
     await markJobStatus(qx, jobId, 'merging')
     const [result] = await qx.select(`SELECT * FROM rank_packages()`)

services/apps/packages_worker/src/packagist/README.md

@@ -3,10 +3,11 @@
 The Packagist worker keeps the PHP/Composer slice of the packages database fresh
 by crawling **packagist.org directly** — deps.dev has no Packagist coverage, so
 unlike npm/maven/pypi there is no BigQuery universe to import from; the registry
-crawl _is_ the universe source. It runs **four Temporal workflows** on the
-`packagist-worker` task queue: three cron schedules registered at worker boot
-(`src/bin/packagist-worker.ts`), plus the metadata drain, which the seed chains
-as a child workflow on completion.
+crawl _is_ the universe source. It runs **six Temporal workflows** on the
+`packagist-worker` task queue: four cron schedules registered at worker boot
+(`src/bin/packagist-worker.ts` — including the transitive backstop), plus two
+event-chained drains — the metadata drain (chained off the seed) and the
+transitive-dependents closure (chained off the metadata drain).
 
 Identity: ecosystem `packagist`, purls `pkg:composer/{vendor}/{name}`
 (namespace = vendor). Audit tag: `packagist` in `audit_field_changes`.
@@ -115,6 +116,11 @@ touching p2.
   `If-Modified-Since` next run).
 - `audit_field_changes` — every changed field above, worker tag `packagist`.
 
+On natural completion (not in `STOP_AFTER_FIRST_PAGE` debug runs) the drain
+**chain-starts the transitive-dependents closure** (§5) — freshly refreshed
+edges are what the closure consumes, so it follows the drain as an event, not
+a clock offset.
+
 ---
 
 ## 3. `ingestPackagistDownloads30d` — monthly rolling-window capture
@@ -166,11 +172,70 @@ last. State: `daily_downloads_last_run_at` + `daily_downloads_run_result`.
 
 ---
 
+## 5. `computePackagistTransitiveDependents` — weekly reverse-closure counts
+
+**Schedule:** primary trigger is the **chain off the metadata drain's natural
+completion** (effectively weekly, after the Sunday drain finishes), the same
+event-not-clock idiom as seed → metadata. A **ledger-gated backstop cron**
+(`packagist-transitive-backstop`, Monday 04:41 UTC) covers broken weeks: it
+no-ops when a run completed within 6 days or while the metadata drain is still
+crawling (whose completion chains the closure itself), otherwise chain-starts
+the same fixed workflow id — so it can never race a live drain and a healthy
+week never pays a second scan. Recover manually with
+`pnpm trigger-packagist:local transitive`.
+**Targets:** every packagist package (the merge zero-fills leaves).
+
+**What it does:** three steps —
+
+1. **Snapshot** — collapses the packagist slice of `package_dependencies` into
+   `staging.packagist_transitive_edges`: distinct package-level pairs, `require`
+   edges only (`require-dev` excluded — Composer does not install dev deps
+   transitively), self-edges dropped. This is the only step touching the full
+   ~1.5B-row table; `package_id` has no index there, so it is a deliberate
+   weekly parallel seq scan (same access pattern the criticality PageRank loader
+   already uses).
+2. **Closure** — the exact reverse transitive closure in one recursive
+   statement into `staging.packagist_transitive_counts` (~29M reachable pairs,
+   ~51 s measured on the full dataset). Cycles terminate; a package is never
+   counted as its own dependent.
+3. **Merge** — keyset batches of 10K into
+   `packages.transitive_dependent_count`, zero-filling packages with no
+   dependents (`0` = "computed, none" vs NULL = "never computed").
+   `IS DISTINCT FROM` keeps re-runs churn-free — `last_synced_at` (the
+   Sequin/Tinybird signal) moves only on real changes.
+
+**Populates:** `packages.transitive_dependent_count` for packagist rows only —
+dependents reachable **only at depth ≥ 2**; direct dependents are excluded,
+matching the deps.dev `MinimumDepth > 1` convention every other ecosystem uses.
+The registry-reported `dependent_count` is untouched. **No audit rows** — bulk
+derived analytics, same policy as the deps.dev dependent-counts merges. Run
+state: one `packagist_transitive_runs` row per run
+(`pending → merging → done | failed`, with graph sizes and merge totals) —
+per-purl `packagist_package_state` doesn't fit a whole-ecosystem batch, and
+`osspckgs_ingest_jobs` is the BQ-ingest ledger. An empty edge snapshot
+hard-aborts the run (`failed`) instead of writing zeros over good data; the
+merge itself refuses an empty counts table — the staging tables are UNLOGGED
+(truncated by crash recovery), and a mid-drain truncation must never be
+zero-filled over real counts — and a merge phase that fails permanently marks
+the run `failed` rather than leaving it in `merging`.
+
+**Why:** deps.dev has zero Packagist coverage, so no external source can supply
+this signal. `rank_packages()` already has `transitive_dependents` wired in but
+drops it while the ecosystem total is zero — the first pass after this lane
+runs picks it up automatically, and since `is_critical` is a BOOL_OR across
+signals it can only add critical packages, never remove any.
+
+**Known undercounts (conservative by design):** edges exist only where the
+dependency target resolved to a known packages row (unresolved targets are
+skipped at ingest), and packages with no stored edges count as leaves.
+
+---
+
 ## What this worker deliberately does NOT write
 
-- `transitive_dependent_count`, `dependent_repos_count` — not computable from
-  the registry; needs a reverse-closure over our stored direct edges
-  (future work, see ADR-0009 risks).
+- `dependent_repos_count` — not computable from the registry; needs a
+  package→repo mapping across the dependent set (deps.dev provides this for
+  its ecosystems; Packagist has no equivalent).
 - Advisories — the OSV worker owns security data platform-wide.
 - `is_critical` / `criticality_score` / ranking columns — the shared
   criticality worker; this worker only _reads_ `is_critical` for scoping.
@@ -199,14 +264,17 @@ DEV=1 ./scripts/cli service packagist-worker up
 # trigger on demand instead of waiting for the crons
 cd services/apps/packages_worker
 pnpm trigger-packagist:local seed             # discovery (chain-starts metadata!)
-pnpm trigger-packagist:local metadata         # enrichment: info + versions + deps
+pnpm trigger-packagist:local metadata         # enrichment: info + versions + deps (chain-starts transitive!)
 pnpm trigger-packagist:local downloads-30d    # monthly rolling-window capture
 pnpm trigger-packagist:local downloads-daily  # daily capture, critical slice
+pnpm trigger-packagist:local transitive       # reverse-closure transitive dependent counts
 ```
 
 Note: triggering `seed` also chain-starts the full `metadata` drain (set
 `CROWD_PACKAGES_PACKAGIST_STOP_AFTER_FIRST_PAGE=true` locally to bound it).
-Local smoke order: seed → metadata → (rank) → downloads lanes for
-critical-scoped writes. State lives in `packagist_package_state`
-(migration `V1784314023__packagist_worker.sql`); design decisions in
+Local smoke order: seed → metadata → transitive → (rank) → downloads lanes for
+critical-scoped writes. Per-purl state lives in `packagist_package_state`
+(migration `V1784314023__packagist_worker.sql`); the transitive lane tracks its
+runs in `packagist_transitive_runs` (migration
+`V1785740540__packagist_transitive_runs.sql`). Design decisions in
 `docs/adr/0009-packagist-worker-design-decisions.md`.