From 1dad69008ec0cb73cb5c06f6e6e42092c664503d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gergely=20B=C3=A9k=C3=A9si?= Date: Fri, 28 Aug 2026 14:28:56 +0200 Subject: [PATCH 1/8] feat: implement rolling feed --- src/bee.ts | 9 +- src/modules/rollingFeed.ts | 227 ++++++++++++++++++++++++++ test/integration/rolling-feed.spec.ts | 117 +++++++++++++ 3 files changed, 352 insertions(+), 1 deletion(-) create mode 100644 src/modules/rollingFeed.ts create mode 100644 test/integration/rolling-feed.spec.ts diff --git a/src/bee.ts b/src/bee.ts index 2bf1520d..0013b6cf 100644 --- a/src/bee.ts +++ b/src/bee.ts @@ -1,4 +1,3 @@ -import { Objects } from 'cafe-utility' import { BatchId, Bytes, @@ -11,6 +10,7 @@ import { makeContentAddressedChunk, unmarshalContentAddressedChunk, } from '@ethersphere/core-sdk' +import { Objects } from 'cafe-utility' import { postEnvelope } from './api/envelope' import { rchash } from './api/rchash' import { SingleOwnerChunk, makeSOCAddress, makeSingleOwnerChunk, unmarshalSingleOwnerChunk } from './chunk/soc' @@ -27,6 +27,7 @@ import { File as FileNamespace } from './modules/file' import { Grantee } from './modules/grantee' import { Messaging } from './modules/messaging' import { Pin } from './modules/pin' +import { RollingFeed } from './modules/rollingFeed' import { Settlement } from './modules/settlement' import { Soc } from './modules/soc' import { Stake } from './modules/stake' @@ -140,6 +141,7 @@ export class Bee { this.chunk = new ChunkNamespace(context) this.file = new FileNamespace(context) this.collection = new Collection(context) + this.rollingFeed = new RollingFeed(context) } /** @@ -247,6 +249,11 @@ export class Bee { */ public readonly collection: Collection + /** + * Rolling feed operations. + */ + public readonly rollingFeed: RollingFeed + /** * Creates a Content Addressed Chunk. * diff --git a/src/modules/rollingFeed.ts b/src/modules/rollingFeed.ts new file mode 100644 index 00000000..2ddb2e5a --- /dev/null +++ b/src/modules/rollingFeed.ts @@ -0,0 +1,227 @@ +import { BatchId, EthAddress, keccak256, numberToUint64, PrivateKey, Reference, Topic } from '@ethersphere/core-sdk' +import { BeeError, Bytes } from '..' +import { + FeedPayloadResult, + FeedReferenceResult, + FeedUpdateOptions, + fetchLatestFeedUpdate, + probeFeed, +} from '../api/feed' +import { uploadSingleOwnerChunkWithWrappedChunk } from '../chunk/soc' +import { + downloadFeedUpdate, + downloadFeedUpdateAsCAC, + FeedUploadOptions, + updateFeedWithPayload, + updateFeedWithReference, +} from '../feed' +import { makeFeedIdentifier } from '../feed/identifier' +import { BeeRequestOptions, UploadResult } from '../types' +import { BeeResponseError } from '../utils/error' +import { BeeContext } from './context' + +// Not yet finalized (see ROLLING_FEED.md "Open parameters"): bounds how far catchUp/isCaughtUp +// will scan/backfill so a long-dead writer can't trigger an unbounded loop. +const MAX_CATCH_UP_LOOKBACK = 1000 + +function periodIndex(t: number, periodLength: number): number { + if (periodLength <= 0) { + throw new BeeError('Period length must be greater than zero!') + } + + return Math.floor(t / periodLength) +} + +function topicFor(baseTopic: Topic, periodIdx: number): Topic { + const baseTopicBytes = baseTopic.toUint8Array() + + return new Topic(keccak256(Bytes.concat(baseTopicBytes, numberToUint64(BigInt(periodIdx), 'BE')))) +} + +async function isPeriodPopulated(requestOptions: BeeRequestOptions, owner: EthAddress, topic: Topic): Promise { + try { + await probeFeed(requestOptions, owner, topic) + + return true + } catch (e) { + if (e instanceof BeeResponseError) { + return false + } + throw e + } +} + +async function fetchLatestReference( + requestOptions: BeeRequestOptions, + owner: EthAddress, + topic: Topic, +): Promise { + const { feedIndex } = await probeFeed(requestOptions, owner, topic) + const update = await downloadFeedUpdate(requestOptions, owner, topic, feedIndex, true) + + return { + reference: new Reference(update.payload.toUint8Array()), + feedIndex, + feedIndexNext: feedIndex.next(), + } +} + +export class RollingFeed { + constructor(private readonly context: BeeContext) {} + + makeWriter(baseTopic: Topic, signer: PrivateKey, periodLength: number): RollingFeedWriter { + return new RollingFeedWriter(this.context, baseTopic, signer, periodLength) + } + + makeReader(baseTopic: Topic, owner: EthAddress, periodLength: number): RollingFeedReader { + return new RollingFeedReader(this.context, baseTopic, owner, periodLength) + } +} + +export class RollingFeedWriter { + constructor( + private readonly context: BeeContext, + private readonly baseTopic: Topic, + private readonly signer: PrivateKey, + private readonly periodLength: number, + ) {} + + async uploadPayload( + postageBatchId: string | BatchId, + payload: Uint8Array | string, + options?: FeedUploadOptions, + ): Promise { + const requestOptions = this.context.getRequestOptionsForCall() + const stamp = new BatchId(postageBatchId) + const { currentTopic, nextTopic, mirrorOptions } = this.currentAndNextTopics(options) + + const [result] = await Promise.all([ + updateFeedWithPayload(requestOptions, this.signer, currentTopic, payload, stamp, options), + updateFeedWithPayload(requestOptions, this.signer, nextTopic, payload, stamp, mirrorOptions), + ]) + + return result + } + + async uploadReference( + postageBatchId: string | BatchId, + reference: Reference | string | Uint8Array, + options?: FeedUploadOptions, + ): Promise { + const requestOptions = this.context.getRequestOptionsForCall() + const stamp = new BatchId(postageBatchId) + const { currentTopic, nextTopic, mirrorOptions } = this.currentAndNextTopics(options) + + const [result] = await Promise.all([ + updateFeedWithReference(requestOptions, this.signer, currentTopic, reference, stamp, options), + updateFeedWithReference(requestOptions, this.signer, nextTopic, reference, stamp, mirrorOptions), + ]) + + return result + } + + /** + * True unless the period right before `periodIdx` (default: current) never got mirrored + * forward into it, i.e. the writer went silent for at least one whole period. + */ + async isCaughtUp(periodIdx?: number): Promise { + const requestOptions = this.context.getRequestOptionsForCall() + const targetPeriod = periodIdx ?? periodIndex(Date.now() / 1000, this.periodLength) + const owner = this.signer.publicKey().address() + + return isPeriodPopulated(requestOptions, owner, topicFor(this.baseTopic, targetPeriod)) + } + + /** + * Backfills every period from the last populated one (exclusive) up to `periodIdx` + * (default: current) with that period's last known payload/reference. + */ + async catchUp(postageBatchId: string | BatchId, periodIdx?: number): Promise { + const requestOptions = this.context.getRequestOptionsForCall() + const stamp = new BatchId(postageBatchId) + const owner = this.signer.publicKey().address() + const targetPeriod = periodIdx ?? periodIndex(Date.now() / 1000, this.periodLength) + + let lastGoodPeriod = targetPeriod - 1 + + while (lastGoodPeriod >= targetPeriod - MAX_CATCH_UP_LOOKBACK) { + if (await isPeriodPopulated(requestOptions, owner, topicFor(this.baseTopic, lastGoodPeriod))) { + break + } + lastGoodPeriod-- + } + + if (lastGoodPeriod < targetPeriod - MAX_CATCH_UP_LOOKBACK) { + throw new BeeError(`No populated period found within ${MAX_CATCH_UP_LOOKBACK} periods to catch up from!`) + } + + const sourceTopic = topicFor(this.baseTopic, lastGoodPeriod) + 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 identifier = makeFeedIdentifier(topicFor(this.baseTopic, period), 0) + await uploadSingleOwnerChunkWithWrappedChunk(requestOptions, this.signer, stamp, identifier, sourceChunk) + } + } + + private currentAndNextTopics(options?: FeedUploadOptions): { + currentTopic: Topic + nextTopic: Topic + mirrorOptions: FeedUploadOptions + } { + const currentPeriod = periodIndex(Date.now() / 1000, this.periodLength) + + return { + currentTopic: topicFor(this.baseTopic, currentPeriod), + nextTopic: topicFor(this.baseTopic, currentPeriod + 1), + mirrorOptions: { ...options, index: undefined }, + } + } +} + +export class RollingFeedReader { + constructor( + private readonly context: BeeContext, + private readonly baseTopic: Topic, + private readonly owner: EthAddress, + private readonly periodLength: number, + ) {} + + /** + * Reads the current period's feed; falls back to the previous period once if empty, + * to tolerate clock skew between writer and reader. + */ + async downloadPayload(options?: Omit): Promise { + const requestOptions = this.context.getRequestOptionsForCall() + const currentPeriod = periodIndex(Date.now() / 1000, this.periodLength) + + try { + return await fetchLatestFeedUpdate(requestOptions, this.owner, topicFor(this.baseTopic, currentPeriod), options) + } catch (e) { + if (!(e instanceof BeeResponseError)) { + throw e + } + + return fetchLatestFeedUpdate(requestOptions, this.owner, topicFor(this.baseTopic, currentPeriod - 1), options) + } + } + + /** + * Same as `downloadPayload`, but for a reference to data uploaded elsewhere. + */ + async downloadReference(): Promise { + const requestOptions = this.context.getRequestOptionsForCall() + const currentPeriod = periodIndex(Date.now() / 1000, this.periodLength) + + try { + return await fetchLatestReference(requestOptions, this.owner, topicFor(this.baseTopic, currentPeriod)) + } catch (e) { + if (!(e instanceof BeeResponseError)) { + throw e + } + + return fetchLatestReference(requestOptions, this.owner, topicFor(this.baseTopic, currentPeriod - 1)) + } + } +} diff --git a/test/integration/rolling-feed.spec.ts b/test/integration/rolling-feed.spec.ts new file mode 100644 index 00000000..95f08d70 --- /dev/null +++ b/test/integration/rolling-feed.spec.ts @@ -0,0 +1,117 @@ +import { Strings, System } from 'cafe-utility' +import { PrivateKey, Topic } from '@ethersphere/core-sdk' +import { batch, makeBee } from '../utils' + +const bee = makeBee() + +const PERIOD_LENGTH = 3 // seconds + +function currentPeriod(): number { + return Math.floor(Date.now() / 1000 / PERIOD_LENGTH) +} + +async function waitForPeriod(target: number): Promise { + while (currentPeriod() < target) { + await System.sleepMillis(150) + } +} + +function makeWriterAndReader() { + const privateKey = new PrivateKey(Strings.randomHex(64)) + const owner = privateKey.publicKey().address() + const baseTopic = new Topic(Strings.randomHex(64)) + + return { + writer: bee.rollingFeed.makeWriter(baseTopic, privateKey, PERIOD_LENGTH), + reader: bee.rollingFeed.makeReader(baseTopic, owner, PERIOD_LENGTH), + } +} + +test('uploadPayload / downloadPayload roundtrip', async () => { + const { writer, reader } = makeWriterAndReader() + + await writer.uploadPayload(batch(), 'Hello rolling feed', { deferred: false }) + + const result = await reader.downloadPayload() + expect(result.payload.toUtf8()).toBe('Hello rolling feed') +}) + +test('uploadPayload mirrors into the next period', async () => { + const { writer, reader } = makeWriterAndReader() + const startPeriod = currentPeriod() + + await writer.uploadPayload(batch(), 'Mirrored payload', { deferred: false }) + await waitForPeriod(startPeriod + 1) + + const result = await reader.downloadPayload() + expect(result.payload.toUtf8()).toBe('Mirrored payload') +}) + +test('downloadPayload falls back to the previous period when the current one is empty', async () => { + const { writer, reader } = makeWriterAndReader() + + await writer.uploadPayload(batch(), 'Last known payload', { deferred: false }) + + // wait for the one-period-empty gap (current unpopulated, previous populated); if a slow + // call overshoots past it, re-seed from wherever we land instead of racing a fixed wait + for (;;) { + const period = currentPeriod() + const previousPopulated = await writer.isCaughtUp(period - 1) + const currentPopulated = await writer.isCaughtUp(period) + + if (previousPopulated && !currentPopulated) { + break + } + + if (!previousPopulated) { + await writer.uploadPayload(batch(), 'Last known payload', { deferred: false }) + } + + await System.sleepMillis(150) + } + + const result = await reader.downloadPayload() + expect(result.payload.toUtf8()).toBe('Last known payload') +}) + +test('uploadReference / downloadReference roundtrip', async () => { + const { writer, reader } = makeWriterAndReader() + + const uploaded = await bee.data.upload(batch(), 'Referenced content') + await writer.uploadReference(batch(), uploaded.reference, { deferred: false }) + + const result = await reader.downloadReference() + expect(result.reference.toHex()).toBe(uploaded.reference.toHex()) +}) + +test('isCaughtUp is true for a written/mirrored period and false past a gap', async () => { + const { writer } = makeWriterAndReader() + const startPeriod = currentPeriod() + + await writer.uploadPayload(batch(), 'Still going', { deferred: false }) + + expect(await writer.isCaughtUp(startPeriod)).toBe(true) + expect(await writer.isCaughtUp(startPeriod + 1)).toBe(true) + expect(await writer.isCaughtUp(startPeriod + 5)).toBe(false) +}) + +test('catchUp backfills a gap so the reader resolves it again', async () => { + const { writer, reader } = makeWriterAndReader() + const startPeriod = currentPeriod() + const gapPeriod = startPeriod + 2 + + await writer.uploadPayload(batch(), 'Backfilled payload', { deferred: false }) + expect(await writer.isCaughtUp(gapPeriod)).toBe(false) + + await writer.catchUp(batch(), gapPeriod) + expect(await writer.isCaughtUp(gapPeriod)).toBe(true) + + // the calls above take real time; keep the backfill current with whatever period + // we actually land on before reading it back + for (let period = currentPeriod(); !(await writer.isCaughtUp(period)); period = currentPeriod()) { + await writer.catchUp(batch(), period) + } + + const result = await reader.downloadPayload() + expect(result.payload.toUtf8()).toBe('Backfilled payload') +}) From 6db057e3ce21a08d2c1a2da8ada948a558f2eba0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gergely=20B=C3=A9k=C3=A9si?= Date: Fri, 28 Aug 2026 14:38:18 +0200 Subject: [PATCH 2/8] fix: remove sleeps to speed up tests --- test/integration/rolling-feed.spec.ts | 74 +++++++++++---------------- 1 file changed, 30 insertions(+), 44 deletions(-) diff --git a/test/integration/rolling-feed.spec.ts b/test/integration/rolling-feed.spec.ts index 95f08d70..5720a5ae 100644 --- a/test/integration/rolling-feed.spec.ts +++ b/test/integration/rolling-feed.spec.ts @@ -1,19 +1,24 @@ -import { Strings, System } from 'cafe-utility' +import { Strings } from 'cafe-utility' import { PrivateKey, Topic } from '@ethersphere/core-sdk' import { batch, makeBee } from '../utils' const bee = makeBee() -const PERIOD_LENGTH = 3 // seconds +const PERIOD_LENGTH = 5 // seconds -function currentPeriod(): number { - return Math.floor(Date.now() / 1000 / PERIOD_LENGTH) -} +let dateNowSpy: jest.SpiedFunction | undefined -async function waitForPeriod(target: number): Promise { - while (currentPeriod() < target) { - await System.sleepMillis(150) - } +afterEach(() => { + dateNowSpy?.mockRestore() + dateNowSpy = undefined +}) + +// Topics are a pure function of the period index, which the SDK derives from Date.now() alone +// (no server-side clock involved) - freezing it lets tests jump between periods instantly +// instead of waiting on the real wall clock. +function setPeriod(period: number): void { + dateNowSpy?.mockRestore() + dateNowSpy = jest.spyOn(Date, 'now').mockReturnValue(period * PERIOD_LENGTH * 1000) } function makeWriterAndReader() { @@ -29,6 +34,7 @@ function makeWriterAndReader() { test('uploadPayload / downloadPayload roundtrip', async () => { const { writer, reader } = makeWriterAndReader() + setPeriod(1000) await writer.uploadPayload(batch(), 'Hello rolling feed', { deferred: false }) @@ -38,11 +44,11 @@ test('uploadPayload / downloadPayload roundtrip', async () => { test('uploadPayload mirrors into the next period', async () => { const { writer, reader } = makeWriterAndReader() - const startPeriod = currentPeriod() + setPeriod(1000) await writer.uploadPayload(batch(), 'Mirrored payload', { deferred: false }) - await waitForPeriod(startPeriod + 1) + setPeriod(1001) const result = await reader.downloadPayload() expect(result.payload.toUtf8()).toBe('Mirrored payload') }) @@ -50,32 +56,18 @@ test('uploadPayload mirrors into the next period', async () => { test('downloadPayload falls back to the previous period when the current one is empty', async () => { const { writer, reader } = makeWriterAndReader() + setPeriod(1000) await writer.uploadPayload(batch(), 'Last known payload', { deferred: false }) - // wait for the one-period-empty gap (current unpopulated, previous populated); if a slow - // call overshoots past it, re-seed from wherever we land instead of racing a fixed wait - for (;;) { - const period = currentPeriod() - const previousPopulated = await writer.isCaughtUp(period - 1) - const currentPopulated = await writer.isCaughtUp(period) - - if (previousPopulated && !currentPopulated) { - break - } - - if (!previousPopulated) { - await writer.uploadPayload(batch(), 'Last known payload', { deferred: false }) - } - - await System.sleepMillis(150) - } - + // period 1001 got mirrored, period 1002 was never written at all + setPeriod(1002) const result = await reader.downloadPayload() expect(result.payload.toUtf8()).toBe('Last known payload') }) test('uploadReference / downloadReference roundtrip', async () => { const { writer, reader } = makeWriterAndReader() + setPeriod(1000) const uploaded = await bee.data.upload(batch(), 'Referenced content') await writer.uploadReference(batch(), uploaded.reference, { deferred: false }) @@ -86,32 +78,26 @@ test('uploadReference / downloadReference roundtrip', async () => { test('isCaughtUp is true for a written/mirrored period and false past a gap', async () => { const { writer } = makeWriterAndReader() - const startPeriod = currentPeriod() + setPeriod(1000) await writer.uploadPayload(batch(), 'Still going', { deferred: false }) - expect(await writer.isCaughtUp(startPeriod)).toBe(true) - expect(await writer.isCaughtUp(startPeriod + 1)).toBe(true) - expect(await writer.isCaughtUp(startPeriod + 5)).toBe(false) + expect(await writer.isCaughtUp(1000)).toBe(true) + expect(await writer.isCaughtUp(1001)).toBe(true) + expect(await writer.isCaughtUp(1005)).toBe(false) }) test('catchUp backfills a gap so the reader resolves it again', async () => { const { writer, reader } = makeWriterAndReader() - const startPeriod = currentPeriod() - const gapPeriod = startPeriod + 2 + setPeriod(1000) await writer.uploadPayload(batch(), 'Backfilled payload', { deferred: false }) - expect(await writer.isCaughtUp(gapPeriod)).toBe(false) - await writer.catchUp(batch(), gapPeriod) - expect(await writer.isCaughtUp(gapPeriod)).toBe(true) - - // the calls above take real time; keep the backfill current with whatever period - // we actually land on before reading it back - for (let period = currentPeriod(); !(await writer.isCaughtUp(period)); period = currentPeriod()) { - await writer.catchUp(batch(), period) - } + expect(await writer.isCaughtUp(1003)).toBe(false) + await writer.catchUp(batch(), 1003) + expect(await writer.isCaughtUp(1003)).toBe(true) + setPeriod(1003) const result = await reader.downloadPayload() expect(result.payload.toUtf8()).toBe('Backfilled payload') }) From 58085c0790ac6d95645e2b9d40fcacfd40b480f3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gergely=20B=C3=A9k=C3=A9si?= Date: Fri, 28 Aug 2026 15:45:46 +0200 Subject: [PATCH 3/8] fix: edge-case bug --- README.md | 24 +++++++++++++++++++ src/index.ts | 1 + src/modules/rollingFeed.ts | 7 ++++-- test/integration/rolling-feed.spec.ts | 34 +++++++++++++++++++++++++++ 4 files changed, 64 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index f11cac9a..c424d25f 100644 --- a/README.md +++ b/README.md @@ -127,6 +127,8 @@ The `toString` method uses `toHex`. | SOCReader | SingleOwnerChunk reader | `bee.soc.makeReader` | | FeedWriter | Feed writer | `bee.feed.makeWriter` | | FeedReader | Feed reader | `bee.feed.makeReader` | +| RollingFeedWriter | Rolling feed writer | `bee.rollingFeed.makeWriter` | +| RollingFeedReader | Rolling feed reader | `bee.rollingFeed.makeReader` | ### Bee API @@ -332,6 +334,28 @@ const bee = new Bee('http://localhost:1633') const uploadResult = await bee.collection.uploadFromDirectory(batchId, './path/to/gallery/') ``` +### Rolling feed (periodically-restarting sequential feed) + +A rolling feed avoids the unbounded growth of a plain sequential feed by restarting it every +`periodLength` seconds, so old postage-batch eviction never breaks the latest update. See +[ROLLING_FEED.md](./ROLLING_FEED.md) for the full design. + +```js +import { Bee, PrivateKey, Topic } from '@ethersphere/bee-js' + +const bee = new Bee('http://localhost:1633') +const topic = Topic.fromString('my-feed') +const signer = new PrivateKey('...') +const periodLength = 600 // 10 minutes + +const writer = bee.rollingFeed.makeWriter(topic, signer, periodLength) +await writer.uploadPayload(batchId, 'Hello, World!') + +const reader = bee.rollingFeed.makeReader(topic, signer.publicKey().address(), periodLength) +const result = await reader.downloadPayload() +console.log(result.payload.toUtf8()) // prints 'Hello, World!' +``` + ### Customize http/https agent and headers ```js diff --git a/src/index.ts b/src/index.ts index 321bbc19..c7898373 100644 --- a/src/index.ts +++ b/src/index.ts @@ -21,6 +21,7 @@ export { export type { Chunk } from '@ethersphere/core-sdk' export type { SingleOwnerChunk } from './chunk/soc' export { MantarayNode } from './manifest/manifest' +export { RollingFeedReader, RollingFeedWriter } from './modules/rollingFeed' export { SUPPORTED_BEE_VERSION, SUPPORTED_BEE_VERSION_EXACT } from './version' export * from './types' export * from './utils/constants' diff --git a/src/modules/rollingFeed.ts b/src/modules/rollingFeed.ts index 2ddb2e5a..0131f25c 100644 --- a/src/modules/rollingFeed.ts +++ b/src/modules/rollingFeed.ts @@ -142,16 +142,19 @@ export class RollingFeedWriter { const owner = this.signer.publicKey().address() const targetPeriod = periodIdx ?? periodIndex(Date.now() / 1000, this.periodLength) + // periods before 0 can't exist (period index is derived from Unix time), so the scan + // must not probe them even when the lookback window would otherwise reach that far + const lookbackFloor = Math.max(0, targetPeriod - MAX_CATCH_UP_LOOKBACK) let lastGoodPeriod = targetPeriod - 1 - while (lastGoodPeriod >= targetPeriod - MAX_CATCH_UP_LOOKBACK) { + while (lastGoodPeriod >= lookbackFloor) { if (await isPeriodPopulated(requestOptions, owner, topicFor(this.baseTopic, lastGoodPeriod))) { break } lastGoodPeriod-- } - if (lastGoodPeriod < targetPeriod - MAX_CATCH_UP_LOOKBACK) { + if (lastGoodPeriod < lookbackFloor) { throw new BeeError(`No populated period found within ${MAX_CATCH_UP_LOOKBACK} periods to catch up from!`) } diff --git a/test/integration/rolling-feed.spec.ts b/test/integration/rolling-feed.spec.ts index 5720a5ae..792093db 100644 --- a/test/integration/rolling-feed.spec.ts +++ b/test/integration/rolling-feed.spec.ts @@ -101,3 +101,37 @@ test('catchUp backfills a gap so the reader resolves it again', async () => { const result = await reader.downloadPayload() expect(result.payload.toUtf8()).toBe('Backfilled payload') }) + +test('catchUp is not fooled by an older buried gap', async () => { + const { writer, reader } = makeWriterAndReader() + + setPeriod(0) + await writer.uploadPayload(batch(), 'Old payload', { deferred: false }) + // period 2 is a deliberate buried gap: never written, never mirrored into + + setPeriod(3) + await writer.uploadPayload(batch(), 'Recent payload', { deferred: false }) + // populated so far: 0, 1, 3, 4 -- with a hole at 2 + + expect(await writer.isCaughtUp(6)).toBe(false) + await writer.catchUp(batch(), 6) + expect(await writer.isCaughtUp(6)).toBe(true) + + setPeriod(6) + const result = await reader.downloadPayload() + // must resume from the recent (period 4) content, not the one from before the buried gap + expect(result.payload.toUtf8()).toBe('Recent payload') +}) + +test('catchUp fails cleanly, without scanning past period 0, when nothing was ever written', async () => { + const { writer } = makeWriterAndReader() + + await expect(writer.catchUp(batch(), 5)).rejects.toThrow('No populated period found') +}) + +test('a non-positive periodLength is rejected', async () => { + const privateKey = new PrivateKey(Strings.randomHex(64)) + const writer = bee.rollingFeed.makeWriter(new Topic(Strings.randomHex(64)), privateKey, 0) + + await expect(writer.isCaughtUp()).rejects.toThrow('Period length must be greater than zero') +}) From af7fb35b95b4d6b76117a7d5c6ba2d68188d4186 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gergely=20B=C3=A9k=C3=A9si?= Date: Fri, 28 Aug 2026 15:58:55 +0200 Subject: [PATCH 4/8] fix: stabilize tests --- test/integration/rolling-feed.spec.ts | 35 +++++++++++++++------------ 1 file changed, 19 insertions(+), 16 deletions(-) diff --git a/test/integration/rolling-feed.spec.ts b/test/integration/rolling-feed.spec.ts index 792093db..5e3044ee 100644 --- a/test/integration/rolling-feed.spec.ts +++ b/test/integration/rolling-feed.spec.ts @@ -1,4 +1,4 @@ -import { Strings } from 'cafe-utility' +import { Dates, Strings, System } from 'cafe-utility' import { PrivateKey, Topic } from '@ethersphere/core-sdk' import { batch, makeBee } from '../utils' @@ -21,6 +21,12 @@ function setPeriod(period: number): void { dateNowSpy = jest.spyOn(Date, 'now').mockReturnValue(period * PERIOD_LENGTH * 1000) } +// A write's HTTP response returning is not a guarantee it's already retrievable elsewhere on +// the node (this varies by Bee build/environment) - wait for it rather than assume it. +async function waitUntil(predicate: () => Promise): Promise { + await System.waitFor(predicate, { attempts: 30, waitMillis: Dates.seconds(1) }) +} + function makeWriterAndReader() { const privateKey = new PrivateKey(Strings.randomHex(64)) const owner = privateKey.publicKey().address() @@ -38,8 +44,7 @@ test('uploadPayload / downloadPayload roundtrip', async () => { await writer.uploadPayload(batch(), 'Hello rolling feed', { deferred: false }) - const result = await reader.downloadPayload() - expect(result.payload.toUtf8()).toBe('Hello rolling feed') + await waitUntil(async () => (await reader.downloadPayload()).payload.toUtf8() === 'Hello rolling feed') }) test('uploadPayload mirrors into the next period', async () => { @@ -49,8 +54,7 @@ test('uploadPayload mirrors into the next period', async () => { await writer.uploadPayload(batch(), 'Mirrored payload', { deferred: false }) setPeriod(1001) - const result = await reader.downloadPayload() - expect(result.payload.toUtf8()).toBe('Mirrored payload') + await waitUntil(async () => (await reader.downloadPayload()).payload.toUtf8() === 'Mirrored payload') }) test('downloadPayload falls back to the previous period when the current one is empty', async () => { @@ -61,8 +65,7 @@ test('downloadPayload falls back to the previous period when the current one is // period 1001 got mirrored, period 1002 was never written at all setPeriod(1002) - const result = await reader.downloadPayload() - expect(result.payload.toUtf8()).toBe('Last known payload') + await waitUntil(async () => (await reader.downloadPayload()).payload.toUtf8() === 'Last known payload') }) test('uploadReference / downloadReference roundtrip', async () => { @@ -72,8 +75,7 @@ test('uploadReference / downloadReference roundtrip', async () => { const uploaded = await bee.data.upload(batch(), 'Referenced content') await writer.uploadReference(batch(), uploaded.reference, { deferred: false }) - const result = await reader.downloadReference() - expect(result.reference.toHex()).toBe(uploaded.reference.toHex()) + await waitUntil(async () => (await reader.downloadReference()).reference.toHex() === uploaded.reference.toHex()) }) test('isCaughtUp is true for a written/mirrored period and false past a gap', async () => { @@ -82,7 +84,7 @@ test('isCaughtUp is true for a written/mirrored period and false past a gap', as await writer.uploadPayload(batch(), 'Still going', { deferred: false }) - expect(await writer.isCaughtUp(1000)).toBe(true) + await waitUntil(async () => writer.isCaughtUp(1000)) expect(await writer.isCaughtUp(1001)).toBe(true) expect(await writer.isCaughtUp(1005)).toBe(false) }) @@ -92,14 +94,14 @@ test('catchUp backfills a gap so the reader resolves it again', async () => { setPeriod(1000) await writer.uploadPayload(batch(), 'Backfilled payload', { deferred: false }) + await waitUntil(async () => writer.isCaughtUp(1000)) expect(await writer.isCaughtUp(1003)).toBe(false) await writer.catchUp(batch(), 1003) - expect(await writer.isCaughtUp(1003)).toBe(true) + await waitUntil(async () => writer.isCaughtUp(1003)) setPeriod(1003) - const result = await reader.downloadPayload() - expect(result.payload.toUtf8()).toBe('Backfilled payload') + await waitUntil(async () => (await reader.downloadPayload()).payload.toUtf8() === 'Backfilled payload') }) test('catchUp is not fooled by an older buried gap', async () => { @@ -107,20 +109,21 @@ test('catchUp is not fooled by an older buried gap', async () => { setPeriod(0) await writer.uploadPayload(batch(), 'Old payload', { deferred: false }) + await waitUntil(async () => writer.isCaughtUp(0)) // period 2 is a deliberate buried gap: never written, never mirrored into setPeriod(3) await writer.uploadPayload(batch(), 'Recent payload', { deferred: false }) + await waitUntil(async () => writer.isCaughtUp(3)) // populated so far: 0, 1, 3, 4 -- with a hole at 2 expect(await writer.isCaughtUp(6)).toBe(false) await writer.catchUp(batch(), 6) - expect(await writer.isCaughtUp(6)).toBe(true) + await waitUntil(async () => writer.isCaughtUp(6)) setPeriod(6) - const result = await reader.downloadPayload() // must resume from the recent (period 4) content, not the one from before the buried gap - expect(result.payload.toUtf8()).toBe('Recent payload') + await waitUntil(async () => (await reader.downloadPayload()).payload.toUtf8() === 'Recent payload') }) test('catchUp fails cleanly, without scanning past period 0, when nothing was ever written', async () => { From 8d7757c863e72f071517aa05dafca8109cf22d76 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gergely=20B=C3=A9k=C3=A9si?= Date: Mon, 31 Aug 2026 12:48:38 +0200 Subject: [PATCH 5/8] fix: backfill should only write periods before current --- src/modules/rollingFeed.ts | 7 +++--- test/integration/rolling-feed.spec.ts | 31 ++++++++++++++++++++++++--- 2 files changed, 32 insertions(+), 6 deletions(-) diff --git a/src/modules/rollingFeed.ts b/src/modules/rollingFeed.ts index 0131f25c..f8c0b5dc 100644 --- a/src/modules/rollingFeed.ts +++ b/src/modules/rollingFeed.ts @@ -133,8 +133,9 @@ export class RollingFeedWriter { } /** - * Backfills every period from the last populated one (exclusive) up to `periodIdx` - * (default: current) with that period's last known payload/reference. + * 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. */ async catchUp(postageBatchId: string | BatchId, periodIdx?: number): Promise { const requestOptions = this.context.getRequestOptionsForCall() @@ -162,7 +163,7 @@ 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++) { + for (let period = lastGoodPeriod + 1; period < targetPeriod; period++) { const identifier = makeFeedIdentifier(topicFor(this.baseTopic, period), 0) await uploadSingleOwnerChunkWithWrappedChunk(requestOptions, this.signer, stamp, identifier, sourceChunk) } diff --git a/test/integration/rolling-feed.spec.ts b/test/integration/rolling-feed.spec.ts index 5e3044ee..fd756080 100644 --- a/test/integration/rolling-feed.spec.ts +++ b/test/integration/rolling-feed.spec.ts @@ -98,12 +98,34 @@ test('catchUp backfills a gap so the reader resolves it again', async () => { expect(await writer.isCaughtUp(1003)).toBe(false) await writer.catchUp(batch(), 1003) - await waitUntil(async () => writer.isCaughtUp(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) 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('catchUp does not clobber fresh data already written to its target period', async () => { + const { writer, reader } = makeWriterAndReader() + + setPeriod(0) + await writer.uploadPayload(batch(), 'Old payload', { deferred: false }) + await waitUntil(async () => writer.isCaughtUp(0)) + // period 1 is mirrored from period 0; periods 2 are a gap; the writer resumes at 3 + + setPeriod(3) + await writer.uploadPayload(batch(), 'Fresh payload', { deferred: false }) + await waitUntil(async () => (await reader.downloadPayload()).payload.toUtf8() === 'Fresh payload') + + // catching up to the very period that was just written with fresh data must not overwrite it + await writer.catchUp(batch(), 3) + + const result = await reader.downloadPayload() + expect(result.payload.toUtf8()).toBe('Fresh payload') +}) + test('catchUp is not fooled by an older buried gap', async () => { const { writer, reader } = makeWriterAndReader() @@ -119,10 +141,13 @@ test('catchUp is not fooled by an older buried gap', async () => { expect(await writer.isCaughtUp(6)).toBe(false) await writer.catchUp(batch(), 6) - await waitUntil(async () => writer.isCaughtUp(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) setPeriod(6) - // must resume from the recent (period 4) content, not the one from before the buried gap + // 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 await waitUntil(async () => (await reader.downloadPayload()).payload.toUtf8() === 'Recent payload') }) From 4b87a049c3b7b736f502fe16d66d95d7e65148f5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gergely=20B=C3=A9k=C3=A9si?= Date: Mon, 31 Aug 2026 12:57:02 +0200 Subject: [PATCH 6/8] fix: heartbeat bug --- src/modules/rollingFeed.ts | 12 +++++++---- test/integration/rolling-feed.spec.ts | 31 +++++++++++++++++++-------- 2 files changed, 30 insertions(+), 13 deletions(-) diff --git a/src/modules/rollingFeed.ts b/src/modules/rollingFeed.ts index f8c0b5dc..351eab8d 100644 --- a/src/modules/rollingFeed.ts +++ b/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 { 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) } diff --git a/test/integration/rolling-feed.spec.ts b/test/integration/rolling-feed.spec.ts index fd756080..f37d23e9 100644 --- a/test/integration/rolling-feed.spec.ts +++ b/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') }) From 8a4d63ff8d83c5b2a513ccf82d3d4d8398fe0e40 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gergely=20B=C3=A9k=C3=A9si?= Date: Mon, 31 Aug 2026 13:54:06 +0200 Subject: [PATCH 7/8] fix: backfill logic --- src/modules/rollingFeed.ts | 19 ++++++++------ test/integration/rolling-feed.spec.ts | 36 +++++++++++++++++---------- 2 files changed, 34 insertions(+), 21 deletions(-) diff --git a/src/modules/rollingFeed.ts b/src/modules/rollingFeed.ts index 351eab8d..b516668b 100644 --- a/src/modules/rollingFeed.ts +++ b/src/modules/rollingFeed.ts @@ -133,10 +133,12 @@ export class RollingFeedWriter { } /** - * 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`. + * 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 - keeping it populated during silence is the caller's job (a + * periodic `uploadPayload`/`uploadReference` heartbeat), not catchUp's. A conditional + * "write it if empty" would race that heartbeat: there's no compare-and-swap on a SOC + * address, so a stale write can still land after a concurrent fresh one and shadow it. */ async catchUp(postageBatchId: string | BatchId, periodIdx?: number): Promise { const requestOptions = this.context.getRequestOptionsForCall() @@ -160,14 +162,15 @@ export class RollingFeedWriter { throw new BeeError(`No populated period found within ${MAX_CATCH_UP_LOOKBACK} periods to catch up from!`) } + if (lastGoodPeriod === targetPeriod - 1) { + return // no gap - nothing to backfill + } + const sourceTopic = topicFor(this.baseTopic, lastGoodPeriod) const { feedIndex: sourceIndex } = await probeFeed(requestOptions, owner, sourceTopic) const sourceChunk = await downloadFeedUpdateAsCAC(requestOptions, owner, sourceTopic, sourceIndex) - const targetAlreadyWritten = await isPeriodPopulated(requestOptions, owner, topicFor(this.baseTopic, targetPeriod)) - const backfillEnd = targetAlreadyWritten ? targetPeriod : targetPeriod + 1 - - for (let period = lastGoodPeriod + 1; period < backfillEnd; period++) { + for (let period = lastGoodPeriod + 1; period < targetPeriod; period++) { const identifier = makeFeedIdentifier(topicFor(this.baseTopic, period), 0) await uploadSingleOwnerChunkWithWrappedChunk(requestOptions, this.signer, stamp, identifier, sourceChunk) } diff --git a/test/integration/rolling-feed.spec.ts b/test/integration/rolling-feed.spec.ts index f37d23e9..be7ba256 100644 --- a/test/integration/rolling-feed.spec.ts +++ b/test/integration/rolling-feed.spec.ts @@ -89,7 +89,7 @@ test('isCaughtUp is true for a written/mirrored period and false past a gap', as expect(await writer.isCaughtUp(1005)).toBe(false) }) -test('catchUp backfills a gap so the reader resolves it again', async () => { +test('catchUp backfills a gap without touching its target period', async () => { const { writer, reader } = makeWriterAndReader() setPeriod(1000) @@ -98,43 +98,51 @@ test('catchUp backfills a gap so the reader resolves it again', async () => { expect(await writer.isCaughtUp(1003)).toBe(false) await writer.catchUp(batch(), 1003) - // 1003 wasn't written yet, so catchUp fills it too, not just the gap strictly before it - await waitUntil(async () => writer.isCaughtUp(1003)) + // catchUp never writes its own target (1003) - only the gap strictly before it + await waitUntil(async () => writer.isCaughtUp(1002)) + expect(await writer.isCaughtUp(1003)).toBe(false) 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 () => { +test('heartbeat (periodic uploadPayload) plus catchUp recovers cleanly after silence', 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) + // two periods of silence: 1001 is mirrored, 1002 is a genuine gap, 1003 is where we resume + setPeriod(1003) expect(await writer.isCaughtUp()).toBe(false) + + // per ROLLING_FEED.md: keeping the current period alive during silence is the caller's + // job (a heartbeat republish), catchUp only backfills what's strictly before it (1002 here) + await writer.uploadPayload(batch(), 'Heartbeat payload', { deferred: false }) await writer.catchUp(batch()) - await waitUntil(async () => writer.isCaughtUp()) + await waitUntil(async () => writer.isCaughtUp(1002)) + expect(await writer.isCaughtUp()).toBe(true) 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 () => { +test('catchUp cannot clobber a concurrent write to its target period', async () => { const { writer, reader } = makeWriterAndReader() setPeriod(0) await writer.uploadPayload(batch(), 'Old payload', { deferred: false }) await waitUntil(async () => writer.isCaughtUp(0)) - // period 1 is mirrored from period 0; periods 2 are a gap; the writer resumes at 3 + // period 1 is mirrored from period 0; period 2 is a gap; the writer resumes at 3 setPeriod(3) await writer.uploadPayload(batch(), 'Fresh payload', { deferred: false }) await waitUntil(async () => (await reader.downloadPayload()).payload.toUtf8() === 'Fresh payload') - // catching up to the very period that was just written with fresh data must not overwrite it + // catching up to the very period that was just written with fresh data must not touch it, + // regardless of call order - there's no way to safely check-then-write around a concurrent writer await writer.catchUp(batch(), 3) const result = await reader.downloadPayload() @@ -156,11 +164,13 @@ test('catchUp is not fooled by an older buried gap', async () => { expect(await writer.isCaughtUp(6)).toBe(false) await writer.catchUp(batch(), 6) - // 6 wasn't written yet, so catchUp fills it too, not just the gap strictly before it - await waitUntil(async () => writer.isCaughtUp(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) setPeriod(6) - // must resume from the recent (period 4) content, not the one from before the buried gap + // 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 await waitUntil(async () => (await reader.downloadPayload()).payload.toUtf8() === 'Recent payload') }) From b7603030a7e7a2788bc45bc73a4ee2faff995205 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gergely=20B=C3=A9k=C3=A9si?= Date: Tue, 1 Sep 2026 11:16:53 +0200 Subject: [PATCH 8/8] fix: bound catchUp's scan and backfill depth --- src/modules/rollingFeed.ts | 47 ++++++++++++++++++--------- test/integration/rolling-feed.spec.ts | 27 +++++++++++++++ 2 files changed, 59 insertions(+), 15 deletions(-) diff --git a/src/modules/rollingFeed.ts b/src/modules/rollingFeed.ts index b516668b..b9a28f19 100644 --- a/src/modules/rollingFeed.ts +++ b/src/modules/rollingFeed.ts @@ -20,9 +20,9 @@ import { BeeRequestOptions, UploadResult } from '../types' import { BeeResponseError } from '../utils/error' import { BeeContext } from './context' -// Not yet finalized (see ROLLING_FEED.md "Open parameters"): bounds how far catchUp/isCaughtUp -// will scan/backfill so a long-dead writer can't trigger an unbounded loop. -const MAX_CATCH_UP_LOOKBACK = 1000 +// RollingFeedReader.downloadPayload/downloadReference only ever fall back one period, so +// backfilling deeper than that serves no reader by default - see catchUp's maxBackfill param. +const DEFAULT_MAX_BACKFILL = 1 function periodIndex(t: number, periodLength: number): number { if (periodLength <= 0) { @@ -133,14 +133,23 @@ 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 - keeping it populated during silence is the caller's job (a + * Backfills up to `maxBackfill` periods strictly between the last populated one and + * `periodIdx` (default: current) with that period's last known payload/reference. Never + * writes `periodIdx` itself - keeping it populated during silence is the caller's job (a * periodic `uploadPayload`/`uploadReference` heartbeat), not catchUp's. A conditional * "write it if empty" would race that heartbeat: there's no compare-and-swap on a SOC * address, so a stale write can still land after a concurrent fresh one and shadow it. + * + * `maxBackfill` bounds both the backward scan and the backfill itself, since scanning + * further back than you're willing to backfill only wastes round trips. Defaults to 1, + * matching the reader's one-period fallback - nothing deeper is ever read anyway. Throws + * if no populated period is found within that bound. */ - async catchUp(postageBatchId: string | BatchId, periodIdx?: number): Promise { + async catchUp( + postageBatchId: string | BatchId, + periodIdx?: number, + maxBackfill = DEFAULT_MAX_BACKFILL, + ): Promise { const requestOptions = this.context.getRequestOptionsForCall() const stamp = new BatchId(postageBatchId) const owner = this.signer.publicKey().address() @@ -148,18 +157,18 @@ export class RollingFeedWriter { // periods before 0 can't exist (period index is derived from Unix time), so the scan // must not probe them even when the lookback window would otherwise reach that far - const lookbackFloor = Math.max(0, targetPeriod - MAX_CATCH_UP_LOOKBACK) + const scanFloor = Math.max(0, targetPeriod - 1 - maxBackfill) let lastGoodPeriod = targetPeriod - 1 - while (lastGoodPeriod >= lookbackFloor) { + while (lastGoodPeriod >= scanFloor) { if (await isPeriodPopulated(requestOptions, owner, topicFor(this.baseTopic, lastGoodPeriod))) { break } lastGoodPeriod-- } - if (lastGoodPeriod < lookbackFloor) { - throw new BeeError(`No populated period found within ${MAX_CATCH_UP_LOOKBACK} periods to catch up from!`) + if (lastGoodPeriod < scanFloor) { + throw new BeeError(`No populated period found within ${maxBackfill} periods to catch up from!`) } if (lastGoodPeriod === targetPeriod - 1) { @@ -170,10 +179,18 @@ 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 identifier = makeFeedIdentifier(topicFor(this.baseTopic, period), 0) - await uploadSingleOwnerChunkWithWrappedChunk(requestOptions, this.signer, stamp, identifier, sourceChunk) - } + const periodsToBackfill = Array.from( + { length: targetPeriod - 1 - lastGoodPeriod }, + (_, i) => lastGoodPeriod + 1 + i, + ) + + await Promise.all( + periodsToBackfill.map(async period => { + const identifier = makeFeedIdentifier(topicFor(this.baseTopic, period), 0) + + return uploadSingleOwnerChunkWithWrappedChunk(requestOptions, this.signer, stamp, identifier, sourceChunk) + }), + ) } private currentAndNextTopics(options?: FeedUploadOptions): { diff --git a/test/integration/rolling-feed.spec.ts b/test/integration/rolling-feed.spec.ts index be7ba256..71e07d68 100644 --- a/test/integration/rolling-feed.spec.ts +++ b/test/integration/rolling-feed.spec.ts @@ -180,6 +180,33 @@ test('catchUp fails cleanly, without scanning past period 0, when nothing was ev await expect(writer.catchUp(batch(), 5)).rejects.toThrow('No populated period found') }) +test('catchUp respects the default maxBackfill: a 2-period gap is out of range', async () => { + const { writer } = makeWriterAndReader() + + setPeriod(1000) + await writer.uploadPayload(batch(), 'Old payload', { deferred: false }) + await waitUntil(async () => writer.isCaughtUp(1000)) + // populated: 1000, 1001 (mirror). Resuming at 1004 needs a 2-period backfill (1002, 1003), + // which is past the default maxBackfill of 1 + + await expect(writer.catchUp(batch(), 1004)).rejects.toThrow('No populated period found') +}) + +test('catchUp with an explicit maxBackfill bridges a deeper gap', async () => { + const { writer, reader } = makeWriterAndReader() + + setPeriod(1000) + await writer.uploadPayload(batch(), 'Old payload', { deferred: false }) + await waitUntil(async () => writer.isCaughtUp(1000)) + + await writer.catchUp(batch(), 1004, 2) + await waitUntil(async () => writer.isCaughtUp(1003)) + + setPeriod(1004) + // reader falls back from the still-empty 1004 to the now-backfilled 1003 + await waitUntil(async () => (await reader.downloadPayload()).payload.toUtf8() === 'Old payload') +}) + test('a non-positive periodLength is rejected', async () => { const privateKey = new PrivateKey(Strings.randomHex(64)) const writer = bee.rollingFeed.makeWriter(new Topic(Strings.randomHex(64)), privateKey, 0)