Skip to content

Commit 5937852

Browse files
epipavepipav
committed
fix: address pypi worker review
Signed-off-by: anilb <epipav@gmail.com>
1 parent c915a75 commit 5937852

8 files changed

Lines changed: 31 additions & 28 deletions

File tree

‎services/apps/packages_worker/src/npm/upsertPackage.ts‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,9 @@
11
import {
22
getOrCreateRepoByUrl,
33
upsertNpmFundingLinks,
4-
upsertNpmMaintainers,
54
upsertNpmPackage,
65
upsertNpmVersions,
6+
upsertPackageMaintainers,
77
upsertPackageRepo,
88
} from '@crowd/data-access-layer/src/packages'
99
import type { QueryExecutor } from '@crowd/data-access-layer/src/queryExecutor'
@@ -106,7 +106,7 @@ export async function upsertPackage(
106106
verChanged.forEach((f) => changed.add(f))
107107

108108
if (maintainers.length > 0) {
109-
const mChanged = await upsertNpmMaintainers(t, pkgId, maintainers)
109+
const mChanged = await upsertPackageMaintainers(t, pkgId, maintainers)
110110
mChanged.forEach((f) => changed.add(f))
111111
}
112112

‎services/apps/packages_worker/src/pypi/downloads/ingestPypiDownloads.ts‎

Lines changed: 8 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -45,12 +45,9 @@ const { getCriticalPypiCount } = proxyActivities<typeof pypiDownloadsActivities>
4545
retry: { maximumAttempts: 3 },
4646
})
4747

48-
// Per-window 30-day scans can reach ~1.5 TiB worst case; daily scans are far smaller but a wide
49-
// backfill multiplies by the day count. Defaults guard against runaway scans and are overridable
50-
// per kind via BQ_DATASET_INGEST_PYPI_DOWNLOADS_{30D,DAILY}_MAX_BQ_GB.
51-
// A single 30d window scans ~31 day-partitions; measured at ~4.56 TB (≈147 GB/day averaged over
52-
// a month — weekdays are heavier than the weekend sample). Ceiling sits above that with headroom;
53-
// raise it if a future month exceeds it. Daily scans its 2-day trailing window (~300 GB).
48+
// Per-kind ceilings guard against runaway BQ scans; override via
49+
// BQ_DATASET_INGEST_PYPI_DOWNLOADS_{30D,DAILY}_MAX_BQ_GB. Defaults sized from the measured
50+
// ~4.56 TB/30d window and ~300 GB daily 2-day window, with headroom.
5451
const MAX_BYTES_GB_30D = 6000
5552
const MAX_BYTES_GB_DAILY = 2000
5653

@@ -181,6 +178,11 @@ export async function ingestPypiDownloadsDaily(opts: {
181178
const runId = start.toISOString().replace(/[:.]/g, '-')
182179
const today = start.toISOString().slice(0, 10)
183180

181+
// A backfill must supply BOTH bounds; a single bound is a mistake, not a partial range —
182+
// fail loudly rather than silently scanning the default 2-day window.
183+
if (Boolean(opts.startDate) !== Boolean(opts.endDate)) {
184+
throw new Error('ingestPypiDownloadsDaily: startDate and endDate must be provided together')
185+
}
184186
const range =
185187
opts.startDate && opts.endDate
186188
? { startDate: opts.startDate, endDate: opts.endDate }

‎services/apps/packages_worker/src/pypi/fetchProject.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@ const REGISTRY = 'https://pypi.org/pypi'
66
const USER_AGENT = 'lfx-packages-worker/0.1 (+https://lfx.linuxfoundation.org)'
77

88
// Fetch a project's metadata from the PyPI JSON API.
9-
// Error's handled with respect to their types (retryable or not)
9+
// Errors are handled per their type (retryable or not):
1010
// 404 → NOT_FOUND (skip)
1111
// 429 → RATE_LIMIT and 5xx/network → TRANSIENT (Temporal retries)
1212
// malformed body → MALFORMED (skip).

‎services/apps/packages_worker/src/pypi/proxies.ts‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,8 @@
1-
import { parseProxies, type ProxyEndpoint } from '../proxies'
1+
import { type ProxyEndpoint, parseProxies } from '../proxies'
22

3-
// Off by default: when disabled thesingle PyPI lane egresses directly (no ProxyAgent).
4-
// The proxy list is shared with workers via CROWD_PACKAGES_PROXIES
5-
// only the enable flag (CROWD_PACKAGES_PYPI_PROXIES_ENABLED) is PyPI-specific.
3+
// Off by default: when disabled, the single PyPI lane egresses directly (no ProxyAgent).
4+
// The proxy list is shared with other workers via CROWD_PACKAGES_PROXIES; only the enable
5+
// flag (CROWD_PACKAGES_PYPI_PROXIES_ENABLED) is PyPI-specific.
66
export function pypiProxiesEnabled(): boolean {
77
const raw = (process.env.CROWD_PACKAGES_PYPI_PROXIES_ENABLED ?? '').trim().toLowerCase()
88
return raw === 'true' || raw === '1'

‎services/apps/packages_worker/src/pypi/upsertProject.ts‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
import {
22
getOrCreateRepoByUrl,
33
upsertNpmFundingLinks,
4-
upsertNpmMaintainers,
4+
upsertPackageMaintainers,
55
upsertPackageRepo,
66
upsertPypiPackage,
77
upsertPypiVersions,
@@ -91,12 +91,12 @@ export async function upsertProject(
9191
}
9292

9393
if (versionRows.length > 0) {
94-
const verChanged = await upsertPypiVersions(t, pkgId, versionRows)
94+
const verChanged = await upsertPypiVersions(t, pkgId, versionRows, latestVersion)
9595
verChanged.forEach((f) => changed.add(f))
9696
}
9797

9898
if (maintainers.length > 0) {
99-
const mChanged = await upsertNpmMaintainers(t, pkgId, maintainers, 'pypi')
99+
const mChanged = await upsertPackageMaintainers(t, pkgId, maintainers, 'pypi')
100100
mChanged.forEach((f) => changed.add(f))
101101
}
102102

‎services/libs/data-access-layer/src/packages/maintainers.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,7 @@ export interface NpmMaintainerInput {
77
role: 'author' | 'maintainer'
88
}
99

10-
export async function upsertNpmMaintainers(
10+
export async function upsertPackageMaintainers(
1111
qx: QueryExecutor,
1212
packageId: string,
1313
maintainers: NpmMaintainerInput[],

‎services/libs/data-access-layer/src/packages/packages.ts‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -72,8 +72,8 @@ export async function upsertNpmPackage(
7272
THEN packages.versions_count
7373
ELSE EXCLUDED.versions_count END,
7474
latest_version = EXCLUDED.latest_version,
75-
first_release_at = EXCLUDED.first_release_at,
76-
latest_release_at = EXCLUDED.latest_release_at,
75+
first_release_at = COALESCE(EXCLUDED.first_release_at, packages.first_release_at),
76+
latest_release_at = COALESCE(EXCLUDED.latest_release_at, packages.latest_release_at),
7777
ingestion_source = EXCLUDED.ingestion_source,
7878
last_synced_at = EXCLUDED.last_synced_at
7979
RETURNING id, namespace, name, status, registry_url, description, homepage,
@@ -173,8 +173,8 @@ export async function upsertPypiPackage(
173173
THEN packages.versions_count
174174
ELSE EXCLUDED.versions_count END,
175175
latest_version = EXCLUDED.latest_version,
176-
first_release_at = EXCLUDED.first_release_at,
177-
latest_release_at = EXCLUDED.latest_release_at,
176+
first_release_at = COALESCE(EXCLUDED.first_release_at, packages.first_release_at),
177+
latest_release_at = COALESCE(EXCLUDED.latest_release_at, packages.latest_release_at),
178178
ingestion_source = EXCLUDED.ingestion_source,
179179
last_synced_at = EXCLUDED.last_synced_at
180180
RETURNING id, namespace, name, status, registry_url, description, homepage,

‎services/libs/data-access-layer/src/packages/versions.ts‎

Lines changed: 8 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -82,6 +82,7 @@ export async function upsertPypiVersions(
8282
qx: QueryExecutor,
8383
packageId: string,
8484
versions: PypiVersionInput[],
85+
latestNumber: string | null,
8586
): Promise<string[]> {
8687
if (versions.length === 0) return []
8788
const row: { changed_fields: string[] } = await qx.selectOne(
@@ -138,17 +139,17 @@ export async function upsertPypiVersions(
138139
},
139140
)
140141

141-
// Clear a stale is_latest on any OTHER version of this package — e.g. a previously-latest
142-
// version whose files were all deleted, so it is not in this batch and would otherwise keep
143-
// is_latest = true alongside the new latest.
144-
const latestNumbers = versions.filter((v) => v.isLatest).map((v) => v.number)
145-
if (latestNumbers.length > 0) {
142+
// Clear a stale is_latest on every OTHER version of this package. Anchored on the declared
143+
// latest (info.version) — NOT on what's in this batch — so a previously-latest version whose
144+
// files were all deleted (and is therefore omitted from the batch) can't keep is_latest = true
145+
// alongside the new latest. When no latest is known, leave flags untouched rather than wipe all.
146+
if (latestNumber != null) {
146147
await qx.result(
147148
`UPDATE versions SET is_latest = false
148149
WHERE package_id = $(packageId)::bigint
149150
AND is_latest = true
150-
AND NOT (number = ANY($(latestNumbers)::text[]))`,
151-
{ packageId, latestNumbers },
151+
AND number <> $(latestNumber)`,
152+
{ packageId, latestNumber },
152153
)
153154
}
154155

0 commit comments

Comments
 (0)