Watch

Follow a board's writes as they land, with the design behind catch-up, retention, and resync.

watch follows a board’s changes as an async iterable: it replays from a watermark you hold, then stays live. The surface is identical embedded and over HTTP. There is no second log behind it. Under writePath: "wal" the feed reads the same WAL that makes writes durable (see write paths); under writePath: "lake" it reads DuckLake’s own change tracking.

Pick a surface

Usage

for await (const event of engine.watch("acme_roadmap", { after })) {
  if (event.resync) {
    // re-read board state with a query, then continue.
    // The iterator already moved its cursor to event.from.
    continue;
  }
  console.log(event.records.map((r) => r.kind));
}

import { connect } from "lakefront/client";

const board = connect({ url }).board("acme_roadmap");

for await (const event of board.watch({ after })) {
if (event.resync) {
continue; // same recovery as above
}
console.log(event.records.map((r) => r.kind));
}

after defaults to the current head, so a bare watch() follows the future. Pass an older watermark to replay first. The loop ends when signal (an AbortSignal) aborts. A read failure throws.

Both iterators yield the same two shapes:

Resync is a value the loop yields and moves past, not an error to catch. The section below explains when it happens.

Latency

A write on the node you are watching wakes the iterator immediately. A write landing on a different node arrives on the next poll, every 500ms by default (pollMs on engine.watch). The feed is complete either way; the wakeup only sharpens latency for local writes.

Change records

Each record carries a seq (the watermark it landed at, comparable with the watermarks reads and writes report), a kind, a payload, and an optional actor:

interface ChangeRecord {
  seq: number;
  kind: "insert" | "upsert" | "update" | "delete" | "drop";
  payload: unknown; // shape depends on kind
  actor?: string;
}

insert and upsert payloads carry the full rows written. update carries the id and the patch, not a computed postimage: a consumer sees exactly what the caller sent. delete carries the deleted ids.

drop is a tombstone. It means the board was dropped or truncated, and it voids everything before it. A consumer that sees one should treat prior state as gone. This is also how a replica on another node learns a board disappeared (the consistency page covers how watermarks tie reads to this feed).

Retention and resync

The WAL is not an archive. Once the flusher drains a board’s records into the lake, it trims them from the WAL table: the lake already holds that history, and a second durable copy would serve no reader. Under writePath: "lake", snapshot expiry bounds the feed the same way.

A consumer whose cursor falls behind that bound cannot be handed a contiguous feed. Instead of an error, the feed yields { resync: true, from } and continues from the new floor.

The trade asks two things of a consumer:

• Applying a change record must be idempotent, because the recovery is “read current state, then continue”.
• There must be a rebuild path that re-reads state rather than assuming the tail is always available.

The engine’s own sync machinery uses the same recovery internally: a replica that falls behind retention rebuilds from the lake rather than silently missing trimmed records.

Cursor paging

watch() is engine.changes(boardId, { after, limit? }) in a loop, plus the wakeup and poll logic. Call it directly when a one-shot page fits better:

const page = (await engine.changes("tasks", { after: 0, limit: 500 })).unwrap("changes");
if (!page.resync) {
  for (const record of page.records) {
    // apply record
  }
  // next call: engine.changes("tasks", { after: page.next })
}

Over HTTP this is the changes.read procedure. BoardClient does not wrap it; reach it with the lower-level RPC client:

import { createRpcClient } from "lakefront/contract";

const rpc = createRpcClient({ url, boardId: "acme_roadmap", token });
const page = await rpc["changes.read"]({ after: 0, limit: 500 });

onChange

engine.onChange(listener) fires once per touched board, on a microtask, after a write’s ack, and only for writes this process performed. A write landing on another node does not fire it. Its job is powering watch’s instant local wakeup; consumers that need every write should watch instead.

const unsubscribe = engine.onChange((event) => {
  // event.boardId, event.watermark, event.ops
});

Over the wire

A server node exposes the feed as an SSE stream at GET /watch, which board.watch() consumes. The stream carries change and resync events plus a : hb heartbeat, is authorized with the read verb, and takes the same after query parameter. Exact frames and headers are in the wire protocol reference.

One more contract worth knowing: a replayed idempotent write (a mutation whose idempotencyKey the WAL has already seen) fires no change event on any surface. Its event fired when the original attempt landed, and firing again would double every trigger a retried duplicate touches. The idempotency guide covers the exactly-once contract.