From e2d60a1b4da13ddea78d0f857b163cbb09cf0d68 Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Thu, 9 Jul 2026 15:12:48 +0100 Subject: [PATCH 01/17] feat: add RubyGems types and HTTP client Signed-off-by: Mouad BANI --- .../packages_worker/src/rubygems/client.ts | 61 ++++++++++++++++++ .../packages_worker/src/rubygems/types.ts | 62 +++++++++++++++++++ 2 files changed, 123 insertions(+) create mode 100644 services/apps/packages_worker/src/rubygems/client.ts create mode 100644 services/apps/packages_worker/src/rubygems/types.ts diff --git a/services/apps/packages_worker/src/rubygems/client.ts b/services/apps/packages_worker/src/rubygems/client.ts new file mode 100644 index 0000000000..bb613a4267 --- /dev/null +++ b/services/apps/packages_worker/src/rubygems/client.ts @@ -0,0 +1,61 @@ +import axios from 'axios' + +import { + RubyGemsFetchError, + RubyGemsFetchResult, + RubyGemsGemResponse, + RubyGemsOwner, + RubyGemsVersionItem, +} from './types' + +function classifyError(err: unknown): RubyGemsFetchError | null { + if (!axios.isAxiosError(err)) return null + const status = err.response?.status + if (status === 404) return { kind: 'NOT_FOUND', status, message: err.message } + if (status === 429) return { kind: 'RATE_LIMIT', status, message: err.message } + return null +} + +export async function fetchGem(name: string): Promise> { + try { + const resp = await axios.get( + `https://rubygems.org/api/v1/gems/${encodeURIComponent(name)}.json`, + { timeout: 15000 }, + ) + return resp.data + } catch (err) { + const classified = classifyError(err) + if (classified) return classified + throw err + } +} + +export async function fetchVersions( + name: string, +): Promise> { + try { + const resp = await axios.get( + `https://rubygems.org/api/v1/versions/${encodeURIComponent(name)}.json`, + { timeout: 15000 }, + ) + return resp.data + } catch (err) { + const classified = classifyError(err) + if (classified) return classified + throw err + } +} + +export async function fetchOwners(name: string): Promise> { + try { + const resp = await axios.get( + `https://rubygems.org/api/v1/gems/${encodeURIComponent(name)}/owners.json`, + { timeout: 15000 }, + ) + return resp.data + } catch (err) { + const classified = classifyError(err) + if (classified) return classified + throw err + } +} diff --git a/services/apps/packages_worker/src/rubygems/types.ts b/services/apps/packages_worker/src/rubygems/types.ts new file mode 100644 index 0000000000..f1948ce1d1 --- /dev/null +++ b/services/apps/packages_worker/src/rubygems/types.ts @@ -0,0 +1,62 @@ +export interface BatchResult { + processed: number + skipped: number + error: number + unchanged: number +} + +export interface RubyGemsFetchError { + kind: 'NOT_FOUND' | 'RATE_LIMIT' + status?: number + message: string +} + +export type RubyGemsFetchResult = T | RubyGemsFetchError + +export function isRubyGemsFetchError(r: RubyGemsFetchResult): r is RubyGemsFetchError { + return typeof r === 'object' && r !== null && 'kind' in (r as object) +} + +export interface RubyGemsGemResponse { + name: string + version: string + info?: string | null + homepage_uri?: string | null + source_code_uri?: string | null + licenses?: string[] | null + downloads?: number +} + +export interface RubyGemsVersionItem { + number: string + created_at: string + prerelease?: boolean + licenses?: string[] | null +} + +export interface RubyGemsOwner { + handle?: string | null + email?: string | null +} + +export interface NormalizedRubyGemsPackage { + description: string | null + homepage: string | null + declaredRepositoryUrl: string | null + licenses: string[] | null + licensesRaw: string | null + latestVersion: string | null + totalDownloads: number +} + +export interface NormalizedRubyGemsVersion { + number: string + publishedAt: Date | null + isPrerelease: boolean + licenses: string[] | null +} + +export interface NormalizedRubyGemsOwner { + username: string + email: string | null +} From 6fb979245e8e694c275b2055e13ffe33d6a45656 Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Thu, 9 Jul 2026 15:13:22 +0100 Subject: [PATCH 02/17] feat: add RubyGems normalize functions Signed-off-by: Mouad BANI --- .../packages_worker/src/rubygems/normalize.ts | 63 +++++++++++++++++++ 1 file changed, 63 insertions(+) create mode 100644 services/apps/packages_worker/src/rubygems/normalize.ts diff --git a/services/apps/packages_worker/src/rubygems/normalize.ts b/services/apps/packages_worker/src/rubygems/normalize.ts new file mode 100644 index 0000000000..a7997cc91d --- /dev/null +++ b/services/apps/packages_worker/src/rubygems/normalize.ts @@ -0,0 +1,63 @@ +import { + NormalizedRubyGemsOwner, + NormalizedRubyGemsPackage, + NormalizedRubyGemsVersion, + RubyGemsGemResponse, + RubyGemsOwner, + RubyGemsVersionItem, +} from './types' + +function nonEmpty(value: string | null | undefined): string | null { + if (!value) return null + const trimmed = value.trim() + return trimmed === '' ? null : trimmed +} + +export function normalizeRubyGemsPackage(doc: RubyGemsGemResponse): NormalizedRubyGemsPackage { + const licenses = doc.licenses && doc.licenses.length > 0 ? doc.licenses : null + return { + description: nonEmpty(doc.info), + homepage: nonEmpty(doc.homepage_uri), + declaredRepositoryUrl: nonEmpty(doc.source_code_uri), + licenses, + licensesRaw: licenses ? licenses.join(', ') : null, + latestVersion: nonEmpty(doc.version), + totalDownloads: doc.downloads ?? 0, + } +} + +function parseCreatedAt(value: string | undefined): Date | null { + if (!value) return null + const date = new Date(value) + return isNaN(date.getTime()) ? null : date +} + +export function normalizeRubyGemsVersions( + items: RubyGemsVersionItem[], +): NormalizedRubyGemsVersion[] { + return items.map((item) => ({ + number: item.number, + publishedAt: parseCreatedAt(item.created_at), + isPrerelease: item.prerelease ?? false, + licenses: item.licenses && item.licenses.length > 0 ? item.licenses : null, + })) +} + +export function pickLatestRubyGemsVersion( + versions: NormalizedRubyGemsVersion[], +): NormalizedRubyGemsVersion | null { + if (versions.length === 0) return null + const sorted = [...versions].sort((a, b) => { + const at = a.publishedAt ? a.publishedAt.getTime() : 0 + const bt = b.publishedAt ? b.publishedAt.getTime() : 0 + return bt - at + }) + const stable = sorted.find((v) => !v.isPrerelease) + return stable ?? sorted[0] +} + +export function normalizeRubyGemsOwners(owners: RubyGemsOwner[]): NormalizedRubyGemsOwner[] { + return owners + .filter((o): o is RubyGemsOwner & { handle: string } => !!o.handle && o.handle.trim() !== '') + .map((o) => ({ username: o.handle, email: nonEmpty(o.email) })) +} From bfaf69bbf5e852da0bdc6c3f9b8bb05a7e12a2fd Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Thu, 9 Jul 2026 15:22:17 +0100 Subject: [PATCH 03/17] feat: add generic recordDownloadSnapshot DAL function Signed-off-by: Mouad BANI --- services/libs/data-access-layer/src/index.ts | 1 + .../src/osspckgs/downloads.ts | 58 +++++++++++++++++++ 2 files changed, 59 insertions(+) create mode 100644 services/libs/data-access-layer/src/osspckgs/downloads.ts diff --git a/services/libs/data-access-layer/src/index.ts b/services/libs/data-access-layer/src/index.ts index 2bcbdacc47..70141897f7 100644 --- a/services/libs/data-access-layer/src/index.ts +++ b/services/libs/data-access-layer/src/index.ts @@ -23,4 +23,5 @@ export * from './osspckgs/repos' export * from './osspckgs/stewardships' export * from './osspckgs/versions' export * from './osspckgs/nuget' +export * from './osspckgs/downloads' export * from './osspckgs/api' diff --git a/services/libs/data-access-layer/src/osspckgs/downloads.ts b/services/libs/data-access-layer/src/osspckgs/downloads.ts new file mode 100644 index 0000000000..14340b919c --- /dev/null +++ b/services/libs/data-access-layer/src/osspckgs/downloads.ts @@ -0,0 +1,58 @@ +import { insertDailyDownloads } from '../packages/downloadsDaily' +import { upsertLast30dDownload } from '../packages/downloadsLast30d' +import { QueryExecutor } from '../queryExecutor' + +export async function recordDownloadSnapshot( + qx: QueryExecutor, + params: { + packageId: number + purl: string + totalDownloads: number + today: string // YYYY-MM-DD + }, +): Promise { + const { packageId, purl, totalDownloads, today } = params + const changed: string[] = [] + + const prevRow = await qx.selectOne( + `SELECT total_downloads FROM packages WHERE id = $(packageId)`, + { packageId }, + ) + const prev: number | null = + prevRow?.total_downloads != null ? Number(prevRow.total_downloads) : null + + if (prev !== null && totalDownloads > prev) { + const delta = totalDownloads - prev + const dailyChanged = await insertDailyDownloads(qx, String(packageId), [ + { day: today, downloads: delta }, + ]) + dailyChanged.forEach((f) => changed.push(f)) + } + + const updated = await qx.result( + `UPDATE packages SET total_downloads = $(totalDownloads) + WHERE id = $(packageId) + AND (total_downloads IS NULL OR total_downloads < $(totalDownloads))`, + { totalDownloads, packageId }, + ) + if (updated > 0) changed.push('packages.total_downloads') + + const sumRow = await qx.selectOne( + `SELECT COALESCE(SUM(count), 0)::bigint AS total + FROM downloads_daily + WHERE package_id = $(packageId)::bigint + AND date > $(today)::date - INTERVAL '30 days' + AND date <= $(today)::date`, + { packageId, today }, + ) + const monthlySum = Number(sumRow.total) + + const thirtyDaysAgo = new Date(today) + thirtyDaysAgo.setDate(thirtyDaysAgo.getDate() - 29) + const startDate = thirtyDaysAgo.toISOString().split('T')[0] + + const monthlyChanged = await upsertLast30dDownload(qx, purl, startDate, today, monthlySum, true) + monthlyChanged.forEach((f) => changed.push(f)) + + return changed +} From 3cabf60a005728deb4936e7b8ee44c22a2702098 Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Thu, 9 Jul 2026 15:26:00 +0100 Subject: [PATCH 04/17] feat: add RubyGems package-selection DAL functions Signed-off-by: Mouad BANI --- services/libs/data-access-layer/src/index.ts | 1 + .../src/osspckgs/rubygems.ts | 68 +++++++++++++++++++ 2 files changed, 69 insertions(+) create mode 100644 services/libs/data-access-layer/src/osspckgs/rubygems.ts diff --git a/services/libs/data-access-layer/src/index.ts b/services/libs/data-access-layer/src/index.ts index 70141897f7..49dbb9e361 100644 --- a/services/libs/data-access-layer/src/index.ts +++ b/services/libs/data-access-layer/src/index.ts @@ -24,4 +24,5 @@ export * from './osspckgs/stewardships' export * from './osspckgs/versions' export * from './osspckgs/nuget' export * from './osspckgs/downloads' +export * from './osspckgs/rubygems' export * from './osspckgs/api' diff --git a/services/libs/data-access-layer/src/osspckgs/rubygems.ts b/services/libs/data-access-layer/src/osspckgs/rubygems.ts new file mode 100644 index 0000000000..f29921ae80 --- /dev/null +++ b/services/libs/data-access-layer/src/osspckgs/rubygems.ts @@ -0,0 +1,68 @@ +import { QueryExecutor } from '../queryExecutor' + +export type RubyGemsPackageToSync = { + id: number + purl: string + name: string + latestVersion: string | null +} + +const RUBYGEMS_WORKER_OUTCOMES = ['rubygems-registry', 'rubygems_not_found', 'rubygems_error'] + +export async function listRubyGemsPackagesToSync( + qx: QueryExecutor, + options: { limit: number }, +): Promise { + const { limit } = options + return qx.select( + ` + SELECT + p.id, + p.purl, + p.name, + p.latest_version AS "latestVersion" + FROM packages p + WHERE + p.ecosystem = 'rubygems' + AND ( + p.ingestion_source IS NULL + OR p.ingestion_source <> ALL($(workerOutcomes)::text[]) + OR p.last_synced_at < NOW() - INTERVAL '1 day' + ) + ORDER BY + p.dependent_count DESC NULLS LAST, + p.id ASC + LIMIT $(limit) + `, + { limit, workerOutcomes: RUBYGEMS_WORKER_OUTCOMES }, + ) +} + +export type RubyGemsCriticalPackageToSync = { + id: number + purl: string + name: string +} + +export async function listRubyGemsCriticalPackagesToSync( + qx: QueryExecutor, + options: { limit: number }, +): Promise { + const { limit } = options + return qx.select( + ` + SELECT p.id, p.purl, p.name + FROM packages p + LEFT JOIN LATERAL ( + SELECT MAX(v.last_synced_at) AS last_synced + FROM versions v WHERE v.package_id = p.id + ) v ON true + WHERE p.ecosystem = 'rubygems' + AND p.is_critical + AND (v.last_synced IS NULL OR v.last_synced < NOW() - INTERVAL '1 day') + ORDER BY p.dependent_count DESC NULLS LAST, p.id ASC + LIMIT $(limit) + `, + { limit }, + ) +} From 6d8c9cf5e5ccb90daea8580eb7d9c1b4849b0786 Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Thu, 9 Jul 2026 15:30:02 +0100 Subject: [PATCH 05/17] feat: add RubyGems tier 1 core ingestion loop Signed-off-by: Mouad BANI --- .../src/rubygems/runRubyGemsCoreLoop.ts | 182 ++++++++++++++++++ 1 file changed, 182 insertions(+) create mode 100644 services/apps/packages_worker/src/rubygems/runRubyGemsCoreLoop.ts diff --git a/services/apps/packages_worker/src/rubygems/runRubyGemsCoreLoop.ts b/services/apps/packages_worker/src/rubygems/runRubyGemsCoreLoop.ts new file mode 100644 index 0000000000..9176b8faa7 --- /dev/null +++ b/services/apps/packages_worker/src/rubygems/runRubyGemsCoreLoop.ts @@ -0,0 +1,182 @@ +import { + QueryExecutor, + RubyGemsPackageToSync, + listRubyGemsPackagesToSync, + logAuditFieldChange, + recordDownloadSnapshot, + upsertPackage, +} from '@crowd/data-access-layer' +import { getServiceChildLogger } from '@crowd/logging' + +import { fetchGem } from './client' +import { normalizeRubyGemsPackage } from './normalize' +import { BatchResult, isRubyGemsFetchError } from './types' + +const log = getServiceChildLogger('rubygems-core') + +async function withDeadlockRetry(fn: () => Promise, maxAttempts = 4): Promise { + for (let attempt = 1; ; attempt++) { + try { + return await fn() + } catch (err) { + const code = (err as { code?: string }).code + const isDeadlock = + code === '40P01' || /deadlock detected/i.test(String((err as Error)?.message)) + if (isDeadlock && attempt < maxAttempts) { + await new Promise((r) => setTimeout(r, 50 * attempt + Math.random() * 100)) + log.debug({ attempt }, 'Deadlock detected — retrying transaction') + continue + } + throw err + } + } +} + +function rubyGemsRegistryUrl(name: string): string { + return `https://rubygems.org/gems/${name}` +} + +async function markPackageError(qx: QueryExecutor, pkg: RubyGemsPackageToSync): Promise { + await upsertPackage(qx, { + purl: pkg.purl, + ecosystem: 'rubygems', + namespace: null, + name: pkg.name, + description: null, + homepage: null, + declaredRepositoryUrl: null, + licenses: null, + licensesRaw: null, + latestVersion: pkg.latestVersion, + ingestionSource: 'rubygems_error', + }) +} + +type PackageStatus = 'processed' | 'skipped' | 'error' | 'unchanged' + +export type RubyGemsCoreConfig = { batchSize: number; concurrency: number; groupDelayMs: number } + +async function processPackage( + qx: QueryExecutor, + pkg: RubyGemsPackageToSync, + today: string, +): Promise { + const gemResult = await fetchGem(pkg.name) + + if (isRubyGemsFetchError(gemResult)) { + if (gemResult.kind === 'NOT_FOUND') { + await upsertPackage(qx, { + purl: pkg.purl, + ecosystem: 'rubygems', + namespace: null, + name: pkg.name, + description: null, + homepage: null, + declaredRepositoryUrl: null, + licenses: null, + licensesRaw: null, + latestVersion: pkg.latestVersion, + registryUrl: rubyGemsRegistryUrl(pkg.name), + ingestionSource: 'rubygems_not_found', + }) + log.warn( + { purl: pkg.purl }, + 'Package not found on RubyGems registry — writing minimal record', + ) + return 'skipped' + } + if (gemResult.kind === 'RATE_LIMIT') { + log.warn({ purl: pkg.purl }, 'Rate limited by RubyGems registry — will retry next pass') + await markPackageError(qx, pkg) + return 'error' + } + throw new Error(`Transient error fetching ${pkg.purl}: ${gemResult.message}`) + } + + const normalized = normalizeRubyGemsPackage(gemResult) + + await withDeadlockRetry(() => + qx.tx(async (t) => { + const changed = new Set() + + const { id: packageDbId, changedFields: pkgChanged } = await upsertPackage(t, { + purl: pkg.purl, + ecosystem: 'rubygems', + namespace: null, + name: pkg.name, + description: normalized.description, + homepage: normalized.homepage, + declaredRepositoryUrl: normalized.declaredRepositoryUrl, + licenses: normalized.licenses, + licensesRaw: normalized.licensesRaw, + latestVersion: normalized.latestVersion, + registryUrl: rubyGemsRegistryUrl(pkg.name), + ingestionSource: 'rubygems-registry', + }) + pkgChanged.forEach((f) => changed.add(f)) + + if (normalized.totalDownloads > 0) { + const dlChanged = await recordDownloadSnapshot(t, { + packageId: packageDbId, + purl: pkg.purl, + totalDownloads: normalized.totalDownloads, + today, + }) + dlChanged.forEach((f) => changed.add(f)) + } + + await logAuditFieldChange(t, 'rubygems', pkg.purl, Array.from(changed)) + + log.debug({ purl: pkg.purl, totalDownloads: normalized.totalDownloads }, 'ok') + }), + ) + + return 'processed' +} + +export async function processBatch( + qx: QueryExecutor, + config: RubyGemsCoreConfig, + today: string, +): Promise { + const packages = await listRubyGemsPackagesToSync(qx, { limit: config.batchSize }) + + if (packages.length === 0) return { processed: 0, skipped: 0, error: 0, unchanged: 0 } + + log.info({ count: packages.length }, 'Batch started') + + const counts = { processed: 0, skipped: 0, error: 0, unchanged: 0 } + + for (let batchStart = 0; batchStart < packages.length; batchStart += config.concurrency) { + const group = packages.slice(batchStart, batchStart + config.concurrency) + + if (config.groupDelayMs > 0 && batchStart > 0) { + await new Promise((r) => setTimeout(r, config.groupDelayMs)) + } + + await Promise.all( + group.map(async (pkg) => { + try { + const status = await processPackage(qx, pkg, today) + counts[status]++ + } catch (err) { + const message = err instanceof Error ? err.message : String(err) + log.error({ purl: pkg.purl, error: message }, 'Unexpected error processing package') + try { + await markPackageError(qx, pkg) + } catch { + // best-effort — don't let a failed mark crash the batch + } + counts.error++ + } + }), + ) + + const done = batchStart + group.length + if (done % 1000 === 0 || done === packages.length) { + log.info({ done, total: packages.length, ...counts }, 'Progress') + } + } + + return counts +} From ec1aa9341a9f0dc72a2d2808da83fa4772f9f282 Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Thu, 9 Jul 2026 15:31:18 +0100 Subject: [PATCH 06/17] feat: add RubyGems tier 2 critical-details enrichment loop Signed-off-by: Mouad BANI --- .../src/rubygems/runRubyGemsCriticalLoop.ts | 182 ++++++++++++++++++ services/libs/data-access-layer/src/index.ts | 1 + 2 files changed, 183 insertions(+) create mode 100644 services/apps/packages_worker/src/rubygems/runRubyGemsCriticalLoop.ts diff --git a/services/apps/packages_worker/src/rubygems/runRubyGemsCriticalLoop.ts b/services/apps/packages_worker/src/rubygems/runRubyGemsCriticalLoop.ts new file mode 100644 index 0000000000..a22d9d1cb5 --- /dev/null +++ b/services/apps/packages_worker/src/rubygems/runRubyGemsCriticalLoop.ts @@ -0,0 +1,182 @@ +import { + IDbVersionUpsert, + QueryExecutor, + RubyGemsCriticalPackageToSync, + listRubyGemsCriticalPackagesToSync, + logAuditFieldChange, + replacePackageMaintainers, + upsertMaintainer, + upsertPackage, + upsertVersionsBatch, +} from '@crowd/data-access-layer' +import { getServiceChildLogger } from '@crowd/logging' + +import { fetchOwners, fetchVersions } from './client' +import { + normalizeRubyGemsOwners, + normalizeRubyGemsVersions, + pickLatestRubyGemsVersion, +} from './normalize' +import { BatchResult, isRubyGemsFetchError } from './types' + +const log = getServiceChildLogger('rubygems-critical') + +const BATCH_SIZE = 5000 +const CONCURRENCY = 4 +const GROUP_DELAY_MS = 1000 + +async function withDeadlockRetry(fn: () => Promise, maxAttempts = 4): Promise { + for (let attempt = 1; ; attempt++) { + try { + return await fn() + } catch (err) { + const code = (err as { code?: string }).code + const isDeadlock = + code === '40P01' || /deadlock detected/i.test(String((err as Error)?.message)) + if (isDeadlock && attempt < maxAttempts) { + await new Promise((r) => setTimeout(r, 50 * attempt + Math.random() * 100)) + log.debug({ attempt }, 'Deadlock detected — retrying transaction') + continue + } + throw err + } + } +} + +type PackageStatus = 'processed' | 'skipped' | 'error' | 'unchanged' + +async function processPackage( + qx: QueryExecutor, + pkg: RubyGemsCriticalPackageToSync, +): Promise { + const [versionsResult, ownersResult] = await Promise.all([ + fetchVersions(pkg.name), + fetchOwners(pkg.name), + ]) + + if (isRubyGemsFetchError(versionsResult)) { + if (versionsResult.kind === 'NOT_FOUND') { + log.warn({ purl: pkg.purl }, 'Package not found on RubyGems versions endpoint — skipping') + return 'skipped' + } + if (versionsResult.kind === 'RATE_LIMIT') { + log.warn({ purl: pkg.purl }, 'Rate limited fetching RubyGems versions — will retry next pass') + return 'error' + } + throw new Error(`Transient error fetching versions for ${pkg.purl}: ${versionsResult.message}`) + } + + const normalizedVersions = normalizeRubyGemsVersions(versionsResult) + const latest = pickLatestRubyGemsVersion(normalizedVersions) + const owners = isRubyGemsFetchError(ownersResult) ? [] : normalizeRubyGemsOwners(ownersResult) + + await withDeadlockRetry(() => + qx.tx(async (t) => { + const changed = new Set() + + const { id: packageDbId, changedFields: pkgChanged } = await upsertPackage(t, { + purl: pkg.purl, + ecosystem: 'rubygems', + namespace: null, + name: pkg.name, + description: null, + homepage: null, + declaredRepositoryUrl: null, + licenses: null, + licensesRaw: null, + latestVersion: latest?.number ?? null, + versionsCount: normalizedVersions.length > 0 ? normalizedVersions.length : null, + latestReleaseAt: latest?.publishedAt ?? null, + ingestionSource: 'rubygems-registry', + }) + pkgChanged.forEach((f) => changed.add(f)) + + if (normalizedVersions.length > 0) { + const versionRows: IDbVersionUpsert[] = normalizedVersions.map((v) => ({ + packageId: packageDbId, + ecosystem: 'rubygems', + namespace: null, + name: pkg.name, + number: v.number, + isLatest: v.number === latest?.number, + isPrerelease: v.isPrerelease, + license: v.licenses && v.licenses.length > 0 ? v.licenses[0] : null, + })) + const verChanged = await upsertVersionsBatch(t, versionRows) + verChanged.forEach((f) => changed.add(f)) + } + + const maintainerLinks: Array<{ maintainerId: number; role: 'maintainer' }> = [] + for (const owner of owners) { + const { id: maintainerId, changedFields: mChanged } = await upsertMaintainer(t, { + ecosystem: 'rubygems', + username: owner.username, + displayName: owner.username, + url: null, + email: owner.email, + }) + mChanged.forEach((f) => changed.add(f)) + maintainerLinks.push({ maintainerId, role: 'maintainer' }) + } + + if (maintainerLinks.length > 0) { + const pmChanged = await replacePackageMaintainers(t, packageDbId, maintainerLinks) + pmChanged.forEach((f) => changed.add(f)) + } + + await logAuditFieldChange(t, 'rubygems', pkg.purl, Array.from(changed)) + + log.debug( + { + purl: pkg.purl, + versions: normalizedVersions.length, + maintainers: maintainerLinks.length, + }, + 'ok', + ) + }), + ) + + return 'processed' +} + +export async function processBatch(qx: QueryExecutor): Promise { + const packages = await listRubyGemsCriticalPackagesToSync(qx, { limit: BATCH_SIZE }) + + if (packages.length === 0) return { processed: 0, skipped: 0, error: 0, unchanged: 0 } + + log.info({ count: packages.length }, 'Critical batch started') + + const counts = { processed: 0, skipped: 0, error: 0, unchanged: 0 } + + for (let batchStart = 0; batchStart < packages.length; batchStart += CONCURRENCY) { + const group = packages.slice(batchStart, batchStart + CONCURRENCY) + + if (GROUP_DELAY_MS > 0 && batchStart > 0) { + await new Promise((r) => setTimeout(r, GROUP_DELAY_MS)) + } + + await Promise.all( + group.map(async (pkg) => { + try { + const status = await processPackage(qx, pkg) + counts[status]++ + } catch (err) { + const message = err instanceof Error ? err.message : String(err) + log.error( + { purl: pkg.purl, error: message }, + 'Unexpected error processing critical package', + ) + counts.error++ + } + }), + ) + + const done = batchStart + group.length + if (done % 1000 === 0 || done === packages.length) { + log.info({ done, total: packages.length, ...counts }, 'Critical progress') + } + } + + return counts +} diff --git a/services/libs/data-access-layer/src/index.ts b/services/libs/data-access-layer/src/index.ts index 49dbb9e361..9ede0d95cd 100644 --- a/services/libs/data-access-layer/src/index.ts +++ b/services/libs/data-access-layer/src/index.ts @@ -21,6 +21,7 @@ export * from './osspckgs/maintainers' export * from './osspckgs/packages' export * from './osspckgs/repos' export * from './osspckgs/stewardships' +export * from './osspckgs/types' export * from './osspckgs/versions' export * from './osspckgs/nuget' export * from './osspckgs/downloads' From 4d82dce53789ffd1a8b701920439f9fd3848843b Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Thu, 9 Jul 2026 15:33:12 +0100 Subject: [PATCH 07/17] feat: wire RubyGems config, workflows, and activities Signed-off-by: Mouad BANI --- .../apps/packages_worker/src/activities.ts | 1 + services/apps/packages_worker/src/config.ts | 8 ++++++ .../src/rubygems/activities.ts | 26 +++++++++++++++++++ .../packages_worker/src/rubygems/workflows.ts | 26 +++++++++++++++++++ .../packages_worker/src/workflows/index.ts | 1 + 5 files changed, 62 insertions(+) create mode 100644 services/apps/packages_worker/src/rubygems/activities.ts create mode 100644 services/apps/packages_worker/src/rubygems/workflows.ts diff --git a/services/apps/packages_worker/src/activities.ts b/services/apps/packages_worker/src/activities.ts index 5b035b52cb..2985d91f79 100644 --- a/services/apps/packages_worker/src/activities.ts +++ b/services/apps/packages_worker/src/activities.ts @@ -32,6 +32,7 @@ export { } from './pypi/activities' export { getCriticalPypiCount } from './pypi/downloads/getCriticalPypiCount' export { processNuGetBatch } from './nuget/activities' +export { processRubyGemsCoreBatch, processRubyGemsCriticalBatch } from './rubygems/activities' export { processSecurityContactsBatch, ingestSecurityContactsForPurlActivity, diff --git a/services/apps/packages_worker/src/config.ts b/services/apps/packages_worker/src/config.ts index cc5307f161..8b083e9522 100644 --- a/services/apps/packages_worker/src/config.ts +++ b/services/apps/packages_worker/src/config.ts @@ -86,6 +86,14 @@ export function getNuGetConfig() { } } +export function getRubyGemsConfig() { + return { + batchSize: parseInt(process.env.RUBYGEMS_FETCHER_BATCH_SIZE ?? '1000', 10), + concurrency: parseInt(process.env.RUBYGEMS_FETCHER_CONCURRENCY ?? '8', 10), + groupDelayMs: parseInt(process.env.RUBYGEMS_FETCHER_GROUP_DELAY_MS ?? '1000', 10), + } +} + export function getDockerhubConfig() { return { hubBaseUrl: requireEnv('DOCKERHUB_API_BASE_URL'), diff --git a/services/apps/packages_worker/src/rubygems/activities.ts b/services/apps/packages_worker/src/rubygems/activities.ts new file mode 100644 index 0000000000..05ca858a3c --- /dev/null +++ b/services/apps/packages_worker/src/rubygems/activities.ts @@ -0,0 +1,26 @@ +import { getServiceChildLogger } from '@crowd/logging' + +import { getRubyGemsConfig } from '../config' +import { getPackagesDb } from '../db' + +import { processBatch as processCoreBatch } from './runRubyGemsCoreLoop' +import { processBatch as processCriticalBatch } from './runRubyGemsCriticalLoop' +import { BatchResult } from './types' + +const log = getServiceChildLogger('rubygems-activity') + +export async function processRubyGemsCoreBatch(): Promise { + const config = getRubyGemsConfig() + const qx = await getPackagesDb() + const today = new Date().toISOString().split('T')[0] + const result = await processCoreBatch(qx, config, today) + log.info({ ...result }, 'RubyGems core batch complete') + return result +} + +export async function processRubyGemsCriticalBatch(): Promise { + const qx = await getPackagesDb() + const result = await processCriticalBatch(qx) + log.info({ ...result }, 'RubyGems critical batch complete') + return result +} diff --git a/services/apps/packages_worker/src/rubygems/workflows.ts b/services/apps/packages_worker/src/rubygems/workflows.ts new file mode 100644 index 0000000000..80fa5e5b21 --- /dev/null +++ b/services/apps/packages_worker/src/rubygems/workflows.ts @@ -0,0 +1,26 @@ +import { continueAsNew, log, proxyActivities } from '@temporalio/workflow' + +import type * as activities from './activities' + +const acts = proxyActivities({ + startToCloseTimeout: '30 minutes', + retry: { initialInterval: '30 seconds', backoffCoefficient: 2, maximumAttempts: 5 }, +}) + +export async function ingestRubyGemsPackages(): Promise { + const result = await acts.processRubyGemsCoreBatch() + if (result.processed + result.skipped + result.error + result.unchanged === 0) { + log.info('RubyGems core ingestion complete — no more work, exiting.', { ...result }) + return + } + await continueAsNew() +} + +export async function ingestRubyGemsCriticalDetails(): Promise { + const result = await acts.processRubyGemsCriticalBatch() + if (result.processed + result.skipped + result.error + result.unchanged === 0) { + log.info('RubyGems critical ingestion complete — no more work, exiting.', { ...result }) + return + } + await continueAsNew() +} diff --git a/services/apps/packages_worker/src/workflows/index.ts b/services/apps/packages_worker/src/workflows/index.ts index 9b762282b4..a14aee8ed0 100644 --- a/services/apps/packages_worker/src/workflows/index.ts +++ b/services/apps/packages_worker/src/workflows/index.ts @@ -25,6 +25,7 @@ export { ingestPypiDownloadsDaily, } from '../pypi/downloads/ingestPypiDownloads' export { ingestNuGetPackages } from '../nuget/workflows' +export { ingestRubyGemsCriticalDetails, ingestRubyGemsPackages } from '../rubygems/workflows' export { ingestSecurityContacts, ingestSecurityContactsForPurlWorkflow, From f2acf3253aa6d6a66ab11a67de0a6a273b563395 Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Thu, 9 Jul 2026 15:35:01 +0100 Subject: [PATCH 08/17] feat: add RubyGems worker schedules, entrypoint, and package scripts Signed-off-by: Mouad BANI --- scripts/services/rubygems-worker.yaml | 65 ++++++++++++++++ services/apps/packages_worker/package.json | 3 + .../src/bin/rubygems-worker.ts | 9 +++ .../packages_worker/src/rubygems/schedule.ts | 78 +++++++++++++++++++ 4 files changed, 155 insertions(+) create mode 100644 scripts/services/rubygems-worker.yaml create mode 100644 services/apps/packages_worker/src/bin/rubygems-worker.ts create mode 100644 services/apps/packages_worker/src/rubygems/schedule.ts diff --git a/scripts/services/rubygems-worker.yaml b/scripts/services/rubygems-worker.yaml new file mode 100644 index 0000000000..5f7b1ed878 --- /dev/null +++ b/scripts/services/rubygems-worker.yaml @@ -0,0 +1,65 @@ +version: '3.1' + +x-env-args: &env-args + DOCKER_BUILDKIT: 1 + NODE_ENV: docker + SERVICE: rubygems-worker + SHELL: /bin/sh + SUPPRESS_NO_CONFIG_WARNING: 'true' + CROWD_TEMPORAL_TASKQUEUE: rubygems-worker + +services: + rubygems-worker: + build: + context: ../../ + dockerfile: ./scripts/services/docker/Dockerfile.packages + command: 'pnpm run start:rubygems-worker' + working_dir: /usr/crowd/app/services/apps/packages_worker + env_file: + - ../../backend/.env.dist.local + - ../../backend/.env.dist.composed + - ../../backend/.env.override.local + - ../../backend/.env.override.composed + environment: + <<: *env-args + restart: always + networks: + - crowd-bridge + + rubygems-worker-dev: + build: + context: ../../ + dockerfile: ./scripts/services/docker/Dockerfile.packages + command: 'pnpm run dev:rubygems-worker' + working_dir: /usr/crowd/app/services/apps/packages_worker + env_file: + - ../../backend/.env.dist.local + - ../../backend/.env.dist.composed + - ../../backend/.env.override.local + - ../../backend/.env.override.composed + environment: + <<: *env-args + hostname: rubygems-worker + networks: + - crowd-bridge + volumes: + - ../../services/libs/audit-logs/src:/usr/crowd/app/services/libs/audit-logs/src + - ../../services/libs/common/src:/usr/crowd/app/services/libs/common/src + - ../../services/libs/common_services/src:/usr/crowd/app/services/libs/common_services/src + - ../../services/libs/data-access-layer/src:/usr/crowd/app/services/libs/data-access-layer/src + - ../../services/libs/database/src:/usr/crowd/app/services/libs/database/src + - ../../services/libs/integrations/src:/usr/crowd/app/services/libs/integrations/src + - ../../services/libs/logging/src:/usr/crowd/app/services/libs/logging/src + - ../../services/libs/nango/src:/usr/crowd/app/services/libs/nango/src + - ../../services/libs/opensearch/src:/usr/crowd/app/services/libs/opensearch/src + - ../../services/libs/queue/src:/usr/crowd/app/services/libs/queue/src + - ../../services/libs/redis/src:/usr/crowd/app/services/libs/redis/src + - ../../services/libs/snowflake/src:/usr/crowd/app/services/libs/snowflake/src + - ../../services/libs/telemetry/src:/usr/crowd/app/services/libs/telemetry/src + - ../../services/libs/temporal/src:/usr/crowd/app/services/libs/temporal/src + - ../../services/libs/types/src:/usr/crowd/app/services/libs/types/src + - ../../services/apps/packages_worker/src:/usr/crowd/app/services/apps/packages_worker/src + +networks: + crowd-bridge: + external: true diff --git a/services/apps/packages_worker/package.json b/services/apps/packages_worker/package.json index d500ab3301..6d6936478f 100644 --- a/services/apps/packages_worker/package.json +++ b/services/apps/packages_worker/package.json @@ -49,6 +49,9 @@ "start:nuget-worker": "CROWD_TEMPORAL_TASKQUEUE=nuget-worker SERVICE=nuget-worker tsx src/bin/nuget-worker.ts", "dev:nuget-worker": "CROWD_TEMPORAL_TASKQUEUE=nuget-worker SERVICE=nuget-worker LOG_LEVEL=trace nodemon --watch src --watch ../../libs --ext ts --exec tsx --inspect=0.0.0.0:9242 src/bin/nuget-worker.ts", "dev:nuget-worker:local": "set -a && . ../../../backend/.env.dist.local && . ../../../backend/.env.override.local && set +a && CROWD_TEMPORAL_TASKQUEUE=nuget-worker SERVICE=nuget-worker LOG_LEVEL=trace nodemon --watch src --watch ../../libs --ext ts --exec tsx --inspect=0.0.0.0:9242 src/bin/nuget-worker.ts", + "start:rubygems-worker": "CROWD_TEMPORAL_TASKQUEUE=rubygems-worker SERVICE=rubygems-worker tsx src/bin/rubygems-worker.ts", + "dev:rubygems-worker": "CROWD_TEMPORAL_TASKQUEUE=rubygems-worker SERVICE=rubygems-worker LOG_LEVEL=trace nodemon --watch src --watch ../../libs --ext ts --exec tsx --inspect=0.0.0.0:9244 src/bin/rubygems-worker.ts", + "dev:rubygems-worker:local": "set -a && . ../../../backend/.env.dist.local && . ../../../backend/.env.override.local && set +a && CROWD_TEMPORAL_TASKQUEUE=rubygems-worker SERVICE=rubygems-worker LOG_LEVEL=trace nodemon --watch src --watch ../../libs --ext ts --exec tsx --inspect=0.0.0.0:9244 src/bin/rubygems-worker.ts", "backfill:maven": "SERVICE=maven tsx src/bin/maven-backfill.ts", "backfill:maven:local": "set -a && . ../../../backend/.env.dist.local && . ../../../backend/.env.override.local && set +a && SERVICE=maven LOG_LEVEL=info tsx src/bin/maven-backfill.ts", "import:maven-maintainers": "SERVICE=maven tsx src/maven/scripts/importMaintainersFromCsv.ts", diff --git a/services/apps/packages_worker/src/bin/rubygems-worker.ts b/services/apps/packages_worker/src/bin/rubygems-worker.ts new file mode 100644 index 0000000000..cd5415b53c --- /dev/null +++ b/services/apps/packages_worker/src/bin/rubygems-worker.ts @@ -0,0 +1,9 @@ +import { scheduleRubyGemsCriticalIngestion, scheduleRubyGemsIngestion } from '../rubygems/schedule' +import { svc } from '../service' + +setImmediate(async () => { + await svc.init() + await scheduleRubyGemsIngestion() + await scheduleRubyGemsCriticalIngestion() + await svc.start() +}) diff --git a/services/apps/packages_worker/src/rubygems/schedule.ts b/services/apps/packages_worker/src/rubygems/schedule.ts new file mode 100644 index 0000000000..1d06224a3c --- /dev/null +++ b/services/apps/packages_worker/src/rubygems/schedule.ts @@ -0,0 +1,78 @@ +import { ScheduleAlreadyRunning, ScheduleOverlapPolicy } from '@temporalio/client' + +import { svc } from '../service' +import { ingestRubyGemsCriticalDetails, ingestRubyGemsPackages } from '../workflows' + +export async function scheduleRubyGemsIngestion(): Promise { + const { temporal } = svc + if (!temporal) throw new Error('Temporal client not initialized') + + try { + await temporal.schedule.create({ + scheduleId: 'rubygems-registry-ingest', + spec: { + cronExpressions: ['0 7 * * *'], + }, + policies: { + overlap: ScheduleOverlapPolicy.SKIP, + catchupWindow: '1 hour', + }, + action: { + type: 'startWorkflow', + workflowType: ingestRubyGemsPackages, + workflowId: 'rubygems-daily-core', + taskQueue: 'rubygems-worker', + workflowRunTimeout: '24 hours', + retry: { + initialInterval: '30 seconds', + backoffCoefficient: 2, + maximumAttempts: 5, + }, + args: [], + }, + }) + } catch (err) { + if (err instanceof ScheduleAlreadyRunning) { + svc.log.info('Schedule rubygems-registry-ingest already exists, skipping creation.') + } else { + throw err + } + } +} + +export async function scheduleRubyGemsCriticalIngestion(): Promise { + const { temporal } = svc + if (!temporal) throw new Error('Temporal client not initialized') + + try { + await temporal.schedule.create({ + scheduleId: 'rubygems-critical-ingest', + spec: { + cronExpressions: ['0 9 * * *'], + }, + policies: { + overlap: ScheduleOverlapPolicy.SKIP, + catchupWindow: '1 hour', + }, + action: { + type: 'startWorkflow', + workflowType: ingestRubyGemsCriticalDetails, + workflowId: 'rubygems-daily-critical', + taskQueue: 'rubygems-worker', + workflowRunTimeout: '24 hours', + retry: { + initialInterval: '30 seconds', + backoffCoefficient: 2, + maximumAttempts: 5, + }, + args: [], + }, + }) + } catch (err) { + if (err instanceof ScheduleAlreadyRunning) { + svc.log.info('Schedule rubygems-critical-ingest already exists, skipping creation.') + } else { + throw err + } + } +} From 43d82e7fedfeaa2455add14d866958ef89af337a Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Thu, 9 Jul 2026 16:15:03 +0100 Subject: [PATCH 09/17] chore: add rubygems worker to packages.env Signed-off-by: Mouad BANI --- scripts/builders/packages.env | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/scripts/builders/packages.env b/scripts/builders/packages.env index d178e59ec3..7c3dc10b60 100644 --- a/scripts/builders/packages.env +++ b/scripts/builders/packages.env @@ -1,4 +1,4 @@ DOCKERFILE="./services/docker/Dockerfile.packages" CONTEXT="../" REPO="sjc.ocir.io/axbydjxa5zuh/packages" -SERVICES="github-repos-enricher bq-dataset-ingest npm-worker pypi-worker maven-worker osv-worker dockerhub-sync cargo-worker go-worker nuget-worker security-contacts-worker" +SERVICES="github-repos-enricher bq-dataset-ingest npm-worker pypi-worker maven-worker osv-worker dockerhub-sync cargo-worker go-worker nuget-worker security-contacts-worker rubygems-worker" From ac19f22a582aa0da434b993288b89e6818a4b1b6 Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Thu, 9 Jul 2026 16:18:12 +0100 Subject: [PATCH 10/17] fix: correct owners-fetch handling and add config to rubygems critical loop Signed-off-by: Mouad BANI --- services/apps/packages_worker/src/config.ts | 8 ++++++ .../src/rubygems/activities.ts | 5 ++-- .../src/rubygems/runRubyGemsCriticalLoop.ts | 28 +++++++++++-------- 3 files changed, 28 insertions(+), 13 deletions(-) diff --git a/services/apps/packages_worker/src/config.ts b/services/apps/packages_worker/src/config.ts index 8b083e9522..6b587381cf 100644 --- a/services/apps/packages_worker/src/config.ts +++ b/services/apps/packages_worker/src/config.ts @@ -94,6 +94,14 @@ export function getRubyGemsConfig() { } } +export function getRubyGemsCriticalConfig() { + return { + batchSize: parseInt(process.env.RUBYGEMS_CRITICAL_FETCHER_BATCH_SIZE ?? '5000', 10), + concurrency: parseInt(process.env.RUBYGEMS_CRITICAL_FETCHER_CONCURRENCY ?? '4', 10), + groupDelayMs: parseInt(process.env.RUBYGEMS_CRITICAL_FETCHER_GROUP_DELAY_MS ?? '1000', 10), + } +} + export function getDockerhubConfig() { return { hubBaseUrl: requireEnv('DOCKERHUB_API_BASE_URL'), diff --git a/services/apps/packages_worker/src/rubygems/activities.ts b/services/apps/packages_worker/src/rubygems/activities.ts index 05ca858a3c..4dd4ba1f26 100644 --- a/services/apps/packages_worker/src/rubygems/activities.ts +++ b/services/apps/packages_worker/src/rubygems/activities.ts @@ -1,6 +1,6 @@ import { getServiceChildLogger } from '@crowd/logging' -import { getRubyGemsConfig } from '../config' +import { getRubyGemsConfig, getRubyGemsCriticalConfig } from '../config' import { getPackagesDb } from '../db' import { processBatch as processCoreBatch } from './runRubyGemsCoreLoop' @@ -19,8 +19,9 @@ export async function processRubyGemsCoreBatch(): Promise { } export async function processRubyGemsCriticalBatch(): Promise { + const config = getRubyGemsCriticalConfig() const qx = await getPackagesDb() - const result = await processCriticalBatch(qx) + const result = await processCriticalBatch(qx, config) log.info({ ...result }, 'RubyGems critical batch complete') return result } diff --git a/services/apps/packages_worker/src/rubygems/runRubyGemsCriticalLoop.ts b/services/apps/packages_worker/src/rubygems/runRubyGemsCriticalLoop.ts index a22d9d1cb5..f148a55656 100644 --- a/services/apps/packages_worker/src/rubygems/runRubyGemsCriticalLoop.ts +++ b/services/apps/packages_worker/src/rubygems/runRubyGemsCriticalLoop.ts @@ -21,9 +21,11 @@ import { BatchResult, isRubyGemsFetchError } from './types' const log = getServiceChildLogger('rubygems-critical') -const BATCH_SIZE = 5000 -const CONCURRENCY = 4 -const GROUP_DELAY_MS = 1000 +export type RubyGemsCriticalConfig = { + batchSize: number + concurrency: number + groupDelayMs: number +} async function withDeadlockRetry(fn: () => Promise, maxAttempts = 4): Promise { for (let attempt = 1; ; attempt++) { @@ -68,7 +70,8 @@ async function processPackage( const normalizedVersions = normalizeRubyGemsVersions(versionsResult) const latest = pickLatestRubyGemsVersion(normalizedVersions) - const owners = isRubyGemsFetchError(ownersResult) ? [] : normalizeRubyGemsOwners(ownersResult) + const ownersFetchFailed = isRubyGemsFetchError(ownersResult) + const owners = ownersFetchFailed ? [] : normalizeRubyGemsOwners(ownersResult) await withDeadlockRetry(() => qx.tx(async (t) => { @@ -119,7 +122,7 @@ async function processPackage( maintainerLinks.push({ maintainerId, role: 'maintainer' }) } - if (maintainerLinks.length > 0) { + if (!ownersFetchFailed) { const pmChanged = await replacePackageMaintainers(t, packageDbId, maintainerLinks) pmChanged.forEach((f) => changed.add(f)) } @@ -140,8 +143,11 @@ async function processPackage( return 'processed' } -export async function processBatch(qx: QueryExecutor): Promise { - const packages = await listRubyGemsCriticalPackagesToSync(qx, { limit: BATCH_SIZE }) +export async function processBatch( + qx: QueryExecutor, + config: RubyGemsCriticalConfig, +): Promise { + const packages = await listRubyGemsCriticalPackagesToSync(qx, { limit: config.batchSize }) if (packages.length === 0) return { processed: 0, skipped: 0, error: 0, unchanged: 0 } @@ -149,11 +155,11 @@ export async function processBatch(qx: QueryExecutor): Promise { const counts = { processed: 0, skipped: 0, error: 0, unchanged: 0 } - for (let batchStart = 0; batchStart < packages.length; batchStart += CONCURRENCY) { - const group = packages.slice(batchStart, batchStart + CONCURRENCY) + for (let batchStart = 0; batchStart < packages.length; batchStart += config.concurrency) { + const group = packages.slice(batchStart, batchStart + config.concurrency) - if (GROUP_DELAY_MS > 0 && batchStart > 0) { - await new Promise((r) => setTimeout(r, GROUP_DELAY_MS)) + if (config.groupDelayMs > 0 && batchStart > 0) { + await new Promise((r) => setTimeout(r, config.groupDelayMs)) } await Promise.all( From e76438f7efa7b1ca2fd6139df00c0f087b302713 Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Fri, 10 Jul 2026 11:43:47 +0100 Subject: [PATCH 11/17] feat: support dependents_count ingestion Signed-off-by: Mouad BANI --- .../apps/packages_worker/src/activities.ts | 6 +- .../src/bin/rubygems-worker.ts | 7 +- services/apps/packages_worker/src/config.ts | 7 ++ .../src/rubygems/activities.ts | 18 ++++- .../packages_worker/src/rubygems/client.ts | 16 ++++ .../src/rubygems/runRubyGemsDependentsLoop.ts | 81 +++++++++++++++++++ .../packages_worker/src/rubygems/schedule.ts | 43 +++++++++- .../packages_worker/src/rubygems/workflows.ts | 9 +++ .../packages_worker/src/workflows/index.ts | 6 +- .../src/osspckgs/rubygems.ts | 36 +++++++++ 10 files changed, 224 insertions(+), 5 deletions(-) create mode 100644 services/apps/packages_worker/src/rubygems/runRubyGemsDependentsLoop.ts diff --git a/services/apps/packages_worker/src/activities.ts b/services/apps/packages_worker/src/activities.ts index 2985d91f79..ddce9cd27a 100644 --- a/services/apps/packages_worker/src/activities.ts +++ b/services/apps/packages_worker/src/activities.ts @@ -32,7 +32,11 @@ export { } from './pypi/activities' export { getCriticalPypiCount } from './pypi/downloads/getCriticalPypiCount' export { processNuGetBatch } from './nuget/activities' -export { processRubyGemsCoreBatch, processRubyGemsCriticalBatch } from './rubygems/activities' +export { + processRubyGemsCoreBatch, + processRubyGemsCriticalBatch, + processRubyGemsDependentsBatch, +} from './rubygems/activities' export { processSecurityContactsBatch, ingestSecurityContactsForPurlActivity, diff --git a/services/apps/packages_worker/src/bin/rubygems-worker.ts b/services/apps/packages_worker/src/bin/rubygems-worker.ts index cd5415b53c..f0d8332dbe 100644 --- a/services/apps/packages_worker/src/bin/rubygems-worker.ts +++ b/services/apps/packages_worker/src/bin/rubygems-worker.ts @@ -1,9 +1,14 @@ -import { scheduleRubyGemsCriticalIngestion, scheduleRubyGemsIngestion } from '../rubygems/schedule' +import { + scheduleRubyGemsCriticalIngestion, + scheduleRubyGemsDependentsIngestion, + scheduleRubyGemsIngestion, +} from '../rubygems/schedule' import { svc } from '../service' setImmediate(async () => { await svc.init() await scheduleRubyGemsIngestion() await scheduleRubyGemsCriticalIngestion() + await scheduleRubyGemsDependentsIngestion() await svc.start() }) diff --git a/services/apps/packages_worker/src/config.ts b/services/apps/packages_worker/src/config.ts index 6b587381cf..3df03e0101 100644 --- a/services/apps/packages_worker/src/config.ts +++ b/services/apps/packages_worker/src/config.ts @@ -102,6 +102,13 @@ export function getRubyGemsCriticalConfig() { } } +export function getRubyGemsDependentsConfig() { + return { + batchSize: parseInt(process.env.RUBYGEMS_DEPENDENTS_BATCH_SIZE ?? '50000', 10), + concurrency: parseInt(process.env.RUBYGEMS_DEPENDENTS_CONCURRENCY ?? '50', 10), + } +} + export function getDockerhubConfig() { return { hubBaseUrl: requireEnv('DOCKERHUB_API_BASE_URL'), diff --git a/services/apps/packages_worker/src/rubygems/activities.ts b/services/apps/packages_worker/src/rubygems/activities.ts index 4dd4ba1f26..3623a076c1 100644 --- a/services/apps/packages_worker/src/rubygems/activities.ts +++ b/services/apps/packages_worker/src/rubygems/activities.ts @@ -1,10 +1,18 @@ import { getServiceChildLogger } from '@crowd/logging' -import { getRubyGemsConfig, getRubyGemsCriticalConfig } from '../config' +import { + getRubyGemsConfig, + getRubyGemsCriticalConfig, + getRubyGemsDependentsConfig, +} from '../config' import { getPackagesDb } from '../db' import { processBatch as processCoreBatch } from './runRubyGemsCoreLoop' import { processBatch as processCriticalBatch } from './runRubyGemsCriticalLoop' +import { + DependentsBatchResult, + processBatch as processDependentsBatch, +} from './runRubyGemsDependentsLoop' import { BatchResult } from './types' const log = getServiceChildLogger('rubygems-activity') @@ -25,3 +33,11 @@ export async function processRubyGemsCriticalBatch(): Promise { log.info({ ...result }, 'RubyGems critical batch complete') return result } + +export async function processRubyGemsDependentsBatch(): Promise { + const config = getRubyGemsDependentsConfig() + const qx = await getPackagesDb() + const result = await processDependentsBatch(qx, config) + log.info({ ...result }, 'RubyGems dependents batch complete') + return result +} diff --git a/services/apps/packages_worker/src/rubygems/client.ts b/services/apps/packages_worker/src/rubygems/client.ts index bb613a4267..5d21528a57 100644 --- a/services/apps/packages_worker/src/rubygems/client.ts +++ b/services/apps/packages_worker/src/rubygems/client.ts @@ -59,3 +59,19 @@ export async function fetchOwners(name: string): Promise> { + try { + const resp = await axios.get( + `https://rubygems.org/api/v1/gems/${encodeURIComponent(name)}/reverse_dependencies.json`, + { timeout: 15000 }, + ) + return resp.data + } catch (err) { + const classified = classifyError(err) + if (classified) return classified + throw err + } +} diff --git a/services/apps/packages_worker/src/rubygems/runRubyGemsDependentsLoop.ts b/services/apps/packages_worker/src/rubygems/runRubyGemsDependentsLoop.ts new file mode 100644 index 0000000000..09cafb5b56 --- /dev/null +++ b/services/apps/packages_worker/src/rubygems/runRubyGemsDependentsLoop.ts @@ -0,0 +1,81 @@ +import { + QueryExecutor, + RubyGemsPackageForDependents, + listRubyGemsPackagesForDependents, + updateRubyGemsDependentCount, +} from '@crowd/data-access-layer' +import { getServiceChildLogger } from '@crowd/logging' + +import { fetchReverseDependencies } from './client' +import { isRubyGemsFetchError } from './types' + +const log = getServiceChildLogger('rubygems-dependents') + +export type RubyGemsDependentsConfig = { + batchSize: number + concurrency: number +} + +export type DependentsBatchResult = { + processed: number + notFound: number + error: number +} + +type PackageStatus = 'processed' | 'notFound' | 'error' + +async function processPackage( + qx: QueryExecutor, + pkg: RubyGemsPackageForDependents, +): Promise { + const result = await fetchReverseDependencies(pkg.name) + + if (isRubyGemsFetchError(result)) { + if (result.kind === 'NOT_FOUND') { + await updateRubyGemsDependentCount(qx, pkg.id, 0) + return 'notFound' + } + log.warn({ name: pkg.name }, 'Rate limited fetching reverse dependencies — will retry') + return 'error' + } + + await updateRubyGemsDependentCount(qx, pkg.id, result.length) + return 'processed' +} + +export async function processBatch( + qx: QueryExecutor, + config: RubyGemsDependentsConfig, +): Promise { + const packages = await listRubyGemsPackagesForDependents(qx, { limit: config.batchSize }) + + if (packages.length === 0) return { processed: 0, notFound: 0, error: 0 } + + log.info({ count: packages.length }, 'Dependents batch started') + + const counts: DependentsBatchResult = { processed: 0, notFound: 0, error: 0 } + + for (let batchStart = 0; batchStart < packages.length; batchStart += config.concurrency) { + const group = packages.slice(batchStart, batchStart + config.concurrency) + + await Promise.all( + group.map(async (pkg) => { + try { + const status = await processPackage(qx, pkg) + counts[status]++ + } catch (err) { + const message = err instanceof Error ? err.message : String(err) + log.error({ name: pkg.name, error: message }, 'Unexpected error fetching dependents') + counts.error++ + } + }), + ) + + const done = batchStart + group.length + if (done % 1000 === 0 || done === packages.length) { + log.info({ done, total: packages.length, ...counts }, 'Dependents progress') + } + } + + return counts +} diff --git a/services/apps/packages_worker/src/rubygems/schedule.ts b/services/apps/packages_worker/src/rubygems/schedule.ts index 1d06224a3c..4567ad6bcd 100644 --- a/services/apps/packages_worker/src/rubygems/schedule.ts +++ b/services/apps/packages_worker/src/rubygems/schedule.ts @@ -1,7 +1,11 @@ import { ScheduleAlreadyRunning, ScheduleOverlapPolicy } from '@temporalio/client' import { svc } from '../service' -import { ingestRubyGemsCriticalDetails, ingestRubyGemsPackages } from '../workflows' +import { + ingestRubyGemsCriticalDetails, + ingestRubyGemsDependents, + ingestRubyGemsPackages, +} from '../workflows' export async function scheduleRubyGemsIngestion(): Promise { const { temporal } = svc @@ -76,3 +80,40 @@ export async function scheduleRubyGemsCriticalIngestion(): Promise { } } } + +export async function scheduleRubyGemsDependentsIngestion(): Promise { + const { temporal } = svc + if (!temporal) throw new Error('Temporal client not initialized') + + try { + await temporal.schedule.create({ + scheduleId: 'rubygems-dependents-ingest', + spec: { + cronExpressions: ['0 3 * * 0'], + }, + policies: { + overlap: ScheduleOverlapPolicy.SKIP, + catchupWindow: '1 hour', + }, + action: { + type: 'startWorkflow', + workflowType: ingestRubyGemsDependents, + workflowId: 'rubygems-weekly-dependents', + taskQueue: 'rubygems-worker', + workflowRunTimeout: '24 hours', + retry: { + initialInterval: '30 seconds', + backoffCoefficient: 2, + maximumAttempts: 5, + }, + args: [], + }, + }) + } catch (err) { + if (err instanceof ScheduleAlreadyRunning) { + svc.log.info('Schedule rubygems-dependents-ingest already exists, skipping creation.') + } else { + throw err + } + } +} diff --git a/services/apps/packages_worker/src/rubygems/workflows.ts b/services/apps/packages_worker/src/rubygems/workflows.ts index 80fa5e5b21..c52afbffe3 100644 --- a/services/apps/packages_worker/src/rubygems/workflows.ts +++ b/services/apps/packages_worker/src/rubygems/workflows.ts @@ -24,3 +24,12 @@ export async function ingestRubyGemsCriticalDetails(): Promise { } await continueAsNew() } + +export async function ingestRubyGemsDependents(): Promise { + const result = await acts.processRubyGemsDependentsBatch() + if (result.processed + result.notFound === 0) { + log.info('RubyGems dependents ingestion complete — no more progress, exiting.', { ...result }) + return + } + await continueAsNew() +} diff --git a/services/apps/packages_worker/src/workflows/index.ts b/services/apps/packages_worker/src/workflows/index.ts index a14aee8ed0..465047c39c 100644 --- a/services/apps/packages_worker/src/workflows/index.ts +++ b/services/apps/packages_worker/src/workflows/index.ts @@ -25,7 +25,11 @@ export { ingestPypiDownloadsDaily, } from '../pypi/downloads/ingestPypiDownloads' export { ingestNuGetPackages } from '../nuget/workflows' -export { ingestRubyGemsCriticalDetails, ingestRubyGemsPackages } from '../rubygems/workflows' +export { + ingestRubyGemsCriticalDetails, + ingestRubyGemsDependents, + ingestRubyGemsPackages, +} from '../rubygems/workflows' export { ingestSecurityContacts, ingestSecurityContactsForPurlWorkflow, diff --git a/services/libs/data-access-layer/src/osspckgs/rubygems.ts b/services/libs/data-access-layer/src/osspckgs/rubygems.ts index f29921ae80..c54425a91a 100644 --- a/services/libs/data-access-layer/src/osspckgs/rubygems.ts +++ b/services/libs/data-access-layer/src/osspckgs/rubygems.ts @@ -66,3 +66,39 @@ export async function listRubyGemsCriticalPackagesToSync( { limit }, ) } + +export type RubyGemsPackageForDependents = { + id: number + name: string +} + +export async function listRubyGemsPackagesForDependents( + qx: QueryExecutor, + options: { limit: number }, +): Promise { + const { limit } = options + return qx.select( + ` + SELECT p.id, p.name + FROM packages p + WHERE p.ecosystem = 'rubygems' + AND p.dependent_count IS NULL + ORDER BY p.id ASC + LIMIT $(limit) + `, + { limit }, + ) +} + +export async function updateRubyGemsDependentCount( + qx: QueryExecutor, + packageId: number, + dependentCount: number, +): Promise { + await qx.result( + `UPDATE packages + SET dependent_count = $(dependentCount) + WHERE id = $(packageId)`, + { packageId, dependentCount }, + ) +} From 9d24e807b1dc8851b38edab9c5c9dcbb1ee8abdd Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Fri, 10 Jul 2026 12:00:47 +0100 Subject: [PATCH 12/17] fix: use proxies Signed-off-by: Mouad BANI --- .../packages_worker/src/rubygems/client.ts | 26 ++++++--- .../packages_worker/src/rubygems/proxies.ts | 11 ++++ .../src/rubygems/runRubyGemsDependentsLoop.ts | 55 ++++++++++++------- 3 files changed, 62 insertions(+), 30 deletions(-) create mode 100644 services/apps/packages_worker/src/rubygems/proxies.ts diff --git a/services/apps/packages_worker/src/rubygems/client.ts b/services/apps/packages_worker/src/rubygems/client.ts index 5d21528a57..434ee03105 100644 --- a/services/apps/packages_worker/src/rubygems/client.ts +++ b/services/apps/packages_worker/src/rubygems/client.ts @@ -1,4 +1,5 @@ import axios from 'axios' +import type { Dispatcher } from 'undici' import { RubyGemsFetchError, @@ -62,16 +63,23 @@ export async function fetchOwners(name: string): Promise> { + const url = `https://rubygems.org/api/v1/gems/${encodeURIComponent(name)}/reverse_dependencies.json` + const abort = new AbortController() + const timer = setTimeout(() => abort.abort(), 15000) + let res: Response try { - const resp = await axios.get( - `https://rubygems.org/api/v1/gems/${encodeURIComponent(name)}/reverse_dependencies.json`, - { timeout: 15000 }, - ) - return resp.data - } catch (err) { - const classified = classifyError(err) - if (classified) return classified - throw err + const init: RequestInit & { dispatcher?: Dispatcher } = { signal: abort.signal } + if (dispatcher) init.dispatcher = dispatcher + res = await fetch(url, init as RequestInit) + } finally { + clearTimeout(timer) } + + if (res.status === 404) return { kind: 'NOT_FOUND', status: 404, message: `${name} not found` } + if (res.status === 429) return { kind: 'RATE_LIMIT', status: 429, message: 'rate limited' } + if (!res.ok) throw new Error(`RubyGems reverse_dependencies HTTP ${res.status} for ${name}`) + + return (await res.json()) as string[] } diff --git a/services/apps/packages_worker/src/rubygems/proxies.ts b/services/apps/packages_worker/src/rubygems/proxies.ts new file mode 100644 index 0000000000..d5fba1ab9d --- /dev/null +++ b/services/apps/packages_worker/src/rubygems/proxies.ts @@ -0,0 +1,11 @@ +import { parseProxies } from '../proxies' + +export function rubyGemsProxiesEnabled(): boolean { + const raw = (process.env.CROWD_PACKAGES_PROXIES_ENABLED ?? '').trim().toLowerCase() + return raw === 'true' || raw === '1' +} + +export function rubyGemsProxyUrls(): string[] { + if (!rubyGemsProxiesEnabled()) return [] + return parseProxies().map((p) => `http://${p.username}:${p.password}@${p.host}:${p.port}`) +} diff --git a/services/apps/packages_worker/src/rubygems/runRubyGemsDependentsLoop.ts b/services/apps/packages_worker/src/rubygems/runRubyGemsDependentsLoop.ts index 09cafb5b56..da357c89e9 100644 --- a/services/apps/packages_worker/src/rubygems/runRubyGemsDependentsLoop.ts +++ b/services/apps/packages_worker/src/rubygems/runRubyGemsDependentsLoop.ts @@ -5,8 +5,10 @@ import { updateRubyGemsDependentCount, } from '@crowd/data-access-layer' import { getServiceChildLogger } from '@crowd/logging' +import { type Dispatcher, ProxyAgent } from 'undici' import { fetchReverseDependencies } from './client' +import { rubyGemsProxyUrls } from './proxies' import { isRubyGemsFetchError } from './types' const log = getServiceChildLogger('rubygems-dependents') @@ -27,8 +29,9 @@ type PackageStatus = 'processed' | 'notFound' | 'error' async function processPackage( qx: QueryExecutor, pkg: RubyGemsPackageForDependents, + dispatcher?: Dispatcher, ): Promise { - const result = await fetchReverseDependencies(pkg.name) + const result = await fetchReverseDependencies(pkg.name, dispatcher) if (isRubyGemsFetchError(result)) { if (result.kind === 'NOT_FOUND') { @@ -51,30 +54,40 @@ export async function processBatch( if (packages.length === 0) return { processed: 0, notFound: 0, error: 0 } - log.info({ count: packages.length }, 'Dependents batch started') + const proxyUrls = rubyGemsProxyUrls() + const dispatchers: Array = proxyUrls.length + ? proxyUrls.map((url) => new ProxyAgent(url)) + : [undefined] + + log.info({ count: packages.length, proxies: proxyUrls.length }, 'Dependents batch started') const counts: DependentsBatchResult = { processed: 0, notFound: 0, error: 0 } - for (let batchStart = 0; batchStart < packages.length; batchStart += config.concurrency) { - const group = packages.slice(batchStart, batchStart + config.concurrency) - - await Promise.all( - group.map(async (pkg) => { - try { - const status = await processPackage(qx, pkg) - counts[status]++ - } catch (err) { - const message = err instanceof Error ? err.message : String(err) - log.error({ name: pkg.name, error: message }, 'Unexpected error fetching dependents') - counts.error++ - } - }), - ) - - const done = batchStart + group.length - if (done % 1000 === 0 || done === packages.length) { - log.info({ done, total: packages.length, ...counts }, 'Dependents progress') + try { + for (let batchStart = 0; batchStart < packages.length; batchStart += config.concurrency) { + const group = packages.slice(batchStart, batchStart + config.concurrency) + + await Promise.all( + group.map(async (pkg, i) => { + const dispatcher = dispatchers[(batchStart + i) % dispatchers.length] + try { + const status = await processPackage(qx, pkg, dispatcher) + counts[status]++ + } catch (err) { + const message = err instanceof Error ? err.message : String(err) + log.error({ name: pkg.name, error: message }, 'Unexpected error fetching dependents') + counts.error++ + } + }), + ) + + const done = batchStart + group.length + if (done % 1000 === 0 || done === packages.length) { + log.info({ done, total: packages.length, ...counts }, 'Dependents progress') + } } + } finally { + await Promise.all(dispatchers.map((d) => d?.close())) } return counts From 4c366d768d598c023de3536e6b57846267cc7d2d Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Fri, 10 Jul 2026 12:25:50 +0100 Subject: [PATCH 13/17] fix: undo proxies and reduce concurrency Signed-off-by: Mouad BANI --- .../packages_worker/src/rubygems/client.ts | 26 +++------ .../packages_worker/src/rubygems/proxies.ts | 11 ---- .../src/rubygems/runRubyGemsDependentsLoop.ts | 55 +++++++------------ 3 files changed, 30 insertions(+), 62 deletions(-) delete mode 100644 services/apps/packages_worker/src/rubygems/proxies.ts diff --git a/services/apps/packages_worker/src/rubygems/client.ts b/services/apps/packages_worker/src/rubygems/client.ts index 434ee03105..5d21528a57 100644 --- a/services/apps/packages_worker/src/rubygems/client.ts +++ b/services/apps/packages_worker/src/rubygems/client.ts @@ -1,5 +1,4 @@ import axios from 'axios' -import type { Dispatcher } from 'undici' import { RubyGemsFetchError, @@ -63,23 +62,16 @@ export async function fetchOwners(name: string): Promise> { - const url = `https://rubygems.org/api/v1/gems/${encodeURIComponent(name)}/reverse_dependencies.json` - const abort = new AbortController() - const timer = setTimeout(() => abort.abort(), 15000) - let res: Response try { - const init: RequestInit & { dispatcher?: Dispatcher } = { signal: abort.signal } - if (dispatcher) init.dispatcher = dispatcher - res = await fetch(url, init as RequestInit) - } finally { - clearTimeout(timer) + const resp = await axios.get( + `https://rubygems.org/api/v1/gems/${encodeURIComponent(name)}/reverse_dependencies.json`, + { timeout: 15000 }, + ) + return resp.data + } catch (err) { + const classified = classifyError(err) + if (classified) return classified + throw err } - - if (res.status === 404) return { kind: 'NOT_FOUND', status: 404, message: `${name} not found` } - if (res.status === 429) return { kind: 'RATE_LIMIT', status: 429, message: 'rate limited' } - if (!res.ok) throw new Error(`RubyGems reverse_dependencies HTTP ${res.status} for ${name}`) - - return (await res.json()) as string[] } diff --git a/services/apps/packages_worker/src/rubygems/proxies.ts b/services/apps/packages_worker/src/rubygems/proxies.ts deleted file mode 100644 index d5fba1ab9d..0000000000 --- a/services/apps/packages_worker/src/rubygems/proxies.ts +++ /dev/null @@ -1,11 +0,0 @@ -import { parseProxies } from '../proxies' - -export function rubyGemsProxiesEnabled(): boolean { - const raw = (process.env.CROWD_PACKAGES_PROXIES_ENABLED ?? '').trim().toLowerCase() - return raw === 'true' || raw === '1' -} - -export function rubyGemsProxyUrls(): string[] { - if (!rubyGemsProxiesEnabled()) return [] - return parseProxies().map((p) => `http://${p.username}:${p.password}@${p.host}:${p.port}`) -} diff --git a/services/apps/packages_worker/src/rubygems/runRubyGemsDependentsLoop.ts b/services/apps/packages_worker/src/rubygems/runRubyGemsDependentsLoop.ts index da357c89e9..09cafb5b56 100644 --- a/services/apps/packages_worker/src/rubygems/runRubyGemsDependentsLoop.ts +++ b/services/apps/packages_worker/src/rubygems/runRubyGemsDependentsLoop.ts @@ -5,10 +5,8 @@ import { updateRubyGemsDependentCount, } from '@crowd/data-access-layer' import { getServiceChildLogger } from '@crowd/logging' -import { type Dispatcher, ProxyAgent } from 'undici' import { fetchReverseDependencies } from './client' -import { rubyGemsProxyUrls } from './proxies' import { isRubyGemsFetchError } from './types' const log = getServiceChildLogger('rubygems-dependents') @@ -29,9 +27,8 @@ type PackageStatus = 'processed' | 'notFound' | 'error' async function processPackage( qx: QueryExecutor, pkg: RubyGemsPackageForDependents, - dispatcher?: Dispatcher, ): Promise { - const result = await fetchReverseDependencies(pkg.name, dispatcher) + const result = await fetchReverseDependencies(pkg.name) if (isRubyGemsFetchError(result)) { if (result.kind === 'NOT_FOUND') { @@ -54,40 +51,30 @@ export async function processBatch( if (packages.length === 0) return { processed: 0, notFound: 0, error: 0 } - const proxyUrls = rubyGemsProxyUrls() - const dispatchers: Array = proxyUrls.length - ? proxyUrls.map((url) => new ProxyAgent(url)) - : [undefined] - - log.info({ count: packages.length, proxies: proxyUrls.length }, 'Dependents batch started') + log.info({ count: packages.length }, 'Dependents batch started') const counts: DependentsBatchResult = { processed: 0, notFound: 0, error: 0 } - try { - for (let batchStart = 0; batchStart < packages.length; batchStart += config.concurrency) { - const group = packages.slice(batchStart, batchStart + config.concurrency) - - await Promise.all( - group.map(async (pkg, i) => { - const dispatcher = dispatchers[(batchStart + i) % dispatchers.length] - try { - const status = await processPackage(qx, pkg, dispatcher) - counts[status]++ - } catch (err) { - const message = err instanceof Error ? err.message : String(err) - log.error({ name: pkg.name, error: message }, 'Unexpected error fetching dependents') - counts.error++ - } - }), - ) - - const done = batchStart + group.length - if (done % 1000 === 0 || done === packages.length) { - log.info({ done, total: packages.length, ...counts }, 'Dependents progress') - } + for (let batchStart = 0; batchStart < packages.length; batchStart += config.concurrency) { + const group = packages.slice(batchStart, batchStart + config.concurrency) + + await Promise.all( + group.map(async (pkg) => { + try { + const status = await processPackage(qx, pkg) + counts[status]++ + } catch (err) { + const message = err instanceof Error ? err.message : String(err) + log.error({ name: pkg.name, error: message }, 'Unexpected error fetching dependents') + counts.error++ + } + }), + ) + + const done = batchStart + group.length + if (done % 1000 === 0 || done === packages.length) { + log.info({ done, total: packages.length, ...counts }, 'Dependents progress') } - } finally { - await Promise.all(dispatchers.map((d) => d?.close())) } return counts From 729e8837d1ec9816ceb3b218e8365cddcf8544e6 Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Fri, 10 Jul 2026 17:29:07 +0100 Subject: [PATCH 14/17] fix: code review suggestions Signed-off-by: Mouad BANI --- .../__tests__/rubygemsDal.integration.test.ts | 130 ++++++++++++++++++ .../src/rubygems/activities.ts | 10 +- .../src/rubygems/runRubyGemsCoreLoop.ts | 1 - .../src/rubygems/runRubyGemsCriticalLoop.ts | 19 ++- .../src/rubygems/runRubyGemsDependentsLoop.ts | 14 +- .../packages_worker/src/rubygems/workflows.ts | 18 +-- .../src/osspckgs/rubygems.ts | 30 ++-- .../data-access-layer/src/osspckgs/types.ts | 1 + .../src/osspckgs/versions.ts | 17 ++- 9 files changed, 194 insertions(+), 46 deletions(-) create mode 100644 services/apps/packages_worker/src/rubygems/__tests__/rubygemsDal.integration.test.ts diff --git a/services/apps/packages_worker/src/rubygems/__tests__/rubygemsDal.integration.test.ts b/services/apps/packages_worker/src/rubygems/__tests__/rubygemsDal.integration.test.ts new file mode 100644 index 0000000000..1d16cbe48d --- /dev/null +++ b/services/apps/packages_worker/src/rubygems/__tests__/rubygemsDal.integration.test.ts @@ -0,0 +1,130 @@ +import { afterAll, beforeAll, beforeEach, describe, expect, it } from 'vitest' + +import { + listRubyGemsCriticalPackagesToSync, + listRubyGemsPackagesToSync, +} from '@crowd/data-access-layer/src/osspckgs/rubygems' +import { QueryExecutor, pgpQx } from '@crowd/data-access-layer/src/queryExecutor' +import { getDbConnection } from '@crowd/database' + +// Integration test: hits the running packages-db. Skipped automatically when +// any of the DB env vars are missing so unit-test runs in CI stay green. +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_PURL_PREFIX = 'pkg:gem/crowd-dev-test-rubygems-fixture-' + +function daysAgo(days: number): Date { + const d = new Date() + d.setDate(d.getDate() - days) + return d +} + +async function cleanupFixtures(qx: QueryExecutor): Promise { + await qx.result( + `DELETE FROM versions WHERE package_id IN (SELECT id FROM packages WHERE purl LIKE $(prefix))`, + { prefix: `${FIXTURE_PURL_PREFIX}%` }, + ) + await qx.result(`DELETE FROM packages WHERE purl LIKE $(prefix)`, { + prefix: `${FIXTURE_PURL_PREFIX}%`, + }) +} + +async function insertPackage( + qx: QueryExecutor, + name: string, + overrides: { + isCritical?: boolean + ingestionSource?: string | null + lastSyncedAt?: Date + } = {}, +): Promise { + const purl = `${FIXTURE_PURL_PREFIX}${name}` + const row = await qx.selectOne( + `INSERT INTO packages (purl, ecosystem, name, is_critical, ingestion_source, last_synced_at, created_at) + VALUES ($(purl), 'rubygems', $(name), $(isCritical), $(ingestionSource), $(lastSyncedAt), NOW()) + RETURNING id`, + { + purl, + name, + isCritical: overrides.isCritical ?? false, + ingestionSource: overrides.ingestionSource ?? null, + lastSyncedAt: (overrides.lastSyncedAt ?? new Date()).toISOString(), + }, + ) + return row.id as number +} + +describe.skipIf(!HAVE_DB)('rubygems DAL — real packages-db', () => { + let qx: QueryExecutor + + 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) + }, 30_000) + + beforeEach(async () => { + await cleanupFixtures(qx) + }) + + afterAll(async () => { + if (qx) await cleanupFixtures(qx) + }) + + describe('listRubyGemsPackagesToSync', () => { + it('returns never-synced and stale packages, excludes fresh ones', async () => { + await insertPackage(qx, 'never-synced', { ingestionSource: null }) + await insertPackage(qx, 'fresh', { + ingestionSource: 'rubygems-registry', + lastSyncedAt: new Date(), + }) + await insertPackage(qx, 'stale', { + ingestionSource: 'rubygems-registry', + lastSyncedAt: daysAgo(2), + }) + + const rows = await listRubyGemsPackagesToSync(qx, { limit: 100 }) + const names = rows.map((r) => r.name) + + expect(names).toContain('never-synced') + expect(names).toContain('stale') + expect(names).not.toContain('fresh') + }) + }) + + describe('listRubyGemsCriticalPackagesToSync', () => { + it('returns is_critical rubygems packages and excludes non-critical', async () => { + await insertPackage(qx, 'critical-a', { isCritical: true, ingestionSource: 'rubygems-registry' }) + await insertPackage(qx, 'critical-b', { isCritical: true, ingestionSource: null }) + await insertPackage(qx, 'non-critical', { isCritical: false, ingestionSource: null }) + + const rows = await listRubyGemsCriticalPackagesToSync(qx, { limit: 100 }) + const names = rows.map((r) => r.name) + + expect(names).toContain('critical-a') + expect(names).toContain('critical-b') + expect(names).not.toContain('non-critical') + }) + + it('paginates by id cursor via afterId', async () => { + const firstId = await insertPackage(qx, 'critical-cursor-1', { isCritical: true }) + await insertPackage(qx, 'critical-cursor-2', { isCritical: true }) + + const rows = await listRubyGemsCriticalPackagesToSync(qx, { limit: 100, afterId: firstId }) + const ids = rows.map((r) => r.id) + + expect(ids.every((id) => id > firstId)).toBe(true) + expect(rows.map((r) => r.name)).not.toContain('critical-cursor-1') + }) + }) +}) diff --git a/services/apps/packages_worker/src/rubygems/activities.ts b/services/apps/packages_worker/src/rubygems/activities.ts index 3623a076c1..9e97f1a85d 100644 --- a/services/apps/packages_worker/src/rubygems/activities.ts +++ b/services/apps/packages_worker/src/rubygems/activities.ts @@ -26,18 +26,20 @@ export async function processRubyGemsCoreBatch(): Promise { return result } -export async function processRubyGemsCriticalBatch(): Promise { +export async function processRubyGemsCriticalBatch( + afterId = 0, +): Promise { const config = getRubyGemsCriticalConfig() const qx = await getPackagesDb() - const result = await processCriticalBatch(qx, config) + const result = await processCriticalBatch(qx, config, afterId) log.info({ ...result }, 'RubyGems critical batch complete') return result } -export async function processRubyGemsDependentsBatch(): Promise { +export async function processRubyGemsDependentsBatch(afterId = 0): Promise { const config = getRubyGemsDependentsConfig() const qx = await getPackagesDb() - const result = await processDependentsBatch(qx, config) + const result = await processDependentsBatch(qx, config, afterId) log.info({ ...result }, 'RubyGems dependents batch complete') return result } diff --git a/services/apps/packages_worker/src/rubygems/runRubyGemsCoreLoop.ts b/services/apps/packages_worker/src/rubygems/runRubyGemsCoreLoop.ts index 9176b8faa7..89ccef94dc 100644 --- a/services/apps/packages_worker/src/rubygems/runRubyGemsCoreLoop.ts +++ b/services/apps/packages_worker/src/rubygems/runRubyGemsCoreLoop.ts @@ -87,7 +87,6 @@ async function processPackage( } if (gemResult.kind === 'RATE_LIMIT') { log.warn({ purl: pkg.purl }, 'Rate limited by RubyGems registry — will retry next pass') - await markPackageError(qx, pkg) return 'error' } throw new Error(`Transient error fetching ${pkg.purl}: ${gemResult.message}`) diff --git a/services/apps/packages_worker/src/rubygems/runRubyGemsCriticalLoop.ts b/services/apps/packages_worker/src/rubygems/runRubyGemsCriticalLoop.ts index f148a55656..5e71cca388 100644 --- a/services/apps/packages_worker/src/rubygems/runRubyGemsCriticalLoop.ts +++ b/services/apps/packages_worker/src/rubygems/runRubyGemsCriticalLoop.ts @@ -104,6 +104,7 @@ async function processPackage( isLatest: v.number === latest?.number, isPrerelease: v.isPrerelease, license: v.licenses && v.licenses.length > 0 ? v.licenses[0] : null, + publishedAt: v.publishedAt, })) const verChanged = await upsertVersionsBatch(t, versionRows) verChanged.forEach((f) => changed.add(f)) @@ -146,12 +147,18 @@ async function processPackage( export async function processBatch( qx: QueryExecutor, config: RubyGemsCriticalConfig, -): Promise { - const packages = await listRubyGemsCriticalPackagesToSync(qx, { limit: config.batchSize }) - - if (packages.length === 0) return { processed: 0, skipped: 0, error: 0, unchanged: 0 } + afterId: number, +): Promise { + const packages = await listRubyGemsCriticalPackagesToSync(qx, { + limit: config.batchSize, + afterId, + }) + + if (packages.length === 0) { + return { processed: 0, skipped: 0, error: 0, unchanged: 0, lastId: null } + } - log.info({ count: packages.length }, 'Critical batch started') + log.info({ count: packages.length, afterId }, 'Critical batch started') const counts = { processed: 0, skipped: 0, error: 0, unchanged: 0 } @@ -184,5 +191,5 @@ export async function processBatch( } } - return counts + return { ...counts, lastId: packages[packages.length - 1].id } } diff --git a/services/apps/packages_worker/src/rubygems/runRubyGemsDependentsLoop.ts b/services/apps/packages_worker/src/rubygems/runRubyGemsDependentsLoop.ts index 09cafb5b56..893376ef93 100644 --- a/services/apps/packages_worker/src/rubygems/runRubyGemsDependentsLoop.ts +++ b/services/apps/packages_worker/src/rubygems/runRubyGemsDependentsLoop.ts @@ -20,6 +20,7 @@ export type DependentsBatchResult = { processed: number notFound: number error: number + lastId: number | null } type PackageStatus = 'processed' | 'notFound' | 'error' @@ -46,14 +47,18 @@ async function processPackage( export async function processBatch( qx: QueryExecutor, config: RubyGemsDependentsConfig, + afterId: number, ): Promise { - const packages = await listRubyGemsPackagesForDependents(qx, { limit: config.batchSize }) + const packages = await listRubyGemsPackagesForDependents(qx, { + limit: config.batchSize, + afterId, + }) - if (packages.length === 0) return { processed: 0, notFound: 0, error: 0 } + if (packages.length === 0) return { processed: 0, notFound: 0, error: 0, lastId: null } - log.info({ count: packages.length }, 'Dependents batch started') + log.info({ count: packages.length, afterId }, 'Dependents batch started') - const counts: DependentsBatchResult = { processed: 0, notFound: 0, error: 0 } + const counts: DependentsBatchResult = { processed: 0, notFound: 0, error: 0, lastId: null } for (let batchStart = 0; batchStart < packages.length; batchStart += config.concurrency) { const group = packages.slice(batchStart, batchStart + config.concurrency) @@ -77,5 +82,6 @@ export async function processBatch( } } + counts.lastId = packages[packages.length - 1].id return counts } diff --git a/services/apps/packages_worker/src/rubygems/workflows.ts b/services/apps/packages_worker/src/rubygems/workflows.ts index c52afbffe3..086c1d7a50 100644 --- a/services/apps/packages_worker/src/rubygems/workflows.ts +++ b/services/apps/packages_worker/src/rubygems/workflows.ts @@ -16,20 +16,20 @@ export async function ingestRubyGemsPackages(): Promise { await continueAsNew() } -export async function ingestRubyGemsCriticalDetails(): Promise { - const result = await acts.processRubyGemsCriticalBatch() - if (result.processed + result.skipped + result.error + result.unchanged === 0) { +export async function ingestRubyGemsCriticalDetails(afterId = 0): Promise { + const result = await acts.processRubyGemsCriticalBatch(afterId) + if (result.lastId === null) { log.info('RubyGems critical ingestion complete — no more work, exiting.', { ...result }) return } - await continueAsNew() + await continueAsNew(result.lastId) } -export async function ingestRubyGemsDependents(): Promise { - const result = await acts.processRubyGemsDependentsBatch() - if (result.processed + result.notFound === 0) { - log.info('RubyGems dependents ingestion complete — no more progress, exiting.', { ...result }) +export async function ingestRubyGemsDependents(afterId = 0): Promise { + const result = await acts.processRubyGemsDependentsBatch(afterId) + if (result.lastId === null) { + log.info('RubyGems dependents ingestion complete — no more work, exiting.', { ...result }) return } - await continueAsNew() + await continueAsNew(result.lastId) } diff --git a/services/libs/data-access-layer/src/osspckgs/rubygems.ts b/services/libs/data-access-layer/src/osspckgs/rubygems.ts index c54425a91a..0f601177b7 100644 --- a/services/libs/data-access-layer/src/osspckgs/rubygems.ts +++ b/services/libs/data-access-layer/src/osspckgs/rubygems.ts @@ -27,7 +27,7 @@ export async function listRubyGemsPackagesToSync( AND ( p.ingestion_source IS NULL OR p.ingestion_source <> ALL($(workerOutcomes)::text[]) - OR p.last_synced_at < NOW() - INTERVAL '1 day' + OR p.last_synced_at < date_trunc('day', NOW()) ) ORDER BY p.dependent_count DESC NULLS LAST, @@ -46,24 +46,20 @@ export type RubyGemsCriticalPackageToSync = { export async function listRubyGemsCriticalPackagesToSync( qx: QueryExecutor, - options: { limit: number }, + options: { limit: number; afterId?: number }, ): Promise { - const { limit } = options + const { limit, afterId = 0 } = options return qx.select( ` SELECT p.id, p.purl, p.name FROM packages p - LEFT JOIN LATERAL ( - SELECT MAX(v.last_synced_at) AS last_synced - FROM versions v WHERE v.package_id = p.id - ) v ON true WHERE p.ecosystem = 'rubygems' AND p.is_critical - AND (v.last_synced IS NULL OR v.last_synced < NOW() - INTERVAL '1 day') - ORDER BY p.dependent_count DESC NULLS LAST, p.id ASC + AND p.id > $(afterId) + ORDER BY p.id ASC LIMIT $(limit) `, - { limit }, + { limit, afterId }, ) } @@ -74,19 +70,19 @@ export type RubyGemsPackageForDependents = { export async function listRubyGemsPackagesForDependents( qx: QueryExecutor, - options: { limit: number }, + options: { limit: number; afterId?: number }, ): Promise { - const { limit } = options + const { limit, afterId = 0 } = options return qx.select( ` SELECT p.id, p.name FROM packages p WHERE p.ecosystem = 'rubygems' - AND p.dependent_count IS NULL + AND p.id > $(afterId) ORDER BY p.id ASC LIMIT $(limit) `, - { limit }, + { limit, afterId }, ) } @@ -97,8 +93,10 @@ export async function updateRubyGemsDependentCount( ): Promise { await qx.result( `UPDATE packages - SET dependent_count = $(dependentCount) - WHERE id = $(packageId)`, + SET dependent_count = $(dependentCount), + last_synced_at = NOW() + WHERE id = $(packageId) + AND dependent_count IS DISTINCT FROM $(dependentCount)`, { packageId, dependentCount }, ) } diff --git a/services/libs/data-access-layer/src/osspckgs/types.ts b/services/libs/data-access-layer/src/osspckgs/types.ts index c5de57a8f1..f0552adc3d 100644 --- a/services/libs/data-access-layer/src/osspckgs/types.ts +++ b/services/libs/data-access-layer/src/osspckgs/types.ts @@ -84,6 +84,7 @@ export type IDbVersionUpsert = { isLatest: boolean isPrerelease: boolean license: string | null + publishedAt?: Date | null } // ─── repos ──────────────────────────────────────────────────────────────────── diff --git a/services/libs/data-access-layer/src/osspckgs/versions.ts b/services/libs/data-access-layer/src/osspckgs/versions.ts index 67d4e1d244..b3f32c8a0c 100644 --- a/services/libs/data-access-layer/src/osspckgs/versions.ts +++ b/services/libs/data-access-layer/src/osspckgs/versions.ts @@ -39,15 +39,16 @@ export async function upsertVersionsBatch( const row: { changed_fields: string[] } = await qx.selectOne( ` WITH old AS ( - SELECT number, is_latest, is_prerelease, licenses + SELECT number, is_latest, is_prerelease, licenses, published_at FROM versions WHERE package_id = $(packageId)::bigint AND number = ANY($(numbers)::text[]) ), ins AS ( - INSERT INTO versions (package_id, ecosystem, namespace, name, number, is_latest, is_prerelease, licenses, last_synced_at, created_at) + INSERT INTO versions (package_id, ecosystem, namespace, name, number, is_latest, is_prerelease, licenses, published_at, last_synced_at, created_at) SELECT $(packageId)::bigint, t.ecosystem, t.namespace, t.name, t.number, t.is_latest, t.is_prerelease, CASE WHEN t.license IS NULL THEN NULL ELSE ARRAY[t.license] END, + t.published_at::timestamptz, NOW(), NOW() FROM UNNEST( $(ecosystems)::text[], @@ -56,21 +57,24 @@ export async function upsertVersionsBatch( $(numbers)::text[], $(isLatests)::bool[], $(isPreleases)::bool[], - $(licenses)::text[] - ) AS t(ecosystem, namespace, name, number, is_latest, is_prerelease, license) + $(licenses)::text[], + $(publishedAts)::text[] + ) AS t(ecosystem, namespace, name, number, is_latest, is_prerelease, license, published_at) ON CONFLICT (package_id, number) DO UPDATE SET namespace = COALESCE(EXCLUDED.namespace, versions.namespace), is_latest = EXCLUDED.is_latest, is_prerelease = EXCLUDED.is_prerelease, licenses = COALESCE(EXCLUDED.licenses, versions.licenses), + published_at = COALESCE(EXCLUDED.published_at, versions.published_at), last_synced_at = NOW() - RETURNING number, is_latest, is_prerelease, licenses + RETURNING number, is_latest, is_prerelease, licenses, published_at ) SELECT array_remove(ARRAY[ CASE WHEN bool_or(o.number IS NULL) THEN 'versions.number' END, CASE WHEN bool_or(o.is_latest IS DISTINCT FROM ins.is_latest) THEN 'versions.is_latest' END, CASE WHEN bool_or(o.is_prerelease IS DISTINCT FROM ins.is_prerelease) THEN 'versions.is_prerelease' END, - CASE WHEN bool_or(o.licenses IS DISTINCT FROM ins.licenses) THEN 'versions.licenses' END + CASE WHEN bool_or(o.licenses IS DISTINCT FROM ins.licenses) THEN 'versions.licenses' END, + CASE WHEN bool_or(o.published_at IS DISTINCT FROM ins.published_at) THEN 'versions.published_at' END ], NULL) AS changed_fields FROM ins LEFT JOIN old o ON o.number = ins.number `, @@ -83,6 +87,7 @@ export async function upsertVersionsBatch( isLatests: versions.map((v) => v.isLatest), isPreleases: versions.map((v) => v.isPrerelease), licenses: versions.map((v) => v.license), + publishedAts: versions.map((v) => (v.publishedAt ? v.publishedAt.toISOString() : null)), }, ) return row.changed_fields From f256e91a0b8749824a7659181ac6874a2748bc97 Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Fri, 10 Jul 2026 17:46:59 +0100 Subject: [PATCH 15/17] chore: remove test Signed-off-by: Mouad BANI --- .../__tests__/rubygemsDal.integration.test.ts | 130 ------------------ .../packages_worker/src/rubygems/client.ts | 99 ++++++------- .../packages_worker/src/rubygems/normalize.ts | 9 +- .../src/rubygems/rateLimiter.ts | 22 +++ .../src/rubygems/runRubyGemsCoreLoop.ts | 6 + 5 files changed, 70 insertions(+), 196 deletions(-) delete mode 100644 services/apps/packages_worker/src/rubygems/__tests__/rubygemsDal.integration.test.ts create mode 100644 services/apps/packages_worker/src/rubygems/rateLimiter.ts diff --git a/services/apps/packages_worker/src/rubygems/__tests__/rubygemsDal.integration.test.ts b/services/apps/packages_worker/src/rubygems/__tests__/rubygemsDal.integration.test.ts deleted file mode 100644 index 1d16cbe48d..0000000000 --- a/services/apps/packages_worker/src/rubygems/__tests__/rubygemsDal.integration.test.ts +++ /dev/null @@ -1,130 +0,0 @@ -import { afterAll, beforeAll, beforeEach, describe, expect, it } from 'vitest' - -import { - listRubyGemsCriticalPackagesToSync, - listRubyGemsPackagesToSync, -} from '@crowd/data-access-layer/src/osspckgs/rubygems' -import { QueryExecutor, pgpQx } from '@crowd/data-access-layer/src/queryExecutor' -import { getDbConnection } from '@crowd/database' - -// Integration test: hits the running packages-db. Skipped automatically when -// any of the DB env vars are missing so unit-test runs in CI stay green. -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_PURL_PREFIX = 'pkg:gem/crowd-dev-test-rubygems-fixture-' - -function daysAgo(days: number): Date { - const d = new Date() - d.setDate(d.getDate() - days) - return d -} - -async function cleanupFixtures(qx: QueryExecutor): Promise { - await qx.result( - `DELETE FROM versions WHERE package_id IN (SELECT id FROM packages WHERE purl LIKE $(prefix))`, - { prefix: `${FIXTURE_PURL_PREFIX}%` }, - ) - await qx.result(`DELETE FROM packages WHERE purl LIKE $(prefix)`, { - prefix: `${FIXTURE_PURL_PREFIX}%`, - }) -} - -async function insertPackage( - qx: QueryExecutor, - name: string, - overrides: { - isCritical?: boolean - ingestionSource?: string | null - lastSyncedAt?: Date - } = {}, -): Promise { - const purl = `${FIXTURE_PURL_PREFIX}${name}` - const row = await qx.selectOne( - `INSERT INTO packages (purl, ecosystem, name, is_critical, ingestion_source, last_synced_at, created_at) - VALUES ($(purl), 'rubygems', $(name), $(isCritical), $(ingestionSource), $(lastSyncedAt), NOW()) - RETURNING id`, - { - purl, - name, - isCritical: overrides.isCritical ?? false, - ingestionSource: overrides.ingestionSource ?? null, - lastSyncedAt: (overrides.lastSyncedAt ?? new Date()).toISOString(), - }, - ) - return row.id as number -} - -describe.skipIf(!HAVE_DB)('rubygems DAL — real packages-db', () => { - let qx: QueryExecutor - - 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) - }, 30_000) - - beforeEach(async () => { - await cleanupFixtures(qx) - }) - - afterAll(async () => { - if (qx) await cleanupFixtures(qx) - }) - - describe('listRubyGemsPackagesToSync', () => { - it('returns never-synced and stale packages, excludes fresh ones', async () => { - await insertPackage(qx, 'never-synced', { ingestionSource: null }) - await insertPackage(qx, 'fresh', { - ingestionSource: 'rubygems-registry', - lastSyncedAt: new Date(), - }) - await insertPackage(qx, 'stale', { - ingestionSource: 'rubygems-registry', - lastSyncedAt: daysAgo(2), - }) - - const rows = await listRubyGemsPackagesToSync(qx, { limit: 100 }) - const names = rows.map((r) => r.name) - - expect(names).toContain('never-synced') - expect(names).toContain('stale') - expect(names).not.toContain('fresh') - }) - }) - - describe('listRubyGemsCriticalPackagesToSync', () => { - it('returns is_critical rubygems packages and excludes non-critical', async () => { - await insertPackage(qx, 'critical-a', { isCritical: true, ingestionSource: 'rubygems-registry' }) - await insertPackage(qx, 'critical-b', { isCritical: true, ingestionSource: null }) - await insertPackage(qx, 'non-critical', { isCritical: false, ingestionSource: null }) - - const rows = await listRubyGemsCriticalPackagesToSync(qx, { limit: 100 }) - const names = rows.map((r) => r.name) - - expect(names).toContain('critical-a') - expect(names).toContain('critical-b') - expect(names).not.toContain('non-critical') - }) - - it('paginates by id cursor via afterId', async () => { - const firstId = await insertPackage(qx, 'critical-cursor-1', { isCritical: true }) - await insertPackage(qx, 'critical-cursor-2', { isCritical: true }) - - const rows = await listRubyGemsCriticalPackagesToSync(qx, { limit: 100, afterId: firstId }) - const ids = rows.map((r) => r.id) - - expect(ids.every((id) => id > firstId)).toBe(true) - expect(rows.map((r) => r.name)).not.toContain('critical-cursor-1') - }) - }) -}) diff --git a/services/apps/packages_worker/src/rubygems/client.ts b/services/apps/packages_worker/src/rubygems/client.ts index 5d21528a57..06c6ef4cdb 100644 --- a/services/apps/packages_worker/src/rubygems/client.ts +++ b/services/apps/packages_worker/src/rubygems/client.ts @@ -1,77 +1,58 @@ import axios from 'axios' +import { acquireRubyGemsSlot, parseRetryAfterMs } from './rateLimiter' import { - RubyGemsFetchError, RubyGemsFetchResult, RubyGemsGemResponse, RubyGemsOwner, RubyGemsVersionItem, } from './types' -function classifyError(err: unknown): RubyGemsFetchError | null { - if (!axios.isAxiosError(err)) return null - const status = err.response?.status - if (status === 404) return { kind: 'NOT_FOUND', status, message: err.message } - if (status === 429) return { kind: 'RATE_LIMIT', status, message: err.message } - return null -} +const MAX_RATE_LIMIT_RETRIES = 5 -export async function fetchGem(name: string): Promise> { - try { - const resp = await axios.get( - `https://rubygems.org/api/v1/gems/${encodeURIComponent(name)}.json`, - { timeout: 15000 }, - ) - return resp.data - } catch (err) { - const classified = classifyError(err) - if (classified) return classified - throw err +async function rubyGemsGet(url: string): Promise> { + for (let attempt = 0; ; attempt++) { + await acquireRubyGemsSlot() + try { + const resp = await axios.get(url, { timeout: 15000 }) + return resp.data + } catch (err) { + if (axios.isAxiosError(err)) { + const status = err.response?.status + if (status === 404) return { kind: 'NOT_FOUND', status, message: err.message } + if (status === 429) { + if (attempt >= MAX_RATE_LIMIT_RETRIES) { + return { kind: 'RATE_LIMIT', status, message: err.message } + } + await new Promise((r) => setTimeout(r, parseRetryAfterMs(err.response?.headers['retry-after']))) + continue + } + } + throw err + } } } -export async function fetchVersions( - name: string, -): Promise> { - try { - const resp = await axios.get( - `https://rubygems.org/api/v1/versions/${encodeURIComponent(name)}.json`, - { timeout: 15000 }, - ) - return resp.data - } catch (err) { - const classified = classifyError(err) - if (classified) return classified - throw err - } +export function fetchGem(name: string): Promise> { + return rubyGemsGet( + `https://rubygems.org/api/v1/gems/${encodeURIComponent(name)}.json`, + ) } -export async function fetchOwners(name: string): Promise> { - try { - const resp = await axios.get( - `https://rubygems.org/api/v1/gems/${encodeURIComponent(name)}/owners.json`, - { timeout: 15000 }, - ) - return resp.data - } catch (err) { - const classified = classifyError(err) - if (classified) return classified - throw err - } +export function fetchVersions(name: string): Promise> { + return rubyGemsGet( + `https://rubygems.org/api/v1/versions/${encodeURIComponent(name)}.json`, + ) } -export async function fetchReverseDependencies( - name: string, -): Promise> { - try { - const resp = await axios.get( - `https://rubygems.org/api/v1/gems/${encodeURIComponent(name)}/reverse_dependencies.json`, - { timeout: 15000 }, - ) - return resp.data - } catch (err) { - const classified = classifyError(err) - if (classified) return classified - throw err - } +export function fetchOwners(name: string): Promise> { + return rubyGemsGet( + `https://rubygems.org/api/v1/gems/${encodeURIComponent(name)}/owners.json`, + ) +} + +export function fetchReverseDependencies(name: string): Promise> { + return rubyGemsGet( + `https://rubygems.org/api/v1/gems/${encodeURIComponent(name)}/reverse_dependencies.json`, + ) } diff --git a/services/apps/packages_worker/src/rubygems/normalize.ts b/services/apps/packages_worker/src/rubygems/normalize.ts index a7997cc91d..7f8b651bf4 100644 --- a/services/apps/packages_worker/src/rubygems/normalize.ts +++ b/services/apps/packages_worker/src/rubygems/normalize.ts @@ -47,13 +47,8 @@ export function pickLatestRubyGemsVersion( versions: NormalizedRubyGemsVersion[], ): NormalizedRubyGemsVersion | null { if (versions.length === 0) return null - const sorted = [...versions].sort((a, b) => { - const at = a.publishedAt ? a.publishedAt.getTime() : 0 - const bt = b.publishedAt ? b.publishedAt.getTime() : 0 - return bt - at - }) - const stable = sorted.find((v) => !v.isPrerelease) - return stable ?? sorted[0] + const stable = versions.find((v) => !v.isPrerelease) + return stable ?? versions[0] } export function normalizeRubyGemsOwners(owners: RubyGemsOwner[]): NormalizedRubyGemsOwner[] { diff --git a/services/apps/packages_worker/src/rubygems/rateLimiter.ts b/services/apps/packages_worker/src/rubygems/rateLimiter.ts new file mode 100644 index 0000000000..7d7513bed2 --- /dev/null +++ b/services/apps/packages_worker/src/rubygems/rateLimiter.ts @@ -0,0 +1,22 @@ +const MAX_RPS = Math.max(1, parseInt(process.env.RUBYGEMS_MAX_RPS ?? '10', 10)) +const INTERVAL_MS = 1000 / MAX_RPS + +let nextSlot = 0 + +export async function acquireRubyGemsSlot(): Promise { + const now = Date.now() + const slot = Math.max(now, nextSlot) + nextSlot = slot + INTERVAL_MS + const wait = slot - now + if (wait > 0) await new Promise((r) => setTimeout(r, wait)) +} + +export function parseRetryAfterMs(header: unknown): number { + const FALLBACK_MS = 1000 + if (typeof header !== 'string') return FALLBACK_MS + const seconds = Number(header) + if (Number.isFinite(seconds)) return Math.max(0, seconds * 1000) + const date = new Date(header) + if (!Number.isNaN(date.getTime())) return Math.max(0, date.getTime() - Date.now()) + return FALLBACK_MS +} diff --git a/services/apps/packages_worker/src/rubygems/runRubyGemsCoreLoop.ts b/services/apps/packages_worker/src/rubygems/runRubyGemsCoreLoop.ts index 89ccef94dc..e3bfe0c1a9 100644 --- a/services/apps/packages_worker/src/rubygems/runRubyGemsCoreLoop.ts +++ b/services/apps/packages_worker/src/rubygems/runRubyGemsCoreLoop.ts @@ -177,5 +177,11 @@ export async function processBatch( } } + if (counts.processed === 0 && counts.skipped === 0 && counts.error > 0) { + throw new Error( + `RubyGems core batch made no progress (${counts.error} errors, likely rate-limited) — failing to trigger backoff`, + ) + } + return counts } From 7bebc2712fac619029c704ecf3ccfef6d868902a Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Fri, 10 Jul 2026 18:00:32 +0100 Subject: [PATCH 16/17] fix: activity timeout Signed-off-by: Mouad BANI --- services/apps/packages_worker/src/config.ts | 6 ++---- services/apps/packages_worker/src/rubygems/client.ts | 4 +++- .../packages_worker/src/rubygems/runRubyGemsCoreLoop.ts | 6 +----- .../packages_worker/src/rubygems/runRubyGemsCriticalLoop.ts | 5 ----- 4 files changed, 6 insertions(+), 15 deletions(-) diff --git a/services/apps/packages_worker/src/config.ts b/services/apps/packages_worker/src/config.ts index 3df03e0101..cbb9826ebb 100644 --- a/services/apps/packages_worker/src/config.ts +++ b/services/apps/packages_worker/src/config.ts @@ -88,9 +88,8 @@ export function getNuGetConfig() { export function getRubyGemsConfig() { return { - batchSize: parseInt(process.env.RUBYGEMS_FETCHER_BATCH_SIZE ?? '1000', 10), + batchSize: parseInt(process.env.RUBYGEMS_FETCHER_BATCH_SIZE ?? '10000', 10), concurrency: parseInt(process.env.RUBYGEMS_FETCHER_CONCURRENCY ?? '8', 10), - groupDelayMs: parseInt(process.env.RUBYGEMS_FETCHER_GROUP_DELAY_MS ?? '1000', 10), } } @@ -98,13 +97,12 @@ export function getRubyGemsCriticalConfig() { return { batchSize: parseInt(process.env.RUBYGEMS_CRITICAL_FETCHER_BATCH_SIZE ?? '5000', 10), concurrency: parseInt(process.env.RUBYGEMS_CRITICAL_FETCHER_CONCURRENCY ?? '4', 10), - groupDelayMs: parseInt(process.env.RUBYGEMS_CRITICAL_FETCHER_GROUP_DELAY_MS ?? '1000', 10), } } export function getRubyGemsDependentsConfig() { return { - batchSize: parseInt(process.env.RUBYGEMS_DEPENDENTS_BATCH_SIZE ?? '50000', 10), + batchSize: parseInt(process.env.RUBYGEMS_DEPENDENTS_BATCH_SIZE ?? '10000', 10), concurrency: parseInt(process.env.RUBYGEMS_DEPENDENTS_CONCURRENCY ?? '50', 10), } } diff --git a/services/apps/packages_worker/src/rubygems/client.ts b/services/apps/packages_worker/src/rubygems/client.ts index 06c6ef4cdb..3eb2adb5c2 100644 --- a/services/apps/packages_worker/src/rubygems/client.ts +++ b/services/apps/packages_worker/src/rubygems/client.ts @@ -24,7 +24,9 @@ async function rubyGemsGet(url: string): Promise> { if (attempt >= MAX_RATE_LIMIT_RETRIES) { return { kind: 'RATE_LIMIT', status, message: err.message } } - await new Promise((r) => setTimeout(r, parseRetryAfterMs(err.response?.headers['retry-after']))) + await new Promise((r) => + setTimeout(r, parseRetryAfterMs(err.response?.headers['retry-after'])), + ) continue } } diff --git a/services/apps/packages_worker/src/rubygems/runRubyGemsCoreLoop.ts b/services/apps/packages_worker/src/rubygems/runRubyGemsCoreLoop.ts index e3bfe0c1a9..775c2d6108 100644 --- a/services/apps/packages_worker/src/rubygems/runRubyGemsCoreLoop.ts +++ b/services/apps/packages_worker/src/rubygems/runRubyGemsCoreLoop.ts @@ -54,7 +54,7 @@ async function markPackageError(qx: QueryExecutor, pkg: RubyGemsPackageToSync): type PackageStatus = 'processed' | 'skipped' | 'error' | 'unchanged' -export type RubyGemsCoreConfig = { batchSize: number; concurrency: number; groupDelayMs: number } +export type RubyGemsCoreConfig = { batchSize: number; concurrency: number } async function processPackage( qx: QueryExecutor, @@ -149,10 +149,6 @@ export async function processBatch( for (let batchStart = 0; batchStart < packages.length; batchStart += config.concurrency) { const group = packages.slice(batchStart, batchStart + config.concurrency) - if (config.groupDelayMs > 0 && batchStart > 0) { - await new Promise((r) => setTimeout(r, config.groupDelayMs)) - } - await Promise.all( group.map(async (pkg) => { try { diff --git a/services/apps/packages_worker/src/rubygems/runRubyGemsCriticalLoop.ts b/services/apps/packages_worker/src/rubygems/runRubyGemsCriticalLoop.ts index 5e71cca388..13818ea482 100644 --- a/services/apps/packages_worker/src/rubygems/runRubyGemsCriticalLoop.ts +++ b/services/apps/packages_worker/src/rubygems/runRubyGemsCriticalLoop.ts @@ -24,7 +24,6 @@ const log = getServiceChildLogger('rubygems-critical') export type RubyGemsCriticalConfig = { batchSize: number concurrency: number - groupDelayMs: number } async function withDeadlockRetry(fn: () => Promise, maxAttempts = 4): Promise { @@ -165,10 +164,6 @@ export async function processBatch( for (let batchStart = 0; batchStart < packages.length; batchStart += config.concurrency) { const group = packages.slice(batchStart, batchStart + config.concurrency) - if (config.groupDelayMs > 0 && batchStart > 0) { - await new Promise((r) => setTimeout(r, config.groupDelayMs)) - } - await Promise.all( group.map(async (pkg) => { try { From f87aa20723ed4135f95cf32aa56ea6b986532bc9 Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Fri, 10 Jul 2026 18:05:12 +0100 Subject: [PATCH 17/17] fix: id type Signed-off-by: Mouad BANI --- .../src/rubygems/activities.ts | 8 ++++--- .../src/rubygems/runRubyGemsCriticalLoop.ts | 4 ++-- .../src/rubygems/runRubyGemsDependentsLoop.ts | 4 ++-- .../packages_worker/src/rubygems/workflows.ts | 4 ++-- .../src/osspckgs/rubygems.ts | 22 +++++++++---------- 5 files changed, 22 insertions(+), 20 deletions(-) diff --git a/services/apps/packages_worker/src/rubygems/activities.ts b/services/apps/packages_worker/src/rubygems/activities.ts index 9e97f1a85d..b124e45210 100644 --- a/services/apps/packages_worker/src/rubygems/activities.ts +++ b/services/apps/packages_worker/src/rubygems/activities.ts @@ -27,8 +27,8 @@ export async function processRubyGemsCoreBatch(): Promise { } export async function processRubyGemsCriticalBatch( - afterId = 0, -): Promise { + afterId = '0', +): Promise { const config = getRubyGemsCriticalConfig() const qx = await getPackagesDb() const result = await processCriticalBatch(qx, config, afterId) @@ -36,7 +36,9 @@ export async function processRubyGemsCriticalBatch( return result } -export async function processRubyGemsDependentsBatch(afterId = 0): Promise { +export async function processRubyGemsDependentsBatch( + afterId = '0', +): Promise { const config = getRubyGemsDependentsConfig() const qx = await getPackagesDb() const result = await processDependentsBatch(qx, config, afterId) diff --git a/services/apps/packages_worker/src/rubygems/runRubyGemsCriticalLoop.ts b/services/apps/packages_worker/src/rubygems/runRubyGemsCriticalLoop.ts index 13818ea482..167caed7b4 100644 --- a/services/apps/packages_worker/src/rubygems/runRubyGemsCriticalLoop.ts +++ b/services/apps/packages_worker/src/rubygems/runRubyGemsCriticalLoop.ts @@ -146,8 +146,8 @@ async function processPackage( export async function processBatch( qx: QueryExecutor, config: RubyGemsCriticalConfig, - afterId: number, -): Promise { + afterId: string, +): Promise { const packages = await listRubyGemsCriticalPackagesToSync(qx, { limit: config.batchSize, afterId, diff --git a/services/apps/packages_worker/src/rubygems/runRubyGemsDependentsLoop.ts b/services/apps/packages_worker/src/rubygems/runRubyGemsDependentsLoop.ts index 893376ef93..95a2888c86 100644 --- a/services/apps/packages_worker/src/rubygems/runRubyGemsDependentsLoop.ts +++ b/services/apps/packages_worker/src/rubygems/runRubyGemsDependentsLoop.ts @@ -20,7 +20,7 @@ export type DependentsBatchResult = { processed: number notFound: number error: number - lastId: number | null + lastId: string | null } type PackageStatus = 'processed' | 'notFound' | 'error' @@ -47,7 +47,7 @@ async function processPackage( export async function processBatch( qx: QueryExecutor, config: RubyGemsDependentsConfig, - afterId: number, + afterId: string, ): Promise { const packages = await listRubyGemsPackagesForDependents(qx, { limit: config.batchSize, diff --git a/services/apps/packages_worker/src/rubygems/workflows.ts b/services/apps/packages_worker/src/rubygems/workflows.ts index 086c1d7a50..6983562983 100644 --- a/services/apps/packages_worker/src/rubygems/workflows.ts +++ b/services/apps/packages_worker/src/rubygems/workflows.ts @@ -16,7 +16,7 @@ export async function ingestRubyGemsPackages(): Promise { await continueAsNew() } -export async function ingestRubyGemsCriticalDetails(afterId = 0): Promise { +export async function ingestRubyGemsCriticalDetails(afterId = '0'): Promise { const result = await acts.processRubyGemsCriticalBatch(afterId) if (result.lastId === null) { log.info('RubyGems critical ingestion complete — no more work, exiting.', { ...result }) @@ -25,7 +25,7 @@ export async function ingestRubyGemsCriticalDetails(afterId = 0): Promise await continueAsNew(result.lastId) } -export async function ingestRubyGemsDependents(afterId = 0): Promise { +export async function ingestRubyGemsDependents(afterId = '0'): Promise { const result = await acts.processRubyGemsDependentsBatch(afterId) if (result.lastId === null) { log.info('RubyGems dependents ingestion complete — no more work, exiting.', { ...result }) diff --git a/services/libs/data-access-layer/src/osspckgs/rubygems.ts b/services/libs/data-access-layer/src/osspckgs/rubygems.ts index 0f601177b7..d72b65b8ee 100644 --- a/services/libs/data-access-layer/src/osspckgs/rubygems.ts +++ b/services/libs/data-access-layer/src/osspckgs/rubygems.ts @@ -1,7 +1,7 @@ import { QueryExecutor } from '../queryExecutor' export type RubyGemsPackageToSync = { - id: number + id: string purl: string name: string latestVersion: string | null @@ -39,23 +39,23 @@ export async function listRubyGemsPackagesToSync( } export type RubyGemsCriticalPackageToSync = { - id: number + id: string purl: string name: string } export async function listRubyGemsCriticalPackagesToSync( qx: QueryExecutor, - options: { limit: number; afterId?: number }, + options: { limit: number; afterId?: string }, ): Promise { - const { limit, afterId = 0 } = options + const { limit, afterId = '0' } = options return qx.select( ` SELECT p.id, p.purl, p.name FROM packages p WHERE p.ecosystem = 'rubygems' AND p.is_critical - AND p.id > $(afterId) + AND p.id > $(afterId)::bigint ORDER BY p.id ASC LIMIT $(limit) `, @@ -64,21 +64,21 @@ export async function listRubyGemsCriticalPackagesToSync( } export type RubyGemsPackageForDependents = { - id: number + id: string name: string } export async function listRubyGemsPackagesForDependents( qx: QueryExecutor, - options: { limit: number; afterId?: number }, + options: { limit: number; afterId?: string }, ): Promise { - const { limit, afterId = 0 } = options + const { limit, afterId = '0' } = options return qx.select( ` SELECT p.id, p.name FROM packages p WHERE p.ecosystem = 'rubygems' - AND p.id > $(afterId) + AND p.id > $(afterId)::bigint ORDER BY p.id ASC LIMIT $(limit) `, @@ -88,14 +88,14 @@ export async function listRubyGemsPackagesForDependents( export async function updateRubyGemsDependentCount( qx: QueryExecutor, - packageId: number, + packageId: string, dependentCount: number, ): Promise { await qx.result( `UPDATE packages SET dependent_count = $(dependentCount), last_synced_at = NOW() - WHERE id = $(packageId) + WHERE id = $(packageId)::bigint AND dependent_count IS DISTINCT FROM $(dependentCount)`, { packageId, dependentCount }, )