Live job dashboard | Durability

A tenant can import orders and watch their state in a browser. The example deliberately separates the object that owns job execution from the object that owns the dashboard read model:

Pipeline owns importOrder operations and their results.
Dashboard owns an SQL projection and WebSocket sessions.
The browser reads and resynchronizes only that projection; it never treats it as the authoritative pipeline.

This is a useful shape when clients need a responsive view, but the business result must remain with the worker that executed it.

Source and prerequisites

Worker entry, Pipeline, Dashboard, and browser client
Shared types and Wrangler configuration
Provision the ORDERS service binding and deploy both Pipeline and Dashboard Durable Object classes. The example’s browser host is intentionally a placeholder and must be replaced.

The composition uses operations, a shared scheduler, fanout, tracker, socket, and the browser client.

Architecture

1/10 · One Pipeline per tenant

The front Worker names the Pipeline after the tenant and calls importOrder() over RPC. The tenant’s Dashboard is a different object with its own storage and failure boundary.

Open source ↗

/**
 * Front Worker for the `job-dashboard` example. Browsers upgrade to the
 * tenant's `Dashboard`. A plain HTTP import talks to the `Pipeline` directly:
 * register, then wait on the handle in the same pipelined round trip.
 */
import type { Env } from './shared';

export { Dashboard } from './dashboard';
export { Pipeline } from './pipeline';

export default {
  async fetch(request: Request, env: Env) {
    const url = new URL(request.url);
    const match = /^\/tenants\/([^/]+)\/(ws|imports)$/.exec(url.pathname);
    if (!match?.[1]) {
      return new Response('Not found', { status: 404 });
    }
    const [, tenant, action] = match;

    if (action === 'ws') {
      return env.DASHBOARD.getByName(tenant).fetch(request);
    }
    const orderId = url.searchParams.get('order');
    if (request.method !== 'POST' || !orderId) {
      return new Response('Expected POST ?order=<id>', { status: 400 });
    }
    const pipeline = env.PIPELINE.getByName(tenant);
    await pipeline.importOrder(orderId);
    const outcome = await pipeline.track(orderId).wait({ timeoutMs: 10_000 });
    return Response.json(outcome, {
      status: outcome.status === 'pending' ? 202 : 200,
    });
  },
};

/**
 * Producer side of the `job-dashboard` example. Runs durable operations, fans
 * their lifecycle out to the tenant's `Dashboard`, and hands out tracker
 * handles so callers can wait for a result over Workers RPC. Browsers never
 * connect here.
 */
import { DurableObject } from 'cloudflare:workers';
import {
  Durability,
  DurabilityFanout,
  DurabilityScheduler,
  type DurableCall,
} from 'durability';
import { DurabilityTracker } from 'durability/experimental';
import type { Env, JobEvent, OrderSummary } from './shared';

export class Pipeline extends DurableObject<Env> {
  private readonly scheduler = new DurabilityScheduler({ context: this.ctx });

  private readonly durability = new Durability({
    scheduler: this.scheduler,
    handlers: {
      importOrder: async ({
        id,
        payload,
        signal,
      }: DurableCall<{ orderId: string }>): Promise<OrderSummary> => {
        const response = await this.env.ORDERS.fetch(
          `https://orders/${payload.orderId}`,
          { headers: { 'Idempotency-Key': id }, signal }
        );
        if (!response.ok) {
          throw new Error(`Order lookup failed with ${response.status}`);
        }
        return response.json();
      },
    },
    // Completed lifecycle events carry the summary the dashboard publishes.
    includeResults: true,
  });

  private readonly tracker = new DurabilityTracker({
    sources: { jobs: this.durability },
  });

  // Push side. Delivery is durable: the dashboard gets the batch again after
  // an eviction or a failed call, until `ingest` returns without throwing.
  private readonly fanout = new DurabilityFanout<JobEvent>({
    scheduler: this.scheduler,
    targets: {
      dashboard: {
        deliver: (messages) =>
          this.env.DASHBOARD.getByName(this.tenant).ingest(
            messages.map((message) => message.body)
          ),
        ackOnReturn: true,
      },
    },
    batchDelayMs: 100,
  });

  constructor(ctx: DurableObjectState, env: Env) {
    super(ctx, env);
    // Lifecycle hooks are best-effort. If the object dies between a job
    // settling and this enqueue committing, that one event is lost until the
    // dashboard resyncs. Enqueue from inside the handler, keyed by `id`, when
    // that gap matters.
    this.durability.lifecycle((event) => {
      if (event.type === 'registered') {
        return this.publish(event, { status: 'pending' });
      }
      if (
        event.type === 'attempt_settled' &&
        event.outcome === 'completed' &&
        event.result !== undefined
      ) {
        return this.publish(event, {
          status: 'completed',
          result: event.result,
        });
      }
      if (
        (event.type === 'attempt_settled' && event.outcome === 'failed') ||
        event.type === 'terminal'
      ) {
        return this.publish(event, { status: 'failed', error: event.error });
      }
      return undefined;
    });
  }

  private get tenant(): string {
    return this.ctx.id.name ?? this.ctx.id.toString();
  }

  private async publish(
    event: { id: string; generation: string; timestamp: number },
    state: JobEvent['state']
  ): Promise<void> {
    await this.fanout.enqueue([
      {
        id: `${event.id}:${event.generation}:${state.status}`,
        body: {
          id: event.id,
          name: 'importOrder',
          generation: event.generation,
          createdAt: event.timestamp,
          state,
        },
      },
    ]);
  }

  /** Registers the import; resolves on acceptance, not completion. */
  importOrder(orderId: string) {
    return this.durability.importOrder({
      id: `import:${orderId}`,
      payload: { orderId },
    });
  }

  /**
   * Pull side. Over Workers RPC the handle's `get` and `wait` become stubs, so
   * `stub.track(id).wait({ timeoutMs })` waits in one pipelined round trip.
   */
  async track(orderId: string) {
    const handle = await this.tracker.get('jobs', `import:${orderId}`);
    if (!handle) {
      throw new Error(`Order "${orderId}" was never imported`);
    }
    return handle;
  }

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

/**
 * Hub of the `job-dashboard` example. Three kinds of peers talk to it:
 *
 * - Browsers, over `DurabilitySocket`. `work` serves `tracker.records()` as
 *   the snapshot (plain JSON, no functions); every ingested change is pushed
 *   as a `job` event, which the socket persists and replays after reconnects.
 *   `importOrder` starts a job on the tenant's `Pipeline` and waits for its
 *   result through a tracker handle, in one pipelined RPC round trip.
 * - The tenant's `Pipeline`, through fanout deliveries into `ingest`.
 *
 * The ingested table is itself a `TrackedSource`, so `this.tracker.get` also
 * works locally: its `watch` is settled by `ingest`.
 */
import { DurableObject } from 'cloudflare:workers';
import {
  DurabilityTracker,
  type TrackedRecord,
  type TrackedSource,
  type TrackedStatus,
} from 'durability/experimental';
import { DurabilitySocket, event } from 'durability/socket';
import { z } from 'zod';
import type { Env, JobEvent, Jobs } from './shared';

type JobRow = {
  id: string;
  name: string;
  generation: string;
  created_at: number;
  status: string;
  state: string;
};

const toRecord = (row: JobRow): TrackedRecord<Jobs> => ({
  id: row.id,
  name: 'importOrder',
  generation: row.generation,
  createdAt: row.created_at,
  state: JSON.parse(row.state),
});

const workQuery = z.object({
  cursor: z.string().optional(),
  status: z.enum(['pending', 'completed', 'failed']).optional(),
});

export class Dashboard extends DurableObject<Env> {
  private readonly sql = this.ctx.storage.sql;
  private readonly watchers = new Map<
    string,
    Set<(settled?: TrackedRecord<Jobs>) => void>
  >();

  /** The ingested table, read through the tracker's source contract. */
  private readonly jobs: TrackedSource<Jobs> = {
    get: async (id) => {
      const row = this.readRow(id);
      return row && toRecord(row);
    },
    list: async ({ limit, cursor, status, name }) => {
      const [createdAt, afterId] =
        cursor === undefined
          ? [-1, '']
          : (JSON.parse(cursor) as [number, string]);
      const rows = this.sql
        .exec<JobRow>(
          `SELECT * FROM jobs
           WHERE (? IS NULL OR status = ?)
             AND (? IS NULL OR name = ?)
             AND (created_at > ? OR (created_at = ? AND id > ?))
           ORDER BY created_at, id
           LIMIT ?`,
          status ?? null,
          status ?? null,
          name ?? null,
          name ?? null,
          createdAt,
          createdAt,
          afterId,
          limit + 1
        )
        .toArray();
      const page = rows.slice(0, limit);
      const last = page[page.length - 1];
      const records = page.map(toRecord);
      return rows.length > limit && last
        ? { records, cursor: JSON.stringify([last.created_at, last.id]) }
        : { records };
    },
    watch: (id, listener) => {
      const listeners = this.watchers.get(id) ?? new Set();
      this.watchers.set(id, listeners.add(listener));
      return () => {
        listeners.delete(listener);
      };
    },
  };

  private readonly tracker = new DurabilityTracker({
    sources: { jobs: this.jobs },
  });

  readonly socket = new DurabilitySocket({
    context: this.ctx,
    inbound: {
      importOrder: {
        input: z.string().min(1),
        handle: async (orderId: string) => {
          const pipeline = this.env.PIPELINE.getByName(this.tenant);
          await pipeline.importOrder(orderId);
          return pipeline.track(orderId).wait({ timeoutMs: 30_000 });
        },
      },
      work: {
        input: workQuery,
        handle: (query: {
          cursor?: string | undefined;
          status?: TrackedStatus | undefined;
        }) => this.tracker.records(query),
      },
    },
    outbound: { job: event<TrackedRecord<Jobs>>() },
  });

  constructor(ctx: DurableObjectState, env: Env) {
    super(ctx, env);
    this.sql.exec(
      `CREATE TABLE IF NOT EXISTS jobs (
         id TEXT PRIMARY KEY,
         name TEXT NOT NULL,
         generation TEXT NOT NULL,
         created_at INTEGER NOT NULL,
         status TEXT NOT NULL,
         state TEXT NOT NULL
       )`
    );
  }

  private get tenant(): string {
    return this.ctx.id.name ?? this.ctx.id.toString();
  }

  /** Called by the pipeline's fanout. Returning without throwing acks the batch. */
  ingest(events: JobEvent[]): void {
    for (const job of events) {
      // Deliveries can arrive out of order on retry: a late `pending` must not
      // overwrite a settled row of the same generation.
      this.sql.exec(
        `INSERT INTO jobs (id, name, generation, created_at, status, state)
         VALUES (?, ?, ?, ?, ?, ?)
         ON CONFLICT(id) DO UPDATE SET
           generation = excluded.generation,
           created_at = excluded.created_at,
           status = excluded.status,
           state = excluded.state
         WHERE excluded.generation != jobs.generation OR jobs.status = 'pending'`,
        job.id,
        job.name,
        job.generation,
        job.createdAt,
        job.state.status,
        JSON.stringify(job.state)
      );
      const row = this.readRow(job.id);
      if (!row) {
        continue;
      }
      const record = toRecord(row);
      this.socket.job(record);
      for (const listener of this.watchers.get(job.id) ?? []) {
        listener();
      }
    }
  }

  private readRow(id: string): JobRow | undefined {
    return this.sql
      .exec<JobRow>('SELECT * FROM jobs WHERE id = ?', id)
      .toArray()[0];
  }

  override fetch(request: Request) {
    return this.socket.fetch(request);
  }

  override webSocketMessage(ws: WebSocket, message: string | ArrayBuffer) {
    return this.socket.webSocketMessage(ws, message);
  }

  override webSocketClose(ws: WebSocket) {
    this.socket.webSocketClose(ws);
  }
}

/**
 * Browser side of the `job-dashboard` example. Loads a snapshot of tracked
 * work through `work`, applies `job` events as they arrive, and starts
 * imports through the dashboard, which waits on the pipeline for the result.
 */
import { DurabilitySocketClient } from 'durability/client';
import type { TrackedRecord } from 'durability/experimental';
import type { Dashboard } from './dashboard';
import type { Jobs } from './shared';

const client = new DurabilitySocketClient<Dashboard['socket']>({
  url: (tenant) => `wss://jobs.example.com/tenants/${tenant}/ws`,
});

export const openDashboard = (
  tenant: string,
  render: (jobs: TrackedRecord<Jobs>[]) => void
) => {
  const conn = client.connect(tenant);
  const jobs = new Map<string, TrackedRecord<Jobs>>();

  const reload = async () => {
    jobs.clear();
    let cursor: string | undefined;
    do {
      // eslint-disable-next-line no-await-in-loop
      const page = await conn.rpc.work({ cursor });
      for (const record of page.records) {
        jobs.set(record.id, record);
      }
      cursor = page.cursor;
    } while (cursor !== undefined);
    render([...jobs.values()]);
  };

  // Fires on first connect and whenever missed events could not be replayed.
  conn.on.resync(() => void reload());
  conn.on.job((record) => {
    jobs.set(record.id, record);
    render([...jobs.values()]);
  });

  return {
    /** Resolves with the job's outcome; other tabs see it through `job`. */
    importOrder: (orderId: string) => conn.rpc.importOrder(orderId),
    close: () => conn.close(),
  };
};

import type { StandardSchemaV1 } from '@standard-schema/spec';
import { trace, traceFunction, traceMethods, traceStorage } from './dev.js';
import { reportFailure, serializeError } from './errors.js';
import {
  durabilitySocketMigrations,
  migrateCapability,
  type DurabilityMigrationResult,
} from './migrations.js';
import { deserialize, serialize } from './serialization.js';
import {
  parseClientFrame,
  socketMaxFrameLength,
  socketProtocolErrorCode,
  socketProtocolVersion,
  type SocketCall,
  type SocketClientFrame,
  type SocketInboundHandler,
  type SocketInboundMap,
  type SocketOutboundMap,
  type SocketOutboundPayload,
  type SocketServerFrame,
  type SocketInboundMetadata,
  type SocketShape,
  socketShape,
} from './socket-protocol.js';

/** The Durable Object state a socket needs: SQLite storage and hibernatable WebSockets. */
export type DurabilitySocketContext = Pick<
  DurableObjectState,
  'storage' | 'acceptWebSocket' | 'getWebSockets' | 'setWebSocketAutoResponse'
>;

export type DurabilitySocketConfig<
  Inbound extends SocketInboundMap,
  Outbound extends SocketOutboundMap,
  Metadata extends SocketInboundMetadata<Inbound>,
> = {
  context: DurabilitySocketContext;
  /** Methods clients may call, keyed by the names they become on the client. */
  inbound: Inbound;
  /** Events the object may push, keyed by the methods they become on this instance. */
  outbound: Outbound;
  /**
   * Captures trusted per-connection data from the upgrade request. Stored in
   * the WebSocket attachment, so it survives hibernation without re-auth.
   */
  metadata?: (request: Request) => Metadata | Promise<Metadata>;
  /** Outbound events kept for replay to reconnecting clients. Defaults to 1000. */
  retention?: number;
};

/** Thrown to the caller when an inbound payload fails its schema. */
export class SocketInputError extends Error {
  readonly issues: readonly StandardSchemaV1.Issue[];

  constructor(method: string, issues: readonly StandardSchemaV1.Issue[]) {
    super(
      `Invalid input for "${method}": ${issues.map((issue) => issue.message).join('; ')}`
    );
    this.name = 'SocketInputError';
    this.issues = issues;
  }
}

type Attachment<Metadata> = {
  metadata: Metadata;
  names: string[];
};

type EventRow = {
  seq: number;
  name: string;
  payload: string;
};

declare const WebSocketPair: {
  new (): { 0: WebSocket; 1: WebSocket };
};

declare const WebSocketRequestResponsePair: {
  new (request: string, response: string): WebSocketRequestResponsePair;
};

const reservedOutboundNames = new Set(['status', 'resync']);

const sendFrame = (ws: WebSocket, frame: SocketServerFrame): void => {
  trace('socket.send', { frame });
  let text = JSON.stringify(frame);
  if (text.length > socketMaxFrameLength) {
    if (frame.type === 'event') {
      reportFailure(
        'durability.socket.event_too_large',
        { name: frame.name, seq: frame.seq },
        new Error('Outbound event exceeds the frame limit')
      );
      return;
    }
    text = JSON.stringify({
      v: socketProtocolVersion,
      type: 'error',
      id: frame.id,
      error: {
        name: 'SocketFrameTooLargeError',
        message: `Response to "${frame.type}" exceeds ${socketMaxFrameLength} characters`,
      },
    } satisfies SocketServerFrame);
  }
  try {
    ws.send(text);
  } catch (error) {
    reportFailure('durability.socket.send_failed', { type: frame.type }, error);
  }
};

const readAttachment = <Metadata>(ws: WebSocket): Attachment<Metadata> => {
  const stored: unknown = ws.deserializeAttachment();
  if (typeof stored !== 'object' || stored === null) {
    return { metadata: undefined as Metadata, names: [] };
  }
  const names = Reflect.get(stored, 'names');
  return {
    metadata: Reflect.get(stored, 'metadata') as Metadata,
    names: Array.isArray(names)
      ? names.filter((name): name is string => typeof name === 'string')
      : [],
  };
};

class DurabilitySocketCore<
  Inbound extends SocketInboundMap,
  Outbound extends SocketOutboundMap,
  Metadata extends SocketInboundMetadata<Inbound>,
> {
  declare readonly [socketShape]: SocketShape<Inbound, Outbound>;

  readonly #context: DurabilitySocketContext;
  readonly #storage: DurableObjectStorage;
  readonly #inbound: Map<string, SocketInboundHandler<never>>;
  readonly #outboundNames: Set<string>;
  readonly #metadata: DurabilitySocketConfig<
    Inbound,
    Outbound,
    Metadata
  >['metadata'];
  readonly #retention: number;

  constructor(config: DurabilitySocketConfig<Inbound, Outbound, Metadata>) {
    const retention = config.retention ?? 1_000;
    if (!Number.isSafeInteger(retention) || retention <= 0) {
      throw new RangeError('retention must be a positive safe integer');
    }

    this.#context = config.context;
    this.#storage = traceStorage(config.context.storage);
    // Only own keys are callable; nothing on the prototype chain is reachable.
    this.#inbound = new Map(Object.entries(config.inbound));
    this.#metadata = config.metadata;
    this.#retention = retention;
    this.#outboundNames = new Set(Object.keys(config.outbound));

    const reserved = new Set([
      ...Object.getOwnPropertyNames(DurabilitySocketCore.prototype),
      ...reservedOutboundNames,
    ]);
    for (const name of this.#outboundNames) {
      if (reserved.has(name)) {
        throw new Error(`"${name}" is reserved by DurabilitySocket`);
      }
      Object.defineProperty(this, name, {
        value: traceFunction(`DurabilitySocket.${name}`, (payload: unknown) =>
          this.#publish(name, payload)
        ),
        enumerable: true,
      });
    }

    migrateCapability(
      this.#storage,
      'socket',
      durabilitySocketMigrations,
      undefined
    );
    // Keep-alive pings are answered by the runtime without waking a hibernated object.
    this.#context.setWebSocketAutoResponse(
      new WebSocketRequestResponsePair(
        JSON.stringify({ v: socketProtocolVersion, type: 'ping' }),
        JSON.stringify({ v: socketProtocolVersion, type: 'pong' })
      )
    );
  }

  /** Applies or reverts the `durability_socket_events` schema. */
  static migrate(
    context: Pick<DurableObjectState, 'storage'>,
    target?: string | null
  ): DurabilityMigrationResult {
    return migrateCapability(
      context.storage,
      'socket',
      durabilitySocketMigrations,
      target
    );
  }

  /** Accepts a WebSocket upgrade; call from the Durable Object's `fetch`. */
  async fetch(request: Request): Promise<Response> {
    if (request.headers.get('Upgrade')?.toLowerCase() !== 'websocket') {
      return new Response('Expected a WebSocket upgrade', { status: 426 });
    }

    const metadata = await this.#metadata?.(request);
    const pair = new WebSocketPair();
    const client = pair[0];
    const server = pair[1];
    this.#context.acceptWebSocket(server);
    server.serializeAttachment({
      metadata,
      names: [],
    } satisfies Attachment<Metadata | undefined>);

    return new Response(null, { status: 101, webSocket: client });
  }

  /** Handles one inbound frame; call from the Durable Object's `webSocketMessage`. */
  async webSocketMessage(
    ws: WebSocket,
    message: string | ArrayBuffer
  ): Promise<void> {
    const text =
      typeof message === 'string' ? message : new TextDecoder().decode(message);
    const frame = parseClientFrame(text);
    if (!frame) {
      trace('socket.close', {
        code: socketProtocolErrorCode,
        reason: 'Invalid frame',
      });
      ws.close(socketProtocolErrorCode, 'Invalid frame');
      return;
    }
    if (frame.type === 'subscribe') {
      this.#subscribe(ws, frame);
      return;
    }
    await this.#call(ws, frame);
  }

  /**
   * Call from the Durable Object's `webSocketClose`. Subscriptions live in the
   * socket's attachment, so nothing needs cleaning up; the runtime requires
   * the method to exist for hibernatable WebSockets.
   */
  webSocketClose(_ws: WebSocket): void {}

  async #call(
    ws: WebSocket,
    frame: Extract<SocketClientFrame, { type: 'call' }>
  ): Promise<void> {
    const handler = this.#inbound.get(frame.method);
    if (!handler) {
      sendFrame(ws, {
        v: socketProtocolVersion,
        type: 'error',
        id: frame.id,
        error: {
          name: 'SocketUnknownMethodError',
          message: `Unknown method "${frame.method}"`,
        },
      });
      return;
    }

    const call: SocketCall<never> = {
      metadata: readAttachment<never>(ws).metadata,
    };
    try {
      const output =
        typeof handler === 'function'
          ? await handler(frame.input, call)
          : await handler.handle(
              (await this.#validate(
                frame.method,
                handler.input,
                frame.input
              )) as never,
              call
            );
      sendFrame(ws, {
        v: socketProtocolVersion,
        type: 'result',
        id: frame.id,
        output,
      });
    } catch (error) {
      sendFrame(ws, {
        v: socketProtocolVersion,
        type: 'error',
        id: frame.id,
        error: serializeError(error),
      });
    }
  }

  async #validate(
    method: string,
    schema: StandardSchemaV1,
    input: unknown
  ): Promise<unknown> {
    const result = await schema['~standard'].validate(input);
    if (result.issues) {
      throw new SocketInputError(method, result.issues);
    }
    return result.value;
  }

  #subscribe(
    ws: WebSocket,
    frame: Extract<SocketClientFrame, { type: 'subscribe' }>
  ): void {
    const unknown = frame.names.find((name) => !this.#outboundNames.has(name));
    if (unknown !== undefined) {
      sendFrame(ws, {
        v: socketProtocolVersion,
        type: 'error',
        id: frame.id,
        error: {
          name: 'SocketUnknownEventError',
          message: `Unknown event "${unknown}"`,
        },
      });
      return;
    }

    const attachment = readAttachment<Metadata>(ws);
    ws.serializeAttachment({
      ...attachment,
      names: frame.names,
    } satisfies Attachment<Metadata>);

    const head = this.#headSeq();
    const replayed = this.#replay(ws, frame.names, frame.since, head);
    sendFrame(ws, {
      v: socketProtocolVersion,
      type: 'subscribed',
      id: frame.id,
      seq: head,
      replayed,
    });
  }

  /**
   * Replays retained events after `since` and reports whether the client is
   * now caught up. A client ahead of the log or behind its oldest row must
   * rebuild its state from a fresh snapshot instead.
   */
  #replay(
    ws: WebSocket,
    names: string[],
    since: number | undefined,
    head: number
  ): boolean {
    if (since === undefined || since > head) {
      return false;
    }
    if (since === head) {
      return true;
    }
    const oldest = this.#storage.sql
      .exec<{ seq: number }>(
        'SELECT MIN(seq) AS seq FROM durability_socket_events'
      )
      .one().seq;
    if (since + 1 < oldest) {
      return false;
    }

    const rows = this.#storage.sql
      .exec<EventRow>(
        'SELECT seq, name, payload FROM durability_socket_events WHERE seq > ? ORDER BY seq',
        since
      )
      .toArray();
    const wanted = new Set(names);
    for (const row of rows) {
      if (!wanted.has(row.name)) {
        continue;
      }
      sendFrame(ws, {
        v: socketProtocolVersion,
        type: 'event',
        seq: row.seq,
        name: row.name,
        payload: deserialize(row.payload),
      });
    }
    return true;
  }

  #headSeq(): number {
    const row = this.#storage.sql
      .exec<{ seq: number | null }>(
        'SELECT MAX(seq) AS seq FROM durability_socket_events'
      )
      .one();
    return row.seq ?? 0;
  }

  #publish(name: string, payload: unknown): void {
    const serialized = serialize(payload);
    const sql = this.#storage.sql;
    const { seq } = sql
      .exec<{ seq: number }>(
        'INSERT INTO durability_socket_events (name, payload, created_at) VALUES (?, ?, ?) RETURNING seq',
        name,
        serialized,
        Date.now()
      )
      .one();
    sql.exec(
      'DELETE FROM durability_socket_events WHERE seq <= ?',
      seq - this.#retention
    );

    const frame: SocketServerFrame = {
      v: socketProtocolVersion,
      type: 'event',
      seq,
      name,
      payload,
    };
    for (const ws of this.#context.getWebSockets()) {
      if (ws.readyState !== WebSocket.OPEN) {
        continue;
      }
      if (!readAttachment(ws).names.includes(name)) {
        continue;
      }
      sendFrame(ws, frame);
    }
  }
}

type OutboundMethods<Outbound extends SocketOutboundMap> = {
  [Name in keyof Outbound]: (
    payload: SocketOutboundPayload<Outbound[Name]>
  ) => void;
};

/**
 * Typed WebSocket RPC and event push for one Durable Object, over hibernatable
 * WebSockets.
 *
 * Every `outbound` event becomes a method on the instance that persists the
 * event for replay and pushes it to subscribed clients. Clients call `inbound`
 * handlers through `durability/client`.
 *
 * @example
 * ```ts
 * class TodoList extends DurableObject<Env> {
 *   readonly socket = new DurabilitySocket({
 *     context: this.ctx,
 *     inbound: {
 *       list: () => this.repo.list(),
 *       create: {
 *         input: TodoSchema,
 *         handle: async (todo: Todo) => {
 *           await this.repo.insert(todo);
 *           this.socket.created(todo);
 *         },
 *       },
 *     },
 *     outbound: { created: event<Todo>() },
 *   });
 *
 *   fetch(request: Request) {
 *     return this.socket.fetch(request);
 *   }
 *
 *   webSocketMessage(ws: WebSocket, message: string | ArrayBuffer) {
 *     return this.socket.webSocketMessage(ws, message);
 *   }
 *
 *   webSocketClose(ws: WebSocket) {
 *     this.socket.webSocketClose(ws);
 *   }
 * }
 * ```
 */
export type DurabilitySocket<
  Inbound extends SocketInboundMap,
  Outbound extends SocketOutboundMap,
  Metadata extends SocketInboundMetadata<Inbound> =
    SocketInboundMetadata<Inbound>,
> = DurabilitySocketCore<Inbound, Outbound, Metadata> &
  OutboundMethods<Outbound>;

export interface DurabilitySocketConstructor {
  new <
    Inbound extends SocketInboundMap,
    Outbound extends SocketOutboundMap,
    Metadata extends SocketInboundMetadata<Inbound> =
      SocketInboundMetadata<Inbound>,
  >(
    config: DurabilitySocketConfig<Inbound, Outbound, Metadata>
  ): DurabilitySocket<Inbound, Outbound, Metadata>;
  migrate: typeof DurabilitySocketCore.migrate;
}

traceMethods(DurabilitySocketCore.prototype, 'DurabilitySocket');

// TypeScript cannot declare class members derived from a generic event map,
// so the constructor is typed separately; the constructor installs and
// reserves every outbound method at runtime.
export const DurabilitySocket =
  DurabilitySocketCore as unknown as DurabilitySocketConstructor;

export { durabilitySocketMigrations } from './migrations.js';
export {
  event,
  type SocketCall,
  type SocketInboundHandler,
  type SocketInboundMap,
  type SocketOutboundEvent,
  type SocketOutboundMap,
} from './socket-protocol.js';

Normal projection handoffs

Details

Each handoff has its own owner and storage; the Dashboard’s SQL view is not the Pipeline’s operation record.

Durable state: Nothing is written yet.
In-flight / in memory: The HTTP request waits on the RPC call.

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.

JavaScript is disabled. Read the complete example source or follow the complete ordered walkthrough below.

examples/job-dashboard/worker.ts

examples/job-dashboard/pipeline.ts

examples/job-dashboard/dashboard.ts

examples/job-dashboard/client.ts

packages/durability/src/socket.ts

Normal projection handoffs

One Pipeline per tenant. The front Worker names the Pipeline after the tenant and calls importOrder() over RPC. The tenant’s Dashboard is a different object with its own storage and failure boundary. Read these source lines. Durable state: Nothing is written yet. In memory: The HTTP request waits on the RPC call.
Register the import. importOrder registers an operation with the stable ID import:<orderId> in durability_calls and resolves on acceptance, not completion. Repeating the import targets the same row. Read these source lines. Durable state: A pending import row commits with the Pipeline’s alarm. In memory: If a permit is free, the attempt starts in the background now.
One scheduler, one alarm. The Pipeline’s scheduler drives both its operations and its fanout from the object’s one alarm, and alarm() delegates to it. The Dashboard has no scheduler: SQL, its tracker source, and the socket need no background work. Read these source lines. Durable state: The alarm is set for the earliest pending row. In memory: The scheduler and its permits are rebuilt with every new instance.
Fetch the order with the operation ID. The handler sends the operation ID as Idempotency-Key and passes the attempt’s abort signal. ORDERS must honour the key, because a retry can repeat the request. Read these source lines. Durable state: The import row stays pending while the request runs. In memory: The fetch and its abort signal.
Expose operation results to the hook. Durability owns the import operation and its result. This option puts a completed summary on lifecycle events; it does not register anything itself. Read these source lines. Durable state: The Pipeline retains the operation and its result under the stable import ID. In memory: Lifecycle hooks run in memory after the change commits; nothing records that they ran.
Select a projection state. The hook maps registered to pending, a completed attempt with its result to completed, and a terminal failure to failed. Read these source lines. Durable state: The operation has settled before this callback is invoked. In memory: The callback is best-effort work in the Pipeline instance.
Commit the fanout obligation. The delivery ID includes operation ID, generation, and status, so a committed fanout message can be retried safely. The enqueue commits in the Pipeline’s storage and sets its alarm. Read these source lines. Durable state: After enqueue commits, the fanout retains the delivery until Dashboard.ingest returns without throwing. In memory: The hook call that started the enqueue is not itself a durable handoff.
Deliver over RPC to the Dashboard. After up to 100 ms of batching, the fanout calls ingest() on the tenant’s Dashboard. With ackOnReturn the batch is acknowledged when ingest returns; a throw or an eviction leaves the rows to be retried. Read these source lines. Durable state: The delivery rows stay in the Pipeline until the ack. In memory: The RPC call between the two objects.
Project, then publish. ingest upserts the Dashboard’s jobs table and only then emits the socket job event, with no await in between. The predicate stops a late pending event from overwriting a settled row of the same generation; a different generation always passes, so it is not a freshness order. Read these source lines. Durable state: The jobs table and socket event history are Dashboard-owned storage, separate from the Pipeline. In memory: Watcher callbacks and current socket connections are in memory.
Render the SQL snapshot. On resync the client pages work() records, replaces its local map, then applies incremental job events. Read these source lines. Durable state: The snapshot comes from the Dashboard’s jobs table through its tracker records API. In memory: The browser map is rebuilt and can be discarded at any time.

Why a direct-push dashboard loses updates

Naive: push to browsers from the hook. A naive hook sends to sockets directly, so an unreachable object or browser loses the update. Here the hook only enqueues a fanout message with a stable ID, and delivery to the Dashboard is retried from the Pipeline’s storage. Read these source lines. Durable state: A delivery row per projection event. In memory: Nothing is sent from the Pipeline to a browser.
Naive: the browser holds the only copy. The naive browser keeps the only copy of job state. Here ingest upserts the jobs table in the Dashboard’s own storage, so the browser’s map can be thrown away. Read these source lines. Durable state: One projected row per job in the Dashboard. In memory: The browser map is only a cache.
Naive: a disconnected browser misses updates. Here each job event is stored in durability_socket_events before it is sent, so a reconnecting client can replay what it missed while the event is still retained. Read these source lines. Durable state: The last 1,000 events, by default. In memory: Open WebSockets receive the frame immediately.
One gap remains before the enqueue. All three fixes start once the enqueue has committed. The hook itself is best effort; the “Lost projection before fanout commits” scenario steps through that gap. Read these source lines. Durable state: Nothing is stored until the enqueue commits. In memory: The hook call can be lost with the instance.

Lost projection before fanout commits

The operation settles first. A completed operation has an authoritative retained result in durability_calls before the lifecycle hook tries to publish it. Read these source lines. Durable state: The Pipeline has the completed operation result. In memory: The hook has not enqueued anything yet.
Eviction can lose the handoff. If the Pipeline is evicted after settlement but before the enqueue commits, there is no fanout row to retry. The source comment names this lifecycle hook as best effort. Read these source lines. Durable state: No fanout row exists in this failure window. In memory: The interrupted hook call disappears with the instance.
Resync reads only Dashboard state. The client reacts to resync by calling work() and rebuilding its map. That cannot repair the missing event, because work() reads the Dashboard’s jobs table, where no row was ever ingested. Read these source lines. Durable state: Existing Dashboard rows remain; the absent projection has no row and no socket event. In memory: The reconnect and the browser cache never consult the Pipeline.
Reconciliation remains application work. Production needs an explicit process that reads authoritative Pipeline history and idempotently writes missing rows into the projection. These source files do not implement it. Read these source lines. Durable state: A reconciler would need to write the missing Dashboard row itself. In memory: No client resync or fanout retry can invent an event that was never enqueued.

Recover a dropped connection

Ingest returns or the batch stays. If ingest throws, or either object is evicted during the call, the fanout keeps the rows in the Pipeline and retries the batch. Once ingest returns, the batch is acknowledged. Read these source lines. Durable state: Delivery rows stay until the ack commits. In memory: The RPC call can be lost; the rows cannot.
Store the event, then send it. Each job event gets a sequence number in durability_socket_events, events beyond the retention limit are deleted, and the frame goes to every open WebSocket subscribed to job. Read these source lines. Durable state: The event and its sequence number. In memory: The frame on each open WebSocket.
The connection drops. While disconnected, the browser misses job events. The client remembers the last sequence number it saw; its map is only a cache. Read these source lines. Durable state: The Dashboard keeps storing events and rows. In memory: Missed frames and the stale browser map.
Replay from the last sequence. On reconnect the client subscribes with that sequence number. Retained events after it are replayed; if it is older than the oldest retained event, or ahead of the log, the subscription reports that replay was impossible. Read these source lines. Durable state: Replay reads durability_socket_events. In memory: The reconnecting socket’s subscription.
Rebuild from the Dashboard snapshot. On first connect and whenever replay was impossible, resync fires: the client pages work(), which reads tracker records over the jobs table, and replaces its map. This recovers transport and replay gaps, not a missing enqueue. Read these source lines. Durable state: The snapshot is the Dashboard’s jobs table. In memory: A new browser map.

There is one named Pipeline and one named Dashboard per tenant. They are separate failure and persistence boundaries. The pipeline’s scheduler drives both its Durability operations and DurabilityFanout; its alarm() delegates to that one scheduler. The dashboard has no scheduler because SQL, its custom tracker source, and DurabilitySocket do not need background alarm work in this example.

State or responsibility	Owner	Identity or recovery boundary
Order execution and result	`Pipeline`	`import:<orderId>` operation record
Tracker source for execution	`Pipeline` `DurabilityTracker`	operation state retained by Durability
Dashboard delivery	Pipeline fanout	`id:generation:status` message until acknowledged
Browser read model	`Dashboard` SQL	tenant dashboard object and SQL upsert
Connected-browser cache	Browser	reloads from Dashboard `work` snapshot

No row in this table turns the cross-object handoff into a transaction. The identity makes retries safe at each boundary; reconciliation is still needed where an event is never handed off.

End-to-end flow

The front worker routes POST /tenants/:tenant/imports?order=:id to that tenant’s Pipeline; /ws upgrades against its Dashboard.
Pipeline.importOrder(orderId) registers an operation with the stable ID import:<orderId>. Registration means accepted, not completed. Repeating the same import targets the same operation record.
The operation fetches ORDERS using the durable operation ID as Idempotency-Key. The external service must honor that key because retries can repeat a request.
The pipeline’s DurabilityTracker exposes its operations as source jobs. track(orderId) gets the handle for import:<orderId>; the front worker and dashboard can call wait() in one pipelined Workers RPC path.
includeResults: true puts a completed operation’s order summary on its lifecycle event. Lifecycle callbacks publish that pending, completed, or failed JobEvent; each fanout message has a stable identity: id:generation:status.
Fanout shares the pipeline scheduler and batches for up to 100 ms. It invokes Dashboard.ingest(messages) and acknowledges only when that method returns (ackOnReturn: true), so a failed call or eviction is redelivered.
Dashboard.ingest upserts its own jobs SQL table, emits a socket job event, and wakes local tracker watchers. Its TrackedSource is the SQL table plus the in-memory watcher map; it is not the pipeline’s source.
On connection or a replay gap, the browser calls socket RPC work, pages through tracker.records(), and replaces its local map with the Dashboard snapshot. Incremental job events then update that map.

The connecting code

The stable operation identity makes registration retries safe:

importOrder(orderId: string) {
  return this.durability.importOrder({
    id: `import:${orderId}`,
    payload: { orderId },
  });
}

Lifecycle events become independently retried fanout messages. Include the generation in the delivery ID: reusing an operation ID for a new generation must not collide with its earlier projection event.

await this.fanout.enqueue([
  {
    id: `${event.id}:${event.generation}:${state.status}`,
    body: {
      id: event.id,
      name: 'importOrder',
      generation: event.generation,
      createdAt: event.timestamp,
      state,
    },
  },
]);

The dashboard only lets a same-generation event replace a pending row. A different generation can replace it, regardless of age; the SQL does not order generations. A late same-generation pending cannot overwrite a settled row, but a delayed older generation is not fenced by this predicate.

ON CONFLICT(id) DO UPDATE SET status = excluded.status, state = excluded.state
WHERE excluded.generation != jobs.generation OR jobs.status = 'pending'

A naive design and its break

A tempting version sends a WebSocket message directly from the operation’s lifecycle callback and treats the client as the state store. It loses updates whenever a browser, object, or network call disappears. Fanout fixes delivery after enqueue, and the socket fixes a retained-event replay gap, but there is an earlier gap in this example.

The walkthrough’s “Why a direct-push dashboard loses updates” scenario steps through the lines of this composition that replace each naive step, and its “Lost projection before fanout commits” scenario steps through the remaining gap.

The hook is best effort even though fanout.enqueue is durable once it commits. An eviction between settling the operation and committing that enqueue can lose the completed projection permanently. Socket replay/resync cannot repair it: work reads Dashboard SQL, not authoritative Pipeline operations. Reconnecting to a stale dashboard only reproduces stale state.

For correctness-sensitive production dashboards, reconcile from an authoritative pipeline/history source: periodically enumerate or query authoritative jobs, compare generations and states with the projection, and idempotently write missing rows. Alternatively, stage the projection handoff from inside the operation handler with an application-owned durable record keyed by operation ID, then drain/reconcile it. Either approach needs defined retention and conflict rules.

Recovery boundaries

The walkthrough’s “Recover a dropped connection” scenario steps through this: the fanout retries a batch until ingest returns, the socket stores each event before sending it, and a reconnecting client either replays what it missed or rebuilds from the work snapshot.

If ingest throws before returning, fanout retains and retries the batch. If it returns, that batch is acknowledged. A client that missed socket history receives a resync signal and rebuilds from SQL. These mechanisms recover transport and replay gaps, not the missing lifecycle-to-fanout enqueue described above.

Failure and semantics table

Situation	What this composition does	Remaining concern
Duplicate import request	Stable `import:<orderId>` ID deduplicates at the Pipeline operation boundary.	Reusing an ID has generation semantics; choose that API deliberately.
`ORDERS` call retries	Sends the operation ID as `Idempotency-Key`.	`ORDERS` must implement idempotency.
Dashboard call fails or object is evicted	Fanout retries until `ingest` returns.	The original lifecycle enqueue must already have committed.
Late `pending` delivery	SQL refuses to replace a settled row of the same generation.	This is not a total ordering protocol.
Failed attempt followed by retry	A failed `attempt_settled` event can project `failed` while the operation is still pending retry. The SQL can then reject a later same-generation `completed` update.	This is a concrete recipe limitation: project terminal failure only after checking authoritative state, and add ordered reconciliation.
Old generation arrives late	Any different generation passes the SQL update predicate.	Add an authoritative generation/version order; inequality alone is not a freshness check.
Browser misses events	Socket replay, then `resync` and `work` snapshot recover the Dashboard view.	Snapshot cannot discover a pipeline event never projected.
RPC waiter disconnects or times out	The operation continues; a later caller can query/wait again.	`wait()` is an ephemeral request, not a durable subscription.

Production work left to the application

Authenticate tenant selection, authorize imports and dashboard reads, validate order IDs, and replace wss://jobs.example.com in the client with the deployed host. Add rate limits and limits to work queries. Define projection reconciliation, retention, and explicit same-generation terminal conflict semantics before treating the dashboard as a business record. For durable user notifications, store subscription intent separately; an RPC wait is only a live connection’s convenience.