Key-value state in depth

This page is the full reference for Key-value state. Start there for the short version.

How the store works

laser.kv(namespace) opens a store within one namespace. Keys are unique within a namespace, and scans and bulk deletes stay inside it. Every managed write goes through a mutation topic on the log first. A stored entry records the log record that wrote it in source, so you can find the source message while the log still holds it.

The store needs the managed services of Laser Stack or LaserData Cloud. Key-value operations, compare-and-swap, fenced leases, and forks have separate capabilities. Check them before you depend on them. Without a capability, the matching call returns an unsupported error before anything is sent.

Names and limits

A client with a default stream scopes the namespace to that stream, so laser.kv("config") addresses stream:<stream>/config. Lease and fence namespaces and fork IDs follow the same rule. kv_namespaces() and forks() return your stream's names without the prefix. namespace on a handle returns the name you gave it, and resource_namespace (TypeScript: resourceNamespace) returns the name it sends. See Stream-scoped resource names for the opt out.

Keys are bytes. Rust takes anything byte-like, Python takes str or bytes, and TypeScript takes Uint8Array. A key or value over its limit fails with an invalid error before the request is sent.

Write and read values

set(key) starts a write. Encode with .json(value) or .msgpack(value), pass raw bytes with .bytes(payload), or use any codec with encode_with (TypeScript: encodeWith). Add an optional expiry with .ttl(..), which counts from now, or with .expires_at(epoch_micros) for an absolute time (TypeScript: expiresAt). Then call .send().

Read bytes with get(key), decode JSON with get_typed(key), or decode with any codec through get_as (TypeScript: getAs). TypeScript's getTyped(key, decode) also takes a decode function that checks the value. get_entry(key) (TypeScript: getEntry) returns the value with its key, version, expiry, and source. A missing or expired key reads as empty.

MessagePack encodes structured values as named maps. TypeScript emits the shortest integer encoding and rejects a bigint outside the signed and unsigned 64-bit ranges. Small decoded integers are JavaScript numbers, and values encoded as 64-bit integers decode as bigint.

Compare-and-swap

Compare-and-swap rejects an update based on an outdated version:

1. Read the value and version with get_entry(key).
2. Set expect_version(version) (TypeScript: expectVersion) on the next write and finish with .commit().
3. If the version changed, the write fails with a version conflict that carries the current version. Read again before you retry.

expect_absent() (TypeScript: expectAbsent) permits a write only when the key is absent. Use guarded writes for counters and any value with more than one writer.

.commit() applies a guarded write and returns the new version. It needs expect_version or expect_absent, or it fails with an invalid error. .send() writes unconditionally. If you set a precondition and then call .send(), all three SDKs fail with an invalid error that tells you to call .commit(), so a precondition is never dropped.

Check a version conflict with error.is_version_conflict() in Rust, error.version_conflict in Python, or isVersionConflict(error) in TypeScript, which checks for a KvExecutionError whose detail.kind is "versionConflict".

More operations

All three SDKs provide these, with snake_case in Rust and Python and camelCase in TypeScript:

• Call delete(key) to remove a key. It reports whether it removed a live entry. exists(key) returns the version, expiry, and size without the value, or nothing when the key is absent.
• Call expire(key, ttl) to change the expiry without rewriting the value or changing its version. The TTL counts from now. Leave it out (Rust: pass None) to clear the expiry. expire_at(key, epoch_micros) (TypeScript: expireAt) sets an absolute expiry the same way.
• Call patch(key, patch) to apply a codec-specific merge patch. It returns the new version. For JSON values, pass JSON merge-patch bytes.
• Call copy_to(key, to_key) or move_to(key, to_key) to copy or rename in one transaction. Both return the new version of the destination. The destination is overwritten and the value keeps its remaining expiry. An absent or expired source returns a not-found error. For another namespace, chain into_namespace(ns) (TypeScript: intoNamespace) and finish with .send(). Python also accepts to_namespace= and can await the request directly.
• Call get_many(keys) to read several keys in one round trip. It returns one optional value per key, in key order.
• Call delete_many() and select entries with .prefix(p), .range(start, end), or .key_contains(s). .send() returns the number removed. Without a bound, it clears the namespace.
• Call scan() with the same bounds plus .limit(n) and .cursor(c). Finish with .fetch() for one page and its cursor, or .entries() to walk every page. range includes the start and excludes the end. key_contains skips keys that are not UTF-8.
• Call laser.kv_namespaces() (TypeScript: kvNamespaces) to list the namespaces that hold entries for the caller.

const key = new TextEncoder().encode("service:auth")
await kv.expire(key, 3_600_000) // one hour from now
await kv.expireAt(key, BigInt(Date.now()) * 1000n + 86_400_000_000n) // an absolute time
await kv.expire(key) // clear the expiry

const removed = await kv.deleteMany().prefix(new TextEncoder().encode("session:")).send()

use std::time::{Duration, SystemTime, UNIX_EPOCH};

kv.expire("service:auth", Some(Duration::from_secs(3_600))).await?; // one hour from now
let tomorrow = SystemTime::now().duration_since(UNIX_EPOCH)? + Duration::from_secs(86_400);
kv.expire_at("service:auth", Some(tomorrow.as_micros() as u64)).await?; // an absolute time
kv.expire("service:auth", None).await?; // clear the expiry

let removed = kv.delete_many().prefix("session:").send().await?;

import time

await store.expire("service:auth", 3_600_000)  # one hour from now
tomorrow = time.time_ns() // 1_000 + 86_400_000_000
await store.expire_at("service:auth", tomorrow)  # an absolute time
await store.expire("service:auth")  # clear the expiry

removed = await store.delete_many().prefix("session:").send()

Locks and fenced writes

A lease grants temporary ownership of a key. A fencing token identifies the current grant, so a stale holder cannot keep writing. A lease belongs to a holder, a coordination namespace, and a lease key. Acquire it with lease(key, holder, ttl). The call returns the token, the granted TTL, and the commit position of the grant. TypeScript spells the other calls renewLease, getEntryAtLeast, and casFenced.

The store can grant less TTL than you asked for, never more. A live lease always conflicts with a new acquisition, so extend it with renew_lease, not by acquiring again.

Before planning a protected update, call get_entry_at_least(key, lease.position). It waits until the read view includes the lease grant. If the view does not catch up in time, the read fails with a stale error, which you can retry. Then write with cas_fenced(key, fence_namespace, fence_key, token), which takes the coordination namespace and lease key and needs a precondition. The write succeeds only while the lease is held and the token is current.

const kv = laser.kv("config")
const key = new TextEncoder().encode("service:auth")
const lock = new TextEncoder().encode("lease:service:auth")
const holder = "worker-a"
const lease = await kv.lease(lock, holder, 30_000)
try {
  const entry = await kv.getEntryAtLeast(key, lease.position)
  if (entry === undefined) throw new Error("config entry missing")
  await kv
    .casFenced(key, "config", lock, lease.token)
    .json({ log_level: "debug" })
    .expectVersion(entry.version)
    .commit()
  await kv.renewLease(lock, holder, lease.token, 30_000)
} finally {
  await kv.release(lock, holder, lease.token)
}

let kv = laser.kv("config");
let lease = kv
    .lease("lease:service:auth", "worker-a", Duration::from_secs(30))
    .await?;
let entry = kv
    .get_entry_at_least("service:auth", lease.position)
    .await?
    .ok_or_else(|| LaserError::Invalid("config entry missing".to_owned()))?;
kv.cas_fenced("service:auth", "config", "lease:service:auth", lease.token)
    .json(&serde_json::json!({"log_level": "debug"}))?
    .expect_version(entry.version)
    .commit()
    .await?;
kv.renew_lease("lease:service:auth", "worker-a", lease.token, Duration::from_secs(30))
    .await?;
kv.release("lease:service:auth", "worker-a", lease.token).await?;

store = laser.kv("config")
lease = await store.lease("lease:service:auth", "worker-a", 30_000)
try:
    entry = await store.get_entry_at_least("service:auth", lease.position)
    if entry is None:
        raise RuntimeError("config entry missing")
    await (
        store.cas_fenced("service:auth", "config", "lease:service:auth", lease.token)
        .json({"log_level": "debug"})
        .expect_version(entry.version)
        .commit()
    )
    await store.renew_lease("lease:service:auth", "worker-a", lease.token, 30_000)
finally:
    await store.release("lease:service:auth", "worker-a", lease.token)

The example needs an existing entry and the fenced-lease capability.

Renewal keeps the same token. Release revokes it at once and returns true when a held lease was released, or false when none was held. A stale holder's next write fails with lease-lost, even before another holder acquires the lease. Check it with error.is_lease_lost() in Rust, error.lease_lost in Python, or isLeaseLost(error) in TypeScript. Renew before expiry and stop protected effects if renewal fails. Give each independently running worker its own holder ID, and use a stable worker ID rather than a connection value.

If the outcome of an acquisition is unknown, for example because the reply was lost, the SDK waits for the requested TTL and then returns an ambiguous-mutation error. The wait lets any grant the server made expire, so you can acquire again afterwards. lease runs over a dedicated coordination connection, one acquisition at a time per client. That gate stays closed through the wait, even if the caller dropped the call. The dedicated connection needs a Laser that connected from a connection string. A Laser wrapped around your own Iggy client returns a configuration error, so use FencedLeaseClient with your own transport there.

Lease TTL, read consistency, and value TTL are separate settings.

Prepared coordination requests

FencedLeaseClient exposes the two-phase coordination API. Use it when you need to keep the prepared request and read its recovery action. connect_dedicated(connection_string) (TypeScript: connectDedicated) builds a client over its own connection, opened on first use. To use another transport, pass it to the constructor. A custom transport supplies send and reset, and can add ready and close. reset must stop all in-flight work before it returns.

First call prepare_acquire, prepare_renew, prepare_release, or prepare_cas_fenced. Each validates its request once and returns a PreparedMutation bound to that client and operation. Then pass it to acquire, renew, release, or cas_fenced. Rust takes the KvLease, KvLeaseRenew, KvRelease, and KvCasFenced structs. Python takes dictionaries with the same field names, bytes keys, and "v": 1. TypeScript takes objects with camelCase fields, Uint8Array keys, and bigint numbers. Repeating a prepared object sends the same operation ID and the same request bytes. A prepared object from another client, or one passed to the wrong operation, is refused with an invalid error.

import { FencedLeaseClient, isAmbiguousMutation } from "@laserdata/laser-sdk"

const client = FencedLeaseClient.connectDedicated(connectionString).withAttemptTimeout(5_000)
const key = new TextEncoder().encode("lease:service:auth")
try {
  const lease = await client.acquire(
    client.prepareAcquire({
      namespace: "config",
      key,
      holderId: "worker-a",
      leaseTtlMicros: 30_000_000n
    })
  )

  const renewal = client.prepareRenew({
    namespace: "config",
    key,
    holderId: "worker-a",
    leaseToken: lease.token,
    leaseTtlMicros: 30_000_000n
  })
  try {
    await client.renew(renewal)
  } catch (error) {
    if (!isAmbiguousMutation(error)) throw error
    // renewal.ambiguousRecovery is { kind: "repeatPrepared" }
    await client.renew(renewal)
  }
} finally {
  await client.close()
}

use laser_sdk::kv::{FencedLeaseClient, KV_LEASE_OP_VERSION, KvLease, KvLeaseRenew};
use std::time::Duration;

let client = FencedLeaseClient::connect_dedicated(connection_string)
    .with_attempt_timeout(Duration::from_secs(5));
let acquire = client.prepare_acquire(&KvLease {
    v: KV_LEASE_OP_VERSION,
    namespace: "config".to_owned(),
    key: b"lease:service:auth".to_vec(),
    lease_ttl_micros: 30_000_000,
    holder_id: "worker-a".to_owned(),
    subject_user_id: None,
})?;
let lease = client.acquire(&acquire).await?;

let renewal = client.prepare_renew(&KvLeaseRenew {
    v: KV_LEASE_OP_VERSION,
    namespace: "config".to_owned(),
    key: b"lease:service:auth".to_vec(),
    holder_id: "worker-a".to_owned(),
    subject_user_id: None,
    lease_token: lease.token,
    lease_ttl_micros: 30_000_000,
})?;
if let Err(error) = client.renew(&renewal).await {
    if !error.is_ambiguous_mutation() {
        return Err(error);
    }
    // renewal.ambiguous_recovery() is AmbiguousMutationRecovery::RepeatPrepared
    client.renew(&renewal).await?;
}
client.close().await;

import laser_sdk as ls

client = ls.FencedLeaseClient.connect_dedicated(connection_string).with_attempt_timeout(5_000)
try:
    lease = await client.acquire(
        client.prepare_acquire({
            "v": 1,
            "namespace": "config",
            "key": b"lease:service:auth",
            "lease_ttl_micros": 30_000_000,
            "holder_id": "worker-a",
        })
    )

    renewal = client.prepare_renew({
        "v": 1,
        "namespace": "config",
        "key": b"lease:service:auth",
        "holder_id": "worker-a",
        "lease_token": lease.token,
        "lease_ttl_micros": 30_000_000,
    })
    try:
        await client.renew(renewal)
    except ls.LaserError as error:
        if not error.ambiguous_mutation:
            raise
        # renewal.ambiguous_recovery == ls.AmbiguousMutationRecovery.repeat_prepared()
        await client.renew(renewal)
finally:
    await client.close()

An attempt times out after 10 seconds by default. Override it with with_attempt_timeout(Duration) in Rust, with_attempt_timeout(timeout_ms) in Python, or withAttemptTimeout(timeoutMs) in TypeScript, both in milliseconds. A zero timeout is refused before anything is sent. A timeout or an uncertain transport failure retires the connection and returns an ambiguous-mutation error. Read ambiguous_recovery() in Rust, the ambiguous_recovery property in Python, or ambiguousRecovery in TypeScript to learn what to do next. Python returns an AmbiguousMutationRecovery whose kind names the action and which compares equal to the matching factory.

The low-level client reports the action and leaves the acquisition wait to you. The ordinary lease call performs the wait. get(request) is the barriered read on the same dedicated connection. A timed-out get returns a plain timeout, not an ambiguous-mutation error.

Call close() when finished. Python supports async with and TypeScript supports await using. Closing the client or its DedicatedKvTransport is final, and later calls fail before anything is sent. reset() on a transport retires its connection and allows later reuse. Readiness, authentication, unsupported-capability, and request validation failures happen before the request is sent, so they are definite failures.

Local state stores

A state store is the small get, set, and delete interface that agents use for checkpoints, cursor offsets, and duplicate-suppression keys. Keys are strings and values are bytes. InMemoryStore keeps entries in process and loses them on restart. FileStore(root) writes one file per key under a directory, with the key hex-encoded in the file name, and survives restarts. Both work with Iggy alone and have no size limits.

Rust defines the StateStore trait in laser_sdk::state_store behind the agent feature, and Rust's Kv implements it, so laser.kv(namespace) is the managed drop-in. A Kv used as a StateStore writes without expiry and keeps the key and value limits. TypeScript exports a StateStore interface, and new KvStore(kv) is the managed store behind it. Python exposes a StateStore base class that InMemoryStore, FileStore, and KvStore(kv) extend. In every SDK the managed store's set never expires.

import { FileStore, type StateStore } from "@laserdata/laser-sdk"

const store: StateStore = new FileStore("/var/lib/agent-state")
await store.set("cursor:metrics", new TextEncoder().encode("42"))
const saved = await store.get("cursor:metrics")
await store.delete("cursor:metrics")

use laser_sdk::state_store::{FileStore, StateStore};

let store = FileStore::new("/var/lib/agent-state");
store.set("cursor:metrics", b"42".to_vec()).await?;
let saved = store.get("cursor:metrics").await?;
store.delete("cursor:metrics").await?;

store = ls.FileStore("/var/lib/agent-state")
await store.set("cursor:metrics", b"42")
saved = await store.get("cursor:metrics")
await store.delete("cursor:metrics")

Forks

A fork is a copy-on-write branch of the materialized views. It stores its rows apart from the main rows, so you can try a change before you apply it. Forks branch view tables, not the key-value store, so kv.set(..) is never redirected into a fork. Choose the branch behavior when you create it:

• Continuous, the default, keeps seeing later appends to the main data while it keeps its own writes.
• Severed, selected with .severed(), captures the main data at its current offsets and hides later appends. .tables([...]) narrows the snapshot to some tables. An empty list captures every table.

The fork lifecycle:

• Create the branch with fork(id).create(). It returns its metadata: ID, kind, parent, row count, status, owner, and creation time. Rust and TypeScript finish it with .send() and select the default explicitly with .continuous(). .parent(id) records lineage for audit only, and the data still branches from the main data. Python passes severed, continuous, parent, and tables as keywords to create(..), and setting both severed and continuous raises an invalid error.
• Write a row with put_row(table, partition, offset) (TypeScript: putRow, with the offset as a bigint). It writes at an exact source position. Add .field(..), .payload(..), .embedding(..), .metadata(..), or .projection(id, version), or call .tombstone() to hide the main row at that position. .embedding(..) takes numbers, and a value that is not finite is an invalid error. TypeScript raises it from .embedding(..), and Rust and Python raise it from .send().
• Read the branch with query(index).fork(id). Ordinary reads and queries of the main data do not see its rows until promotion.
• Call promote() to apply the fork's writes to the main data. It closes the fork and returns the number of rows applied.
• Call squash() to discard the fork. It returns true when an open fork was removed.
• Call laser.forks() to list the open forks of the authenticated user. fork.id returns the fork ID (Rust: fork.id()), and resource_id (TypeScript: resourceId) returns the scoped ID to pass to other requests.

Before you run this example, register and bind a view whose index is named service_config through Queries and views.

const fork = laser.fork("experiment-1")
await fork.squash()
await fork.create().severed().tables(["service_config"]).send()
await fork.putRow("service_config", 0, 0n).field("log_level", "trace").send()
const applied = await fork.promote()

let fork = laser.fork("experiment-1");
fork.squash().await?;
fork.create().severed().tables(["service_config"]).send().await?;
fork.put_row("service_config", 0, 0)
    .field("log_level", "trace")
    .send()
    .await?;
let applied = fork.promote().await?;

fork = laser.fork("experiment-1")
await fork.squash()
await fork.create(severed=True, tables=["service_config"])
await fork.put_row("service_config", 0, 0).field("log_level", "trace").send()
applied = await fork.promote()

Session links

A key-value write can name the session it belongs to. in_session(reference) (TypeScript: inSession) on a handle stamps every set, compare-and-swap, delete, and patch with the session's stream and ID. Inside a session, session.kv(namespace) returns a handle that is already linked. The link names the stream, never a numeric ID.

The server checks the link before it stores it. The caller needs its usual key-value grant, session:write on the session's stream, and send permission on that stream's agent.sessions. A managed deployment then lists the key on the session, and the session's state view shows the write as a kv.set entry. The write has no position on the session timeline, because it travels through the mutation log rather than the session's stream.

const flags = session.kv("ticket-flags")
await flags.set(new TextEncoder().encode("ticket:42")).json({ escalated: true }).send()

const linked = laser.kv("ticket-flags").inSession(session.reference())

let flags = session.kv("ticket-flags")?;
flags.set("ticket:42").json(&serde_json::json!({ "escalated": true }))?.send().await?;

let linked = laser.kv("ticket-flags").in_session(session.reference()?);

flags = session.kv("ticket-flags")
await flags.set("ticket:42").json({"escalated": True}).send()

linked = laser.kv("ticket-flags").in_session(session.reference())

Session documents and shared key-value resources

Session state, the JSON document a session keeps on its lane, is a different thing from the key-value store. It is built from the session's state_delta and state_snapshot records. See Sessions in depth.

A key-value namespace lives on the managed mutation log and can be shared by independent sessions. It has its own versions and conflict rules. A session link records which work wrote or used a key. It does not copy the value into the session document.

Errors

Rust returns these as LaserError with helpers such as is_version_conflict(), is_lease_lost(), is_stale(), is_not_found(), and is_ambiguous_mutation(). Python raises KvError for store failures, plus InvalidError, UnsupportedError, and AmbiguousMutationError, and each carries flags such as version_conflict and lease_lost. TypeScript throws KvExecutionError, whose detail.kind names the failure, plus InvalidError, UnsupportedError, and AmbiguousMutationError.

Key operations