Skip to Content
RealtimeLive Subscriptions

Realtime Subscriptions

The Rust SDK supports live queries over a shared WebSocket connection.

Ownership boundary:

  • link-common owns the shared connection, reconnect loop, replay filtering, and SeqId checkpoint tracking.
  • Your app owns the async loop over live.next() or events.next().

Live SQL must follow the strict supported shape: SELECT ... FROM ... WHERE ....

Do not include ORDER BY or LIMIT in live subscription SQL. Apply ordering or row caps in application code after rows arrive, or use LiveRowsConfig::limit.

Call client.connect().await? before opening subscriptions so they multiplex over one socket (unless you rely on lazy connect on first subscription).

Which API to use

APIUse when
live() / live_with_config()You want the current materialized row list (recommended for UI and most services)
live_events() / live_events_with_config()You need raw ChangeEvent frames for custom reconciliation or debugging

Unlike execute_query, live APIs take the final SQL string directly — there is no separate params argument. Embed constants in the WHERE clause or use separate subscriptions per filter.

Materialized live rows

RUST
use kalam_client::{    LiveRowsConfig, LiveRowsEvent, SubscriptionConfig, SubscriptionOptions,}; client.connect().await?; let mut config = SubscriptionConfig::new(    "inbox",    "SELECT id, room, body FROM app.messages WHERE room = 'main'",);config.options = Some(    SubscriptionOptions::new()        .with_last_rows(20)        .with_batch_size(50),); let mut live = client    .live_with_config(        config,        LiveRowsConfig {            limit: Some(20),            ..LiveRowsConfig::default()        },    )    .await?; while let Some(event) = live.next().await {    match event? {        LiveRowsEvent::Rows { rows, last_seq_id, .. } => {            println!("{} rows, last_seq_id={last_seq_id:?}", rows.len());            for row in rows {                println!("{}", row.get("body").and_then(|v| v.as_text()).unwrap_or(""));            }        }        LiveRowsEvent::Error { code, message, .. } => {            eprintln!("live error {code}: {message}");            break;        }    }} live.close().await?;

Shorthand when defaults are enough:

RUST
let mut live = client.live("SELECT * FROM app.tasks WHERE status = 'open'").await?;

Control knobs

KnobLayerEffect
with_batch_size(n)Server snapshotChunks initial rows from the server
with_last_rows(n)Server rewindHow much history to fetch on connect
LiveRowsConfig { limit: Some(n), .. }Client materializerCaps rows kept in memory after reconciliation
with_from(seq_id)ResumeContinue from a persisted SeqId
RUST
use kalam_client::{SeqId, SubscriptionOptions}; let options = SubscriptionOptions::new()    .with_last_rows(200)    .with_from(SeqId::from(42_i64));

Each LiveRowsEvent::Rows includes last_seq_id for durable checkpoints.

Low-level change events

RUST
use kalam_client::{ChangeEvent, SubscriptionConfig}; let config = SubscriptionConfig::new(    "raw-inbox",    "SELECT * FROM app.messages WHERE room = 'main'",); let mut events = client.live_events_with_config(config).await?; while let Some(change) = events.next().await {    match change? {        ChangeEvent::Insert { rows, .. } => println!("inserted {}", rows.len()),        ChangeEvent::Update { rows, .. } => println!("updated {}", rows.len()),        ChangeEvent::Delete { old_rows, .. } => println!("deleted {}", old_rows.len()),        other => println!("event: {:?}", other),    }} events.close().await?;

Inspect active subscriptions

RUST
let subs = client.subscriptions().await;for info in subs {    println!("{} — {}", info.id, info.query);}

Cancel one explicitly:

RUST
client.cancel_subscription("subscription-id").await?;

Topic consumers (optional)

Topic workers are behind the consumer feature — not a separate crate.

Cargo.toml consumerTOML
kalam-client = { version = "0.8.0", features = ["native-sdk", "consumer"] }
RUST
use std::time::Duration;use kalam_client::AutoOffsetReset; let mut consumer = client    .consumer()    .group_id("billing-workers")    .topic("orders.events")    .auto_offset_reset(AutoOffsetReset::Earliest)    .enable_auto_commit(false)    .poll_timeout(Duration::from_secs(10))    .build()?; let records = consumer.poll().await?;for record in &records {    // process record.payload    consumer.mark_processed(record);}consumer.commit_sync().await?;consumer.close().await?;

Full walkthrough: Topic Consumer example.

Next

SHARED table ownership and resume domains

Clustered servers default to shared_shards = 4. Each SHARED table owns exactly one Raft group. CREATE persists table_options.shared_shard_id and a fresh history_incarnation in replicated table metadata. Increasing the configured count places new tables across the expanded set; existing owners do not change. Do not remove a shard that owns tables. Legacy metadata without an owner stays on shard 0. Moving existing data requires explicit export/import into a fresh history.

_version = (raft_log_index << 16) | ordinal is ordered only within its table’s domain. Different tables on the same shard can interleave, so per-table versions can have gaps. Transactions spanning different owner groups are rejected.

A SHARED subscription acknowledgement includes batch_control.version_domain with history_incarnation, table_id, scope_id (the shared owner), and partition_id (currently null). Persist this complete envelope together with the last safely consumed version. Resume by supplying both options.from and options.version_domain. Missing envelopes, another table/group/history, and unsupported partitions are rejected as stale cursors. Take a fresh snapshot after CURSOR_EXPIRED; do not reuse the old domain.

The Rust transport preserves the acknowledgement domain for automatic reconnect. In TypeScript, use versionDomain with from when restoring a persisted checkpoint; it serializes as version_domain. Treat the domain as opaque and never infer it from the numeric version.

SQL QueriesTopic Consumers & ACK