Skip to content

Commit 659a559

Browse files
skwowetskwowet
authored
Merge branch 'main' into CM-1266-public-api-org-domains
2 parents 0e5c7dc + 658f21d commit 659a559

15 files changed

Lines changed: 244 additions & 64 deletions

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

Lines changed: 111 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,111 @@
1+
-- Add the Sonatype popularity score as a 4th ranking signal in rank_packages().
2+
--
3+
-- Sonatype could not deliver raw Maven download counts, so they provide a
4+
-- popularity score (0-100, normalized per ecosystem) as the substitute for the
5+
-- missing downloads signal. Maven previously had only dependent_count and
6+
-- transitive_dependent_count feeding criticality.
7+
--
8+
-- The signal slots into the existing cumulative-coverage machinery unchanged:
9+
-- * cumulative-share is scale-invariant, so a 0-100 score sorts identically
10+
-- to a raw count — no normalization needed;
11+
-- * non-maven rows are all-NULL -> ecosystem total = 0 -> the WHERE clause
12+
-- drops the signal for them, so npm/pypi ranking is untouched.
13+
--
14+
-- No tier short-circuit here (measurement-first): is_critical is a BOOL_OR, so
15+
-- adding a signal can only make more packages critical, never fewer. Whether a
16+
-- hard "P0 always critical" guarantee is needed is decided after measuring the
17+
-- impact shift on real data.
18+
19+
CREATE OR REPLACE FUNCTION rank_packages(
20+
coverage_cutoff numeric DEFAULT 0.90,
21+
ecosystems text[] DEFAULT NULL
22+
)
23+
RETURNS TABLE(processed_rows int)
24+
LANGUAGE plpgsql AS $$
25+
DECLARE
26+
processed_count int;
27+
effective_ecosystems text[];
28+
BEGIN
29+
IF ecosystems IS NULL THEN
30+
SELECT ARRAY_AGG(DISTINCT ecosystem)
31+
INTO effective_ecosystems
32+
FROM packages;
33+
ELSE
34+
effective_ecosystems := ecosystems;
35+
END IF;
36+
37+
WITH base AS (
38+
SELECT
39+
id,
40+
ecosystem,
41+
COALESCE(downloads_last_30d, 0) AS downloads,
42+
COALESCE(dependent_count, 0) AS direct_dependents,
43+
COALESCE(transitive_dependent_count, 0) AS transitive_dependents,
44+
COALESCE(sonatype_popularity_score, 0) AS sonatype_popularity,
45+
SUM(COALESCE(downloads_last_30d, 0)) OVER (PARTITION BY ecosystem) AS ecosystem_total_downloads,
46+
SUM(COALESCE(dependent_count, 0)) OVER (PARTITION BY ecosystem) AS ecosystem_total_direct_dependents,
47+
SUM(COALESCE(transitive_dependent_count, 0)) OVER (PARTITION BY ecosystem) AS ecosystem_total_transitive_dependents,
48+
SUM(COALESCE(sonatype_popularity_score, 0)) OVER (PARTITION BY ecosystem) AS ecosystem_total_sonatype
49+
FROM packages
50+
WHERE ecosystem = ANY(effective_ecosystems)
51+
),
52+
walked AS (
53+
-- One row per (package × signal). Signals with a zero ecosystem total are
54+
-- excluded so they don't factor into the average (e.g. downloads for maven,
55+
-- sonatype for npm/pypi).
56+
-- cumulative_share_exclusive for the top-ranked package equals 0 by arithmetic
57+
-- (sum of rows above it is 0), so the top package is always inside the critical set.
58+
SELECT
59+
id,
60+
ecosystem,
61+
SUM(signal_value) OVER coverage_window / ecosystem_signal_total::numeric AS cumulative_share_inclusive,
62+
(SUM(signal_value) OVER coverage_window - signal_value) / ecosystem_signal_total::numeric AS cumulative_share_exclusive
63+
FROM base
64+
CROSS JOIN LATERAL (VALUES
65+
('downloads', downloads, ecosystem_total_downloads),
66+
('direct_dependents', direct_dependents, ecosystem_total_direct_dependents),
67+
('transitive_dependents', transitive_dependents, ecosystem_total_transitive_dependents),
68+
('sonatype_popularity', sonatype_popularity, ecosystem_total_sonatype)
69+
) AS signal(signal_name, signal_value, ecosystem_signal_total)
70+
WHERE ecosystem_signal_total > 0
71+
WINDOW coverage_window AS (
72+
PARTITION BY ecosystem, signal_name
73+
ORDER BY signal_value DESC, id
74+
ROWS UNBOUNDED PRECEDING
75+
)
76+
),
77+
combined AS (
78+
SELECT
79+
id,
80+
ecosystem,
81+
AVG(1.0 - cumulative_share_inclusive)::numeric(10, 4) AS new_impact,
82+
BOOL_OR(cumulative_share_exclusive < coverage_cutoff) AS new_is_critical
83+
FROM walked
84+
GROUP BY id, ecosystem
85+
),
86+
final AS (
87+
SELECT
88+
combined.id,
89+
combined.new_impact,
90+
combined.new_is_critical OR (spotlight.package_id IS NOT NULL) AS new_is_critical,
91+
ROW_NUMBER() OVER (
92+
PARTITION BY combined.ecosystem
93+
ORDER BY combined.new_impact DESC NULLS LAST, combined.id
94+
) AS new_rank_in_ecosystem
95+
FROM combined
96+
LEFT JOIN package_criticality_spotlight spotlight ON spotlight.package_id = combined.id
97+
)
98+
UPDATE packages p
99+
SET impact = final.new_impact,
100+
is_critical = final.new_is_critical,
101+
rank_in_ecosystem = final.new_rank_in_ecosystem,
102+
last_rank_pass_at = NOW(),
103+
last_synced_at = NOW()
104+
FROM final
105+
WHERE p.id = final.id;
106+
107+
GET DIAGNOSTICS processed_count = ROW_COUNT;
108+
109+
RETURN QUERY SELECT processed_count;
110+
END;
111+
$$;

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

Lines changed: 47 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -16,27 +16,61 @@ const log = getServiceChildLogger('go')
1616
const PROXY_SOURCE = 'go-proxy'
1717
const PKGGODEV_SOURCE = 'pkg-go-dev'
1818

19-
// TODO: filter to critical packages once computed
19+
export interface GoScanCursor {
20+
criticalAfter: string
21+
after: string
22+
}
23+
24+
type GoRow = { purl: string; name: string }
25+
26+
// Two independent purl-keyset cursors — one for critical packages, one for everything else —
27+
// each ordered/paginated purely by purl so WHERE and ORDER BY always match (no gaps, no
28+
// duplicates). A single query sorted by is_critical DESC with one shared purl cursor was tried
29+
// and rejected: the cursor advances to the last row's purl, and when a batch is critical-heavy
30+
// that purl can be far ahead of unprocessed non-critical rows, permanently excluding them for
31+
// the rest of the run.
2032
async function getGoBatch(
2133
qx: QueryExecutor,
34+
isCritical: boolean,
2235
afterPurl: string,
2336
batchSize: number,
24-
): Promise<Array<{ purl: string; name: string }>> {
37+
): Promise<GoRow[]> {
38+
if (batchSize <= 0) return []
2539
return qx.select(
2640
`SELECT purl, name FROM packages
27-
WHERE ecosystem = 'go' AND purl > $(after)
28-
ORDER BY last_synced_at ASC NULLS FIRST, purl ASC
41+
WHERE ecosystem = 'go' AND is_critical = $(isCritical) AND purl > $(after)
42+
ORDER BY purl ASC
2943
LIMIT $(limit)`,
30-
{ after: afterPurl, limit: batchSize },
44+
{ isCritical, after: afterPurl, limit: batchSize },
3145
)
3246
}
3347

48+
// Drains not-yet-processed critical packages first, then tops up the rest of the batch with
49+
// non-critical ones — so a rate-limit run only ever starves the non-critical tail.
50+
async function getGoPriorityBatch(
51+
qx: QueryExecutor,
52+
cursor: GoScanCursor,
53+
batchSize: number,
54+
): Promise<{ rows: GoRow[]; nextCursor: GoScanCursor }> {
55+
const critical = await getGoBatch(qx, true, cursor.criticalAfter, batchSize)
56+
const nonCritical = await getGoBatch(qx, false, cursor.after, batchSize - critical.length)
57+
58+
return {
59+
rows: [...critical, ...nonCritical],
60+
nextCursor: {
61+
criticalAfter:
62+
critical.length > 0 ? critical[critical.length - 1].purl : cursor.criticalAfter,
63+
after: nonCritical.length > 0 ? nonCritical[nonCritical.length - 1].purl : cursor.after,
64+
},
65+
}
66+
}
67+
3468
export async function enrichGoVersionsBatch(
35-
afterPurl: string,
69+
cursor: GoScanCursor,
3670
batchSize: number,
37-
): Promise<string | null> {
71+
): Promise<GoScanCursor | null> {
3872
const qx = await getPackagesDb()
39-
const rows = await getGoBatch(qx, afterPurl, batchSize)
73+
const { rows, nextCursor } = await getGoPriorityBatch(qx, cursor, batchSize)
4074
if (rows.length === 0) return null
4175

4276
const { fetchTimeoutMs, proxyConcurrency } = getGoConfig()
@@ -89,15 +123,15 @@ export async function enrichGoVersionsBatch(
89123
}
90124

91125
log.info({ count: rows.length, concurrency: proxyConcurrency }, 'Enriched go versions batch')
92-
return rows[rows.length - 1].purl
126+
return nextCursor
93127
}
94128

95129
export async function enrichGoStatusBatch(
96-
afterPurl: string,
130+
cursor: GoScanCursor,
97131
batchSize: number,
98-
): Promise<string | null> {
132+
): Promise<GoScanCursor | null> {
99133
const qx = await getPackagesDb()
100-
const rows = await getGoBatch(qx, afterPurl, batchSize)
134+
const { rows, nextCursor } = await getGoPriorityBatch(qx, cursor, batchSize)
101135
if (rows.length === 0) return null
102136

103137
const { fetchTimeoutMs } = getGoConfig()
@@ -137,5 +171,5 @@ export async function enrichGoStatusBatch(
137171
}
138172

139173
log.info({ count: rows.length }, 'Enriched go status batch')
140-
return rows[rows.length - 1].purl
174+
return nextCursor
141175
}

‎services/apps/packages_worker/src/go/proxyClient.ts‎

Lines changed: 52 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,11 @@ import { FetchError, GoProxyLatest } from './types'
22

33
const BASE = process.env.GO_PROXY_BASE_URL ?? 'https://proxy.golang.org'
44
const ZERO_TIME = '0001-01-01T00:00:00Z'
5+
const MAX_429_RETRIES = 5
6+
7+
function sleep(ms: number): Promise<void> {
8+
return new Promise((r) => setTimeout(r, ms))
9+
}
510

611
// GOPROXY spec: uppercase letters in a module path are escaped as '!' + lowercase.
712
export function escapeModulePath(module: string): string {
@@ -13,40 +18,55 @@ export async function fetchLatest(
1318
timeoutMs: number,
1419
): Promise<GoProxyLatest | FetchError> {
1520
const url = `${BASE}/${escapeModulePath(module)}/@latest`
16-
const controller = new AbortController()
17-
const timer = setTimeout(() => controller.abort(), timeoutMs)
18-
19-
let res: Response
20-
try {
21-
res = await fetch(url, { signal: controller.signal })
22-
} catch (e) {
23-
return { kind: 'TRANSIENT', message: `network error: ${(e as Error).message}` }
24-
} finally {
25-
clearTimeout(timer)
26-
}
2721

28-
if (res.status === 429) {
29-
return { kind: 'RATE_LIMIT', statusCode: 429, message: 'rate limited' }
30-
}
31-
// Any other 4xx is permanent (unknown/invalid module path) — skip, don't retry.
32-
if (res.status >= 400 && res.status < 500) {
33-
return { kind: 'NOT_FOUND', statusCode: res.status, message: `${res.status}` }
34-
}
35-
if (res.status !== 200) {
36-
return { kind: 'TRANSIENT', statusCode: res.status, message: `unexpected status ${res.status}` }
37-
}
22+
for (let attempt = 0; attempt <= MAX_429_RETRIES; attempt++) {
23+
const controller = new AbortController()
24+
const timer = setTimeout(() => controller.abort(), timeoutMs)
3825

39-
let body: { Version?: string; Time?: string; Origin?: { URL?: string } }
40-
try {
41-
body = (await res.json()) as { Version?: string; Time?: string; Origin?: { URL?: string } }
42-
} catch {
43-
return { kind: 'MALFORMED', message: 'invalid json' }
44-
}
45-
if (!body.Version) return { kind: 'MALFORMED', message: 'missing Version' }
26+
let res: Response
27+
try {
28+
res = await fetch(url, { signal: controller.signal })
29+
} catch (e) {
30+
return { kind: 'TRANSIENT', message: `network error: ${(e as Error).message}` }
31+
} finally {
32+
clearTimeout(timer)
33+
}
34+
35+
if (res.status === 429) {
36+
if (attempt === MAX_429_RETRIES) {
37+
return { kind: 'RATE_LIMIT', statusCode: 429, message: '429 after retries' }
38+
}
39+
const retryAfterSec = parseInt(res.headers.get('retry-after') ?? '', 10)
40+
const waitMs = Number.isNaN(retryAfterSec) ? 1000 * 2 ** attempt : retryAfterSec * 1000
41+
await sleep(waitMs)
42+
continue
43+
}
44+
// Any other 4xx is permanent (unknown/invalid module path) — skip, don't retry.
45+
if (res.status >= 400 && res.status < 500) {
46+
return { kind: 'NOT_FOUND', statusCode: res.status, message: `${res.status}` }
47+
}
48+
if (res.status !== 200) {
49+
return {
50+
kind: 'TRANSIENT',
51+
statusCode: res.status,
52+
message: `unexpected status ${res.status}`,
53+
}
54+
}
4655

47-
return {
48-
version: body.Version,
49-
releaseAt: body.Time && body.Time !== ZERO_TIME ? body.Time : null,
50-
repoUrl: body.Origin?.URL || null,
56+
let body: { Version?: string; Time?: string; Origin?: { URL?: string } }
57+
try {
58+
body = (await res.json()) as { Version?: string; Time?: string; Origin?: { URL?: string } }
59+
} catch {
60+
return { kind: 'MALFORMED', message: 'invalid json' }
61+
}
62+
if (!body.Version) return { kind: 'MALFORMED', message: 'missing Version' }
63+
64+
return {
65+
version: body.Version,
66+
releaseAt: body.Time && body.Time !== ZERO_TIME ? body.Time : null,
67+
repoUrl: body.Origin?.URL || null,
68+
}
5169
}
70+
71+
return { kind: 'RATE_LIMIT', statusCode: 429, message: '429 after retries' }
5272
}

‎services/apps/packages_worker/src/go/workflows.ts‎

Lines changed: 5 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -15,26 +15,22 @@ const acts = proxyActivities<typeof activities>({
1515
const BATCH = 100
1616
const ROUNDS_PER_RUN = 200
1717

18-
interface ScanState {
19-
cursor: string
20-
}
18+
const START_CURSOR = { criticalAfter: '', after: '' }
2119

22-
export async function enrichGoVersions(state: ScanState = { cursor: '' }): Promise<void> {
23-
let { cursor } = state
20+
export async function enrichGoVersions(cursor = START_CURSOR): Promise<void> {
2421
for (let r = 0; r < ROUNDS_PER_RUN; r++) {
2522
const next = await acts.enrichGoVersionsBatch(cursor, BATCH)
2623
if (next === null) return
2724
cursor = next
2825
}
29-
await continueAsNew<typeof enrichGoVersions>({ cursor })
26+
await continueAsNew<typeof enrichGoVersions>(cursor)
3027
}
3128

32-
export async function enrichGoStatus(state: ScanState = { cursor: '' }): Promise<void> {
33-
let { cursor } = state
29+
export async function enrichGoStatus(cursor = START_CURSOR): Promise<void> {
3430
for (let r = 0; r < ROUNDS_PER_RUN; r++) {
3531
const next = await acts.enrichGoStatusBatch(cursor, BATCH)
3632
if (next === null) return
3733
cursor = next
3834
}
39-
await continueAsNew<typeof enrichGoStatus>({ cursor })
35+
await continueAsNew<typeof enrichGoStatus>(cursor)
4036
}

‎services/libs/tinybird/pipes/activityRepositories_filtered.pipe‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,7 @@ SQL >
1717
JOIN
1818
repositories r FINAL ON r.insightsProjectId = i.id AND isNull (r.deletedAt) AND r.enabled = true
1919
where
20-
isNull (i.deletedAt)
20+
isNull (i.deletedAt) AND i.enabled = 1
2121
{% if defined(repos) %}
2222
AND r.url
2323
IN {{ Array(repos, 'String', description="Filter activity repo list", required=False) }}

‎services/libs/tinybird/pipes/collections_filtered.pipe‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -7,13 +7,21 @@ SQL >
77
SELECT
88
collections.*,
99
SUM(
10-
CASE WHEN collectionsInsightsProjects.insightsProjectId != '' THEN 1 ELSE 0 END
10+
CASE
11+
WHEN collectionsInsightsProjects.insightsProjectId != '' AND ip.enabled = 1
12+
THEN 1
13+
ELSE 0
14+
END
1115
) as "projectCount"
1216
FROM collections FINAL
1317
left join
1418
collectionsInsightsProjects
1519
on collectionsInsightsProjects.collectionId = collections.id
1620
and isNull (collectionsInsightsProjects.deletedAt)
21+
left join
22+
insightsProjects ip FINAL
23+
on ip.id = collectionsInsightsProjects.insightsProjectId
24+
and isNull (ip.deletedAt)
1725
where
1826
isNull (collections.deletedAt) AND isNull (collections.ssoUserId)
1927
{% if defined(slug) %}

‎services/libs/tinybird/pipes/contributors_leaderboard.pipe‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,7 @@ SQL >
4343
FROM insightsProjects final
4444
WHERE
4545
isNull (deletedAt)
46+
AND enabled = 1
4647
and slug = {{ String(project, description="Filter by project slug", required=True) }}
4748

4849
NODE contributors_leaderboard_2

‎services/libs/tinybird/pipes/health_score_copy.pipe‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,7 @@ DESCRIPTION >
55
SQL >
66
SELECT id, segmentId, slug, widgets
77
from insightsProjects FINAL
8-
WHERE isNull (deletedAt) and segmentId != ''
8+
WHERE isNull (deletedAt) AND enabled = 1 AND segmentId != ''
99

1010
NODE health_score_copy_data
1111
SQL >

0 commit comments

Comments
 (0)