Skip to content
Durability

Replayable event stream

Partition retained events, move cold segments to R2, and advance consumer cursors only after work succeeds.

Updated

This recipe builds a topic from Durable Object partitions rather than treating a topic as a global queue. A partition accepts ordered appends, holds a bounded hot log, writes older segments to R2, and stores an independent cursor for each consumer group. Producers route batches to one partition; consumers read, perform side effects, then commit the cursor.

Read the complete implementation at examples/event-topics/worker.ts and deploy its Durable Object and R2 bindings from examples/event-topics/wrangler.jsonc. These sources are recipes, not a hosted event-stream product: authentication, leases, reconciliation, and external retention policy are application responsibilities.

Use case and boundaries

Use this shape when events need replay within a topic partition, multiple consumer groups need independent progress, and a finite hot window should spill to cheaper cold storage. It provides ordering within one topic:partition object only. It has no global ordering, no cross-partition atomicity, and no distributed transaction between the log, an external consumer side effect, and R2.

There are a fixed eight partitions per topic. The shard name is topic:partition, where a stable FNV-style hash of partitionKey selects the partition. Topics are created by writing to those object names; there is no partition registry or rebalance. Changing the partition count or hash is therefore an application data-migration decision, not a safe configuration tweak.

Architecture

How the event stream uses its Durable Objects

Shows Producer. Focuses on Producer.Producerfetch POST

1/7 · Group events by partition

A topic has eight partitions. The Worker hashes each event’s partitionKey to one of them and groups the events by topic:partition name, because one push may carry events for one partition only.

Open source ↗
import { DurableObject } from 'cloudflare:workers';
import { timeout } from '@durability/transforms';
import {
  DurabilityAlarms,
  DurabilityLog,
  DurabilityScheduler,
  type LogAppendInput,
  type LogPage,
  type LogRecord,
} from 'durability';
import { DurabilityRouting } from 'durability/routing';

type Env = {
  COLD: R2Bucket;
  PARTITIONS: DurableObjectNamespace<Partition>;
};

type Event = {
  topic: string;
  /** Events sharing a partition key land on the same partition. */
  partitionKey: string;
  /** Already-encoded body: a log stores bytes, not the producer's types. */
  payload: string;
};

type Append = {
  events: LogAppendInput<Event>[];
  region?: DurableObjectLocationHint;
};

/** Partitions per topic. Adding one is a routing decision, not a deploy. */
const partitionCount = 8;
const coldPrefix = 'segments/';
// Retention deletes by age and count. It is not Kafka-style key compaction,
// which keeps the latest record per key and is not implemented here.
const retentionSweepMs = 60_000;

const partitionFor = (key: string): number => {
  let hash = 2_166_136_261;
  for (let index = 0; index < key.length; index += 1) {
    hash ^= key.charCodeAt(index);
    hash = Math.imul(hash, 16_777_619);
  }
  return Math.abs(hash) % partitionCount;
};

/**
 * One partition of one topic. The application owns this class and composes the
 * capabilities it wants: a log for retained history, alarms for compaction, and
 * routing's load observer so producers can see this partition's backlog.
 */
export class Partition extends DurableObject<Env> {
  private readonly scheduler = new DurabilityScheduler({ context: this.ctx });
  private readonly routing = new DurabilityRouting();
  private readonly log = new DurabilityLog<Event>({
    scheduler: this.scheduler,
    routing: this.routing,
    retention: {
      maxAgeMs: 7 * 24 * 60 * 60 * 1_000,
      maxRecords: 50_000,
      // The object keeps a hot window; everything older lives in R2.
      maxBytes: 256 * 1024 * 1024,
    },
    // Flushed segments stay readable, so a consumer that falls behind the hot
    // window is served from R2 instead of losing records.
    cold: {
      write: async ({ records, firstOffset }) => {
        const locator = `${coldPrefix}${this.ctx.id.toString()}/${firstOffset}`;
        await this.env.COLD.put(locator, JSON.stringify(records), {
          onlyIf: { etagDoesNotMatch: '*' },
        });
        return locator;
      },
      read: async (locator) => {
        const stored = await this.env.COLD.get(locator);
        return stored ? await stored.json<LogRecord<Event>[]>() : undefined;
      },
    },
  });

  // Retention needs a schedule, and a named alarm is that schedule.
  private readonly alarms = new DurabilityAlarms({
    scheduler: this.scheduler,
    handlers: {
      enforceRetention: async () => {
        await this.log.trim();
        // Compare the hot window, not the oldest readable offset: flushed
        // segments keep that one low forever, which would sweep forever.
        const { hotOffset, nextOffset } = await this.log.bounds();
        if (hotOffset < nextOffset) {
          await this.alarms.enforceRetention(Date.now() + retentionSweepMs);
          return;
        }
        // Nothing left to expire, so stop re-arming and let the partition
        // hibernate until the next append schedules another sweep.
        await this.ctx.storage.delete('sweeping');
      },
    },
  });

  /** Producers reach this through routing; the result carries acceptance + load. */
  async append(batch: Append) {
    const result = await this.log.append(batch.events);
    // Scheduling a named alarm replaces its pending time, so re-arming on
    // every append would postpone retention for as long as traffic continues.
    if (result.success && !(await this.ctx.storage.get<boolean>('sweeping'))) {
      await this.ctx.storage.put('sweeping', true);
      await this.alarms.enforceRetention(Date.now() + retentionSweepMs);
    }
    return result;
  }

  /**
   * Consumers pull. A group's cursor is the only thing that advances, and a
   * cursor below the hot window is served from R2 rather than failing.
   */
  async consume(group: string, limit: number): Promise<LogPage<Event>> {
    // A new group starts at the oldest readable record, flushed or not.
    const from =
      (await this.log.cursor(group)) ?? (await this.log.bounds()).oldestOffset;
    return this.log.read({ from, limit });
  }

  /**
   * Mirrors the bucket's lifecycle rule. R2 expires the objects on its own
   * schedule, so the index must forget the same segments or a read would
   * chase a locator whose object is gone.
   */
  expireCold(before: number) {
    return this.log.forgetColdBefore(before);
  }

  /** Acknowledging is separate from reading, so a crash re-reads the page. */
  commit(group: string, offset: number) {
    return this.log.commit(group, offset);
  }

  cursors() {
    return this.log.cursors();
  }

  bounds() {
    return this.log.bounds();
  }

  load() {
    return this.log.load();
  }

  alarm(info?: AlarmInvocationInfo) {
    return this.scheduler.alarm(info);
  }
}

/**
 * Stands in for the consumer's real work. Records arrive in offset order
 * within a partition, and may be redelivered after a failure.
 */
const handle = async (records: LogRecord<Event>[]): Promise<void> => {
  for (const record of records) {
    console.log(`handled ${record.offset} of ${record.body.topic}`);
  }
};

/**
 * Producer side. The shard key is `topic:partition`, so topics are created by
 * writing to them: no partition table, no rebalance, and idle partitions
 * hibernate. `invoke` picks the application method, so routing works with the
 * log exactly as it does with fanout.
 */
const topics = DurabilityRouting.client({
  target: Partition,
  invoke: (stub, input: Append) => stub.append(input),
  sharding: (input) => {
    const first = input.events[0]?.body;
    if (!first) {
      throw new Error('An append must carry at least one event');
    }
    const shard = `${first.topic}:${partitionFor(first.partitionKey)}`;
    return input.region === undefined
      ? shard
      : { shard, locationHint: input.region };
  },
});

export default {
  async fetch(request, env): Promise<Response> {
    const url = new URL(request.url);
    const topic = url.searchParams.get('topic') ?? 'events';

    // Demo input only: authenticate and validate untrusted payloads.
    if (request.method === 'POST') {
      const events = await request.json<
        Array<{ partitionKey: string; payload: string; key?: string }>
      >();
      // A batch must share a partition, so group by key before pushing.
      const byPartition = new Map<string, LogAppendInput<Event>[]>();
      for (const event of events) {
        const shard = `${topic}:${partitionFor(event.partitionKey)}`;
        const body: Event = {
          topic,
          partitionKey: event.partitionKey,
          payload: event.payload,
        };
        byPartition.set(shard, [
          ...(byPartition.get(shard) ?? []),
          event.key === undefined
            ? { body }
            : { deduplicationKey: event.key, body },
        ]);
      }
      const accepted = await Promise.all(
        [...byPartition.values()].map((group) =>
          topics.with(timeout, 5_000).push({ events: group })
        )
      );
      // A rejected push is definite; a thrown one is uncertain and may retry.
      return Response.json(accepted, {
        status: accepted.every((result) => result.success) ? 202 : 409,
      });
    }

    if (request.method !== 'GET') {
      return new Response('POST events or GET ?group=', { status: 405 });
    }

    // Each partition is polled independently: a slow group never blocks others.
    const group = url.searchParams.get('group') ?? 'default';
    const partitions = [...Array(partitionCount).keys()].map((index) =>
      env.PARTITIONS.getByName(`${topic}:${index}`)
    );
    const pages = await Promise.all(
      partitions.map(async (stub) => {
        const page = await stub.consume(group, 100);
        if (page.records.length === 0) {
          return { nextOffset: page.nextOffset, lag: page.lag, handled: 0 };
        }
        // Commit only once the work succeeded. Committing first would make
        // this at-most-once, silently skipping a page on a crash. Throwing
        // instead re-reads it, so processing must tolerate duplicates.
        await handle(page.records);
        await stub.commit(group, page.nextOffset);
        return {
          nextOffset: page.nextOffset,
          lag: page.lag,
          handled: page.records.length,
        };
      })
    );
    return Response.json(pages);
  },
} satisfies ExportedHandler<Env>;
Append a batch and schedule retention
Details

Follow one producer POST into one partition’s storage, and the first append that schedules a retention sweep.

Durable state
Nothing is written while grouping.
In-flight / in memory
The byPartition Map exists only in this request.

Click the diagram to open a full-screen explorer with the complete source files. Select a component to jump to its code, or select a linked code line to highlight the affected architecture. Use the ← / → keys or on-screen arrows to move through the flow; each step’s explanation sits under the diagram. Escape or the close button returns to the guide. Failure scenarios, the full step list, and durable/in-memory state notes are available under Details. These are explanatory traces of the real source, not executable simulations or guarantees about an external service.

The DurabilityRouting observer is in memory. It helps route a live producer request and report load; it is not a durable routing table. The DurabilityLog and named retention alarm share one scheduler in each Partition, because that object has one physical alarm. Different partitions are independent objects, not members of one scheduler or transaction.

Producer path: one batch, one shard

  1. A producer POST supplies events with a partitionKey, encoded payload, and optionally a producer key used as a log deduplicationKey. The demo endpoint must be replaced with real authentication, authorization, size limits, and schema validation.
  2. The Worker selects the topic and calculates the fixed partition for every event.
  3. It groups events by topic:partition. This is essential: the routing client’s shard function uses the first event, so each push contains events for exactly one shard.
  4. It makes one routed push for each group, with a five-second timeout. Every group is accepted or rejected separately; a POST spanning partitions has no atomic outcome.
  5. In the partition, log.append() persists the batch. A return with success: false is a definite rejection. A thrown/timeout result is uncertain and a producer may retry.
  6. On a successful append, the partition sets a sweeping flag and registers a named retention alarm only when one is not already pending. It does not re-arm on every append, because replacing a named alarm continually would defer trimming forever under constant traffic.
  7. The append response exposes acceptance and load. Acceptance means the log persisted the input, not that every consumer handled it or a cold object has later expired.

The grouping and single-shard push are the crucial producer-side constraint:

byPartition.set(shard, [...(byPartition.get(shard) ?? []), input]);
const accepted = await Promise.all(
  [...byPartition.values()].map((events) => topics.push({ events }))
);

Give retries the same deduplicationKey when the producer can safely identify one logical event. A timeout alone cannot prove whether an append was accepted, so keys must be stable across a retry and their retention must match the producer’s retry horizon.

Retention, hot storage, and cold storage

The log’s configured hot bounds are seven days, 50,000 records, and 256 MiB. Those settings define what trim() may move out of the hot window; they do not automatically schedule retention. The enforceRetention alarm calls log.trim() every 60 seconds while records remain hot. If no hot records remain, it removes sweeping and lets the object hibernate.

When trimming flushes records, the cold writer stores JSON in R2 under segments/<object-id>/<firstOffset>. Its etagDoesNotMatch: '*' conditional write makes a repeat flush for that first offset non-destructive. A reader can serve a cursor below the hot window from the stored segment, so lagging groups can replay it.

The alarm compares hotOffset with nextOffset, not oldestOffset. The latter includes cold-readable records and can remain low indefinitely; using it would keep scheduling an empty retention sweep forever.

R2 object expiration is external policy. Configure and monitor a bucket lifecycle rule (or another application process), then call expireCold(before) / log.forgetColdBefore to remove matching index entries. Expiring R2 objects alone leaves locators that reads may follow to missing data; forgetting first makes retained data unreachable prematurely.

Consumer path: read, handle, commit

  1. A poller chooses a consumer group and polls every one of the topic’s eight object names independently. A slow partition or group does not block the others.
  2. consume(group, limit) uses that group’s cursor, or oldestOffset for a new group. It reads from hot storage or a cold segment as needed.
  3. The poller calls its business handler for the returned page in offset order within that partition.
  4. Only after the handler succeeds does it call commit(group, page.nextOffset).
  5. A crash before commit causes the next poll to read the same page. The handler must tolerate duplicate records, normally with an idempotency key or an inbox record.
  6. Each group has its own cursor, so one group can replay or lag without advancing another.

The source deliberately places commit after side effects:

const page = await stub.consume(group, 100);
await handle(page.records);
await stub.commit(group, page.nextOffset);

Do not run two uncoordinated pollers for the same (topic, partition, group). Both can read the same cursor before either commit, perform side effects, and race to advance it. Use a lease/fencing-token protocol, a single assigned poller, or an idempotent handler designed for concurrent duplicate delivery.

Naive failure: commit first or rely on memory

A plausible naive worker keeps an in-memory queue and advances a consumer offset before calling the downstream service. Restarting loses the queue; a crash after the early commit silently loses the event. This illustrates that design’s failure sequence, not a statement about generic competing products.

The walkthrough’s “An in-memory queue that commits first” scenario steps through that design’s gaps and the lines of this composition that close them.

The composition persists records and separates cursor movement from read. A failure after the external side effect but before commit produces redelivery instead of silent loss, so the receiver must deduplicate the record.

The walkthrough’s “Crash after work, before commit” scenario steps through this recovery: the poll reads a page, the handler’s side effects complete, the request dies before commit(), the next poll reads the same page from the unchanged cursor, and only the receiver can avoid applying it twice.

Failure and recovery decisions

Failure or edge What this composition does Application decision still needed
Producer timeout The acceptance result is uncertain; retry can reach the same partition. Supply stable deduplicationKey values and decide producer retry limits.
Batch spans partitions Groups and pushes separately; there is no cross-partition atomicity. Design compensations or a single partition for work needing one atomic boundary.
Consumer crashes before commit It re-reads the page from its independent cursor. Make side effects idempotent and add leases if pollers can overlap.
Consumer commits before handling Not done by this recipe; doing so makes delivery at-most-once. Preserve handle-then-commit ordering and monitor handler failures.
Hot retention runs trim() moves eligible records to R2 and periodically re-arms only while hot data exists. Operate a reconciliation process; retention is not automatic from configuration alone.
R2 lifecycle expires cold data The recipe offers expireCold to forget matching indexes. Set external lifecycle policy and coordinate its cutoff with replay commitments.
sweeping scheduling race Flag write and alarm registration are not one atomic application transaction. Reconcile partitions in production: inspect bounds/flag and schedule a missed sweep.
Retention sweep keeps failing After five failed attempts, or one over the five-minute timeout, the alarm row is failed. sweeping stays set, so appends stop scheduling sweeps; the same reconciliation applies.

Those last two rows are unresolved source gaps, not automatic recovery; the walkthrough’s “Where retention can stop” scenario steps through both. For example, a deployment can periodically enumerate known topic/partition names, query bounds, and schedule a sweep when retained hot data exists without a valid sweep schedule. Make that repair idempotent and observable; it cannot be inferred solely from an in-memory routing observer.

Provisioning and security checklist

  • Bind and migrate the Partition Durable Object, and bind the COLD R2 bucket exactly as shown in the Wrangler config. Restrict access to both bindings.
  • Authenticate producers and consumers; authorize topic and consumer-group names rather than letting request parameters become unrestricted Durable Object names.
  • Enforce event schemas, encoded-byte limits, per-topic quotas, and safe payload logging.
  • Keep partition count and hash algorithm versioned. A change requires a read/write migration plan because it changes ordering and ownership.
  • Use stable producer keys, an idempotent consumer inbox/effect key, and a lease or fencing token for same-group pollers.
  • Configure R2 lifecycle retention independently, run cold-index expiration, and document the maximum replay window promised to each group.
  • Monitor append rejections/timeouts, per-group lag, hot/cold bounds, trim failures, R2 conditional-write conflicts, cursor age, and reconciliation repairs.

Primitive map

This design composes routing, log, alarms, and the shared scheduler. Those primitives provide durable local boundaries; they do not create a global ordered log or distributed transaction across partitions and external services.

Edit this page
Durability

Type to search…

↑↓ navigate↵ selectEsc close