feat: rolling feed

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

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.
    *

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'

src/modules/rollingFeed.ts

@@ -0,0 +1,255 @@
+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'
+
+// 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) {
+    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<boolean> {
+  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<FeedReferenceResult> {
+  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<UploadResult> {
+    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<UploadResult> {
+    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<boolean> {
+    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 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,
+    maxBackfill = DEFAULT_MAX_BACKFILL,
+  ): Promise<void> {
+    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)
+
+    // 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 scanFloor = Math.max(0, targetPeriod - 1 - maxBackfill)
+    let lastGoodPeriod = targetPeriod - 1
+
+    while (lastGoodPeriod >= scanFloor) {
+      if (await isPeriodPopulated(requestOptions, owner, topicFor(this.baseTopic, lastGoodPeriod))) {
+        break
+      }
+      lastGoodPeriod--
+    }
+
+    if (lastGoodPeriod < scanFloor) {
+      throw new BeeError(`No populated period found within ${maxBackfill} 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 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): {
+    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<FeedUpdateOptions, 'index'>): Promise<FeedPayloadResult> {
+    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<FeedReferenceResult> {
+    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))
+    }
+  }
+}