Skip to content
Durability

DurabilityLog

DurabilityLog API, behavior, and recovery contract.

Import DurabilityLog from durability. Use retained records when consumers need independent positions and replay. See the event-stream composition.

How it uses your Durable Object

DurabilityLog adds no stream service of its own. It uses the Durable Object it is constructed in:

  • ctx.storage.sql holds the hot records, the offset sequence, every consumer’s cursor, the index of flushed segments, and, with leases, each held range. A partition’s history and positions belong to that object alone.
  • The object’s one alarm is never set by the log. It registers no due work with the scheduler, so retention runs only when your code calls trim(), typically from a DurabilityAlarms handler whose alarm() your class delegates to scheduler.alarm().
  • RPC from outside: producers call your object to append and pollers call it to read and commit. Cold segments are the only external I/O, written by trim() and read back for consumers behind the hot window.
  • The live instance holds only configuration. Lease expiry is a stored timestamp rather than a timer, so eviction loses nothing the log needs. A producer whose reply was lost retries with the same deduplication keys.

How DurabilityLog uses your Durable Object

Shows Partition, DurabilityLog. Focuses on DurabilityLog.One Durable ObjectPartitionyour Durable ObjectDurabilityLogholds config onlylog calls

1/8 · The log belongs to the object instance

Partition builds a scheduler from this.ctx and attaches the log to it. The log reads and writes only this object’s storage; another partition has its own records and cursors.

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, read, and commit
Details

Follow one batch of events into a partition and one consumer group reading it.

Durable state
Nothing is written yet.
In-flight / in memory
The helper holds only its configuration and is rebuilt with every new instance.

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.

Behind the scenes

Durable state: durability_log_records holds hot records by offset, with a unique retained deduplication key. Cursor, sequence/byte-count, and cold-segment metadata tables track readers and storage. Optional leasing adds durable lease/allocation tables.

append() allocates offsets and inserts the batch atomically. Reading does not acknowledge anything: commit() only advances one consumer cursor. Optional lease() persists an exclusive range before returning records; fresh batch IDs fence expired or superseded holders, and acknowledgements advance only across contiguous settled ranges.

Not an automatic background loop: the log can share a scheduler’s storage engine but does not register a due-work participant. Nothing periodically calls trim() for you. Compose a named alarm when retention needs a clock.

Cold-storage failure: the external segment is written before its locator is committed and hot records are removed. A failed write preserves hot data. A crash after the remote write may repeat it or leave an orphan; use idempotent segment keys. The cold store’s lifecycle and removal of obsolete locators are application policy.

Lifetime: retention ignores consumer cursors. A slow reader can lose readable history and gets LogTruncatedError. Append deduplication ends when the original hot record is flushed or deleted. Processing can repeat after an effect succeeds but before cursor/lease settlement, so effects still need idempotency.

Implementation: log.ts (appendRecords, read, trim, lease, ack, nack).

Retained log

Fanout deletes a record once every target settles it, which is what a work queue wants. A log wants the opposite: records are retained so a consumer can rewind, a new consumer can start from the beginning, and several consumers can read the same history at different speeds. DurabilityLog is that second shape, and it attaches to the same scheduler.

import { DurabilityLog, DurabilityScheduler, type LogRecord } from 'durability';

export class Partition extends DurableObject<Env> {
  private readonly scheduler = new DurabilityScheduler({ context: this.ctx });
  private readonly log = new DurabilityLog<Event>({
    scheduler: this.scheduler,
    retention: {
      maxAgeMs: 7 * 24 * 60 * 60 * 1_000,
      maxRecords: 100_000,
      maxBytes: 512 * 1024 * 1024,
    },
    // Flushed records stay readable: `write` returns a locator the log keeps,
    // and `read` hands the segment back when a consumer asks for it.
    cold: {
      write: async ({ records, firstOffset }) => {
        const locator = `segment/${this.ctx.id}/${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;
      },
    },
  });

  append(events: Event[]) {
    return this.log.append(events.map((body) => ({ body })));
  }
}

append mirrors enqueue: the batch commits atomically, definite input rejections return success: false, and storage or commit uncertainty throws. It returns each record’s offset in input order. Passing a deduplicationKey makes an append idempotent — re-appending an existing key returns its original offset and consumes no new one — so a producer that retries after an uncertain commit does not duplicate records.

Consumers pull and own their position:

const from = (await log.cursor('search')) ?? (await log.bounds()).oldestOffset;
const page = await log.read({ from, limit: 100 });
await index(page.records);
await log.commit('search', page.nextOffset);

commit only ever advances, so a late or duplicated commit cannot rewind a cursor. Cursors are independent: log.cursors() returns every consumer’s position, and log.load() reports the furthest-behind cursor as backlog so routing can weigh partitions by consumer lag rather than by queue depth.

Retention is not acknowledgement

Records leave only through trim(), which enforces maxAgeMs and maxRecords, oldest first. Retention is deliberately not automatic, because no capability should schedule work an application did not ask for. A DurabilityAlarms handler is the composition that gives it a schedule, and it keeps alarm() a pure delegation:

private readonly alarms = new DurabilityAlarms({
  scheduler: this.scheduler,
  handlers: { compact: () => this.log.trim() },
});

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

maxBytes is the budget that actually matters, because a Durable Object holds a hot window rather than an unbounded log. It counts payload bytes held in the object and is capped at maxLogBytes (1 GiB), leaving room for the other capabilities sharing that object’s storage.

Flushing to cold storage, and reading it back

Retention has two possible destinations. Without cold, expiring records are deleted. With cold, they are flushed and stay readable:

type LogColdStorage<Body> = {
  write(segment: LogSegment<Body>): Promise<string>;
  read(locator: string): Promise<LogRecord<Body>[] | undefined>;
};

write receives one contiguous run of records, oldest first, and returns a locator — any string identifying where the segment went. The log stores that locator in a durable segment index before deleting the records, so a crash in between leaves a written, readable segment rather than records pointing nowhere. A failed write propagates with the records still in the object, so configured storage cannot be skipped, and writes are at-least-once, so key segments by first offset and refuse to overwrite.

read is what makes a flush rehydratable. A consumer reading a flushed offset gets its records back transparently, so falling behind the hot window is no longer fatal; bounds() reports oldestOffset (oldest readable, including flushed segments) alongside hotOffset (lowest offset still in the object). Cold reads cost a round trip and open input gates, so they serve only consumers that fell behind — never the hot path.

External storage expires on its own schedule, so mirror a lifecycle rule with forgetColdBefore(offset) to drop stale index entries. Returning undefined from read is also honoured: a segment the index outlived surfaces as truncation rather than a silent gap.

Retention still ignores cursors, exactly as a log should: a consumer slower than everything readable loses records. It learns loudly rather than silently skipping, because reading below the oldest readable offset throws LogTruncatedError carrying requestedOffset and oldestOffset. Leased consumers learn the same way: when a range awaiting redelivery (nacked or expired) was trimmed, lease() throws LogTruncatedError once, drops the trimmed offsets from the range, and the next call carries on from what is still retained. A leased consumer’s cursor moves past offsets trimmed before anyone leased them. Compare bounds() against a cursor to detect the risk before it happens.

Leases for parallel consumers

read plus commit is enough for one worker per consumer. It is not enough when two pollers share a consumer name: both read the same page and both process it. Leases close that gap.

const lease = await log.lease('search', 100);
if (lease) {
  try {
    await index(lease.records);
    await log.ack('search', lease.batchId);
  } catch {
    await log.nack('search', lease.batchId);
  }
}

lease claims a range of offsets durably before returning the records, so a second caller gets undefined rather than the same offsets. undefined means nothing is available right now — either the consumer is at maxParallelism or the log has no unleased records — so treat it as a signal to back off.

A range is held until it is acked, nacked, or its lease expires after leaseMs. Redelivery always issues a fresh batchId, which fences the previous holder: its ack throws LogLeaseError rather than committing a range someone else now owns. Double settlement throws for the same reason.

maxParallelism lets one consumer hold several ranges at once, and ack commits only over an unbroken run of settled ranges starting at the cursor. Acking a later range while an earlier one is still in flight advances nothing, so parallelism never skips offsets. Leases are durable, so eviction mid-flight does not release a range early, and a range that retention flushed while it was held is rehydrated on redelivery.

A range that keeps failing would be redelivered forever. maxAttempts bounds that, but only together with onPoison: without somewhere to send the records, blocking the consumer beats silently dropping them. With both set, an exhausted range is marked skipped — durably, before the callback runs, so a failing handler cannot resurrect it — and the commit then runs over it like an acked one.

Leases are opt-in. Without the leases option no lease tables are created, read/commit behave exactly as before, and the lease methods throw.

What it is not

There is no per-record retry, timeout, or dead letter here: a log tracks positions, not attempts. When a single record must be retried independently of its neighbours, that is fanout’s job, and the two compose on one object.

There is also no key compaction. deduplicationKey is unique per retained record, which is the opposite of a compaction key, and deduplication lasts only while the original record is retained — re-appending a key after its record was flushed or deleted creates a new record.

Edit this page
Durability

Type to search…

↑↓ navigate↵ selectEsc close