fix: anchor packagist metadata due-selection to a fixed cutoff

Signed-off-by: anilb <epipav@gmail.com>

Author: epipav

services/apps/packages_worker/src/packagist/__tests__/dueSelection.test.ts

@@ -35,15 +35,20 @@ beforeEach(() => {
 })
 
 // Metadata (merged enrichment) due-selection: keyset pagination, refresh window, and
-// the stored Last-Modified for If-Modified-Since replay on the p2 fetch.
+// the stored Last-Modified for If-Modified-Since replay on the p2 fetch. `cutoff` is a
+// pre-computed threshold (this run's fixed start time minus the refresh window), not a
+// live NOW() — a keyset scan only visits each purl once per drain, so due-selection
+// must stay anchored to one stable point in time for the whole run.
 describe('getPackagistMetadataDuePurls', () => {
-  it('returns purls with their stored Last-Modified, scoped by refresh window', async () => {
+  const CUTOFF = '2026-07-08T00:00:00.000Z'
+
+  it('returns purls with their stored Last-Modified, scoped by the fixed cutoff', async () => {
     qx.select.mockResolvedValue([
       { purl: 'pkg:composer/a/x', metadata_last_modified: 'Tue, 30 Jun 2026 00:00:00 GMT' },
       { purl: 'pkg:composer/b/y', metadata_last_modified: null },
     ])
 
-    const candidates = await getPackagistMetadataDuePurls(asQx(qx), '', 50, 7, true)
+    const candidates = await getPackagistMetadataDuePurls(asQx(qx), CUTOFF, '', 50, true)
 
     expect(candidates).toEqual([
       { purl: 'pkg:composer/a/x', metadataLastModified: 'Tue, 30 Jun 2026 00:00:00 GMT' },
@@ -54,11 +59,11 @@ describe('getPackagistMetadataDuePurls', () => {
     expect(sql).toMatch(/is_critical/)
     expect(sql).toMatch(/metadata_last_run_at/)
     expect(sql).toMatch(/ORDER BY\s+p\.purl/i)
-    expect(params).toMatchObject({ batchSize: 50, refreshDays: 7, onlyCritical: true })
+    expect(params).toMatchObject({ cutoff: CUTOFF, batchSize: 50, onlyCritical: true })
   })
 
   it('selects across all packagist packages when onlyCritical is false (the all-packages default)', async () => {
-    await getPackagistMetadataDuePurls(asQx(qx), '', 50, 7, false)
+    await getPackagistMetadataDuePurls(asQx(qx), CUTOFF, '', 50, false)
 
     const [sql, params] = qx.select.mock.calls[0]
     expect(sql).toMatch(/ecosystem = 'packagist'/)

services/apps/packages_worker/src/packagist/__tests__/ingest.test.ts

@@ -414,39 +414,61 @@ describe('ingestPackagistItemsConcurrently', () => {
 
 // All-packages metadata scope: the env default drives due-selection's onlyCritical arg.
 // Default is ALL packages (deps.dev has no Packagist data to fall back on); the env var
-// narrows back to the critical slice.
+// narrows back to the critical slice. The due-cutoff is derived once from the run's
+// fixed cutoff minus the refresh window, not a live NOW() re-evaluated per batch.
 describe('getPackagistMetadataBatch scope', () => {
   const mockMetadataDue = vi.mocked(getPackagistMetadataDuePurls)
+  const RUN_CUTOFF = '2026-07-15T00:00:00.000Z'
+  const DUE_CUTOFF = '2026-07-08T00:00:00.000Z' // RUN_CUTOFF minus the 7-day default
 
   afterEach(() => {
     delete process.env.CROWD_PACKAGES_PACKAGIST_RUN_ONLY_FOR_CRITICAL
+    delete process.env.CROWD_PACKAGES_PACKAGIST_METADATA_REFRESH_DAYS
+  })
+
+  it('derives the due-cutoff from the fixed run cutoff, not a live clock', async () => {
+    process.env.CROWD_PACKAGES_PACKAGIST_METADATA_REFRESH_DAYS = '3'
+    mockMetadataDue.mockResolvedValue([])
+
+    await getPackagistMetadataBatch(RUN_CUTOFF, '', 50)
+
+    // 3 days before the run's own fixed cutoff — never "3 days before whenever this
+    // particular batch happens to execute" — is what makes the keyset scan give every
+    // purl exactly one refresh-eligibility check per drain.
+    expect(mockMetadataDue).toHaveBeenCalledWith(
+      expect.anything(),
+      '2026-07-12T00:00:00.000Z',
+      '',
+      50,
+      false,
+    )
   })
 
   it('selects across ALL packagist packages by default', async () => {
     mockMetadataDue.mockResolvedValue([])
 
-    await getPackagistMetadataBatch('', 50)
+    await getPackagistMetadataBatch(RUN_CUTOFF, '', 50)
 
-    expect(mockMetadataDue).toHaveBeenCalledWith(expect.anything(), '', 50, 7, false)
+    expect(mockMetadataDue).toHaveBeenCalledWith(expect.anything(), DUE_CUTOFF, '', 50, false)
   })
 
   it('narrows to the critical slice when CROWD_PACKAGES_PACKAGIST_RUN_ONLY_FOR_CRITICAL=true', async () => {
     process.env.CROWD_PACKAGES_PACKAGIST_RUN_ONLY_FOR_CRITICAL = 'true'
     mockMetadataDue.mockResolvedValue([])
 
-    await getPackagistMetadataBatch('', 50)
+    await getPackagistMetadataBatch(RUN_CUTOFF, '', 50)
 
-    expect(mockMetadataDue).toHaveBeenCalledWith(expect.anything(), '', 50, 7, true)
+    expect(mockMetadataDue).toHaveBeenCalledWith(expect.anything(), DUE_CUTOFF, '', 50, true)
   })
 
   it('keeps the all-packages default when the env value is unrecognized', async () => {
     // a typo must not silently flip the sweep to critical-only
     process.env.CROWD_PACKAGES_PACKAGIST_RUN_ONLY_FOR_CRITICAL = 'ture'
     mockMetadataDue.mockResolvedValue([])
 
-    await getPackagistMetadataBatch('', 50)
+    await getPackagistMetadataBatch(RUN_CUTOFF, '', 50)
 
-    expect(mockMetadataDue).toHaveBeenCalledWith(expect.anything(), '', 50, 7, false)
+    expect(mockMetadataDue).toHaveBeenCalledWith(expect.anything(), DUE_CUTOFF, '', 50, false)
   })
 })
 

services/apps/packages_worker/src/packagist/activities.ts

@@ -323,16 +323,25 @@ export async function runPackagistPackageSeed(): Promise<{ discovered: number; i
   return { discovered: entries.length, invalid }
 }
 
+// `cutoff` is the drain's own fixed start time (stable across every round via
+// continueAsNew), not a live NOW() — a keyset scan only visits each purl once per
+// drain, so re-deriving "7 days ago" fresh on every batch would silently skip a purl
+// that hasn't quite hit the refresh window yet when the cursor passes it, pushing its
+// effective cadence out toward two refresh cycles instead of one.
 export async function getPackagistMetadataBatch(
+  cutoff: string,
   afterPurl: string,
   batchSize: number,
 ): Promise<{ candidates: PackagistMetadataCandidate[]; nextCursor: string }> {
   const qx = await getPackagesDb()
+  const dueCutoff = new Date(
+    new Date(cutoff).getTime() - metadataRefreshDays() * 24 * 60 * 60 * 1000,
+  ).toISOString()
   const candidates = await getPackagistMetadataDuePurls(
     qx,
+    dueCutoff,
     afterPurl,
     batchSize,
-    metadataRefreshDays(),
     runOnlyForCritical(),
   )
   return {

services/apps/packages_worker/src/packagist/workflows.ts

@@ -23,6 +23,7 @@ const INGEST_BATCH = 50
 const ROUNDS_PER_RUN = 20
 
 interface MetadataState {
+  cutoff?: string
   cursor?: string
 }
 
@@ -54,20 +55,29 @@ export async function seedPackagistPackages(): Promise<void> {
   }
 }
 
+// The cutoff is fixed once per run (deterministic activity), same pattern as the
+// downloads-30d/daily lanes — a keyset scan only ever visits each purl once per drain,
+// so due-selection must be anchored to a stable point in time rather than a live NOW()
+// that would let a purl processed early in the run dodge this cycle's refresh window.
 export async function ingestPackagistMetadata(state: MetadataState = {}): Promise<void> {
+  const cutoff = state.cutoff ?? (await acts.packagistCurrentTimestamp())
   let cursor = state.cursor || ''
   const stopAfterFirstPage = await acts.packagistStopAfterFirstPage()
 
   for (let r = 0; r < ROUNDS_PER_RUN; r++) {
-    const { candidates, nextCursor } = await acts.getPackagistMetadataBatch(cursor, INGEST_BATCH)
+    const { candidates, nextCursor } = await acts.getPackagistMetadataBatch(
+      cutoff,
+      cursor,
+      INGEST_BATCH,
+    )
     if (candidates.length === 0) return
     await acts.ingestPackagistMetadataBatch(candidates)
     cursor = nextCursor
     if (stopAfterFirstPage) return
     if (candidates.length < INGEST_BATCH) return
   }
 
-  await continueAsNew<typeof ingestPackagistMetadata>({ cursor })
+  await continueAsNew<typeof ingestPackagistMetadata>({ cutoff, cursor })
 }
 
 // Monthly capture of the observed rolling 30d window for every packagist package.

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

@@ -68,9 +68,9 @@ export async function markPackagistMetadataScanned(
 
 export async function getPackagistMetadataDuePurls(
   qx: QueryExecutor,
+  cutoff: string,
   afterPurl: string,
   batchSize: number,
-  refreshDays: number,
   onlyCritical: boolean,
 ): Promise<PackagistMetadataCandidate[]> {
   const rows: Array<{ purl: string; metadata_last_modified: string | null }> = await qx.select(
@@ -82,11 +82,11 @@ export async function getPackagistMetadataDuePurls(
         AND p.purl > $(afterPurl)
         AND (
           s.metadata_last_run_at IS NULL
-          OR s.metadata_last_run_at < NOW() - ($(refreshDays) || ' days')::interval
+          OR s.metadata_last_run_at < $(cutoff)::timestamptz
         )
       ORDER BY p.purl
       LIMIT $(batchSize)`,
-    { afterPurl, batchSize, refreshDays, onlyCritical },
+    { cutoff, afterPurl, batchSize, onlyCritical },
   )
   return rows.map((r) => ({
     purl: r.purl,