Changes

Poll a lightweight advancement feed, then query only when the view has moved. It uses the connection you already have.

Built for

Live UIs, cache invalidation, reactive pipelines

How it works

The change feed rides the same projection binding as a normal view, with one extra flag: notify. A binding registered without it only ever serves queries. A binding with it also publishes a change record every time the projector materializes a batch against that index.

Watching targets an index name - the same one you'd query.

• Rust and TypeScript build a small reader: watch().index(name).records().
• Python opens the feed with one flat call: watch(index=name).

Reading it is a drain-and-wait loop, not a callback:

• Call .poll().
• Get back whatever change records landed since the last poll, possibly none.
• Loop with a short sleep if you got nothing.

The SDK tracks feed position between polls. You do not manage offsets during a run. There is no push-callback API, so the application owns the polling loop.

Rust also provides .stream(). It returns a futures::Stream of change records and ends when caught up. Python's async reader follows the same pattern.

The reader is resumable across restarts. offsets() returns the per-partition positions consumed so far. Persist them, then seed a fresh reader with .from_offsets(saved) on the next run, exactly like a named cursor. Without this a restarted watcher starts from the feed's beginning.

A change record describes a committed range. It contains the index, source partition, offset range, and row count. to_offset is the new watermark. Query the index to read the resulting rows.

Malformed records do not block the feed. The reader skips records that do not decode. .index(name) also filters records for other indexes. Open watch().records() without an index to read every index from one feed.

Change feed requires query and watch capabilities. The example checks both before it runs.

Quick example

// orders_v1 declared with notify: true above this block
const feed = await laser.watch().index("orders_v1").records()
while ((await feed.poll()).length > 0) {
  // drain every batch a previous run left behind
}

await laser
  .topic("orders")
  .publish()
  .json({ id: 4, status: "paid", total: 20 })
  .send()

let changes = await feed.poll()
while (changes.length === 0) {
  await new Promise((resolve) => setTimeout(resolve, 200))
  changes = await feed.poll()
}
for (const change of changes) {
  console.log(
    `view advanced: ${change.rows} row(s), ` +
      `source offsets ${change.fromOffset}..${change.toOffset}`
  )
}

let mut feed = laser.watch().index("orders_v1").records()?;
while !feed.poll().await?.is_empty() {
    // drain every batch a previous run left behind
}

laser.topic("orders").publish().json(&order)?.send().await?;

loop {
    let changes = feed.poll().await?;
    if !changes.is_empty() {
        for change in changes {
            println!(
                "view advanced: {} row(s), source offsets {}..{}",
                change.rows, change.from_offset, change.to_offset
            );
        }
        break;
    }
    tokio::time::sleep(Duration::from_millis(200)).await;
}

feed = laser.watch(index="orders_v1")
while await feed.poll():
    pass  # drain every batch a previous run left behind

await laser.topic("orders").publish(
    {"id": 4, "total": 20, "status": "paid"}
).send()

while not (changes := await feed.poll()):
    await asyncio.sleep(0.2)
for change in changes:
    print(
        f"view advanced: {change.rows} row(s), "
        f"source offsets {change.from_offset}..{change.to_offset}"
    )

Full runnable example: Rust · Python · TypeScript

Key operations

Running it

Change feed runs on Laser Stack and LaserData Cloud. Standalone Iggy without laser-plane does not provide it.

Related