fix: heartbeat bug

Author: Gergely Békési

src/modules/rollingFeed.ts

@@ -133,9 +133,10 @@ export class RollingFeedWriter {
   }
 
   /**
-   * Backfills every period strictly between the last populated one and `periodIdx`
-   * (default: current) with that period's last known payload/reference. Never writes
-   * `periodIdx` itself - that period is the caller's to write with fresh data.
+   * Backfills every period from the last populated one (exclusive) up to `periodIdx`
+   * (default: current) with that period's last known payload/reference. Skips `periodIdx`
+   * itself if it's already populated, so it never clobbers fresh data the caller may have
+   * already written there via `uploadPayload`/`uploadReference`.
    */
   async catchUp(postageBatchId: string | BatchId, periodIdx?: number): Promise<void> {
     const requestOptions = this.context.getRequestOptionsForCall()
@@ -163,7 +164,10 @@ export class RollingFeedWriter {
     const { feedIndex: sourceIndex } = await probeFeed(requestOptions, owner, sourceTopic)
     const sourceChunk = await downloadFeedUpdateAsCAC(requestOptions, owner, sourceTopic, sourceIndex)
 
-    for (let period = lastGoodPeriod + 1; period < targetPeriod; period++) {
+    const targetAlreadyWritten = await isPeriodPopulated(requestOptions, owner, topicFor(this.baseTopic, targetPeriod))
+    const backfillEnd = targetAlreadyWritten ? targetPeriod : targetPeriod + 1
+
+    for (let period = lastGoodPeriod + 1; period < backfillEnd; period++) {
       const identifier = makeFeedIdentifier(topicFor(this.baseTopic, period), 0)
       await uploadSingleOwnerChunkWithWrappedChunk(requestOptions, this.signer, stamp, identifier, sourceChunk)
     }

test/integration/rolling-feed.spec.ts

@@ -98,15 +98,30 @@ test('catchUp backfills a gap so the reader resolves it again', async () => {
 
   expect(await writer.isCaughtUp(1003)).toBe(false)
   await writer.catchUp(batch(), 1003)
-  // catchUp never writes its target period itself (1003) - only the gap strictly before it
-  await waitUntil(async () => writer.isCaughtUp(1002))
-  expect(await writer.isCaughtUp(1003)).toBe(false)
+  // 1003 wasn't written yet, so catchUp fills it too, not just the gap strictly before it
+  await waitUntil(async () => writer.isCaughtUp(1003))
 
   setPeriod(1003)
-  // reader falls back one period, from the still-empty 1003 to the now-backfilled 1002
   await waitUntil(async () => (await reader.downloadPayload()).payload.toUtf8() === 'Backfilled payload')
 })
 
+test('heartbeat idiom (isCaughtUp then catchUp with no args) converges after one silent period', async () => {
+  const { writer, reader } = makeWriterAndReader()
+
+  setPeriod(1000)
+  await writer.uploadPayload(batch(), 'Heartbeat payload', { deferred: false })
+  await waitUntil(async () => writer.isCaughtUp(1000))
+
+  // one whole period of silence: 1001 is mirrored, 1002 is not
+  setPeriod(1002)
+  expect(await writer.isCaughtUp()).toBe(false)
+  await writer.catchUp(batch())
+  await waitUntil(async () => writer.isCaughtUp())
+
+  const result = await reader.downloadPayload()
+  expect(result.payload.toUtf8()).toBe('Heartbeat payload')
+})
+
 test('catchUp does not clobber fresh data already written to its target period', async () => {
   const { writer, reader } = makeWriterAndReader()
 
@@ -141,13 +156,11 @@ test('catchUp is not fooled by an older buried gap', async () => {
 
   expect(await writer.isCaughtUp(6)).toBe(false)
   await writer.catchUp(batch(), 6)
-  // catchUp never writes period 6 itself - only fills the gap up to period 5
-  await waitUntil(async () => writer.isCaughtUp(5))
-  expect(await writer.isCaughtUp(6)).toBe(false)
+  // 6 wasn't written yet, so catchUp fills it too, not just the gap strictly before it
+  await waitUntil(async () => writer.isCaughtUp(6))
 
   setPeriod(6)
-  // reader falls back from the still-empty 6 to the now-backfilled 5; must be the recent
-  // (period 4) content, not the one from before the buried gap
+  // must resume from the recent (period 4) content, not the one from before the buried gap
   await waitUntil(async () => (await reader.downloadPayload()).payload.toUtf8() === 'Recent payload')
 })