diff --git a/backend/src/osspckgs/migrations/V1785740540__packagist_transitive_runs.sql b/backend/src/osspckgs/migrations/V1785740540__packagist_transitive_runs.sql new file mode 100644 index 0000000000..c24d52293d --- /dev/null +++ b/backend/src/osspckgs/migrations/V1785740540__packagist_transitive_runs.sql @@ -0,0 +1,20 @@ +-- Packagist transitive-dependents lane: run-level state ledger. +-- +-- The lane is a whole-ecosystem batch (snapshot → closure → keyset merge), so its +-- natural unit of record is a run +-- +-- One row per run (weekly cadence + manual triggers). A 'pending' row is reused +-- across Temporal retries of the prepare activity; 'merging' rows either finish +-- ('done') or are marked 'failed' by the workflow's terminal error handling. +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 +); diff --git a/docs/adr/0009-packagist-worker-design-decisions.md b/docs/adr/0009-packagist-worker-design-decisions.md index 5d77f0b932..ed89fece59 100644 --- a/docs/adr/0009-packagist-worker-design-decisions.md +++ b/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. diff --git a/services/apps/packages_worker/src/activities.ts b/services/apps/packages_worker/src/activities.ts index d0bf0a4d24..6dc028cd0d 100644 --- a/services/apps/packages_worker/src/activities.ts +++ b/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 { diff --git a/services/apps/packages_worker/src/criticality/activities.ts b/services/apps/packages_worker/src/criticality/activities.ts index dc0c18d1ec..8eee237346 100644 --- a/services/apps/packages_worker/src/criticality/activities.ts +++ b/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()`) diff --git a/services/apps/packages_worker/src/packagist/README.md b/services/apps/packages_worker/src/packagist/README.md index 29dcec90d2..eed2b5a6a5 100644 --- a/services/apps/packages_worker/src/packagist/README.md +++ b/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`. diff --git a/services/apps/packages_worker/src/packagist/__tests__/transitiveDependents.test.ts b/services/apps/packages_worker/src/packagist/__tests__/transitiveDependents.test.ts new file mode 100644 index 0000000000..01cb692c20 --- /dev/null +++ b/services/apps/packages_worker/src/packagist/__tests__/transitiveDependents.test.ts @@ -0,0 +1,496 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest' + +import { EmptyPackagistTransitiveCountsError } from '@crowd/data-access-layer/src/packages/transitiveDependents' + +import { + failPackagistTransitiveRun, + finishPackagistTransitiveRun, + mergePackagistTransitiveBatch, + packagistMetadataDrainRunning, + preparePackagistTransitiveCounts, +} from '../activities' +import { + ROUNDS_PER_RUN, + TRANSITIVE_MERGE_BATCH, + backstopPackagistTransitiveDrain, + computePackagistTransitiveDependents, + ingestPackagistMetadata, +} from '../workflows' + +// The metadata drain chain-starts the closure workflow on completion (event, not clock). +// The closure runs prepare → keyset merge drain → finish across continueAsNew generations. + +const h = vi.hoisted(() => ({ + acts: { + packagistCurrentTimestamp: vi.fn(), + packagistStopAfterFirstPage: vi.fn(), + getPackagistMetadataBatch: vi.fn(), + ingestPackagistMetadataBatch: vi.fn(), + preparePackagistTransitiveCounts: vi.fn(), + mergePackagistTransitiveBatch: vi.fn(), + finishPackagistTransitiveRun: vi.fn(), + failPackagistTransitiveRun: vi.fn(), + packagistTransitiveRanRecently: vi.fn(), + packagistMetadataDrainRunning: vi.fn(), + }, + startChild: vi.fn(), + continueAsNew: vi.fn(), + logWarn: vi.fn(), + attempt: vi.fn(), + describeWorkflow: vi.fn(), + snapshot: vi.fn(), + closure: vi.fn(), + mergeDal: vi.fn(), + createRun: vi.fn(), + findPendingRun: vi.fn(), + markMerging: vi.fn(), + finishRun: vi.fn(), + failRun: vi.fn(), + fakeQx: { + select: vi.fn(), + selectOne: vi.fn(), + selectOneOrNone: vi.fn(), + result: vi.fn(), + tx: vi.fn(), + }, +})) + +vi.mock('@temporalio/workflow', () => ({ + proxyActivities: () => h.acts, + startChild: h.startChild, + continueAsNew: h.continueAsNew, + log: { info: vi.fn(), warn: h.logWarn, error: vi.fn(), debug: vi.fn() }, + ParentClosePolicy: { ABANDON: 'ABANDON' }, + WorkflowIdReusePolicy: { ALLOW_DUPLICATE: 'ALLOW_DUPLICATE' }, +})) + +vi.mock('@temporalio/activity', async (importOriginal) => ({ + ...(await importOriginal()), + Context: { + current: () => ({ info: { attempt: h.attempt() }, heartbeat: vi.fn() }), + }, +})) + +vi.mock('@crowd/temporal', async (importOriginal) => ({ + ...(await importOriginal()), + TEMPORAL_CONFIG: () => ({}), + getTemporalClient: async () => ({ + workflow: { getHandle: () => ({ describe: h.describeWorkflow }) }, + }), +})) + +vi.mock('../../db', async (importOriginal) => ({ + ...(await importOriginal()), + getPackagesDb: async () => h.fakeQx, +})) + +vi.mock('@crowd/data-access-layer/src/packages/transitiveDependents', async (importOriginal) => ({ + ...(await importOriginal()), + snapshotPackagistDirectEdges: h.snapshot, + computePackagistTransitiveCounts: h.closure, + mergePackagistTransitiveCounts: h.mergeDal, +})) + +vi.mock( + '@crowd/data-access-layer/src/packages/packagistTransitiveRuns', + async (importOriginal) => ({ + ...(await importOriginal()), + createPackagistTransitiveRun: h.createRun, + findUnfinishedPackagistTransitiveRun: h.findPendingRun, + markPackagistTransitiveRunMerging: h.markMerging, + finishPackagistTransitiveRun: h.finishRun, + failPackagistTransitiveRun: h.failRun, + }), +) + +function metadataCandidates(n: number) { + return Array.from({ length: n }, (_, i) => ({ + purl: `pkg:composer/t/p${i}`, + metadataLastModified: null, + })) +} + +beforeEach(() => { + vi.clearAllMocks() + h.acts.packagistCurrentTimestamp.mockResolvedValue('2026-07-26T00:00:00.000Z') + h.acts.packagistStopAfterFirstPage.mockResolvedValue(false) + h.acts.ingestPackagistMetadataBatch.mockResolvedValue(undefined) + h.acts.finishPackagistTransitiveRun.mockResolvedValue(undefined) + h.acts.failPackagistTransitiveRun.mockResolvedValue(undefined) + h.startChild.mockResolvedValue(undefined) + h.continueAsNew.mockResolvedValue(undefined) + h.attempt.mockReturnValue(1) + h.acts.packagistMetadataDrainRunning.mockResolvedValue(false) +}) + +describe('ingestPackagistMetadata — chaining the transitive drain', () => { + it('chain-starts the transitive drain when the drain completes on an empty batch', async () => { + h.acts.getPackagistMetadataBatch.mockResolvedValue({ candidates: [], nextCursor: '' }) + + await ingestPackagistMetadata({}) + + expect(h.acts.ingestPackagistMetadataBatch).not.toHaveBeenCalled() + expect(h.startChild).toHaveBeenCalledTimes(1) + const [wf, opts] = h.startChild.mock.calls[0] + expect(wf).toBe(computePackagistTransitiveDependents) + expect(opts).toMatchObject({ + workflowId: 'packagist-transitive-drain', + workflowIdReusePolicy: 'ALLOW_DUPLICATE', + parentClosePolicy: 'ABANDON', + args: [{}], + }) + }) + + it('chain-starts after ingesting the final short batch', async () => { + h.acts.getPackagistMetadataBatch.mockResolvedValue({ + candidates: metadataCandidates(3), + nextCursor: 'pkg:composer/t/p2', + }) + + await ingestPackagistMetadata({}) + + expect(h.acts.ingestPackagistMetadataBatch).toHaveBeenCalledTimes(1) + expect(h.startChild).toHaveBeenCalledTimes(1) + // the final batch is fully ingested before the drain is chained + expect(h.acts.ingestPackagistMetadataBatch.mock.invocationCallOrder[0]).toBeLessThan( + h.startChild.mock.invocationCallOrder[0], + ) + }) + + it('does not chain in stopAfterFirstPage debug mode', async () => { + h.acts.packagistStopAfterFirstPage.mockResolvedValue(true) + h.acts.getPackagistMetadataBatch.mockResolvedValue({ + candidates: metadataCandidates(50), + nextCursor: 'pkg:composer/t/p49', + }) + + await ingestPackagistMetadata({}) + + expect(h.acts.ingestPackagistMetadataBatch).toHaveBeenCalledTimes(1) + expect(h.startChild).not.toHaveBeenCalled() + }) + + it('does not chain on an empty batch in stopAfterFirstPage debug mode either', async () => { + h.acts.packagistStopAfterFirstPage.mockResolvedValue(true) + h.acts.getPackagistMetadataBatch.mockResolvedValue({ candidates: [], nextCursor: '' }) + + await ingestPackagistMetadata({}) + + expect(h.startChild).not.toHaveBeenCalled() + }) + + it('does not chain while the drain still has full batches — it continues-as-new instead', async () => { + h.acts.getPackagistMetadataBatch.mockResolvedValue({ + candidates: metadataCandidates(50), + nextCursor: 'pkg:composer/t/p49', + }) + + await ingestPackagistMetadata({}) + + expect(h.continueAsNew).toHaveBeenCalledTimes(1) + expect(h.startChild).not.toHaveBeenCalled() + }) + + it('swallows already-started when a prior transitive drain is still running', async () => { + h.acts.getPackagistMetadataBatch.mockResolvedValue({ candidates: [], nextCursor: '' }) + h.startChild.mockRejectedValue( + Object.assign(new Error('already started'), { + name: 'WorkflowExecutionAlreadyStartedError', + }), + ) + + await expect(ingestPackagistMetadata({})).resolves.toBeUndefined() + expect(h.logWarn).toHaveBeenCalled() + }) +}) + +describe('computePackagistTransitiveDependents — workflow', () => { + it('prepares once, drains merge batches, then finishes the job with totals', async () => { + h.acts.preparePackagistTransitiveCounts.mockResolvedValue({ runId: 7 }) + h.acts.mergePackagistTransitiveBatch + .mockResolvedValueOnce({ processed: TRANSITIVE_MERGE_BATCH, changed: 5, nextCursor: '100' }) + .mockResolvedValueOnce({ processed: 20, changed: 3, nextCursor: '120' }) + + await computePackagistTransitiveDependents({}) + + expect(h.acts.preparePackagistTransitiveCounts).toHaveBeenCalledTimes(1) + expect(h.acts.mergePackagistTransitiveBatch).toHaveBeenNthCalledWith( + 1, + '', + TRANSITIVE_MERGE_BATCH, + ) + expect(h.acts.mergePackagistTransitiveBatch).toHaveBeenNthCalledWith( + 2, + '100', + TRANSITIVE_MERGE_BATCH, + ) + expect(h.acts.finishPackagistTransitiveRun).toHaveBeenCalledWith(7, { + processed: TRANSITIVE_MERGE_BATCH + 20, + changed: 8, + }) + expect(h.continueAsNew).not.toHaveBeenCalled() + }) + + it('finishes immediately when the first batch is already empty', async () => { + h.acts.preparePackagistTransitiveCounts.mockResolvedValue({ runId: 7 }) + h.acts.mergePackagistTransitiveBatch.mockResolvedValue({ + processed: 0, + changed: 0, + nextCursor: '', + }) + + await computePackagistTransitiveDependents({}) + + expect(h.acts.finishPackagistTransitiveRun).toHaveBeenCalledWith(7, { + processed: 0, + changed: 0, + }) + }) + + it('skips prepare when resuming a later generation and accumulates totals', async () => { + h.acts.mergePackagistTransitiveBatch.mockResolvedValue({ + processed: 20, + changed: 2, + nextCursor: '220', + }) + + await computePackagistTransitiveDependents({ + runId: 7, + cursor: '200', + processed: 40, + changed: 6, + }) + + expect(h.acts.preparePackagistTransitiveCounts).not.toHaveBeenCalled() + expect(h.acts.mergePackagistTransitiveBatch).toHaveBeenCalledWith('200', TRANSITIVE_MERGE_BATCH) + expect(h.acts.finishPackagistTransitiveRun).toHaveBeenCalledWith(7, { + processed: 60, + changed: 8, + }) + }) + + it('continues-as-new with carried state when the round cap is hit', async () => { + h.acts.preparePackagistTransitiveCounts.mockResolvedValue({ runId: 7 }) + let call = 0 + h.acts.mergePackagistTransitiveBatch.mockImplementation(async () => { + call += 1 + return { processed: TRANSITIVE_MERGE_BATCH, changed: 1, nextCursor: String(call * 10) } + }) + + await computePackagistTransitiveDependents({}) + + expect(h.acts.finishPackagistTransitiveRun).not.toHaveBeenCalled() + // Pin the cap itself — deriving expectations from the observed call count would + // stay green for any (broken) number of rounds. + expect(h.acts.mergePackagistTransitiveBatch).toHaveBeenCalledTimes(ROUNDS_PER_RUN) + expect(h.continueAsNew).toHaveBeenCalledTimes(1) + expect(h.continueAsNew).toHaveBeenCalledWith({ + runId: 7, + cursor: String(ROUNDS_PER_RUN * 10), + processed: ROUNDS_PER_RUN * TRANSITIVE_MERGE_BATCH, + changed: ROUNDS_PER_RUN, + }) + }) + + it('marks the run failed with the ROOT cause and rethrows when a merge batch fails permanently', async () => { + h.acts.preparePackagistTransitiveCounts.mockResolvedValue({ runId: 7 }) + // Shaped like Temporal's ActivityFailure: a generic wrapper whose cause chain + // carries the real reason — error_message must record the root, not the wrapper. + h.acts.mergePackagistTransitiveBatch.mockRejectedValue( + Object.assign(new Error('Activity task failed'), { + cause: new Error('counts table is empty'), + }), + ) + + await expect(computePackagistTransitiveDependents({})).rejects.toThrow(/Activity task failed/) + + expect(h.acts.failPackagistTransitiveRun).toHaveBeenCalledWith(7, 'counts table is empty') + expect(h.acts.finishPackagistTransitiveRun).not.toHaveBeenCalled() + expect(h.continueAsNew).not.toHaveBeenCalled() + }) + + it('rethrows the ORIGINAL merge error even when fail-marking itself fails', async () => { + h.acts.preparePackagistTransitiveCounts.mockResolvedValue({ runId: 7 }) + h.acts.mergePackagistTransitiveBatch.mockRejectedValue(new Error('merge exploded')) + h.acts.failPackagistTransitiveRun.mockRejectedValueOnce(new Error('ledger write refused')) + + await expect(computePackagistTransitiveDependents({})).rejects.toThrow(/merge exploded/) + }) +}) + +describe('backstopPackagistTransitiveDrain — workflow', () => { + it('does nothing when a run completed recently', async () => { + h.acts.packagistTransitiveRanRecently.mockResolvedValue(true) + + await backstopPackagistTransitiveDrain() + + expect(h.startChild).not.toHaveBeenCalled() + }) + + it('stands down while the metadata drain is still crawling — its completion chains the closure', async () => { + h.acts.packagistTransitiveRanRecently.mockResolvedValue(false) + h.acts.packagistMetadataDrainRunning.mockResolvedValue(true) + + await backstopPackagistTransitiveDrain() + + expect(h.startChild).not.toHaveBeenCalled() + }) + + it('chain-starts the drain (fixed workflow id) when the week had no successful run', async () => { + h.acts.packagistTransitiveRanRecently.mockResolvedValue(false) + + await backstopPackagistTransitiveDrain() + + expect(h.startChild).toHaveBeenCalledTimes(1) + const [wf, opts] = h.startChild.mock.calls[0] + expect(wf).toBe(computePackagistTransitiveDependents) + expect(opts).toMatchObject({ workflowId: 'packagist-transitive-drain' }) + }) + + it('swallows already-started so it can never race a live drain', async () => { + h.acts.packagistTransitiveRanRecently.mockResolvedValue(false) + h.startChild.mockRejectedValueOnce( + Object.assign(new Error('already started'), { + name: 'WorkflowExecutionAlreadyStartedError', + }), + ) + + await expect(backstopPackagistTransitiveDrain()).resolves.toBeUndefined() + }) +}) + +describe('packagistMetadataDrainRunning — activity', () => { + it.each([ + ['RUNNING', true], + ['COMPLETED', false], + ['FAILED', false], + ])('classifies workflow status %s as %s', async (status, expected) => { + h.describeWorkflow.mockResolvedValue({ status: { name: status } }) + + await expect(packagistMetadataDrainRunning()).resolves.toBe(expected) + }) + + it('treats a never-started drain as not running', async () => { + h.describeWorkflow.mockRejectedValue( + Object.assign(new Error('not found'), { name: 'WorkflowNotFoundError' }), + ) + + await expect(packagistMetadataDrainRunning()).resolves.toBe(false) + }) + + it('propagates unexpected describe errors', async () => { + h.describeWorkflow.mockRejectedValueOnce(new Error('temporal unreachable')) + + await expect(packagistMetadataDrainRunning()).rejects.toThrow(/unreachable/) + }) +}) + +describe('preparePackagistTransitiveCounts — activity', () => { + it('creates a run, snapshots edges, computes the closure, and marks the run merging', async () => { + h.findPendingRun.mockResolvedValue(null) + h.createRun.mockResolvedValue(42) + h.snapshot.mockResolvedValue(918346) + h.closure.mockResolvedValue(85600) + + const result = await preparePackagistTransitiveCounts() + + expect(h.createRun).toHaveBeenCalledWith(h.fakeQx) + expect(result).toEqual({ runId: 42 }) + expect(h.closure).toHaveBeenCalledWith(h.fakeQx) + expect(h.markMerging).toHaveBeenCalledWith(h.fakeQx, 42, { + edgeCount: 918346, + packagesWithDependents: 85600, + }) + }) + + it('adopts an unfinished run row from a prior attempt (even one already merging)', async () => { + h.findPendingRun.mockResolvedValue(41) + h.snapshot.mockResolvedValue(10) + h.closure.mockResolvedValue(4) + + const result = await preparePackagistTransitiveCounts() + + expect(h.createRun).not.toHaveBeenCalled() + expect(result.runId).toBe(41) + }) + + it('does not fail-mark the run on a retryable error before the final attempt', async () => { + h.findPendingRun.mockResolvedValue(null) + h.createRun.mockResolvedValue(44) + h.snapshot.mockRejectedValue(new Error('connection reset')) + h.attempt.mockReturnValue(1) + + await expect(preparePackagistTransitiveCounts()).rejects.toThrow(/connection reset/) + + // the retry adopts the same unfinished row — fail-marking it early would strand it + expect(h.failRun).not.toHaveBeenCalled() + }) + + it('fail-marks the run when a retryable error exhausts the final attempt', async () => { + h.findPendingRun.mockResolvedValue(44) + h.snapshot.mockRejectedValue(new Error('connection reset')) + h.attempt.mockReturnValue(3) + + await expect(preparePackagistTransitiveCounts()).rejects.toThrow(/connection reset/) + + expect(h.failRun).toHaveBeenCalledWith(h.fakeQx, 44, 'connection reset') + }) + + it('aborts and marks the run failed when the edge snapshot is empty', async () => { + h.findPendingRun.mockResolvedValue(null) + h.createRun.mockResolvedValue(43) + h.snapshot.mockResolvedValue(0) + + await expect(preparePackagistTransitiveCounts()).rejects.toThrow(/no packagist direct edges/i) + + expect(h.closure).not.toHaveBeenCalled() + expect(h.failRun).toHaveBeenCalledWith( + h.fakeQx, + 43, + expect.stringMatching(/no packagist direct edges/i), + ) + }) + + it('rethrows the ORIGINAL non-retryable abort even when fail-marking itself fails', async () => { + h.findPendingRun.mockResolvedValue(null) + h.createRun.mockResolvedValue(43) + h.snapshot.mockResolvedValue(0) + h.failRun.mockRejectedValueOnce(new Error('ledger write refused')) + + // A masked original would surface as the retryable ledger error and let Temporal + // rerun the full package_dependencies scan. + await expect(preparePackagistTransitiveCounts()).rejects.toThrow(/no packagist direct edges/i) + }) +}) + +describe('finish/fail run — activities', () => { + it('marks the run done with the drain totals', async () => { + await finishPackagistTransitiveRun(7, { processed: 454000, changed: 86000 }) + + expect(h.finishRun).toHaveBeenCalledWith(h.fakeQx, 7, { processed: 454000, changed: 86000 }) + }) + + it('marks the run failed with the error message', async () => { + await failPackagistTransitiveRun(7, 'merge exploded') + + expect(h.failRun).toHaveBeenCalledWith(h.fakeQx, 7, 'merge exploded') + }) +}) + +describe('mergePackagistTransitiveBatch — activity', () => { + it('classifies an empty counts table as non-retryable', async () => { + h.mergeDal.mockRejectedValue(new EmptyPackagistTransitiveCountsError()) + + await expect(mergePackagistTransitiveBatch('', 10)).rejects.toMatchObject({ + nonRetryable: true, + }) + }) + + it('lets other merge errors stay retryable', async () => { + h.mergeDal.mockRejectedValue(new Error('deadlock detected')) + + await expect(mergePackagistTransitiveBatch('', 10)).rejects.toSatisfy( + (err: unknown) => err instanceof Error && !('nonRetryable' in err && err.nonRetryable), + ) + }) +}) diff --git a/services/apps/packages_worker/src/packagist/__tests__/wiring.test.ts b/services/apps/packages_worker/src/packagist/__tests__/wiring.test.ts index dda12c6401..8dc10de532 100644 --- a/services/apps/packages_worker/src/packagist/__tests__/wiring.test.ts +++ b/services/apps/packages_worker/src/packagist/__tests__/wiring.test.ts @@ -19,10 +19,16 @@ describe('package.json worker scripts', () => { }) describe('schedule cadence', () => { - it('defines the three packagist crons with minutes off :00 (crawler guideline)', () => { - // metadata has no cron — the seed workflow chains it as a child on completion + it('defines the four packagist crons with minutes off :00 (crawler guideline)', () => { + // metadata has no cron — the seed workflow chains it as a child on completion; the + // transitive closure has only the ledger-gated backstop cron, not a primary one const crons = Object.entries(PACKAGIST_CRONS) - expect(crons.map(([name]) => name).sort()).toEqual(['downloads30d', 'downloadsDaily', 'seed']) + expect(crons.map(([name]) => name).sort()).toEqual([ + 'downloads30d', + 'downloadsDaily', + 'seed', + 'transitiveBackstop', + ]) for (const [name, cron] of crons) { const minute = cron.split(' ')[0] expect(minute, `${name} cron minute`).toMatch(/^[1-9][0-9]?$/) @@ -30,7 +36,7 @@ describe('schedule cadence', () => { } }) - it('runs seed weekly, the 30d window capture monthly on the 1st, and daily downloads daily', () => { + it('runs seed weekly, 30d monthly on the 1st, daily downloads daily, backstop weekly after the chain', () => { expect(PACKAGIST_CRONS.seed.split(' ')).toHaveLength(5) expect(PACKAGIST_CRONS.seed.split(' ')[4]).not.toBe('*') // monthly, anchored on the 1st so the observed rolling value sits on the boundary @@ -38,6 +44,28 @@ describe('schedule cadence', () => { expect(PACKAGIST_CRONS.downloads30d.split(' ')[4]).toBe('*') expect(PACKAGIST_CRONS.downloadsDaily.split(' ')[2]).toBe('*') expect(PACKAGIST_CRONS.downloadsDaily.split(' ')[4]).toBe('*') + // weekly, a day after the Sunday seed so a healthy chain has already run + expect(PACKAGIST_CRONS.transitiveBackstop.split(' ')[4]).toBe('1') + expect(PACKAGIST_CRONS.transitiveBackstop.split(' ')[2]).toBe('*') + }) +}) + +describe('transitive dependents wiring', () => { + it('exports the transitive workflow from the shared workflows index', async () => { + const workflows = await import('../../workflows/index.js') + expect(workflows.computePackagistTransitiveDependents).toBeTypeOf('function') + }) + + it('re-exports the transitive activities from the shared activities index', () => { + const index = readFileSync('src/activities.ts', 'utf8') + for (const name of [ + 'preparePackagistTransitiveCounts', + 'mergePackagistTransitiveBatch', + 'finishPackagistTransitiveRun', + 'failPackagistTransitiveRun', + ]) { + expect(index).toContain(name) + } }) }) diff --git a/services/apps/packages_worker/src/packagist/activities.ts b/services/apps/packages_worker/src/packagist/activities.ts index ee610e26df..994b4ce037 100644 --- a/services/apps/packages_worker/src/packagist/activities.ts +++ b/services/apps/packages_worker/src/packagist/activities.ts @@ -1,4 +1,4 @@ -import { Context } from '@temporalio/activity' +import { ApplicationFailure, Context } from '@temporalio/activity' import { partition, timeout } from '@crowd/common' import { @@ -18,8 +18,24 @@ import type { PackagistMetadataCandidate, PackagistRunResult, } from '@crowd/data-access-layer/src/packages/packagistPackageState' +import { + createPackagistTransitiveRun, + failPackagistTransitiveRun as failRunInLedger, + findUnfinishedPackagistTransitiveRun, + finishPackagistTransitiveRun as finishRunInLedger, + hasRecentDonePackagistTransitiveRun, + markPackagistTransitiveRunMerging, +} from '@crowd/data-access-layer/src/packages/packagistTransitiveRuns' +import { + EmptyPackagistTransitiveCountsError, + computePackagistTransitiveCounts, + mergePackagistTransitiveCounts, + snapshotPackagistDirectEdges, +} from '@crowd/data-access-layer/src/packages/transitiveDependents' +import type { PackagistTransitiveMergeResult } from '@crowd/data-access-layer/src/packages/transitiveDependents' import type { QueryExecutor } from '@crowd/data-access-layer/src/queryExecutor' import { getServiceChildLogger } from '@crowd/logging' +import { TEMPORAL_CONFIG, getTemporalClient } from '@crowd/temporal' import { getPackagesDb } from '../db' import { mapWithConcurrency } from '../utils/concurrency' @@ -30,7 +46,7 @@ import { expandComposerMetadata } from './expandMetadata' import { fetchPackagistP2, fetchPackagistStats } from './fetchPackage' import { fetchPackagistPackageList, parsePackagistPackageList } from './listPackages' import { normalizePackagistStats, packagistNameFromPurl } from './normalize' -import { INGEST_MAX_ATTEMPTS } from './retryPolicy' +import { INGEST_MAX_ATTEMPTS, TRANSITIVE_PREPARE_MAX_ATTEMPTS } from './retryPolicy' import { FetchError, isFetchError, isP2NotModified } from './types' import { persistPackagistMetadata } from './upsertMetadata' import { persistPackagistPackageInfo } from './upsertPackageInfo' @@ -78,6 +94,15 @@ export async function packagistCurrentTimestamp(): Promise { return new Date().toISOString() } +// Current Temporal attempt, defaulting to 1 when run standalone (tests, scripts). +function activityAttempt(): number { + try { + return Context.current().info.attempt + } catch { + return 1 + } +} + // Fetch with the shared fast-retry contract: transient/429 results throw so Temporal // retries the batch; 4xx/malformed get INGEST_4XX_ATTEMPTS quick in-lane retries and // then surface as a give-up `error` the caller records on the state row. @@ -466,3 +491,102 @@ export async function getCriticalPackagistCount(): Promise { const qx = await getPackagesDb() return getCriticalPackagistPackageCount(qx) } + +// The heavy phase of the transitive lane: snapshot the direct edges, run the closure, +// leave the run in 'merging' for the keyset drain that follows. The graph sizes land +// on the run row; the workflow only needs the id. +export async function preparePackagistTransitiveCounts(): Promise<{ runId: number }> { + const qx = await getPackagesDb() + + const runId = + (await findUnfinishedPackagistTransitiveRun(qx)) ?? (await createPackagistTransitiveRun(qx)) + + const beat = setInterval(() => { + try { + Context.current().heartbeat() + } catch { + /* standalone */ + } + }, 30_000) + + try { + const edgeCount = await snapshotPackagistDirectEdges(qx) + if (edgeCount === 0) { + // A genuinely empty graph means upstream ingestion is broken — retrying won't help. + throw ApplicationFailure.nonRetryable( + 'no packagist direct edges found — snapshot produced an empty graph', + ) + } + const packagesWithDependents = await computePackagistTransitiveCounts(qx) + await markPackagistTransitiveRunMerging(qx, runId, { edgeCount, packagesWithDependents }) + log.info({ runId, edgeCount, packagesWithDependents }, 'packagist transitive closure prepared') + return { runId } + } catch (err) { + const nonRetryable = err instanceof ApplicationFailure && err.nonRetryable + if (nonRetryable || activityAttempt() >= TRANSITIVE_PREPARE_MAX_ATTEMPTS) { + // Best-effort: a failing ledger write must never replace the original error — + // that would turn a non-retryable abort into a retryable one. + try { + await failRunInLedger(qx, runId, (err as Error).message) + } catch (markErr) { + log.warn({ runId, err: String(markErr) }, 'could not fail-mark transitive run') + } + } + throw err + } finally { + clearInterval(beat) + } +} + +export async function mergePackagistTransitiveBatch( + afterId: string, + limit: number, +): Promise { + const qx = await getPackagesDb() + try { + return await mergePackagistTransitiveCounts(qx, afterId, limit) + } catch (err) { + // An empty counts table cannot heal by retrying — fail fast so the workflow + // fail-marks the run instead of burning the retry schedule against it. + if (err instanceof EmptyPackagistTransitiveCountsError) { + throw ApplicationFailure.nonRetryable(err.message) + } + throw err + } +} + +export async function finishPackagistTransitiveRun( + runId: number, + totals: { processed: number; changed: number }, +): Promise { + const qx = await getPackagesDb() + await finishRunInLedger(qx, runId, totals) +} + +// Terminal failure marking for the merge phase — called from the workflow's catch so a +// permanently failed drain reads 'failed' instead of sitting in 'merging' forever. +export async function packagistTransitiveRanRecently(withinDays: number): Promise { + const qx = await getPackagesDb() + return hasRecentDonePackagistTransitiveRun(qx, withinDays) +} + +// Backstop gate: is the fixed-id metadata drain mid-crawl? Its completion chains the +// closure itself, so the backstop must stand down instead of snapshotting changing edges. +export async function packagistMetadataDrainRunning(): Promise { + const client = await getTemporalClient(TEMPORAL_CONFIG()) + try { + const description = await client.workflow.getHandle('packagist-metadata-drain').describe() + return description.status.name === 'RUNNING' + } catch (err) { + if (err instanceof Error && err.name === 'WorkflowNotFoundError') return false + throw err + } +} + +export async function failPackagistTransitiveRun( + runId: number, + errorMessage: string, +): Promise { + const qx = await getPackagesDb() + await failRunInLedger(qx, runId, errorMessage) +} diff --git a/services/apps/packages_worker/src/packagist/retryPolicy.ts b/services/apps/packages_worker/src/packagist/retryPolicy.ts index 08ecd8fd56..90081372e2 100644 --- a/services/apps/packages_worker/src/packagist/retryPolicy.ts +++ b/services/apps/packages_worker/src/packagist/retryPolicy.ts @@ -2,3 +2,8 @@ // activity's give-up threshold so the two never drift: a package is only given up on // (marked scanned-error so the cursor can advance) once Temporal has exhausted these attempts. export const INGEST_MAX_ATTEMPTS = 5 + +// Transitive prepare attempts. Same lockstep contract: the prepare activity only +// fail-marks its run row on the final attempt (or a non-retryable error) — an earlier +// mark would make the row unadoptable and each retry would mint a duplicate. +export const TRANSITIVE_PREPARE_MAX_ATTEMPTS = 3 diff --git a/services/apps/packages_worker/src/packagist/schedule.ts b/services/apps/packages_worker/src/packagist/schedule.ts index f81ba421cf..1ca08f6e59 100644 --- a/services/apps/packages_worker/src/packagist/schedule.ts +++ b/services/apps/packages_worker/src/packagist/schedule.ts @@ -10,6 +10,9 @@ export const PACKAGIST_CRONS = { // Late in the UTC day on purpose: Packagist's `daily` figure is mostly real data by // 22:23 (vs. mostly borrowed from yesterday earlier on), with buffer before midnight. downloadsDaily: '23 22 * * *', + // Monday, ~24h after the Sunday chain normally completes: a ledger-gated no-op on + // healthy weeks, a fresh closure start when the seed→metadata chain broke. + transitiveBackstop: '41 4 * * 1', } // Workflow types by name (not function reference) so this module doesn't pull the @@ -33,6 +36,12 @@ const SCHEDULES = [ workflowType: 'ingestPackagistDownloadsDaily', args: [{}] as unknown[], }, + { + scheduleId: 'packagist-transitive-backstop', + cron: PACKAGIST_CRONS.transitiveBackstop, + workflowType: 'backstopPackagistTransitiveDrain', + args: [] as unknown[], + }, ] export async function schedulePackagistIngest(): Promise { diff --git a/services/apps/packages_worker/src/packagist/workflows.ts b/services/apps/packages_worker/src/packagist/workflows.ts index 52be4453eb..174b83d382 100644 --- a/services/apps/packages_worker/src/packagist/workflows.ts +++ b/services/apps/packages_worker/src/packagist/workflows.ts @@ -8,7 +8,7 @@ import { } from '@temporalio/workflow' import type * as activities from './activities' -import { INGEST_MAX_ATTEMPTS } from './retryPolicy' +import { INGEST_MAX_ATTEMPTS, TRANSITIVE_PREPARE_MAX_ATTEMPTS } from './retryPolicy' const acts = proxyActivities({ startToCloseTimeout: '15 minutes', @@ -20,45 +20,144 @@ const acts = proxyActivities({ }) const INGEST_BATCH = 50 -const ROUNDS_PER_RUN = 20 +export const ROUNDS_PER_RUN = 20 interface MetadataState { cutoff?: string cursor?: string } +export const TRANSITIVE_MERGE_BATCH = 10_000 + +const transitivePrepareActs = proxyActivities({ + startToCloseTimeout: '90 minutes', + heartbeatTimeout: '2 minutes', + retry: { + initialInterval: '1 minute', + backoffCoefficient: 2, + // Lockstep with the activity's terminal fail-marking — see retryPolicy.ts. + maximumAttempts: TRANSITIVE_PREPARE_MAX_ATTEMPTS, + }, +}) + +interface TransitiveState { + runId?: number + cursor?: string + processed?: number + changed?: number +} + +// ActivityFailure's own message is the generic "Activity task failed" — the reason a +// human wants in error_message sits at the bottom of the cause chain. +function rootErrorMessage(err: unknown): string { + let cur = err + for (;;) { + // `cause` is untyped under the es2017 lib this workspace compiles with, but it is + // present at runtime on Node 20 / Temporal failures. + const cause = cur instanceof Error ? (cur as { cause?: unknown }).cause : undefined + if (!(cause instanceof Error)) break + cur = cause + } + return cur instanceof Error ? cur.message : String(cur) +} + +export async function computePackagistTransitiveDependents( + state: TransitiveState = {}, +): Promise { + const runId = + state.runId ?? (await transitivePrepareActs.preparePackagistTransitiveCounts()).runId + let cursor = state.cursor ?? '' + let processed = state.processed ?? 0 + let changed = state.changed ?? 0 + + try { + for (let r = 0; r < ROUNDS_PER_RUN; r++) { + const batch = await acts.mergePackagistTransitiveBatch(cursor, TRANSITIVE_MERGE_BATCH) + processed += batch.processed + changed += batch.changed + if (batch.processed < TRANSITIVE_MERGE_BATCH) { + await acts.finishPackagistTransitiveRun(runId, { processed, changed }) + return + } + cursor = batch.nextCursor + } + } catch (err) { + // Best-effort: fail-marking must never replace the drain's original error. + try { + await acts.failPackagistTransitiveRun(runId, rootErrorMessage(err)) + } catch (markErr) { + log.warn(`could not fail-mark transitive run ${runId}: ${String(markErr)}`) + } + throw err + } + + await continueAsNew({ + runId, + cursor, + processed, + changed, + }) +} + interface DownloadsState { cutoff?: string cursor?: string } -export async function seedPackagistPackages(): Promise { - await acts.runPackagistPackageSeed() - - // Chain the drain off seed completion (not a cron) so newly discovered packages exist - // as rows first. ABANDON so it outlives this workflow; fixed id + ALLOW_DUPLICATE means - // a drain that outlasts the week makes next Sunday's seed skip its start instead of - // doubling the crawl (a still-RUNNING id always throws regardless of reuse policy). +async function chainDrain( + workflow: typeof ingestPackagistMetadata | typeof computePackagistTransitiveDependents, + workflowId: string, + stillRunningMessage: string, +): Promise { try { - await startChild(ingestPackagistMetadata, { - workflowId: 'packagist-metadata-drain', + await startChild(workflow, { + workflowId, workflowIdReusePolicy: WorkflowIdReusePolicy.ALLOW_DUPLICATE, args: [{}], parentClosePolicy: ParentClosePolicy.ABANDON, }) } catch (err) { if (err instanceof Error && err.name === 'WorkflowExecutionAlreadyStartedError') { - log.warn('packagist metadata drain still running from a prior seed — skipping chain-start') + log.warn(stillRunningMessage) return } throw err } } -// The cutoff is fixed once per run (deterministic activity), same pattern as the -// downloads-30d/daily lanes — a keyset scan only ever visits each purl once per drain, -// so due-selection must be anchored to a stable point in time rather than a live NOW() -// that would let a purl processed early in the run dodge this cycle's refresh window. +export async function seedPackagistPackages(): Promise { + await acts.runPackagistPackageSeed() + + // Chain the drain off seed completion (not a cron) so newly discovered packages exist + // as rows first. + await chainDrain( + ingestPackagistMetadata, + 'packagist-metadata-drain', + 'packagist metadata drain still running from a prior seed — skipping chain-start', + ) +} + +const chainTransitiveDrain = (): Promise => + chainDrain( + computePackagistTransitiveDependents, + 'packagist-transitive-drain', + 'packagist transitive drain still running — skipping chain-start', + ) + +const TRANSITIVE_BACKSTOP_FRESH_DAYS = 6 + +// Clock-based safety net for the event chain: a broken seed or metadata drain means no +// chain fired this week — start the closure anyway instead of letting counts go stale. +// Ledger-gated (a healthy week costs no second scan) and routed through the fixed +// workflow id, so it can never race a live drain. +export async function backstopPackagistTransitiveDrain(): Promise { + if (await acts.packagistTransitiveRanRecently(TRANSITIVE_BACKSTOP_FRESH_DAYS)) return + // A mid-crawl metadata drain will chain the closure itself on completion; starting it + // now would snapshot changing edges AND make that completion chain-start bounce. + if (await acts.packagistMetadataDrainRunning()) return + await chainTransitiveDrain() +} + export async function ingestPackagistMetadata(state: MetadataState = {}): Promise { const cutoff = state.cutoff ?? (await acts.packagistCurrentTimestamp()) let cursor = state.cursor || '' @@ -70,11 +169,17 @@ export async function ingestPackagistMetadata(state: MetadataState = {}): Promis cursor, INGEST_BATCH, ) - if (candidates.length === 0) return + if (candidates.length === 0) { + if (!stopAfterFirstPage) await chainTransitiveDrain() + return + } await acts.ingestPackagistMetadataBatch(candidates) cursor = nextCursor if (stopAfterFirstPage) return - if (candidates.length < INGEST_BATCH) return + if (candidates.length < INGEST_BATCH) { + await chainTransitiveDrain() + return + } } await continueAsNew({ cutoff, cursor }) diff --git a/services/apps/packages_worker/src/scripts/triggerPackagistSeed.ts b/services/apps/packages_worker/src/scripts/triggerPackagistSeed.ts index 48da3b31e8..59b90c16f2 100644 --- a/services/apps/packages_worker/src/scripts/triggerPackagistSeed.ts +++ b/services/apps/packages_worker/src/scripts/triggerPackagistSeed.ts @@ -1,6 +1,7 @@ import { TEMPORAL_CONFIG, getTemporalClient } from '@crowd/temporal' import { + computePackagistTransitiveDependents, ingestPackagistDownloads30d, ingestPackagistDownloadsDaily, ingestPackagistMetadata, @@ -8,13 +9,14 @@ import { } from '../packagist/workflows' const HELP = ` -Usage: trigger-packagist [seed|metadata|downloads-30d|downloads-daily] +Usage: trigger-packagist [seed|metadata|downloads-30d|downloads-daily|transitive] Arguments: seed Fetch packagist.org/packages/list.json and seed the packages table (default) metadata Crawl the dynamic (package info) + p2 (versions/dependencies) endpoints downloads-30d Capture the observed rolling 30d window for every package downloads-daily Capture daily downloads for the critical slice + transitive Recompute transitive dependent counts from stored direct edges Examples: pnpm trigger-packagist:local @@ -22,11 +24,22 @@ Examples: pnpm trigger-packagist:local metadata pnpm trigger-packagist:local downloads-30d pnpm trigger-packagist:local downloads-daily + pnpm trigger-packagist:local transitive ` -const TARGETS = ['seed', 'metadata', 'downloads-30d', 'downloads-daily'] as const +const TARGETS = ['seed', 'metadata', 'downloads-30d', 'downloads-daily', 'transitive'] as const type Target = (typeof TARGETS)[number] +// seed is dispatched separately (no state arg); Record keys keep this exhaustive — a +// target added to TARGETS without a mapping fails to compile instead of silently +// starting the wrong workflow. +const WORKFLOWS: Record, (state?: object) => Promise> = { + metadata: ingestPackagistMetadata, + 'downloads-30d': ingestPackagistDownloads30d, + 'downloads-daily': ingestPackagistDownloadsDaily, + transitive: computePackagistTransitiveDependents, +} + async function main(): Promise { const args = process.argv.slice(2) if (args.includes('--help') || args.includes('-h')) { @@ -59,19 +72,28 @@ async function main(): Promise { return } - const workflow = - target === 'metadata' - ? ingestPackagistMetadata - : target === 'downloads-30d' - ? ingestPackagistDownloads30d - : ingestPackagistDownloadsDaily + // The transitive drain must be single-instance: it drops and rebuilds global staging + // tables, so a manual run racing the chained weekly drain would corrupt the merge. + // Reusing the chained drain's fixed workflow id makes Temporal reject the overlap. + const workflowId = + target === 'transitive' ? 'packagist-transitive-drain' : `packagist-${target}-manual-${now}` - const handle = await client.workflow.start(workflow, { - taskQueue: 'packagist-worker', - workflowId: `packagist-${target}-manual-${now}`, - args: [{}], - }) - console.log(`Started workflow ${handle.workflowId}`) + try { + const handle = await client.workflow.start(WORKFLOWS[target], { + taskQueue: 'packagist-worker', + workflowId, + args: [{}], + }) + console.log(`Started workflow ${handle.workflowId}`) + } catch (err) { + if (err instanceof Error && err.name === 'WorkflowExecutionAlreadyStartedError') { + console.error( + `A ${target} drain is already running (workflow id ${workflowId}) — not starting a second.`, + ) + process.exit(1) + } + throw err + } } main() diff --git a/services/apps/packages_worker/src/workflows/index.ts b/services/apps/packages_worker/src/workflows/index.ts index 2ce5b58e60..695a14dcf2 100644 --- a/services/apps/packages_worker/src/workflows/index.ts +++ b/services/apps/packages_worker/src/workflows/index.ts @@ -30,6 +30,8 @@ export { ingestPackagistMetadata, ingestPackagistDownloads30d, ingestPackagistDownloadsDaily, + computePackagistTransitiveDependents, + backstopPackagistTransitiveDrain, } from '../packagist/workflows' export { ingestRubyGemsCriticalDetails, ingestRubyGemsPackages } from '../rubygems/workflows' export { diff --git a/services/libs/data-access-layer/src/osspckgs/ingestJobs.ts b/services/libs/data-access-layer/src/osspckgs/ingestJobs.ts index 02b7de30a9..9ae55e3c89 100644 --- a/services/libs/data-access-layer/src/osspckgs/ingestJobs.ts +++ b/services/libs/data-access-layer/src/osspckgs/ingestJobs.ts @@ -59,6 +59,27 @@ export interface MarkJobStatusFields { exportName?: string } +// Returns the newest pending job id for the kind, so a retried activity can reuse the +// row from its prior attempt instead of creating a duplicate. +export async function findPendingJobByKind( + qx: QueryExecutor, + jobKind: OsspckgsJobKind, +): Promise { + const row = await qx.selectOneOrNone( + ` + SELECT id + FROM osspckgs_ingest_jobs + WHERE job_kind = $(jobKind) + AND status = 'pending' + ORDER BY id DESC + LIMIT 1 + `, + { jobKind }, + ) + // id is bigserial (pg returns int8 as a string) — convert so the declared type is true. + return row ? Number(row.id) : null +} + // Returns the most recent job for the given kind that has already been exported to GCS, // so callers can skip re-running BQ when the user explicitly opts to reuse prior data. export async function findLatestExportedJobByKind( diff --git a/services/libs/data-access-layer/src/packages/index.ts b/services/libs/data-access-layer/src/packages/index.ts index 00fca97d7b..3d7719955d 100644 --- a/services/libs/data-access-layer/src/packages/index.ts +++ b/services/libs/data-access-layer/src/packages/index.ts @@ -9,7 +9,9 @@ export * from './npmWorkerState' export * from './packagistPackageState' export * from './pypiPackageState' export * from './packages' +export * from './packagistTransitiveRuns' export * from './repos' +export * from './transitiveDependents' export * from './versions' export * from './osv' export * from './repoDocker' diff --git a/services/libs/data-access-layer/src/packages/packagistTransitiveRuns.ts b/services/libs/data-access-layer/src/packages/packagistTransitiveRuns.ts new file mode 100644 index 0000000000..36eac3c46a --- /dev/null +++ b/services/libs/data-access-layer/src/packages/packagistTransitiveRuns.ts @@ -0,0 +1,97 @@ +import { QueryExecutor } from '../queryExecutor' + +// Run-level ledger for the packagist transitive-dependents lane (one row per run). +// The lane is a whole-ecosystem batch, so — unlike the per-purl watermarks in +// packagist_package_state — its state is a run lifecycle: +// pending → merging → done | failed. + +export async function createPackagistTransitiveRun(qx: QueryExecutor): Promise { + const row = await qx.selectOne( + `INSERT INTO packagist_transitive_runs (status) VALUES ('pending') RETURNING id`, + ) + return row.id +} + +// Newest unfinished run ('pending' OR 'merging'): a Temporal retry of the prepare +// activity must adopt the row it may have already marked merging — the activity's +// completion can be lost after the DB commit — instead of minting a duplicate and +// stranding the original. Safe because the fixed workflow id keeps the lane +// single-instance, so an unfinished row always belongs to this logical run. +export async function findUnfinishedPackagistTransitiveRun( + qx: QueryExecutor, +): Promise { + const row = await qx.selectOneOrNone( + `SELECT id FROM packagist_transitive_runs + WHERE status IN ('pending', 'merging') + ORDER BY id DESC LIMIT 1`, + ) + return row?.id ?? null +} + +// Backstop gate: has a run completed within the window? Failed runs don't count — +// the backstop exists precisely to retry after a broken week. +export async function hasRecentDonePackagistTransitiveRun( + qx: QueryExecutor, + withinDays: number, +): Promise { + const row = await qx.selectOne( + `SELECT EXISTS ( + SELECT 1 FROM packagist_transitive_runs + WHERE status = 'done' + AND finished_at > NOW() - $(withinDays) * INTERVAL '1 day' + ) AS recent`, + { withinDays }, + ) + return Boolean(row.recent) +} + +export async function markPackagistTransitiveRunMerging( + qx: QueryExecutor, + runId: number, + graph: { edgeCount: number; packagesWithDependents: number }, +): Promise { + // Guarded transitions make terminal states absorbing: a zombie attempt that outlived + // its Temporal timeout can never revive a run another attempt already finished/failed. + await qx.result( + `UPDATE packagist_transitive_runs + SET status = 'merging', + edge_count = $(edgeCount), + packages_with_dependents = $(packagesWithDependents) + WHERE id = $(runId) + AND status IN ('pending', 'merging')`, + { runId, edgeCount: graph.edgeCount, packagesWithDependents: graph.packagesWithDependents }, + ) +} + +export async function finishPackagistTransitiveRun( + qx: QueryExecutor, + runId: number, + totals: { processed: number; changed: number }, +): Promise { + await qx.result( + `UPDATE packagist_transitive_runs + SET status = 'done', + processed_rows = $(processed), + changed_rows = $(changed), + finished_at = NOW() + WHERE id = $(runId) + AND status = 'merging'`, + { runId, processed: totals.processed, changed: totals.changed }, + ) +} + +export async function failPackagistTransitiveRun( + qx: QueryExecutor, + runId: number, + errorMessage: string, +): Promise { + await qx.result( + `UPDATE packagist_transitive_runs + SET status = 'failed', + error_message = $(errorMessage), + finished_at = NOW() + WHERE id = $(runId) + AND status IN ('pending', 'merging')`, + { runId, errorMessage }, + ) +} diff --git a/services/libs/data-access-layer/src/packages/transitiveDependents.integration.test.ts b/services/libs/data-access-layer/src/packages/transitiveDependents.integration.test.ts new file mode 100644 index 0000000000..632324ec46 --- /dev/null +++ b/services/libs/data-access-layer/src/packages/transitiveDependents.integration.test.ts @@ -0,0 +1,517 @@ +import { afterAll, beforeAll, describe, expect, it } from 'vitest' + +import { getDbConnection } from '@crowd/database' + +import { createIngestJob, findPendingJobByKind, markJobStatus } from '../osspckgs/ingestJobs' +import type { QueryExecutor } from '../queryExecutor' +import { pgpQx } from '../queryExecutor' + +import { + createPackagistTransitiveRun, + failPackagistTransitiveRun, + findUnfinishedPackagistTransitiveRun, + finishPackagistTransitiveRun, + hasRecentDonePackagistTransitiveRun, + markPackagistTransitiveRunMerging, +} from './packagistTransitiveRuns' +import { + computePackagistTransitiveCounts, + mergePackagistTransitiveCounts, + snapshotPackagistDirectEdges, +} from './transitiveDependents' + +// Integration test: hits the running packages-db, DESTRUCTIVELY — it drops and rebuilds +// the production-named staging.packagist_transitive_* tables, which would break a live +// transitive drain on a shared DB. Credentials alone are therefore not enough to run it: +// it also requires the explicit CROWD_PACKAGES_TESTS_DESTRUCTIVE=1 opt-in. Skipped +// automatically otherwise so unit-test runs in CI stay green. +const DESTRUCTIVE_OPT_IN = process.env.CROWD_PACKAGES_TESTS_DESTRUCTIVE === '1' +const HAVE_DB = + !!process.env.CROWD_PACKAGES_DB_WRITE_HOST && + !!process.env.CROWD_PACKAGES_DB_PORT && + !!process.env.CROWD_PACKAGES_DB_USERNAME && + !!process.env.CROWD_PACKAGES_DB_DATABASE && + !!process.env.CROWD_PACKAGES_DB_PASSWORD + +const FIXTURE_TAG = 'fable-transitive-dependents-fixture' +const VENDOR = 'fable-tdx' + +// The packagist reverse transitive closure: collapse version-level direct edges to +// package-level pairs (snapshot), compute per-package transitive dependent counts +// (closure), merge them into packages.transitive_dependent_count (merge). +// +// Fixture graph (dep → subj means "dep depends on subj"): +// collapse: col-a v1 → col-x, col-a v2 → col-x (dedup), col-a v1 → col-dev (dev kind), +// col-self → col-self (self), n1 → n2 (npm) +// chain: c1 → c2 → c3 (c3: 1 transitive) +// diamond: d1 → {d2,d3}, {d2,d3} → d4 (d4: 1 transitive — d1 counted once) +// cycle: y1 → y2, y2 → y1, y3 → y1 (y2: 1 transitive — y3 through the cycle) +// multipath: x1 → x3, x1 → x2, x2 → x3 (x3: 0 — x1 is direct, not double-counted) +// leaf: no edges at all (merge zero-fills 0) +describe.skipIf(!HAVE_DB || !DESTRUCTIVE_OPT_IN)( + 'packagist transitive dependents — real packages-db', + () => { + let qx: QueryExecutor + + const ids: Record = {} + const versionIds: Record = {} + const jobIds: number[] = [] + const runIds: number[] = [] + + async function cleanupFixtures(): Promise { + await qx.result( + `DELETE FROM package_dependencies WHERE package_id IN ( + SELECT id FROM packages WHERE ingestion_source = $(tag))`, + { tag: FIXTURE_TAG }, + ) + await qx.result( + `DELETE FROM versions WHERE package_id IN ( + SELECT id FROM packages WHERE ingestion_source = $(tag))`, + { tag: FIXTURE_TAG }, + ) + await qx.result(`DELETE FROM packages WHERE ingestion_source = $(tag)`, { tag: FIXTURE_TAG }) + if (jobIds.length > 0) { + await qx.result(`DELETE FROM osspckgs_ingest_jobs WHERE id = ANY($(jobIds)::int[])`, { + jobIds, + }) + } + if (runIds.length > 0) { + await qx.result(`DELETE FROM packagist_transitive_runs WHERE id = ANY($(runIds)::int[])`, { + runIds, + }) + } + // The suites below destructively replace the REAL staging tables the live drain + // reads — drop them so no fixture-only remnant can ever feed a later merge. + await qx.result(`DROP TABLE IF EXISTS staging.packagist_transitive_edges`) + await qx.result(`DROP TABLE IF EXISTS staging.packagist_transitive_counts`) + } + + async function makePackage(name: string, ecosystem: 'packagist' | 'npm'): Promise { + const purl = + ecosystem === 'packagist' ? `pkg:composer/${VENDOR}/${name}` : `pkg:npm/${VENDOR}-${name}` + const row = await qx.selectOne( + `INSERT INTO packages (purl, ecosystem, namespace, name, registry_url, status, ingestion_source) + VALUES ($(purl), $(ecosystem), $(ns), $(name), 'https://example.test', 'active', $(tag)) + RETURNING id`, + { + purl, + ecosystem, + ns: ecosystem === 'packagist' ? VENDOR : null, + name, + tag: FIXTURE_TAG, + }, + ) + ids[name] = String(row.id) + return ids[name] + } + + async function makeVersion(pkg: string, number: string): Promise { + const row = await qx.selectOne( + `INSERT INTO versions (package_id, ecosystem, number, name, namespace) + SELECT id, ecosystem, $(number), name, namespace FROM packages WHERE id = $(id)::bigint + RETURNING id`, + { id: ids[pkg], number }, + ) + versionIds[`${pkg}#${number}`] = String(row.id) + return versionIds[`${pkg}#${number}`] + } + + async function addEdge( + pkg: string, + number: string, + toPkg: string, + kind: 'direct' | 'dev', + ): Promise { + await qx.result( + `INSERT INTO package_dependencies + (package_id, version_id, depends_on_id, version_constraint, dependency_kind, is_optional, created_at, updated_at) + VALUES ($(pkgId)::bigint, $(verId)::bigint, $(depId)::bigint, '^1.0', $(kind), FALSE, NOW(), NOW())`, + { + pkgId: ids[pkg], + verId: versionIds[`${pkg}#${number}`], + depId: ids[toPkg], + kind, + }, + ) + } + + function minFixtureId(): string { + const all = Object.values(ids).map((v) => BigInt(v)) + return String(all.reduce((a, b) => (a < b ? a : b)) - 1n) + } + + async function drainMerge(limit: number): Promise<{ processed: number; changed: number }> { + let cursor = minFixtureId() + const totals = { processed: 0, changed: 0 } + for (;;) { + const r = await mergePackagistTransitiveCounts(qx, cursor, limit) + totals.processed += r.processed + totals.changed += r.changed + if (r.processed < limit) return totals + cursor = r.nextCursor + } + } + + async function transitiveOf(pkg: string): Promise { + const row = await qx.selectOne( + `SELECT transitive_dependent_count FROM packages WHERE id = $(id)::bigint`, + { id: ids[pkg] }, + ) + return row.transitive_dependent_count === null ? null : Number(row.transitive_dependent_count) + } + + beforeAll(async () => { + const conn = await getDbConnection({ + host: process.env.CROWD_PACKAGES_DB_WRITE_HOST ?? '', + port: parseInt(process.env.CROWD_PACKAGES_DB_PORT ?? '0', 10), + database: process.env.CROWD_PACKAGES_DB_DATABASE ?? '', + user: process.env.CROWD_PACKAGES_DB_USERNAME ?? '', + password: process.env.CROWD_PACKAGES_DB_PASSWORD ?? '', + }) + qx = pgpQx(conn) + await cleanupFixtures() + + const packagist = [ + 'col-a', + 'col-x', + 'col-dev', + 'col-self', + 'c1', + 'c2', + 'c3', + 'd1', + 'd2', + 'd3', + 'd4', + 'y1', + 'y2', + 'y3', + 'x1', + 'x2', + 'x3', + 'leaf', + ] + for (const name of packagist) await makePackage(name, 'packagist') + for (const name of ['n1', 'n2']) await makePackage(name, 'npm') + + for (const name of [...packagist, 'n1', 'n2']) { + if (name === 'leaf') continue + await makeVersion(name, '1.0.0') + } + await makeVersion('col-a', '2.0.0') + + await addEdge('col-a', '1.0.0', 'col-x', 'direct') + await addEdge('col-a', '2.0.0', 'col-x', 'direct') + await addEdge('col-a', '1.0.0', 'col-dev', 'dev') + await addEdge('col-self', '1.0.0', 'col-self', 'direct') + await addEdge('n1', '1.0.0', 'n2', 'direct') + await addEdge('c1', '1.0.0', 'c2', 'direct') + await addEdge('c2', '1.0.0', 'c3', 'direct') + await addEdge('d1', '1.0.0', 'd2', 'direct') + await addEdge('d1', '1.0.0', 'd3', 'direct') + await addEdge('d2', '1.0.0', 'd4', 'direct') + await addEdge('d3', '1.0.0', 'd4', 'direct') + await addEdge('y1', '1.0.0', 'y2', 'direct') + await addEdge('y2', '1.0.0', 'y1', 'direct') + await addEdge('y3', '1.0.0', 'y1', 'direct') + await addEdge('x1', '1.0.0', 'x3', 'direct') + await addEdge('x1', '1.0.0', 'x2', 'direct') + await addEdge('x2', '1.0.0', 'x3', 'direct') + }, 60_000) + + afterAll(async () => { + if (qx) await cleanupFixtures() + }) + + async function snapshotPairCount(dep: string, subj: string): Promise { + const row = await qx.selectOne( + `SELECT COUNT(*) AS n FROM staging.packagist_transitive_edges + WHERE dep = $(dep)::bigint AND subj = $(subj)::bigint`, + { dep: ids[dep], subj: ids[subj] }, + ) + return Number(row.n) + } + + // Runs against the REAL package_dependencies table (plus the fixture rows), so each + // pass scans the full local dataset — hence the generous per-test timeouts. + describe('snapshotPackagistDirectEdges — package-level collapse', () => { + it('collapses version rows to distinct package edges, excluding dev/self/non-packagist', async () => { + const edgeCount = await snapshotPackagistDirectEdges(qx) + + expect(edgeCount).toBeGreaterThan(0) + const total = await qx.selectOne( + `SELECT COUNT(*) AS n FROM staging.packagist_transitive_edges`, + ) + expect(Number(total.n)).toBe(edgeCount) + + // two versions of col-a require col-x → exactly one package-level edge + expect(await snapshotPairCount('col-a', 'col-x')).toBe(1) + // require-dev is not installed transitively by Composer → excluded + expect(await snapshotPairCount('col-a', 'col-dev')).toBe(0) + // self-edges excluded + expect(await snapshotPairCount('col-self', 'col-self')).toBe(0) + // non-packagist ecosystems excluded + const npm = await qx.selectOne( + `SELECT COUNT(*) AS n FROM staging.packagist_transitive_edges WHERE dep = $(id)::bigint`, + { id: ids.n1 }, + ) + expect(Number(npm.n)).toBe(0) + // ordinary direct edge present + expect(await snapshotPairCount('c1', 'c2')).toBe(1) + }, 120_000) + + it('is re-runnable — a second snapshot fully replaces the first', async () => { + const edgeCount = await snapshotPackagistDirectEdges(qx) + const total = await qx.selectOne( + `SELECT COUNT(*) AS n FROM staging.packagist_transitive_edges`, + ) + expect(Number(total.n)).toBe(edgeCount) + expect(await snapshotPairCount('col-a', 'col-x')).toBe(1) + }, 120_000) + }) + + describe('closure + merge — fixture-only edge set', () => { + let packagesWithDependents: number + + // Rebuild the snapshot table with ONLY the fixture's package-level pairs so the + // closure output is fully deterministic (the snapshot suite above owns testing the + // collapse itself; this suite owns the closure/merge arithmetic). + beforeAll(async () => { + await qx.result(`DROP TABLE IF EXISTS staging.packagist_transitive_edges`) + await qx.result( + `CREATE UNLOGGED TABLE staging.packagist_transitive_edges (dep bigint NOT NULL, subj bigint NOT NULL)`, + ) + const pairs: Array<[string, string]> = [ + ['col-a', 'col-x'], + ['c1', 'c2'], + ['c2', 'c3'], + ['d1', 'd2'], + ['d1', 'd3'], + ['d2', 'd4'], + ['d3', 'd4'], + ['y1', 'y2'], + ['y2', 'y1'], + ['y3', 'y1'], + ['x1', 'x3'], + ['x1', 'x2'], + ['x2', 'x3'], + ] + for (const [dep, subj] of pairs) { + await qx.result( + `INSERT INTO staging.packagist_transitive_edges (dep, subj) + VALUES ($(dep)::bigint, $(subj)::bigint)`, + { dep: ids[dep], subj: ids[subj] }, + ) + } + await qx.result(`CREATE INDEX ON staging.packagist_transitive_edges (subj, dep)`) + + packagesWithDependents = await computePackagistTransitiveCounts(qx) + }, 30_000) + + async function countsRow(pkg: string): Promise { + const row = await qx.selectOneOrNone( + `SELECT transitive_dependent_count FROM staging.packagist_transitive_counts + WHERE package_id = $(id)::bigint`, + { id: ids[pkg] }, + ) + return row === null ? null : Number(row.transitive_dependent_count) + } + + it('produces one counts row per package with at least one dependent', async () => { + // subjects with ≥1 dependent: col-x, c2, c3, d2, d3, d4, y1, y2, x2, x3 + expect(packagesWithDependents).toBe(10) + const total = await qx.selectOne( + `SELECT COUNT(*) AS n FROM staging.packagist_transitive_counts`, + ) + expect(Number(total.n)).toBe(10) + // packages nobody depends on have no row (merge zero-fills them) + expect(await countsRow('c1')).toBeNull() + expect(await countsRow('y3')).toBeNull() + }) + + it('counts dependents at depth ≥ 2, excluding direct ones', async () => { + expect(await countsRow('c3')).toBe(1) // c1 via c2 + expect(await countsRow('c2')).toBe(0) // only the direct c1 + expect(await countsRow('col-x')).toBe(0) + }) + + it('dedups diamond paths — one ancestor counted once', async () => { + expect(await countsRow('d4')).toBe(1) // d1, via both d2 and d3 + expect(await countsRow('d2')).toBe(0) + expect(await countsRow('d3')).toBe(0) + }) + + it('terminates on cycles, never counts a package as its own dependent', async () => { + expect(await countsRow('y1')).toBe(0) // y2 and y3 are both direct + expect(await countsRow('y2')).toBe(1) // y3 reaches y2 through the cycle + }) + + it('counts a dependent that is both direct and transitive as direct only', async () => { + expect(await countsRow('x3')).toBe(0) // x1 is direct even though x1→x2→x3 also exists + }) + + it('merges counts into packages, zero-fills edge-less packagist rows, skips other ecosystems', async () => { + await drainMerge(200) + + expect(await transitiveOf('c3')).toBe(1) + expect(await transitiveOf('c2')).toBe(0) + expect(await transitiveOf('d4')).toBe(1) + expect(await transitiveOf('y2')).toBe(1) + expect(await transitiveOf('x3')).toBe(0) + // packages with no dependents (or no edges at all) get 0, not NULL + expect(await transitiveOf('c1')).toBe(0) + expect(await transitiveOf('y3')).toBe(0) + expect(await transitiveOf('leaf')).toBe(0) + expect(await transitiveOf('col-dev')).toBe(0) + expect(await transitiveOf('col-self')).toBe(0) + // other ecosystems are never touched + expect(await transitiveOf('n2')).toBeNull() + }) + + it('is churn-free on re-run — unchanged rows keep their last_synced_at', async () => { + const before = await qx.selectOne( + `SELECT last_synced_at FROM packages WHERE id = $(id)::bigint`, + { id: ids.c3 }, + ) + + const totals = await drainMerge(200) + expect(totals.changed).toBe(0) + + const after = await qx.selectOne( + `SELECT last_synced_at FROM packages WHERE id = $(id)::bigint`, + { id: ids.c3 }, + ) + expect(after.last_synced_at).toEqual(before.last_synced_at) + }) + + it('paginates by keyset — respects the limit and resumes from the cursor', async () => { + const first = await mergePackagistTransitiveCounts(qx, minFixtureId(), 3) + expect(first.processed).toBe(3) + expect(first.nextCursor).not.toBe('') + + const second = await mergePackagistTransitiveCounts(qx, first.nextCursor, 3) + expect(second.processed).toBe(3) + expect(BigInt(second.nextCursor)).toBeGreaterThan(BigInt(first.nextCursor)) + }) + + // Last on purpose: it empties the counts table the earlier cases depend on. The + // UNLOGGED staging table is truncated by crash recovery, and the zero-fill merge + // would otherwise read that as "every package is a leaf" and wipe real counts. + it('refuses to merge when the counts staging table is empty', async () => { + await qx.result(`TRUNCATE staging.packagist_transitive_counts`) + + await expect(mergePackagistTransitiveCounts(qx, minFixtureId(), 3)).rejects.toThrow( + /empty/i, + ) + }) + }) + + describe('findPendingJobByKind', () => { + it('returns the newest pending job of the kind; a finished job is no longer returned', async () => { + const jobA = await createIngestJob(qx, 'ranking', 'ranking', null) + jobIds.push(jobA) + const jobB = await createIngestJob(qx, 'ranking', 'ranking', null) + jobIds.push(jobB) + + // createIngestJob returns the raw bigserial id (a string at runtime); + // findPendingJobByKind normalizes to number — compare accordingly. + const found = await findPendingJobByKind(qx, 'ranking') + expect(found).toBe(Number(jobB)) + + await markJobStatus(qx, jobB, 'done', { finishedAt: new Date() }) + const next = await findPendingJobByKind(qx, 'ranking') + expect(next).not.toBe(Number(jobB)) + }) + }) + + describe('packagist_transitive_runs ledger', () => { + it('walks the run lifecycle: pending (reusable) → merging with graph sizes → done with totals', async () => { + const runA = await createPackagistTransitiveRun(qx) + runIds.push(runA) + const runB = await createPackagistTransitiveRun(qx) + runIds.push(runB) + + // newest unfinished wins — a Temporal retry reuses it instead of minting a third + expect(await findUnfinishedPackagistTransitiveRun(qx)).toBe(runB) + + await markPackagistTransitiveRunMerging(qx, runB, { + edgeCount: 918346, + packagesWithDependents: 85600, + }) + // still adoptable while merging — a retry whose completion was lost after the + // commit must find the row it already marked, not mint a duplicate + expect(await findUnfinishedPackagistTransitiveRun(qx)).toBe(runB) + + await finishPackagistTransitiveRun(qx, runB, { processed: 454455, changed: 86000 }) + // finished runs are no longer adoptable — the older pending row surfaces again + expect(await findUnfinishedPackagistTransitiveRun(qx)).toBe(runA) + + const row = await qx.selectOne( + `SELECT status, edge_count, packages_with_dependents, processed_rows, changed_rows, finished_at + FROM packagist_transitive_runs WHERE id = $(id)`, + { id: runB }, + ) + expect(row.status).toBe('done') + // window arithmetic: the fresh 'done' row is inside a 7-day window; a zero-day + // window can match nothing (nothing finishes in the future) + expect(await hasRecentDonePackagistTransitiveRun(qx, 7)).toBe(true) + expect(await hasRecentDonePackagistTransitiveRun(qx, 0)).toBe(false) + expect(Number(row.edge_count)).toBe(918346) + expect(Number(row.packages_with_dependents)).toBe(85600) + expect(Number(row.processed_rows)).toBe(454455) + expect(Number(row.changed_rows)).toBe(86000) + expect(row.finished_at).not.toBeNull() + }) + + it('terminal states are absorbing — zombie transitions cannot revive a run', async () => { + const zombie = await createPackagistTransitiveRun(qx) + runIds.push(zombie) + await failPackagistTransitiveRun(qx, zombie, 'boom') + + // a timed-out attempt's late markMerging/finish must bounce off the failed row + await markPackagistTransitiveRunMerging(qx, zombie, { + edgeCount: 1, + packagesWithDependents: 1, + }) + await finishPackagistTransitiveRun(qx, zombie, { processed: 1, changed: 1 }) + const failedRow = await qx.selectOne( + `SELECT status, edge_count FROM packagist_transitive_runs WHERE id = $(id)`, + { id: zombie }, + ) + expect(failedRow.status).toBe('failed') + expect(failedRow.edge_count).toBeNull() + + // and a late fail-mark can never overwrite a completed run + const completed = await createPackagistTransitiveRun(qx) + runIds.push(completed) + await markPackagistTransitiveRunMerging(qx, completed, { + edgeCount: 2, + packagesWithDependents: 2, + }) + await finishPackagistTransitiveRun(qx, completed, { processed: 2, changed: 2 }) + await failPackagistTransitiveRun(qx, completed, 'late zombie failure') + const doneRow = await qx.selectOne( + `SELECT status, error_message FROM packagist_transitive_runs WHERE id = $(id)`, + { id: completed }, + ) + expect(doneRow.status).toBe('done') + expect(doneRow.error_message).toBeNull() + }) + + it('records terminal failure with the error message', async () => { + const run = await createPackagistTransitiveRun(qx) + runIds.push(run) + + await failPackagistTransitiveRun(qx, run, 'closure exploded') + + const row = await qx.selectOne( + `SELECT status, error_message, finished_at FROM packagist_transitive_runs WHERE id = $(id)`, + { id: run }, + ) + expect(row.status).toBe('failed') + expect(row.error_message).toBe('closure exploded') + expect(row.finished_at).not.toBeNull() + }) + }) + }, +) diff --git a/services/libs/data-access-layer/src/packages/transitiveDependents.ts b/services/libs/data-access-layer/src/packages/transitiveDependents.ts new file mode 100644 index 0000000000..0457919cd1 --- /dev/null +++ b/services/libs/data-access-layer/src/packages/transitiveDependents.ts @@ -0,0 +1,135 @@ +import { QueryExecutor } from '../queryExecutor' + +export interface PackagistTransitiveMergeResult { + processed: number + changed: number + nextCursor: string +} + +// Typed so the activity layer can classify it non-retryable: an empty counts table +// cannot heal by retrying the merge. +export class EmptyPackagistTransitiveCountsError extends Error { + constructor() { + super( + 'staging.packagist_transitive_counts is empty — refusing to zero-fill packages.transitive_dependent_count (truncated by a crash mid-drain?)', + ) + this.name = 'EmptyPackagistTransitiveCountsError' + } +} + +// Shared scaffold for the two staging builders. The CTAS command tag already carries the +// row count, so no separate COUNT(*) rescan is needed. ANALYZE matters: a just-created +// table has no pg_statistic rows, and both consumers (the recursive closure, the 45-odd +// merge-batch joins) would otherwise plan against default selectivity guesses. +async function rebuildStagingTable( + qx: QueryExecutor, + table: string, + createAsSql: string, + indexColumns: string, +): Promise { + return qx.tx(async (tx) => { + await tx.result(`SET LOCAL max_parallel_workers_per_gather = 4`) + // Temporal timeouts don't kill in-flight SQL — this bounds each statement so a hung + // CTAS dies well inside the 90-min activity deadline (2 statements + slack). + await tx.result(`SET LOCAL statement_timeout = '40min'`) + await tx.result(`DROP TABLE IF EXISTS ${table}`) + const rows = await tx.result(`CREATE UNLOGGED TABLE ${table} AS ${createAsSql}`) + await tx.result(`CREATE INDEX ON ${table} (${indexColumns})`) + await tx.result(`ANALYZE ${table}`) + return rows + }) +} + +// Collapses version-level direct requires into distinct package-level pairs. +// dep = requirer, subj = depended-upon — same naming as the GO closure script in +// packages_worker/src/deps-dev/queries/dependentCountsSql.ts. +// The only query here that touches the ~1.5B-row package_dependencies table; package_id +// has no index, so this is a deliberate weekly parallel seq scan. +export async function snapshotPackagistDirectEdges(qx: QueryExecutor): Promise { + return rebuildStagingTable( + qx, + 'staging.packagist_transitive_edges', + `SELECT DISTINCT pd.package_id AS dep, pd.depends_on_id AS subj + FROM package_dependencies pd + JOIN packages p + ON p.id = pd.package_id + AND p.ecosystem = 'packagist' + WHERE pd.dependency_kind = 'direct' + AND pd.package_id != pd.depends_on_id`, + 'subj, dep', + ) +} + +// Reverse transitive closure over the snapshot: one row per package with ≥1 dependent, +// transitive = distinct reach minus distinct direct. The snapshot excludes self-edges, +// but cycles re-introduce (subj, subj) pairs in reach — hence the dep != subj filter. +export async function computePackagistTransitiveCounts(qx: QueryExecutor): Promise { + return rebuildStagingTable( + qx, + 'staging.packagist_transitive_counts', + `WITH RECURSIVE reach(subj, dep) AS ( + SELECT subj, dep FROM staging.packagist_transitive_edges + UNION + SELECT r.subj, e.dep + FROM reach r + JOIN staging.packagist_transitive_edges e ON e.subj = r.dep + ), + direct AS (SELECT subj, COUNT(*) AS n FROM staging.packagist_transitive_edges GROUP BY subj), + total AS (SELECT subj, COUNT(*) AS n FROM reach WHERE dep != subj GROUP BY subj) + SELECT t.subj AS package_id, t.n - d.n AS transitive_dependent_count + FROM total t + JOIN direct d USING (subj)`, + 'package_id', + ) +} + +// One keyset batch of packagist package ids merged from the counts staging table. +// COALESCE zero-fills packages with no dependents ("computed, none" vs NULL "never +// computed"); IS DISTINCT FROM keeps re-runs churn-free for Sequin/Tinybird. An empty +// afterId means "from the start" so first-generation callers don't need a sentinel. +export async function mergePackagistTransitiveCounts( + qx: QueryExecutor, + afterId: string, + limit: number, +): Promise { + // The zero-fill makes an empty counts table indistinguishable from "every package is + // a leaf" — and the table is UNLOGGED, so a crash-recovery truncation mid-drain would + // otherwise silently wipe every remaining count. A non-empty closure output is + // guaranteed by prepare's own empty-snapshot abort, so empty here is always an error. + const guard = await qx.selectOne( + `SELECT EXISTS (SELECT 1 FROM staging.packagist_transitive_counts) AS populated`, + ) + if (!guard.populated) { + throw new EmptyPackagistTransitiveCountsError() + } + + const row = await qx.selectOne( + `WITH batch AS ( + SELECT id + FROM packages + WHERE ecosystem = 'packagist' + AND id > COALESCE(NULLIF($(afterId), ''), '0')::bigint + ORDER BY id + LIMIT $(limit) + ), + updated AS ( + UPDATE packages p + SET transitive_dependent_count = COALESCE(c.transitive_dependent_count, 0), + last_synced_at = NOW() + FROM batch b + LEFT JOIN staging.packagist_transitive_counts c ON c.package_id = b.id + WHERE p.id = b.id + AND p.transitive_dependent_count IS DISTINCT FROM COALESCE(c.transitive_dependent_count, 0) + RETURNING p.id + ) + SELECT (SELECT COUNT(*) FROM batch) AS processed, + (SELECT COUNT(*) FROM updated) AS changed, + COALESCE((SELECT MAX(id)::text FROM batch), '') AS next_cursor`, + { afterId, limit }, + ) + return { + processed: Number(row.processed), + changed: Number(row.changed), + nextCursor: row.next_cursor, + } +} diff --git a/services/libs/data-access-layer/tsconfig.json b/services/libs/data-access-layer/tsconfig.json index bf7f183850..7faee40f6d 100644 --- a/services/libs/data-access-layer/tsconfig.json +++ b/services/libs/data-access-layer/tsconfig.json @@ -1,4 +1,12 @@ { "extends": "../../base.tsconfig.json", + // target/lib raised over base's es2017 for the BigInt literals in the packagist + // transitive integration test (include pulls *.test.ts into tsc-check). Consumers + // compile DAL sources under their own (es2017) configs, so src itself must stay + // free of post-es2017 syntax. + "compilerOptions": { + "target": "es2020", + "lib": ["es2020", "ES2021.String"] + }, "include": ["src/**/*"] }