Skip to content

Commit 376d9a0

Browse files
mbani01mbani01
committed
feat: criticality worker
Signed-off-by: Mouad BANI <mouad-mb@outlook.com>
1 parent 583f266 commit 376d9a0

8 files changed

Lines changed: 196 additions & 217 deletions

File tree

‎backend/src/osspckgs/migrations/V1780394591__packages_universe_graph_signals.sql‎

Lines changed: 0 additions & 6 deletions
This file was deleted.

‎backend/src/osspckgs/migrations/V1780416481__rank_packages_universe_v2.sql‎

Lines changed: 0 additions & 110 deletions
This file was deleted.

‎backend/src/osspckgs/migrations/V1780589351__package_criticality_spotlight.sql‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,24 @@
1+
-- Manual override table for criticality scoring (ADR-0001 §Spotlight overrides).
2+
-- Packages listed here are forced is_critical = TRUE regardless of computed score.
3+
-- Applied after ranking inside rank_packages_universe() so overrides survive
4+
-- every automated re-rank pass.
5+
--
6+
-- rationale, added_by, added_at are required — the table must stay auditable.
7+
-- namespace is nullable: cargo crates have no namespace, Maven artifacts do.
8+
-- The UNIQUE key uses COALESCE so (ecosystem, NULL namespace, name) is enforced correctly.
9+
10+
CREATE TABLE package_criticality_spotlight (
11+
id bigserial PRIMARY KEY,
12+
ecosystem text NOT NULL,
13+
namespace text,
14+
name text NOT NULL,
15+
rationale text NOT NULL,
16+
added_by text NOT NULL,
17+
added_at timestamptz NOT NULL DEFAULT NOW()
18+
);
19+
20+
-- Functional unique index: COALESCE treats NULL namespace as '' so that
21+
-- (cargo, NULL, tokio) and (cargo, NULL, serde) are unique but a duplicate
22+
-- (cargo, NULL, tokio) entry is rejected.
23+
CREATE UNIQUE INDEX ON package_criticality_spotlight
24+
(ecosystem, COALESCE(namespace, ''), name);

‎backend/src/osspckgs/migrations/V1780589607__rank_packages_universe_v2.sql‎

Lines changed: 142 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,142 @@
1+
-- Renames criticality_score → impact on both packages_universe and packages,
2+
-- and installs rank_packages_universe() with the updated formula.
3+
--
4+
-- Formula (ADR-0001 §Criticality scoring methodology):
5+
-- impact = w_downloads * pct_rank( LOG(1 + downloads_last_30d) ) within ecosystem
6+
-- + w_dep_pkgs * pct_rank( LOG(1 + dependent_packages_count) ) within ecosystem
7+
-- + w_transitive * pct_rank( LOG(1 + transitive_dependent_count) ) within ecosystem
8+
--
9+
-- Default weights: 0.25 / 0.25 / 0.50 (sum to 1.0).
10+
-- All weights and the top-N budget are call-time parameters — tunable without
11+
-- schema or code changes.
12+
--
13+
-- Steps inside the function:
14+
-- 1. Score — compute impact via weighted PERCENT_RANK()
15+
-- 2. Rank — ROW_NUMBER() per ecosystem, flag top-N as is_critical
16+
-- 2.5 Spotlight — force is_critical = TRUE for rows in package_criticality_spotlight
17+
-- 3. Propagate — copy impact + is_critical onto the packages table
18+
19+
-- Graph-derived scoring signals (ADR-0001 §Criticality scoring methodology).
20+
-- Populated by the criticality worker; NULL until first pass.
21+
ALTER TABLE packages_universe
22+
ADD COLUMN IF NOT EXISTS transitive_dependent_count bigint,
23+
ADD COLUMN IF NOT EXISTS centrality_score numeric(10, 8);
24+
25+
ALTER TABLE packages_universe
26+
RENAME COLUMN criticality_score TO impact;
27+
28+
ALTER TABLE packages
29+
RENAME COLUMN criticality_score TO impact;
30+
31+
CREATE OR REPLACE FUNCTION rank_packages_universe(
32+
weight_downloads numeric DEFAULT 0.25,
33+
weight_dependent_packages numeric DEFAULT 0.25,
34+
weight_transitive numeric DEFAULT 0.50,
35+
critical_top_n_by_ecosystem jsonb DEFAULT '{}'::jsonb
36+
)
37+
RETURNS TABLE(scored_rows int, ranked_rows int, propagated_rows int)
38+
LANGUAGE plpgsql AS $$
39+
DECLARE
40+
n_scored int;
41+
n_ranked int;
42+
n_propagated int;
43+
BEGIN
44+
-- ── Step 1: score ──────────────────────────────────────────────────────────
45+
WITH percentile_scores AS (
46+
SELECT
47+
id,
48+
(
49+
weight_downloads * PERCENT_RANK() OVER (
50+
PARTITION BY ecosystem ORDER BY LOG(1 + COALESCE(downloads_last_30d, 0)))
51+
52+
+ weight_dependent_packages * PERCENT_RANK() OVER (
53+
PARTITION BY ecosystem ORDER BY LOG(1 + COALESCE(dependent_packages_count, 0)))
54+
55+
+ weight_transitive * PERCENT_RANK() OVER (
56+
PARTITION BY ecosystem ORDER BY LOG(1 + COALESCE(transitive_dependent_count, 0)))
57+
)::numeric(10, 4) AS new_impact
58+
FROM packages_universe
59+
)
60+
UPDATE packages_universe pu
61+
SET last_rank_pass_at = NOW()
62+
WHERE TRUE;
63+
64+
UPDATE packages_universe pu
65+
SET impact = ps.new_impact
66+
FROM percentile_scores ps
67+
WHERE pu.id = ps.id
68+
AND pu.impact IS DISTINCT FROM ps.new_impact;
69+
70+
GET DIAGNOSTICS n_scored = ROW_COUNT;
71+
72+
-- ── Step 2: rank + flag ────────────────────────────────────────────────────
73+
WITH ranked AS (
74+
SELECT
75+
id, ecosystem,
76+
ROW_NUMBER() OVER (
77+
PARTITION BY ecosystem
78+
ORDER BY impact DESC NULLS LAST, id
79+
) AS r
80+
FROM packages_universe
81+
WHERE purl IS NOT NULL
82+
),
83+
flagged AS (
84+
SELECT
85+
id, r,
86+
COALESCE(
87+
r <= (critical_top_n_by_ecosystem ->> ecosystem)::int,
88+
FALSE
89+
) AS new_is_critical
90+
FROM ranked
91+
)
92+
UPDATE packages_universe pu
93+
SET rank_in_ecosystem = f.r,
94+
is_critical = f.new_is_critical
95+
FROM flagged f
96+
WHERE pu.id = f.id
97+
AND (
98+
pu.rank_in_ecosystem IS DISTINCT FROM f.r
99+
OR pu.is_critical IS DISTINCT FROM f.new_is_critical
100+
);
101+
102+
GET DIAGNOSTICS n_ranked = ROW_COUNT;
103+
104+
-- ── Step 2.5: apply spotlight overrides ───────────────────────────────────
105+
-- Force is_critical = TRUE for any row in package_criticality_spotlight,
106+
-- regardless of computed score or rank. Runs after Step 2 so overrides
107+
-- survive every automated re-rank pass.
108+
-- IS NOT DISTINCT FROM handles the NULL namespace case (e.g. cargo crates).
109+
UPDATE packages_universe pu
110+
SET is_critical = TRUE
111+
FROM package_criticality_spotlight s
112+
WHERE pu.ecosystem = s.ecosystem
113+
AND (pu.namespace IS NOT DISTINCT FROM s.namespace)
114+
AND pu.name = s.name
115+
AND pu.is_critical = FALSE;
116+
117+
-- ── Step 3: propagate to packages ─────────────────────────────────────────
118+
-- last_rank_pass_at is updated unconditionally (schema requirement: every pass,
119+
-- not only when scores change, so staleness checks work reliably).
120+
-- impact and is_critical are guarded by IS DISTINCT FROM to avoid unnecessary WAL writes.
121+
UPDATE packages p
122+
SET last_rank_pass_at = NOW()
123+
FROM packages_universe pu
124+
WHERE p.purl = pu.purl
125+
AND p.ecosystem = pu.ecosystem;
126+
127+
UPDATE packages p
128+
SET impact = pu.impact,
129+
is_critical = pu.is_critical
130+
FROM packages_universe pu
131+
WHERE p.purl = pu.purl
132+
AND p.ecosystem = pu.ecosystem
133+
AND (
134+
p.impact IS DISTINCT FROM pu.impact
135+
OR p.is_critical IS DISTINCT FROM pu.is_critical
136+
);
137+
138+
GET DIAGNOSTICS n_propagated = ROW_COUNT;
139+
140+
RETURN QUERY SELECT n_scored, n_ranked, n_propagated;
141+
END;
142+
$$;

‎services/apps/packages_worker/package.json‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
"private": true,
44
"scripts": {
55
"start:packages-worker": "CROWD_TEMPORAL_TASKQUEUE=packages-worker CROWD_TEMPORAL_NAMESPACE=$CROWD_PACKAGES_TEMPORAL_NAMESPACE SERVICE=packages-worker tsx src/bin/packages-worker.ts",
6+
"run:pagerank": "tsx src/criticality/run-pagerank.ts",
67
"dev:pagerank": "tsx --expose-gc src/criticality/run-pagerank.ts",
78
"start:criticality-worker": "CROWD_TEMPORAL_TASKQUEUE=packages-worker CROWD_TEMPORAL_NAMESPACE=$CROWD_PACKAGES_TEMPORAL_NAMESPACE SERVICE=criticality-worker tsx src/bin/criticality-worker.ts",
89
"start:github-repos-enricher": "SERVICE=github-repos-enricher tsx src/bin/github-repos-enricher.ts",

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

Lines changed: 16 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -9,17 +9,15 @@ import { CentralityInput, CentralityResult } from './types'
99

1010
const log = getServiceChildLogger('criticality')
1111

12-
function getConfig() {
13-
return {
14-
damping: Number(process.env.CRITICALITY_PAGERANK_DAMPING) || 0.85,
15-
maxIter: parseInt(process.env.CRITICALITY_PAGERANK_MAX_ITER ?? '', 10) || 100,
16-
convergence: Number(process.env.CRITICALITY_PAGERANK_CONVERGENCE) || 1e-6,
17-
}
18-
}
12+
const PAGERANK_DAMPING = 0.85
13+
const PAGERANK_MAX_ITER = 100
14+
const PAGERANK_CONVERGENCE = 1e-6
1915

2016
export async function criticalityComputePageRank(input: CentralityInput): Promise<CentralityResult> {
2117
const { ecosystem } = input
22-
const { damping, maxIter, convergence } = getConfig()
18+
const damping = PAGERANK_DAMPING
19+
const maxIter = PAGERANK_MAX_ITER
20+
const convergence = PAGERANK_CONVERGENCE
2321
const start = Date.now()
2422
const qx = await getPackagesDb()
2523

@@ -40,15 +38,18 @@ export async function criticalityComputePageRank(input: CentralityInput): Promis
4038
log.info({ ecosystem, iterations, nodeCount: graph.N }, 'PageRank converged')
4139

4240
// ── Step 4: merge centrality_score into packages_universe
43-
const batch = Array.from(graph.nodeIndex.entries()).map(([packageId, idx]) => ({
44-
packageId,
45-
centralityScore: scores[idx],
46-
}))
47-
41+
// Stream map entries into fixed-size chunks — O(CHUNK) extra memory, not O(N).
4842
const CHUNK = 10_000
49-
for (let i = 0; i < batch.length; i += CHUNK) {
50-
await mergeCentralityScores(qx, batch.slice(i, i + CHUNK))
43+
let buffer: Array<{ packageId: number; centralityScore: number }> = []
44+
45+
for (const [packageId, idx] of graph.nodeIndex) {
46+
buffer.push({ packageId, centralityScore: scores[idx] })
47+
if (buffer.length === CHUNK) {
48+
await mergeCentralityScores(qx, buffer)
49+
buffer = []
50+
}
5151
}
52+
if (buffer.length > 0) await mergeCentralityScores(qx, buffer)
5253

5354
return { ecosystem, nodeCount: graph.N, edgeCount, iterations, durationMs: Date.now() - start }
5455
}

‎services/apps/packages_worker/src/criticality/queries.ts‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,6 @@ export interface DirectEdge {
55
dependsOnId: number
66
}
77

8-
98
// Unique direct package→package edges for the ecosystem.
109
// DISTINCT deduplicates multi-version rows; parallel hint scoped to the transaction.
1110
export async function loadDirectEdges(qx: QueryExecutor, ecosystem: string): Promise<DirectEdge[]> {

0 commit comments

Comments
 (0)