Changes
Get told when a view advances, so you query only when there is new data
The change feed tells you when a view advanced. Instead of re-running a query on a timer, you poll a small feed and query only when the view moved. Use it for live dashboards, cache invalidation, and agents that react to new data. It needs Laser Stack or LaserData Cloud.
Quick example
This assumes the readings_v1 binding from Queries and views with notify() added to the binding builder. The program opens a reader, skips old records, publishes a reading, waits for the view to move, then queries it.
import { Laser } from "@laserdata/laser-sdk"
const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms))
await using laser = await Laser.connectEnv()
const feed = await laser.watch().index("readings_v1").records()
while ((await feed.poll()).length > 0) {
// Skip batches that earlier runs left behind.
}
await laser
.stream("telemetry")
.topic("readings")
.publish()
.json({ host: "node-7", cpu: 82, status: "degraded" })
.send()
let changes = await feed.poll()
while (changes.length === 0) changes = await sleep(200).then(() => feed.poll())
for (const change of changes) {
console.log(`${change.rows} row(s), offsets ${change.fromOffset}..${change.toOffset}`)
}
const latest = await laser.query("readings_v1").whereEq("status", "degraded").fetch()
console.log(`${latest.rows.length} degraded readings`)use laser_sdk::prelude::full::*;
use std::time::Duration;
#[tokio::main]
async fn main() -> Result<(), LaserError> {
let laser = Laser::connect_env().await?;
let mut feed = laser.watch().index("readings_v1").records()?;
while !feed.poll().await?.is_empty() {
// Skip batches that earlier runs left behind.
}
let reading = serde_json::json!({"host": "node-7", "cpu": 82, "status": "degraded"});
laser
.stream("telemetry")
.topic("readings")
.publish()
.json(&reading)?
.send()
.await?;
let mut changes = feed.poll().await?;
while changes.is_empty() {
tokio::time::sleep(Duration::from_millis(200)).await;
changes = feed.poll().await?;
}
for change in &changes {
println!("{} row(s), offsets {}..{}", change.rows, change.from_offset, change.to_offset);
}
let latest = laser
.query("readings_v1")
.where_eq("status", "degraded")
.fetch()
.await?;
println!("{} degraded readings", latest.rows.len());
Ok(())
}import asyncio
import laser_sdk as ls
async def main() -> None:
async with await ls.Laser.connect_env() as laser:
feed = laser.watch(index="readings_v1")
while await feed.poll():
pass # Skip batches that earlier runs left behind.
await laser.stream("telemetry").topic("readings").publish(
{"host": "node-7", "cpu": 82, "status": "degraded"}
).send()
while not (changes := await feed.poll()):
await asyncio.sleep(0.2)
for change in changes:
print(f"{change.rows} row(s), offsets {change.from_offset}..{change.to_offset}")
latest = await laser.query("readings_v1").where_eq("status", "degraded").fetch()
print(f"{len(latest.rows)} degraded readings")
asyncio.run(main())A change record says which index moved, over which source offsets, and how many rows landed. It does not carry the rows. Query the index to read them.
Watch every view
Leave out the index to read changes for every notifying binding on one feed: laser.watch().records() in Rust and TypeScript, laser.watch() in Python. Each record's index field says which view moved.
Resume after a restart
Save the reader's offsets while it runs. After a restart, open a new reader and restore them. Without restored offsets, a new reader starts at the beginning of the feed.
const saved = feed.offsets // ReadonlyMap<number, bigint>
// After a restart
const resumed = (await laser.watch().index("readings_v1").records()).fromOffsets(saved)let saved: Vec<u64> = feed.offsets().to_vec();
// After a restart
let mut resumed = laser
.watch()
.index("readings_v1")
.records()?
.from_offsets(saved);saved = feed.offsets # list[int]
# After a restart
resumed = laser.watch(index="readings_v1", from_offsets=saved)Store the saved offsets wherever your application keeps state, such as a file or key-value state.
Good to know
- Turn the feed on per binding with
notify(). A binding without it sends no change records. - Notifications are best effort. A lost one does not lose data, because the rows are already in the view.
- The feed shows that a view moved. To be sure a query includes your own write, use a read-your-writes query.
- The SDK does not push. Your code calls
poll()and decides how long to wait between polls. - The index filter runs in the client. Use the same client setup, including the default stream, that you query with.
- On a deployment without the feed, opening the reader returns an unsupported error.