fix: guard packagist watermark against re-processing

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

Author: epipav

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

@@ -72,7 +72,7 @@ describe('markPackagistMetadataScanned', () => {
       asQx(qx),
       'pkg:composer/a/x',
       { status: 'success', attempts: 1 },
-      'Wed, 01 Jul 2026 00:00:00 GMT',
+      { metadataLastModified: 'Wed, 01 Jul 2026 00:00:00 GMT' },
     )
 
     const [sql, params] = qx.result.mock.calls[0]

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

@@ -75,6 +75,7 @@ const mockSeedInsert = vi.mocked(insertPackagistPackages)
 const qx = {} as QueryExecutor
 const PURL = 'pkg:composer/monolog/monolog'
 const RUN_DATE = '2026-07-15'
+const SCHEDULED_AT = '2026-07-15T00:00:00.000Z'
 
 const statsJson = {
   package: { name: 'monolog/monolog', downloads: { total: 1000, monthly: 300, daily: 10 } },
@@ -113,7 +114,7 @@ describe('ingestOnePackagistMetadata', () => {
   it('audits phase 1 immediately and phase 2 separately, then marks scanned with the fresh Last-Modified', async () => {
     happyMocks()
 
-    await ingestOnePackagistMetadata(qx, candidate)
+    await ingestOnePackagistMetadata(qx, candidate, SCHEDULED_AT)
 
     expect(mockPersistInfo).toHaveBeenCalledWith(qx, PURL, expect.anything())
     expect(mockFetchP2).toHaveBeenCalledWith('monolog/monolog', 'Tue, 30 Jun 2026 00:00:00 GMT')
@@ -127,15 +128,15 @@ describe('ingestOnePackagistMetadata', () => {
       qx,
       PURL,
       expect.objectContaining({ status: 'success', attempts: 1 }),
-      'Wed, 01 Jul 2026 00:00:00 GMT',
+      { metadataLastModified: 'Wed, 01 Jul 2026 00:00:00 GMT' },
     )
   })
 
   it('records a p2 NOT_MODIFIED as success: info persisted, versions skipped', async () => {
     happyMocks()
     mockFetchP2.mockResolvedValue({ kind: 'NOT_MODIFIED' } as never)
 
-    await ingestOnePackagistMetadata(qx, candidate)
+    await ingestOnePackagistMetadata(qx, candidate, SCHEDULED_AT)
 
     expect(mockPersistInfo).toHaveBeenCalled()
     expect(mockPersistMetadata).not.toHaveBeenCalled()
@@ -144,7 +145,7 @@ describe('ingestOnePackagistMetadata', () => {
       qx,
       PURL,
       expect.objectContaining({ status: 'success' }),
-      null,
+      { metadataLastModified: null },
     )
   })
 
@@ -156,7 +157,7 @@ describe('ingestOnePackagistMetadata', () => {
       message: 'not found',
     } as never)
 
-    const p = ingestOnePackagistMetadata(qx, candidate)
+    const p = ingestOnePackagistMetadata(qx, candidate, SCHEDULED_AT)
     await vi.runAllTimersAsync()
     await p
 
@@ -167,6 +168,7 @@ describe('ingestOnePackagistMetadata', () => {
       qx,
       PURL,
       expect.objectContaining({ status: 'error', errorKind: 'NOT_FOUND', httpStatus: 404 }),
+      { notBefore: SCHEDULED_AT },
     )
   })
 
@@ -177,7 +179,7 @@ describe('ingestOnePackagistMetadata', () => {
       message: 'HTTP 500',
     } as never)
 
-    await expect(ingestOnePackagistMetadata(qx, candidate)).rejects.toThrow()
+    await expect(ingestOnePackagistMetadata(qx, candidate, SCHEDULED_AT)).rejects.toThrow()
     expect(mockMarkMetadata).not.toHaveBeenCalled()
     expect(mockFetchStats).toHaveBeenCalledTimes(1)
   })
@@ -192,7 +194,7 @@ describe('ingestOnePackagistMetadata', () => {
       message: 'not found',
     } as never)
 
-    const p = ingestOnePackagistMetadata(qx, candidate)
+    const p = ingestOnePackagistMetadata(qx, candidate, SCHEDULED_AT)
     await vi.runAllTimersAsync()
     await p
 
@@ -203,8 +205,7 @@ describe('ingestOnePackagistMetadata', () => {
       qx,
       PURL,
       expect.objectContaining({ status: 'error', errorKind: 'NOT_FOUND' }),
-      undefined,
-      false,
+      { bumpLastRunAt: false, notBefore: SCHEDULED_AT },
     )
   })
 
@@ -219,7 +220,7 @@ describe('ingestOnePackagistMetadata', () => {
       message: 'not found',
     } as never)
 
-    const p = ingestOnePackagistMetadata(qx, candidate)
+    const p = ingestOnePackagistMetadata(qx, candidate, SCHEDULED_AT)
     await vi.runAllTimersAsync()
     await p
 
@@ -228,8 +229,7 @@ describe('ingestOnePackagistMetadata', () => {
       qx,
       PURL,
       expect.objectContaining({ status: 'error', errorKind: 'NOT_FOUND' }),
-      undefined,
-      false,
+      { bumpLastRunAt: false, notBefore: SCHEDULED_AT },
     )
   })
 
@@ -238,7 +238,7 @@ describe('ingestOnePackagistMetadata', () => {
     mockPersistInfo.mockResolvedValue({ found: true, changedFields: ['packages.description'] })
     mockFetchP2.mockResolvedValue({ kind: 'TRANSIENT', message: 'HTTP 502' } as never)
 
-    await expect(ingestOnePackagistMetadata(qx, candidate)).rejects.toThrow()
+    await expect(ingestOnePackagistMetadata(qx, candidate, SCHEDULED_AT)).rejects.toThrow()
     expect(mockMarkMetadata).not.toHaveBeenCalled()
     // phase-1 writes are already committed when the throw happens — a retry re-runs
     // phase 1 idempotently and reports no changes, so this audit event can't be deferred
@@ -252,7 +252,7 @@ describe('ingestOnePackagist30dWindow', () => {
     mockFetchStats.mockResolvedValue(statsJson as never)
     mockPersist30d.mockResolvedValue(['downloads_last_30d.count', 'packages.downloads_last_30d'])
 
-    await ingestOnePackagist30dWindow(qx, PURL, RUN_DATE)
+    await ingestOnePackagist30dWindow(qx, PURL, RUN_DATE, SCHEDULED_AT)
 
     expect(mockPersist30d).toHaveBeenCalledWith(qx, PURL, 300, RUN_DATE)
     expect(mockAudit).toHaveBeenCalledWith(qx, 'packagist', PURL, [
@@ -274,7 +274,7 @@ describe('ingestOnePackagist30dWindow', () => {
       message: 'not found',
     } as never)
 
-    const p = ingestOnePackagist30dWindow(qx, PURL, RUN_DATE)
+    const p = ingestOnePackagist30dWindow(qx, PURL, RUN_DATE, SCHEDULED_AT)
     await vi.runAllTimersAsync()
     await p
 
@@ -283,13 +283,14 @@ describe('ingestOnePackagist30dWindow', () => {
       qx,
       PURL,
       expect.objectContaining({ status: 'error', errorKind: 'NOT_FOUND' }),
+      SCHEDULED_AT,
     )
   })
 
   it('throws on a transient result without marking processed', async () => {
     mockFetchStats.mockResolvedValue({ kind: 'TRANSIENT', message: 'HTTP 503' } as never)
 
-    await expect(ingestOnePackagist30dWindow(qx, PURL, RUN_DATE)).rejects.toThrow()
+    await expect(ingestOnePackagist30dWindow(qx, PURL, RUN_DATE, SCHEDULED_AT)).rejects.toThrow()
     expect(mockMark30d).not.toHaveBeenCalled()
   })
 })
@@ -302,7 +303,7 @@ describe('ingestOnePackagistDailyDownload', () => {
     mockFetchStats.mockResolvedValue(statsJson as never)
     mockDaily.mockResolvedValue(['downloads_daily.date', 'downloads_daily.count'])
 
-    await ingestOnePackagistDailyDownload(qx, candidate, RUN_DATE)
+    await ingestOnePackagistDailyDownload(qx, candidate, RUN_DATE, SCHEDULED_AT)
 
     expect(mockDaily).toHaveBeenCalledWith(qx, '7', [{ day: RUN_DATE, downloads: 10 }])
     expect(mockAudit).toHaveBeenCalledWith(qx, 'packagist', PURL, [
@@ -319,7 +320,7 @@ describe('ingestOnePackagistDailyDownload', () => {
   it('marks success without inserting or auditing when the registry reports no daily count', async () => {
     mockFetchStats.mockResolvedValue({ package: { name: 'monolog/monolog' } } as never)
 
-    await ingestOnePackagistDailyDownload(qx, candidate, RUN_DATE)
+    await ingestOnePackagistDailyDownload(qx, candidate, RUN_DATE, SCHEDULED_AT)
 
     expect(mockDaily).not.toHaveBeenCalled()
     expect(mockAudit).not.toHaveBeenCalled()
@@ -333,7 +334,9 @@ describe('ingestOnePackagistDailyDownload', () => {
   it('throws on a transient result without marking processed', async () => {
     mockFetchStats.mockResolvedValue({ kind: 'TRANSIENT', message: 'HTTP 503' } as never)
 
-    await expect(ingestOnePackagistDailyDownload(qx, candidate, RUN_DATE)).rejects.toThrow()
+    await expect(
+      ingestOnePackagistDailyDownload(qx, candidate, RUN_DATE, SCHEDULED_AT),
+    ).rejects.toThrow()
     expect(mockMarkDaily).not.toHaveBeenCalled()
   })
 })

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

@@ -119,6 +119,9 @@ function giveUpResult(error: FetchError, attempts: number): PackagistRunResult {
 export async function ingestOnePackagistMetadata(
   qx: QueryExecutor,
   candidate: PackagistMetadataCandidate,
+  // This batch activity's own scheduledTimestampMs (stable across Temporal retries),
+  // passed through to every give-up write below — see MarkMetadataScannedOptions.notBefore.
+  scheduledAt: string,
 ): Promise<void> {
   const name = packagistNameFromPurl(candidate.purl)
 
@@ -132,7 +135,14 @@ export async function ingestOnePackagistMetadata(
       { purl: candidate.purl, statusCode: info.error.statusCode, kind: info.error.kind },
       'packagist package info 4xx/malformed after fast retries — marking scanned and skipping',
     )
-    await markPackagistMetadataScanned(qx, candidate.purl, giveUpResult(info.error, info.attempts))
+    await markPackagistMetadataScanned(
+      qx,
+      candidate.purl,
+      giveUpResult(info.error, info.attempts),
+      {
+        notBefore: scheduledAt,
+      },
+    )
     return
   }
 
@@ -154,13 +164,10 @@ export async function ingestOnePackagistMetadata(
     // Phase 1 already succeeded — only p2 (versions/deps) failed to refresh. Don't push
     // metadata_last_run_at forward, or due-selection wrongly treats this package as
     // "recently scanned" and skips it for the full refresh window despite stale p2 data.
-    await markPackagistMetadataScanned(
-      qx,
-      candidate.purl,
-      giveUpResult(p2.error, p2.attempts),
-      undefined,
-      false,
-    )
+    await markPackagistMetadataScanned(qx, candidate.purl, giveUpResult(p2.error, p2.attempts), {
+      bumpLastRunAt: false,
+      notBefore: scheduledAt,
+    })
     return
   }
 
@@ -180,13 +187,13 @@ export async function ingestOnePackagistMetadata(
   }
 
   await logAuditFieldChanges(qx, WORKER, candidate.purl, phase2ChangedFields)
-  // markPackagistMetadataScanned treats a null 4th arg the same as omitting it
-  // (COALESCEd against the stored value), so lastModified can always be passed.
   await markPackagistMetadataScanned(
     qx,
     candidate.purl,
     { status: 'success', attempts: p2.attempts },
-    lastModified,
+    {
+      metadataLastModified: lastModified,
+    },
   )
 }
 
@@ -195,6 +202,7 @@ export async function ingestOnePackagist30dWindow(
   qx: QueryExecutor,
   purl: string,
   runDate: string,
+  scheduledAt: string,
 ): Promise<void> {
   const name = packagistNameFromPurl(purl)
 
@@ -207,7 +215,7 @@ export async function ingestOnePackagist30dWindow(
       { purl, statusCode: info.error.statusCode, kind: info.error.kind },
       'packagist 30d downloads 4xx/malformed after fast retries — marking processed and skipping',
     )
-    await markPackagist30dProcessed(qx, purl, giveUpResult(info.error, info.attempts))
+    await markPackagist30dProcessed(qx, purl, giveUpResult(info.error, info.attempts), scheduledAt)
     return
   }
 
@@ -226,6 +234,7 @@ export async function ingestOnePackagistDailyDownload(
   qx: QueryExecutor,
   candidate: PackagistDailyCandidate,
   runDate: string,
+  scheduledAt: string,
 ): Promise<void> {
   const name = packagistNameFromPurl(candidate.purl)
 
@@ -238,7 +247,12 @@ export async function ingestOnePackagistDailyDownload(
       { purl: candidate.purl, statusCode: info.error.statusCode, kind: info.error.kind },
       'packagist daily downloads 4xx/malformed after fast retries — marking processed and skipping',
     )
-    await markPackagistDailyProcessed(qx, candidate.purl, giveUpResult(info.error, info.attempts))
+    await markPackagistDailyProcessed(
+      qx,
+      candidate.purl,
+      giveUpResult(info.error, info.attempts),
+      scheduledAt,
+    )
     return
   }
 
@@ -332,6 +346,8 @@ export async function ingestPackagistMetadataBatch(
   if (candidates.length === 0) return
   const qx = await getPackagesDb()
   const attempt = Context.current().info.attempt
+  // Stable across every Temporal retry of this same batch — see notBefore below.
+  const scheduledAt = new Date(Context.current().info.scheduledTimestampMs).toISOString()
 
   // The merged lane starts every ingest with a DYNAMIC-endpoint fetch, so it is
   // bounded by that endpoint's 10-concurrent limit — not p2's 20. Running hotter
@@ -340,13 +356,16 @@ export async function ingestPackagistMetadataBatch(
     candidates,
     attempt,
     statsConcurrency(),
-    (candidate) => ingestOnePackagistMetadata(qx, candidate),
+    (candidate) => ingestOnePackagistMetadata(qx, candidate, scheduledAt),
     (candidate, err) =>
-      markPackagistMetadataScanned(qx, candidate.purl, {
-        status: 'error',
-        attempts: attempt,
-        message: String(err),
-      }),
+      markPackagistMetadataScanned(
+        qx,
+        candidate.purl,
+        { status: 'error', attempts: attempt, message: String(err) },
+        // An item that already succeeded earlier in this same batch's retry sequence
+        // must not have that success overwritten by an unrelated re-processing failure.
+        { notBefore: scheduledAt },
+      ),
   )
 
   log.info({ count: candidates.length }, 'Ingested Packagist metadata batch')
@@ -366,18 +385,20 @@ export async function ingestPackagist30dBatch(purls: string[], runDate: string):
   if (purls.length === 0) return
   const qx = await getPackagesDb()
   const attempt = Context.current().info.attempt
+  const scheduledAt = new Date(Context.current().info.scheduledTimestampMs).toISOString()
 
   await ingestPackagistItemsConcurrently(
     purls,
     attempt,
     statsConcurrency(),
-    (purl) => ingestOnePackagist30dWindow(qx, purl, runDate),
+    (purl) => ingestOnePackagist30dWindow(qx, purl, runDate, scheduledAt),
     (purl, err) =>
-      markPackagist30dProcessed(qx, purl, {
-        status: 'error',
-        attempts: attempt,
-        message: String(err),
-      }),
+      markPackagist30dProcessed(
+        qx,
+        purl,
+        { status: 'error', attempts: attempt, message: String(err) },
+        scheduledAt,
+      ),
   )
 
   log.info({ count: purls.length }, 'Ingested Packagist 30d downloads batch')
@@ -403,18 +424,20 @@ export async function ingestPackagistDailyBatch(
   if (candidates.length === 0) return
   const qx = await getPackagesDb()
   const attempt = Context.current().info.attempt
+  const scheduledAt = new Date(Context.current().info.scheduledTimestampMs).toISOString()
 
   await ingestPackagistItemsConcurrently(
     candidates,
     attempt,
     statsConcurrency(),
-    (candidate) => ingestOnePackagistDailyDownload(qx, candidate, runDate),
+    (candidate) => ingestOnePackagistDailyDownload(qx, candidate, runDate, scheduledAt),
     (candidate, err) =>
-      markPackagistDailyProcessed(qx, candidate.purl, {
-        status: 'error',
-        attempts: attempt,
-        message: String(err),
-      }),
+      markPackagistDailyProcessed(
+        qx,
+        candidate.purl,
+        { status: 'error', attempts: attempt, message: String(err) },
+        scheduledAt,
+      ),
   )
 
   log.info({ count: candidates.length }, 'Ingested Packagist daily downloads batch')