refactor: simplify the schedule

Signed-off-by: Umberto Sgueglia <usgueglia@contractor.linuxfoundation.org>

Author: ulemons

backend/.env.dist.local

@@ -185,9 +185,9 @@ ENRICHER_BATCH_SIZE=100
 ENRICHER_REPO_UPDATE_INTERVAL_HOURS=24
 ENRICHER_IDLE_SLEEP_SEC=60
 
-OSSPCKGS_GCP_PROJECT=
-OSSPCKGS_GCS_BUCKET=
-OSSPCKGS_GCP_CREDENTIALS_B64=
+OSSPCKGS_GCP_PROJECT=local-dev
+OSSPCKGS_GCS_BUCKET=local-dev
+OSSPCKGS_GCP_CREDENTIALS_B64=e30=
 
 # osv-sync (Temporal-scheduled; see services/apps/packages_worker/src/osv/schedule.ts)
 # OSV_ECOSYSTEMS uses OSV's canonical bucket case (npm lowercase, Maven titlecase) because
@@ -200,13 +200,16 @@ OSV_TMP_DIR=/tmp/osv
 OSV_BATCH_SIZE=500
 OSV_DERIVE_BATCH_SIZE=1000
 # maven enricher
-POM_FETCHER_BATCH_SIZE=50
-POM_FETCHER_CONCURRENCY=3
+
+POM_FETCHER_BATCH_SIZE=2000
+POM_FETCHER_CONCURRENCY=10
 POM_FETCHER_NON_CRITICAL_BATCH_SIZE=500
 POM_FETCHER_NON_CRITICAL_CONCURRENCY=20
 POM_FETCHER_REFRESH_DAYS=1
-POM_FETCHER_GROUP_DELAY_MS=500
-POM_FETCHER_IDLE_SLEEP_SEC=3600
+POM_FETCHER_GROUP_DELAY_MS=100
 # Set to 'true' on first run against a fresh/restored DB to skip the version-unchanged
 # optimisation and force full POM extraction. Set to 'false' after the first pass.
-POM_FETCHER_FORCE_FULL_EXTRACTION=false
+POM_FETCHER_FORCE_FULL_EXTRACTION=true
+POM_FETCHER_MAVEN_BASE_URL=https://maven-central.storage-download.googleapis.com/maven2
+MAVEN_SYNC_SOURCE=both
+MAVEN_DELTA_API_URL=https://maven-fetcher-production.up.railway.app

scripts/services/maven.yaml

@@ -8,10 +8,6 @@ x-env-args: &env-args
   CROWD_TEMPORAL_NAMESPACE: ${CROWD_PACKAGES_TEMPORAL_NAMESPACE}
   SHELL: /bin/sh
   SUPPRESS_NO_CONFIG_WARNING: 'true'
-  POM_FETCHER_BATCH_SIZE: '50'
-  POM_FETCHER_CONCURRENCY: '3'
-  POM_FETCHER_STALE_DAYS: '7'
-  POM_FETCHER_IDLE_SLEEP_SEC: '3600'
 
 services:
   packages-worker:

scripts/services/packages-worker.yaml

@@ -6,23 +6,22 @@
     "start:github-repos-enricher": "SERVICE=github-repos-enricher tsx src/bin/github-repos-enricher.ts",
     "start:packages-worker": "CROWD_TEMPORAL_TASKQUEUE=packages-worker CROWD_TEMPORAL_NAMESPACE=$CROWD_PACKAGES_TEMPORAL_NAMESPACE SERVICE=packages-worker tsx src/bin/packages-worker.ts",
     "start:pom-fetcher": "SERVICE=pom-fetcher tsx src/bin/pom-fetcher.ts",
+    "backfill:maven": "SERVICE=maven tsx src/bin/maven-backfill.ts",
     "dev:packages-worker": "CROWD_TEMPORAL_TASKQUEUE=packages-worker CROWD_TEMPORAL_NAMESPACE=$CROWD_PACKAGES_TEMPORAL_NAMESPACE SERVICE=packages-worker LOG_LEVEL=trace nodemon --watch src --watch ../../libs --ext ts --exec tsx --inspect=0.0.0.0:9233 src/bin/packages-worker.ts",
     "dev:deps-dev-ingest": "CROWD_TEMPORAL_TASKQUEUE=deps-dev-ingest CROWD_TEMPORAL_NAMESPACE=$CROWD_PACKAGES_TEMPORAL_NAMESPACE SERVICE=deps-dev-ingest nodemon --watch src --watch ../../libs --ext ts --exec tsx --inspect=0.0.0.0:9235 src/bin/deps-dev-ingest.ts",
     "start:maven": "SERVICE=maven tsx src/bin/maven.ts",
     "dev:github-repos-enricher": "SERVICE=github-repos-enricher LOG_LEVEL=trace nodemon --watch src --watch ../../libs --ext ts --exec tsx --inspect=0.0.0.0:9234 src/bin/github-repos-enricher.ts",
-    "dev:maven": "SERVICE=maven LOG_LEVEL=info nodemon --watch src --watch ../../libs --ext ts --exec tsx --inspect=0.0.0.0:9235 src/bin/maven.ts",
     "dev:packages-worker:local": "set -a && . ../../../backend/.env.dist.local && . ../../../backend/.env.override.local && set +a && CROWD_TEMPORAL_TASKQUEUE=packages-worker CROWD_TEMPORAL_NAMESPACE=$CROWD_PACKAGES_TEMPORAL_NAMESPACE SERVICE=packages-worker LOG_LEVEL=trace nodemon --watch src --watch ../../libs --ext ts --exec tsx --inspect=0.0.0.0:9233 src/bin/packages-worker.ts",
     "dev:deps-dev-ingest:local": "set -a && . ../../../backend/.env.dist.local && . ../../../backend/.env.override.local && set +a && CROWD_TEMPORAL_TASKQUEUE=deps-dev-ingest CROWD_TEMPORAL_NAMESPACE=$CROWD_PACKAGES_TEMPORAL_NAMESPACE SERVICE=deps-dev-ingest nodemon --watch src --watch ../../libs --ext ts --exec tsx --inspect=0.0.0.0:9235 src/bin/deps-dev-ingest.ts",
     "dev:github-repos-enricher:local": "set -a && . ../../../backend/.env.dist.local && . ../../../backend/.env.override.local && set +a && SERVICE=github-repos-enricher LOG_LEVEL=trace nodemon --watch src --watch ../../libs --ext ts --exec tsx --inspect=0.0.0.0:9234 src/bin/github-repos-enricher.ts",
     "export-to-bucket": "SERVICE=deps-dev-ingest tsx src/scripts/exportToBucket.ts",
     "export-to-bucket:local": "set -a && . ../../../backend/.env.dist.local && . ../../../backend/.env.override.local && set +a && SERVICE=deps-dev-ingest tsx src/scripts/exportToBucket.ts",
-    "monitor:osspckgs": "SERVICE=monitor tsx src/scripts/monitorOsspckgs.ts",
-    "monitor:osspckgs:local": "bash -c 'set -a && . ../../../backend/.env.dist.local && . ../../../backend/.env.override.local && set +a && SERVICE=monitor tsx src/scripts/monitorOsspckgs.ts'",
-    "trigger-bootstrap": "SERVICE=deps-dev-ingest tsx src/scripts/triggerBootstrap.ts",
-    "trigger-bootstrap:local": "set -a && . ../../../backend/.env.dist.local && . ../../../backend/.env.override.local && set +a && SERVICE=deps-dev-ingest tsx src/scripts/triggerBootstrap.ts",
-    "dev:pom-fetcher": "SERVICE=pom-fetcher LOG_LEVEL=info nodemon --watch src --watch ../../libs --ext ts --exec tsx --inspect=0.0.0.0:9235 src/bin/pom-fetcher.ts",
-    "dev:pom-fetcher:local": "set -a && . ../../../backend/.env.dist.local && . ../../../backend/.env.override.local && set +a && SERVICE=pom-fetcher LOG_LEVEL=info nodemon --watch src --watch ../../libs --ext ts --exec tsx --inspect=0.0.0.0:9235 src/bin/pom-fetcher.ts",
-    "dev:maven:local": "set -a && . ../../../backend/.env.dist.local && . ../../../backend/.env.override.local && set +a && SERVICE=maven LOG_LEVEL=info nodemon --watch src --watch ../../libs --ext ts --exec tsx --inspect=0.0.0.0:9235 src/bin/maven.ts",
+    "monitor:osspckgs:local": "bash -c 'set -a && . ../../../backend/.env.dist.local && . ../../../backend/.env.override.local && set +a && node ../../../scripts/monitor-osspckgs.mjs'",
+    "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",
+    "benchmark:maven-delta": "SERVICE=maven tsx src/scripts/benchmarkMavenDelta.ts",
+    "benchmark:maven-delta:local": "set -a && . ../../../backend/.env.dist.local && . ../../../backend/.env.override.local && set +a && SERVICE=maven LOG_LEVEL=info tsx src/scripts/benchmarkMavenDelta.ts",
+    "validate:maven-quality": "SERVICE=maven tsx src/scripts/validateDataQuality.ts",
+    "validate:maven-quality:local": "set -a && . ../../../backend/.env.dist.local && . ../../../backend/.env.override.local && set +a && SERVICE=maven LOG_LEVEL=info tsx src/scripts/validateDataQuality.ts",
     "lint": "npx eslint --ext .ts src --max-warnings=0",
     "format": "npx prettier --write \"src/**/*.ts\"",
     "format-check": "npx prettier --check .",

services/apps/packages_worker/package.json

@@ -0,0 +1,42 @@
+import { getServiceLogger } from '@crowd/logging'
+
+import { getMavenConfig } from '../config'
+import { getPackagesDb } from '../db'
+import { runMavenCriticalBackfill } from '../maven/runMavenEnrichmentLoop'
+
+const log = getServiceLogger()
+
+let shuttingDown = false
+
+// Graceful stop: finish the in-flight batch, then exit. Safe to interrupt — every
+// write is an idempotent upsert and the DB state is the cursor, so re-running
+// resumes where it left off.
+const shutdown = () => {
+  if (shuttingDown) return
+  shuttingDown = true
+  log.info('Shutting down maven backfill (stopping after the current batch)...')
+}
+
+process.on('SIGINT', shutdown)
+process.on('SIGTERM', shutdown)
+
+const main = async () => {
+  log.info('maven backfill starting (one-shot, full extraction)...')
+
+  const config = getMavenConfig()
+  log.info(config, 'Config loaded')
+
+  const qx = await getPackagesDb()
+  await qx.selectOne('SELECT 1')
+  log.info('Connected to packages-db.')
+
+  const totals = await runMavenCriticalBackfill(qx, config, () => shuttingDown)
+
+  log.info({ ...totals }, 'maven backfill complete')
+  process.exit(0)
+}
+
+main().catch((err) => {
+  log.error({ err }, 'maven backfill fatal error')
+  process.exit(1)
+})

services/apps/packages_worker/src/bin/maven-backfill.ts

@@ -1,13 +1,12 @@
+import { scheduleMavenCritical } from '../maven/schedule'
 import { scheduleNpmIngest } from '../npm/schedule'
 import { scheduleOsvSync } from '../osv/schedule'
-import { scheduleMavenCritical, scheduleMavenNonCritical } from '../maven/schedule'
 import { svc } from '../service'
 
 setImmediate(async () => {
   await svc.init()
   await scheduleNpmIngest()
   await scheduleOsvSync()
   await scheduleMavenCritical()
-  await scheduleMavenNonCritical()
   await svc.start()
 })

services/apps/packages_worker/src/bin/maven.ts

@@ -33,15 +33,44 @@ export function getEnricherConfig() {
   }
 }
 
+// Which source drives the critical Maven sync:
+//   'maven' → poll packages_universe by staleness (current behaviour, default/fallback)
+//   'api'   → only enrich what our delta feed reports as changed
+//   'both'  → run both passes in the same Temporal tick
+export type MavenSyncSource = 'api' | 'maven' | 'both'
+
+function parseMavenSyncSource(raw: string | undefined): MavenSyncSource {
+  if (raw === 'api' || raw === 'both') return raw
+  // Anything else (unset, typo, legacy value) falls back to the current behaviour.
+  return 'maven'
+}
+
 export function getMavenConfig() {
+  const syncSource = parseMavenSyncSource(process.env.MAVEN_SYNC_SOURCE)
+  const deltaApiBaseUrl = (process.env.MAVEN_DELTA_API_URL ?? '').replace(/\/+$/, '')
+
+  if (syncSource !== 'maven' && !deltaApiBaseUrl) {
+    throw new Error(`MAVEN_SYNC_SOURCE='${syncSource}' requires MAVEN_DELTA_API_URL to be set`)
+  }
+
   return {
     batchSize: requireEnvInt('POM_FETCHER_BATCH_SIZE'),
     concurrency: requireEnvInt('POM_FETCHER_CONCURRENCY'),
     nonCriticalBatchSize: requireEnvInt('POM_FETCHER_NON_CRITICAL_BATCH_SIZE'),
     nonCriticalConcurrency: requireEnvInt('POM_FETCHER_NON_CRITICAL_CONCURRENCY'),
     refreshDays: requireEnvInt('POM_FETCHER_REFRESH_DAYS'),
     groupDelayMs: requireEnvInt('POM_FETCHER_GROUP_DELAY_MS'),
-    idleSleepSec: requireEnvInt('POM_FETCHER_IDLE_SLEEP_SEC'),
-    forceFullExtraction: requireEnv('POM_FETCHER_FORCE_FULL_EXTRACTION') === 'true',
+    syncSource,
+    deltaApi: {
+      baseUrl: deltaApiBaseUrl,
+      token: process.env.MAVEN_DELTA_API_TOKEN || undefined,
+      pageSize: process.env.MAVEN_DELTA_API_PAGE_SIZE
+        ? parseInt(process.env.MAVEN_DELTA_API_PAGE_SIZE, 10)
+        : 100,
+      lookbackMinutes: process.env.MAVEN_DELTA_API_LOOKBACK_MINUTES
+        ? parseInt(process.env.MAVEN_DELTA_API_LOOKBACK_MINUTES, 10)
+        : 15,
+      includePrerelease: process.env.MAVEN_DELTA_API_INCLUDE_PRERELEASE === 'true',
+    },
   }
 }