Skip to content
Durability

DurabilityTopology

DurabilityTopology API, behavior, and recovery contract.

Import DurabilityTopology from durability/experimental and pin an exact version. Use one coordinator to own a group’s membership and publish it for readers, optionally pushing shared state to members.

How it uses your Durable Object

DurabilityTopology turns one Durable Object into the coordinator of its groups. It uses that object, and only reads or calls the others:

  • ctx.storage.sql holds the membership, state values, and write records in durability_topology_* tables, plus every staged publication and delivery as rows in the shared durability_outbox table. They commit in one transaction with each change.
  • The object’s one alarm is set in that same transaction to the earliest pending outbox row, so publication and deliveries resume after eviction. Your class’s alarm() must delegate to the scheduler the helper is attached to.
  • The live instance holds the in-flight store writes, member calls, and their timeouts. An eviction loses them; the outbox rows and write records in storage are what recovery relies on.
  • Other objects and Workers never share that storage. Members receive state values over RPC and must check versions themselves; readers use DurabilityTopology.client, which reads the published copy and never calls the coordinator.

How DurabilityTopology uses your Durable Object

Shows TopologyCoordinator, DurabilityTopology. Focuses on DurabilityTopology.Coordinator Durable ObjectTopologyCoordinatoryour Durable ObjectDurabilityTopologyin memoryupdate()

1/9 · The helper belongs to the coordinator

DurabilityTopology attaches to the object’s scheduler and reads the object’s ID. Without a Durable Object context it throws, because every snapshot names the coordinator that published it.

Open source ↗
import { z } from 'zod';
import { reportFailure, TopologyError } from './errors.js';
import type { LifecycleHook } from './lifecycle.js';
import {
  durabilityTopologyMigrations,
  migrateCapability,
} from './migrations.js';
import {
  DurabilityOutbox,
  discardWithin,
  publishNow,
  stageWithin,
  type DurabilityOutboxEvent,
  type OutboxEntry,
} from './outbox.js';
import { assertPositiveInteger } from './policy.js';
import {
  DurabilityScheduler,
  engineFor,
  type Engine,
  type SchedulerAttachment,
} from './scheduler.js';
import { canonical, deserialize, serialize } from './serialization.js';
import {
  TopologyClient,
  type TopologyClientConfig,
} from './topology-client.js';
import {
  encodeSnapshot,
  memberInputSchema,
  membersSchema,
  type TopologyMember,
  type TopologyMemberInput,
  type TopologySnapshot,
} from './topology-snapshot.js';

/** `write` channels reach active members; `read` channels reach draining members too. */
export type TopologyRole = 'write' | 'read';

/** Publication and sync never give up: a stale topology must still converge. */
export type TopologyRetryOptions = {
  /** Maximum duration of one write or delivery. Defaults to 30 seconds. */
  attemptTimeoutMs?: number;
  retries?: {
    /** Delay in milliseconds after a failed attempt. Defaults to jittered exponential backoff capped at 5 minutes. */
    delay?: (attempt: number) => number;
  };
};

export type TopologyStateUpdate<Value> = {
  group: string;
  channel: string;
  /** Increases on every update sent to this member. Apply only versions newer than the last one applied. */
  version: number;
  topologyVersion: number;
  value: Value;
};

export type TopologyPublisher<Meta> = TopologyRetryOptions & {
  /** Must be idempotent. Store `encoded` as is: it is what the client reads. */
  write(snapshot: TopologySnapshot<Meta>, encoded: string): Promise<unknown>;
  delete(group: string, version: number): Promise<unknown>;
};

export type TopologyChannel<Value, Meta> = TopologyRetryOptions & {
  /** Defaults to `write`. */
  members?: TopologyRole;
  deliver(
    member: TopologyMember<Meta>,
    update: TopologyStateUpdate<Value>
  ): unknown;
};

/**
 * Publication and sync run on outbox channels: `durability.topology` keyed by
 * group, and `durability.topology.<channel>` keyed by `[group, member]`.
 */
export type DurabilityTopologyEvent = DurabilityOutboxEvent;

export type DurabilityTopologyConfig<
  Meta,
  States extends Record<string, unknown>,
> = SchedulerAttachment &
  TopologyRetryOptions & {
    /** Without a publisher, membership is only visible inside this object. */
    publish?: TopologyPublisher<Meta>;
    states?: { [Name in keyof States]: TopologyChannel<States[Name], Meta> };
    onLifecycleEvent?: (event: DurabilityTopologyEvent) => void | Promise<void>;
  };

export type TopologyCurrent<Meta, States> = {
  members: readonly TopologyMember<Meta>[];
  states: Partial<States>;
};

export type TopologyDraft<Meta, States> = {
  /** Omit to keep the current members. */
  members?: readonly TopologyMemberInput<Meta>[];
  /** Only the channels being set. */
  states?: Partial<States>;
};

type Publication<Meta> =
  | { snapshot: TopologySnapshot<Meta> }
  | { deleted: { group: string; version: number } };

type Delivery<Meta> = {
  group: string;
  member: TopologyMember<Meta>;
  topologyVersion: number;
  value: unknown;
};

type Channel<Meta> = {
  outbox: DurabilityOutbox<Delivery<Meta>>;
  role: TopologyRole;
};

const publicationChannel = 'durability.topology';

const groupRowSchema = z.object({
  group_id: z.string(),
  version: z.number().int().positive(),
  members: z.string(),
  deleted: z.union([z.literal(0), z.literal(1)]),
});

type GroupRow = z.infer<typeof groupRowSchema>;

const stateRowSchema = z.object({
  channel: z.string(),
  value: z.string(),
});

class TopologyTimeoutError extends Error {
  constructor(label: string, timeoutMs: number) {
    super(`Topology ${label} timed out after ${timeoutMs}ms`);
    this.name = 'TopologyTimeoutError';
  }
}

const withTimeout = async <T>(
  work: Promise<T>,
  timeoutMs: number,
  label: string
): Promise<T> => {
  let timer: ReturnType<typeof setTimeout> | undefined;
  try {
    return await Promise.race([
      work,
      new Promise<never>((_resolve, reject) => {
        timer = setTimeout(
          () => reject(new TopologyTimeoutError(label, timeoutMs)),
          timeoutMs
        );
      }),
    ]);
  } finally {
    clearTimeout(timer);
  }
};

// setTimeout overflows past this, and the outbox enforces the same ceiling.
const maxTimeoutMs = 2 ** 31 - 1;

const timeoutOf = (
  options: TopologyRetryOptions | undefined,
  fallback: TopologyRetryOptions
): number => {
  const timeoutMs =
    options?.attemptTimeoutMs ?? fallback.attemptTimeoutMs ?? 30_000;
  assertPositiveInteger('attemptTimeoutMs', timeoutMs);
  if (timeoutMs > maxTimeoutMs) {
    throw new RangeError(`attemptTimeoutMs must be at most ${maxTimeoutMs}`);
  }
  return timeoutMs;
};

/**
 * Each write or delivery has its own deadline inside a batch, so the outbox's
 * batch deadline only bounds the callback itself.
 */
const outboxOptions = (
  timeoutMs: number,
  options: TopologyRetryOptions | undefined,
  fallback: TopologyRetryOptions
) => {
  const delay = options?.retries?.delay ?? fallback.retries?.delay;
  return {
    attemptTimeoutMs: Math.min(timeoutMs * 2, maxTimeoutMs),
    ...(delay === undefined ? {} : { retries: { delay } }),
  };
};

/**
 * How long after its timeout an abandoned publisher write is assumed able to
 * land. KV has no conditional write, so some bound is required.
 */
const landingWindowMs = 5 * 60_000;

const forgetWrite = (sql: SqlStorage, id: string): void => {
  sql.exec('DELETE FROM durability_topology_writes WHERE id = ?', id);
};

/** Drops expired records and returns when the group's last outstanding write can no longer land. */
const outstandingWritesUntil = (
  sql: SqlStorage,
  group: string,
  now: number
): number | undefined => {
  sql.exec(
    'DELETE FROM durability_topology_writes WHERE group_id = ? AND land_by <= ?',
    group,
    now
  );
  const { until } = z.object({ until: z.number().int().nullable() }).parse(
    sql
      .exec(
        `SELECT MAX(land_by) AS until FROM durability_topology_writes
           WHERE group_id = ?`,
        group
      )
      .toArray()[0]
  );
  return until ?? undefined;
};

const assertGroup = (group: unknown): void => {
  if (typeof group !== 'string' || group.length === 0) {
    throw new TopologyError('Group must be a non-empty string');
  }
};

const deliveryKey = (group: string, member: string): string =>
  JSON.stringify([group, member]);

const isThenable = (value: unknown): boolean =>
  value !== null &&
  typeof value === 'object' &&
  typeof Reflect.get(value, 'then') === 'function';

const readGroup = (sql: SqlStorage, group: string): GroupRow | undefined =>
  groupRowSchema.optional().parse(
    sql
      .exec(
        `SELECT group_id, version, members, deleted
           FROM durability_topology_groups WHERE group_id = ?`,
        group
      )
      .toArray()[0]
  );

const readStates = (sql: SqlStorage, group: string) =>
  stateRowSchema.array().parse(
    sql
      .exec(
        `SELECT channel, value FROM durability_topology_states
           WHERE group_id = ?`,
        group
      )
      .toArray()
  );

const parseMembers = <Meta>(row: GroupRow): TopologyMember<Meta>[] =>
  membersSchema.parse(
    JSON.parse(row.members)
  ) as unknown as TopologyMember<Meta>[];

const inRole = (member: TopologyMember<unknown>, role: TopologyRole): boolean =>
  role === 'read' || member.state === 'active';

/**
 * Experimental. Holds the membership of shard groups inside one coordinator
 * Durable Object, publishes each group's snapshot to a read store, and drives
 * every member to the latest value of each state channel. Only this instance
 * can change or publish its groups; readers use {@link DurabilityTopology.client}.
 */
export class DurabilityTopology<
  Meta = undefined,
  States extends Record<string, unknown> = Record<string, never>,
> {
  readonly #engine: Engine;
  readonly #coordinator: string;
  readonly #publication: DurabilityOutbox<Publication<Meta>> | undefined;
  readonly #channels = new Map<string, Channel<Meta>>();

  constructor(config: DurabilityTopologyConfig<Meta, States>) {
    // The outboxes must share this helper's engine, so a private scheduler is
    // created explicitly instead of letting each attach its own.
    const scheduler = config.scheduler ?? new DurabilityScheduler(config);
    this.#engine = engineFor({ scheduler });
    const coordinator = this.#engine.objectId();
    if (coordinator === undefined) {
      throw new TopologyError(
        'DurabilityTopology needs the Durable Object context, which carries its ID'
      );
    }
    this.#coordinator = coordinator;
    this.#engine.migrateSchema('topology', durabilityTopologyMigrations);

    // The outbox only emits outbox events, which is all this hook accepts.
    const onLifecycleEvent = config.onLifecycleEvent as
      | LifecycleHook
      | undefined;
    const lifecycle =
      onLifecycleEvent === undefined ? {} : { onLifecycleEvent };

    const publisher = config.publish;
    if (publisher !== undefined) {
      if (
        typeof publisher.write !== 'function' ||
        typeof publisher.delete !== 'function'
      ) {
        throw new TopologyError('publish needs write and delete functions');
      }
      const timeoutMs = timeoutOf(publisher, config);
      this.#publication = new DurabilityOutbox<Publication<Meta>>({
        scheduler,
        channel: publicationChannel,
        ...outboxOptions(timeoutMs, publisher, config),
        ...lifecycle,
        publish: (entries) =>
          this.#publishSnapshots(publisher, entries, timeoutMs),
      });
    }

    const channels: Record<
      string,
      TopologyChannel<unknown, Meta>
    > = config.states ?? {};
    for (const [name, channel] of Object.entries(channels)) {
      if (!name || typeof channel.deliver !== 'function') {
        throw new TopologyError(`Invalid topology channel "${name}"`);
      }
      const timeoutMs = timeoutOf(channel, config);
      this.#channels.set(name, {
        role: channel.members ?? 'write',
        outbox: new DurabilityOutbox<Delivery<Meta>>({
          scheduler,
          channel: `${publicationChannel}.${name}`,
          ...outboxOptions(timeoutMs, channel, config),
          ...lifecycle,
          publish: (entries) =>
            this.#deliver(name, channel, entries, timeoutMs),
        }),
      });
    }
  }

  /** A read-only client for stateless Workers. It never contacts the coordinator. */
  static client<Meta = undefined>(
    config: TopologyClientConfig<Meta>
  ): TopologyClient<Meta> {
    return new TopologyClient(config);
  }

  /** Reverts only the topology tables; outbox channels migrateCapability with `DurabilityOutbox.migrate`. */
  static migrate(
    context: Pick<DurableObjectState, 'storage'>,
    target?: string | null
  ) {
    return migrateCapability(
      context.storage,
      'topology',
      durabilityTopologyMigrations,
      target
    );
  }

  alarm(info?: AlarmInvocationInfo): Promise<void> {
    return this.#engine.alarm(info);
  }

  /**
   * Applies one change in a single transaction. `next` runs inside it and must
   * return synchronously, so nothing can interleave between reading `current`
   * and writing the draft. After the commit, one publish is attempted:
   * `published: false` means the snapshot is not yet final in the store:
   * the write failed, or an older write may still land over it. The outbox
   * keeps going until it is.
   */
  async update(
    group: string,
    next: (
      current: TopologyCurrent<Meta, States>
    ) => TopologyDraft<Meta, States>
  ): Promise<{ snapshot: TopologySnapshot<Meta>; published: boolean }> {
    assertGroup(group);
    const snapshot = await this.#engine.transaction((sql) => {
      const row = readGroup(sql, group);
      const live = row?.deleted === 0 ? row : undefined;
      const before = live ? parseMembers<Meta>(live) : [];
      const stateRows = readStates(sql, group);
      const states = Object.fromEntries(
        stateRows.map((state) => [state.channel, deserialize(state.value)])
      ) as Partial<States>;

      // A copy, so an updater that mutates `current` cannot make a change look like a no-op.
      const draft: unknown = next({ members: structuredClone(before), states });
      if (isThenable(draft)) {
        throw new TopologyError('update must return its draft synchronously');
      }
      const { members, states: changes } = this.#parseDraft(draft);
      if (!live && members === undefined) {
        throw new TopologyError(
          `Group "${group}" does not exist; set members to create it`
        );
      }

      const now = Date.now();
      const after =
        members === undefined ? before : this.#normalize(members, before, now);
      const changed =
        !live ||
        canonical(JSON.stringify(after)) !== canonical(JSON.stringify(before));
      const version = changed ? (row?.version ?? 0) + 1 : live.version;
      const current: TopologySnapshot<Meta> = {
        group,
        version,
        coordinator: this.#coordinator,
        members: after,
      };
      if (changed) {
        sql.exec(
          `INSERT INTO durability_topology_groups
             (group_id, version, members, deleted, updated_at)
           VALUES (?, ?, ?, 0, ?)
           ON CONFLICT(group_id) DO UPDATE SET
             version = excluded.version,
             members = excluded.members,
             deleted = 0,
             updated_at = excluded.updated_at`,
          group,
          version,
          JSON.stringify(after),
          now
        );
        this.#stagePublication(sql, group, { snapshot: current });
      }

      const changedChannels = this.#writeStates(sql, group, changes, stateRows);
      this.#stageDeliveries(
        sql,
        group,
        version,
        before,
        after,
        changedChannels
      );
      return current;
    });

    return { snapshot, published: await this.#publishNow(group) };
  }

  /**
   * Deletes the group in one transaction: members and state values are dropped
   * and pending deliveries discarded. Versions keep counting, so a recreated
   * group is newer than anything readers cached. One publisher delete is
   * attempted after the commit.
   */
  async delete(group: string): Promise<{ published: boolean }> {
    assertGroup(group);
    await this.#engine.transaction((sql) => {
      const row = readGroup(sql, group);
      if (row?.deleted !== 0) {
        return;
      }
      const version = row.version + 1;
      sql.exec(
        `UPDATE durability_topology_groups
         SET version = ?, members = '[]', deleted = 1, updated_at = ?
         WHERE group_id = ?`,
        version,
        Date.now(),
        group
      );
      sql.exec(
        'DELETE FROM durability_topology_states WHERE group_id = ?',
        group
      );
      for (const member of parseMembers<Meta>(row)) {
        for (const channel of this.#channels.values()) {
          discardWithin(channel.outbox, sql, deliveryKey(group, member.id));
        }
      }
      this.#stagePublication(sql, group, { deleted: { group, version } });
    });
    return { published: await this.#publishNow(group) };
  }

  /** Coordinator-local read. Readers elsewhere use the published copy. */
  snapshot(group: string): TopologySnapshot<Meta> | undefined {
    assertGroup(group);
    const row = readGroup(this.#engine.sql, group);
    if (row?.deleted !== 0) {
      return undefined;
    }
    return {
      group,
      version: row.version,
      coordinator: this.#coordinator,
      members: parseMembers<Meta>(row),
    };
  }

  #publishNow(group: string): Promise<boolean> {
    return this.#publication
      ? publishNow(this.#publication, group)
      : Promise.resolve(true);
  }

  #stagePublication(
    sql: SqlStorage,
    group: string,
    publication: Publication<Meta>
  ): void {
    if (this.#publication) {
      stageWithin(this.#publication, sql, group, publication);
    }
  }

  /**
   * Every write is recorded before it starts and forgotten once it settles.
   * A record that outlives its write (a timeout, or eviction mid-write) means
   * an older snapshot may still land, so a newer write is not acknowledged
   * until that record's landing deadline passes and it has been written again.
   * Records live in storage, so this survives eviction.
   */
  async #publishSnapshots(
    publisher: TopologyPublisher<Meta>,
    entries: OutboxEntry<Publication<Meta>>[],
    timeoutMs: number
  ): Promise<void> {
    const landBy = Date.now() + timeoutMs + landingWindowMs;
    const writes = entries.map((entry) => ({
      entry,
      id: crypto.randomUUID(),
    }));
    await this.#engine.transaction((sql) => {
      for (const { entry, id } of writes) {
        sql.exec(
          `INSERT INTO durability_topology_writes (id, group_id, land_by)
           VALUES (?, ?, ?)`,
          id,
          entry.key,
          landBy
        );
      }
    });

    await Promise.all(
      writes.map(async ({ entry, id }) => {
        const publication = entry.value;
        const write = Promise.resolve().then(() =>
          'deleted' in publication
            ? publisher.delete(
                publication.deleted.group,
                publication.deleted.version
              )
            : publisher.write(
                publication.snapshot,
                encodeSnapshot(publication.snapshot)
              )
        );
        try {
          await withTimeout(write, timeoutMs, 'publish');
        } catch (error) {
          if (error instanceof TopologyTimeoutError) {
            this.#settleLate(entry.key, id, write);
          } else {
            // A rejected write settled before any newer write of this group started.
            await this.#engine.transaction((sql) => forgetWrite(sql, id));
          }
          reportFailure(
            'durability.topology.publish_failed',
            { group: entry.key, attempt: entry.attempt },
            error
          );
          entry.retry();
          return;
        }

        const outstandingUntil = await this.#engine.transaction((sql) => {
          forgetWrite(sql, id);
          return outstandingWritesUntil(sql, entry.key, Date.now());
        });
        if (outstandingUntil !== undefined) {
          entry.retry(Math.min(outstandingUntil - Date.now(), maxTimeoutMs));
        }
      })
    );
  }

  /**
   * An abandoned write that settles while this isolate lives may have landed
   * over a newer one: forget it and restage the latest snapshot together. If
   * the isolate dies first, its record makes the next publication wait instead.
   */
  #settleLate(group: string, id: string, write: Promise<unknown>): void {
    const onError = (error: unknown) =>
      reportFailure('durability.topology.restage_failed', { group }, error);
    const settled = write
      .then(
        () => undefined,
        () => undefined
      )
      .then(() =>
        this.#engine.transaction((sql) => {
          forgetWrite(sql, id);
          this.#stageLatest(sql, group);
        })
      )
      .catch(onError);
    this.#engine.waitUntil(settled, onError);
  }

  /** A fresh version fences any in-flight publication, so the latest snapshot is written again. */
  #stageLatest(sql: SqlStorage, group: string): void {
    const row = readGroup(sql, group);
    if (!row) {
      return;
    }
    this.#stagePublication(
      sql,
      group,
      row.deleted === 1
        ? { deleted: { group, version: row.version } }
        : {
            snapshot: {
              group,
              version: row.version,
              coordinator: this.#coordinator,
              members: parseMembers<Meta>(row),
            },
          }
    );
  }

  async #deliver(
    name: string,
    channel: TopologyChannel<unknown, Meta>,
    entries: OutboxEntry<Delivery<Meta>>[],
    timeoutMs: number
  ): Promise<void> {
    await Promise.all(
      entries.map(async (entry) => {
        const { group, member, topologyVersion, value } = entry.value;
        const update: TopologyStateUpdate<unknown> = {
          group,
          channel: name,
          version: entry.version,
          topologyVersion,
          value,
        };
        try {
          // Members ignore versions they have passed, so an abandoned delivery landing late is harmless.
          await withTimeout(
            Promise.resolve().then(() => channel.deliver(member, update)),
            timeoutMs,
            'delivery'
          );
        } catch (error) {
          reportFailure(
            'durability.topology.delivery_failed',
            { group, channel: name, member: member.id, attempt: entry.attempt },
            error
          );
          entry.retry();
        }
      })
    );
  }

  #parseDraft(draft: unknown): {
    members: readonly TopologyMemberInput<Meta>[] | undefined;
    states: Record<string, unknown>;
  } {
    if (draft === null || typeof draft !== 'object') {
      throw new TopologyError('update must return a draft object');
    }
    const members: unknown = Reflect.get(draft, 'members');
    const states: unknown = Reflect.get(draft, 'states') ?? {};
    if (members !== undefined && !Array.isArray(members)) {
      throw new TopologyError('Draft members must be an array');
    }
    if (
      states === null ||
      typeof states !== 'object' ||
      Array.isArray(states)
    ) {
      throw new TopologyError('Draft states must be an object');
    }
    return {
      members: members as readonly TopologyMemberInput<Meta>[] | undefined,
      states: states as Record<string, unknown>,
    };
  }

  #normalize(
    inputs: readonly TopologyMemberInput<Meta>[],
    before: readonly TopologyMember<Meta>[],
    now: number
  ): TopologyMember<Meta>[] {
    if (inputs.length === 0) {
      throw new TopologyError(
        'A group needs at least one member; use delete() to remove it'
      );
    }
    const joinedAt = new Map(
      before.map((member) => [member.id, member.joinedAt])
    );
    const seen = new Set<string>();
    const members = inputs.map((input) => {
      const parsed = memberInputSchema.safeParse(input);
      if (!parsed.success) {
        throw new TopologyError(`Invalid member: ${parsed.error.message}`);
      }
      const { id, state, locationHint, meta } = parsed.data;
      if (seen.has(id)) {
        throw new TopologyError(`Duplicate member "${id}"`);
      }
      seen.add(id);
      return {
        id,
        state,
        ...(locationHint === undefined ? {} : { locationHint }),
        ...(meta === undefined ? {} : { meta }),
        joinedAt: joinedAt.get(id) ?? now,
      } as unknown as TopologyMember<Meta>;
    });
    return members.sort((a, b) => (a.id < b.id ? -1 : 1));
  }

  /** Returns the channels whose value changed. Unchanged values are not redelivered. */
  #writeStates(
    sql: SqlStorage,
    group: string,
    changes: Record<string, unknown>,
    current: z.infer<typeof stateRowSchema>[]
  ): Set<string> {
    const existing = new Map(current.map((row) => [row.channel, row.value]));
    const changed = new Set<string>();
    for (const [channel, value] of Object.entries(changes)) {
      if (value === undefined) {
        continue;
      }
      if (!this.#channels.has(channel)) {
        throw new TopologyError(`Unknown topology channel "${channel}"`);
      }
      const stored = serialize(value);
      const previous = existing.get(channel);
      if (previous !== undefined && canonical(previous) === canonical(stored)) {
        continue;
      }
      sql.exec(
        `INSERT INTO durability_topology_states (group_id, channel, value)
         VALUES (?, ?, ?)
         ON CONFLICT(group_id, channel) DO UPDATE SET value = excluded.value`,
        group,
        channel,
        stored
      );
      changed.add(channel);
    }
    return changed;
  }

  /**
   * Members that entered a channel's role receive its current value; a changed
   * value goes to every member in the role. Members that left lose their
   * pending delivery.
   */
  #stageDeliveries(
    sql: SqlStorage,
    group: string,
    topologyVersion: number,
    before: readonly TopologyMember<Meta>[],
    after: readonly TopologyMember<Meta>[],
    changed: ReadonlySet<string>
  ): void {
    const values = new Map(
      readStates(sql, group).map((row) => [row.channel, row.value])
    );
    for (const [name, channel] of this.#channels) {
      const previous = new Set(
        before
          .filter((member) => inRole(member, channel.role))
          .map(({ id }) => id)
      );
      const targets = after.filter((member) => inRole(member, channel.role));
      const kept = new Set(targets.map(({ id }) => id));
      for (const id of previous) {
        if (!kept.has(id)) {
          discardWithin(channel.outbox, sql, deliveryKey(group, id));
        }
      }

      const value = values.get(name);
      if (value === undefined) {
        continue;
      }
      for (const member of targets) {
        if (!changed.has(name) && previous.has(member.id)) {
          continue;
        }
        stageWithin(channel.outbox, sql, deliveryKey(group, member.id), {
          group,
          member,
          topologyVersion,
          value: deserialize(value),
        });
      }
    }
  }
}
Create a group and publish it
Details

Follow one create() from the RPC call to a published snapshot and a state value applied on every member.

Durable state
Nothing is written yet.
In-flight / in memory
The helper, its engine, and the coordinator ID string.

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_topology_groups and durability_topology_states hold membership and channel values. durability_topology_writes tracks publication requests that could still land after timeout or eviction. Internal outbox channels drive group publication and per-member state delivery.

update(group, next) evaluates your synchronous draft inside a transaction. It commits membership changes, changed state values, and staged publication/deliveries together. Membership versions advance on actual membership changes; outbox delivery versions prevent applying stale channel values at members.

Recovery: a failed member retries independently through its outbox entry; newer values coalesce. An active member participates in writes and reads; draining members remain in the read set. Already in-flight deliveries may still arrive after removal.

For unconditional publication stores, write tracking uses a bounded landing deadline before repairing uncertain old writes. An older write arriving beyond that window can still overwrite a newer snapshot; only a version-aware sink removes that bound. The client rejects same-coordinator version regressions in its own cache but cannot make a stale store globally consistent.

Reader state is in memory: TTL caches and fallback decisions are not durable membership. The client reads the published store rather than the coordinator, unless application fallback deliberately adds that dependency.

Implementation: topology.ts, topology-client.ts.

Shard topology (experimental)

DurabilityTopology is for work spread over several Durable Objects (shards) whose membership changes over time, where something else needs to reach them: a router picking a shard for a write, a reader querying every shard, or a coordinator pushing configuration to all of them. It lives in durability/experimental, so pin an exact version.

One coordinator Durable Object holds each group. It is the only thing that can change or publish that group. Readers never call it: they read the snapshot it published.

import {
  DurabilityTopology,
  type TopologyStateUpdate,
} from 'durability/experimental';

export class Coordinator extends DurableObject<Env> {
  private readonly scheduler = new DurabilityScheduler({ context: this.ctx });
  private readonly topology = new DurabilityTopology({
    scheduler: this.scheduler,
    publish: {
      write: (snapshot, encoded) =>
        this.env.TOPOLOGY.put(`topology/${snapshot.group}`, encoded),
      delete: (group) => this.env.TOPOLOGY.delete(`topology/${group}`),
    },
    states: {
      limits: {
        deliver: (member, update: TopologyStateUpdate<Limits>) =>
          this.env.SHARDS.getByName(member.id).applyLimits(update),
      },
    },
  });

  createTenant(tenant: string, limits: Limits) {
    return this.topology.update(tenant, () => ({
      members: [0, 1, 2].map((n) => ({
        id: `${tenant}-${n}`,
        state: 'active' as const,
      })),
      states: { limits },
    }));
  }

  retireShard(tenant: string, shard: string) {
    return this.topology.update(tenant, ({ members }) => ({
      members: [
        ...members.map((m) =>
          m.id === shard ? { ...m, state: 'draining' as const } : m
        ),
        { id: `${tenant}-${members.length}`, state: 'active' },
      ],
    }));
  }

  deleteTenant(tenant: string) {
    return this.topology.delete(tenant);
  }

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

The coordinator has three methods:

  • update(group, next) runs next(current) inside one storage transaction and commits its draft, staging publication and deliveries in the same transaction. current holds the members and the current state values. The draft can set members, states, or both, and next must return synchronously, so nothing interleaves between reading and writing. Membership changes bump the group’s version; unchanged drafts do not. After the commit, one publish is attempted: published: false means the snapshot is not yet final in the store, because the write failed or an older write may still land over it. The change is committed either way, and the outbox keeps going until it is. An empty member list is rejected; use delete.
  • delete(group) drops the members and state values, cancels pending syncs, and asks the publisher to delete the snapshot. Versions keep counting, so a recreated group is newer than anything a reader cached.
  • snapshot(group) reads the group inside the coordinator. It is not a lookup path for other Workers.

An active member is in the write set and the read set. A draining member is only in the read set: it receives no new work but still holds data, so queries must still reach it. Members leave the read set only when a draft omits them.

When to use it

Topology is for when something needs the list of shards. A shard that only does work when a caller reaches it, and only talks to a coordinator whose name it can derive, does not need one: create it on demand.

Need Use
A shard does work when a caller sends it something Nothing: address it by name, for example with DurabilityRouting
A shard asks a central object for something it needs, such as permits or capacity DurabilitySemaphore: shards pull, and the holder learns about them from their requests
A caller must pick among a group’s current shards Topology membership
A reader must reach every shard that may hold data Topology membership, including draining members
Every shard must hold the same data before any caller reaches it, such as schedules it runs by itself or configuration it enforces A topology state channel

A state channel sends one value to every member in its role. Values that differ per shard, or that must be accounted for in both directions (granted, used, returned), do not fit it; that is the semaphore’s model.

A shard created on demand can still join its group, so readers can find it later: its first request can call the coordinator, which adds it with one update. Only do this when a reader needs it.

Publishing

publish.write receives the snapshot and an encoded string. Store encoded as is; it is the only format the client reads, and no encoder is exported, so only the coordinator can produce one. It carries the coordinator’s Durable Object ID, because versions only compare within one coordinator: if a group moves to another coordinator, its versions start over, and readers accept the new snapshot. Publication runs on a DurabilityOutbox channel (durability.topology) keyed by group, so it shares the outbox’s guarantees: only the newest snapshot of a group is pending, publish calls for the channel run one at a time, and attempts never stop (retries.delay controls backoff). The outbox expects sinks to keep the highest version, which a plain KV.put cannot, so the topology closes that gap itself, durably:

  • Every write is recorded in storage before it starts and forgotten once it settles.
  • A record that outlives its write, because the write timed out or the object was evicted mid-write, means an older snapshot may still land. A newer write is then not acknowledged until that record’s landing deadline has passed, and is written again at that point.
  • An abandoned write that settles while the object is still running restages the latest snapshot immediately.

The landing deadline is the write’s start plus attemptTimeoutMs plus five minutes. A write that lands later than that can still overwrite a newer snapshot until the next change; only a sink that keeps the highest version avoids that bound. attemptTimeoutMs bounds one write.

State channels

Each entry in states is a channel whose latest value every member converges to. members: 'write' (the default) reaches active members; 'read' reaches draining members too. Setting a value delivers it to every member in the role. A member that enters the role receives the current value in the same transaction that added it, so a new shard never starts with an old value. A member that leaves the role has its pending delivery discarded; one already in flight may still arrive. Each channel is its own outbox channel (durability.topology.<name>) keyed by group and member, so every member is retried independently; attemptTimeoutMs bounds one delivery, and a member that hangs holds up the rest of its batch only until then. Setting a channel to the value it already has delivers nothing.

Only the latest value is delivered, and a timed-out delivery may still arrive after a newer one. version is the outbox version, which only increases. Apply an update only when it is newer than the last one applied, in the same transaction:

applyLimits(update: TopologyStateUpdate<Limits>) {
  this.ctx.storage.transactionSync(() => {
    const applied = this.ctx.storage.kv.get<number>('limits:version') ?? 0;
    if (update.version <= applied) {
      return;
    }
    this.ctx.storage.kv.put('limits', update.value);
    this.ctx.storage.kv.put('limits:version', update.version);
  });
}

onLifecycleEvent receives the outbox events of those channels: channel names the publication or state channel, and id is the group, or [group, member] as JSON for a delivery.

Reading

Stateless Workers read through a client. It has one method:

const topology = DurabilityTopology.client({
  read: (tenant) => env.TOPOLOGY.get(`topology/${tenant}`),
});

const { members } = await topology.get(tenant);
const writable = members.filter((member) => member.state === 'active');
const shard = writable[Math.floor(Math.random() * writable.length)];

Routing, querying every member, and resolving stubs are ordinary code over members.

The client caches each group in the isolate for ttlMs (default 30 seconds, jittered by ±20%), shares one read between concurrent callers, and never replaces a snapshot with an older version from the same coordinator. A missing or failed read is remembered for negativeTtlMs (default 5 seconds). Reads time out after readTimeoutMs (default 1 second).

When the published copy is missing, invalid, slow, or the read throws, get throws TopologyUnavailableError with a reason. Every fallback is opt-in, and they run in this order:

  1. staleIfErrorMs serves this isolate’s last good snapshot for that long after it expired. It never applies to a missing group.
  2. seed(group) derives members locally, without I/O.
  3. fallback(group, reason) is application code and is not cached.

The client never contacts the coordinator. Wiring fallback to it moves every reader onto one Durable Object at the moment the read store is struggling, so prefer failing or serving stale. onFallback reports each fallback and its source.

Edit this page
Durability

Type to search…

↑↓ navigate↵ selectEsc close