Key-value state
Keyed values that agents read and update together, with versions, expiry, and compare-and-swap
Key-value state holds small values your agents read and change by key, such as a service config, a feature flag, or the owner of an incident. Every write goes through the log first, so each value points back to the record that wrote it. Use it when several agents or services need the same current value.
Quick example
This program writes a config value with a one-day expiry, reads it with its version, and updates it with compare-and-swap. It needs Laser Stack or LaserData Cloud.
import { Laser } from "@laserdata/laser-sdk"
await using laser = await Laser.connectWithStream("iggy:[email protected]:8090", "ops")
const kv = laser.kv("config")
const key = new TextEncoder().encode("service:auth")
await kv.set(key).json({ log_level: "info" }).ttl(86_400_000).send()
const entry = await kv.getEntry(key)
if (entry === undefined) throw new Error("service:auth is missing")
const version = await kv.set(key).json({ log_level: "debug" }).expectVersion(entry.version).commit()
console.log(`service:auth is at version ${String(version)}`)use laser_sdk::prelude::*;
use std::time::Duration;
#[tokio::main]
async fn main() -> Result<(), LaserError> {
let laser = Laser::connect_with_stream("iggy:[email protected]:8090", "ops").await?;
let kv = laser.kv("config");
kv.set("service:auth")
.json(&serde_json::json!({ "log_level": "info" }))?
.ttl(Duration::from_secs(86_400))
.send()
.await?;
let entry = kv
.get_entry("service:auth")
.await?
.ok_or_else(|| LaserError::Invalid("service:auth is missing".to_owned()))?;
let version = kv
.set("service:auth")
.json(&serde_json::json!({ "log_level": "debug" }))?
.expect_version(entry.version)
.commit()
.await?;
println!("service:auth is at version {version}");
Ok(())
}import asyncio
import laser_sdk as ls
async def main():
async with await ls.Laser.connect("iggy:[email protected]:8090", stream="ops") as laser:
store = laser.kv("config")
await store.set("service:auth").json({"log_level": "info"}).ttl(86_400_000).send()
entry = await store.get_entry("service:auth")
if entry is None:
raise RuntimeError("service:auth is missing")
version = await (
store.set("service:auth")
.json({"log_level": "debug"})
.expect_version(entry.version)
.commit()
)
print(f"service:auth is at version {version}")
asyncio.run(main())If another writer changed the key after your read, commit() fails with a version conflict that carries the current version. Read again and retry.
Read a value
get(key) returns the raw bytes, get_typed(key) decodes JSON, and get_entry(key) adds the version and expiry. A missing or expired key reads as empty. TypeScript's getTyped(key, decode) takes a function that checks the decoded value.
const config = await kv.getTyped(key, (value) => value as { log_level: string })let config = kv.get_typed::<serde_json::Value>("service:auth").await?;config = await store.get_typed("service:auth")Create a key only once
expect_absent() (TypeScript: expectAbsent) writes only when the key does not exist yet. Two agents that race to claim an incident both call it, and the second one gets a version conflict.
const owner = new TextEncoder().encode("incident:4711:owner")
await kv.set(owner).json({ agent: "triage" }).expectAbsent().commit()kv.set("incident:4711:owner")
.json(&serde_json::json!({ "agent": "triage" }))?
.expect_absent()
.commit()
.await?;await store.set("incident:4711:owner").json({"agent": "triage"}).expect_absent().commit()Expire values
.ttl(..) on a write sets how long the value lives, counted from now. Rust takes a Duration, and TypeScript and Python take milliseconds. expire(key, ttl) changes the expiry of an existing value without rewriting it. Leave out the TTL (Rust: pass None) to clear the expiry.
Share state between agents
Agents share state by opening the same namespace on the same stream, in any of the three languages. Use compare-and-swap when several agents update one key. When one worker must own a task for a while, use a lease and fenced writes, covered in Locks and fenced writes. Inside a session, session.kv(namespace) links every write to that session.
To try a change on a copy of your view data before you apply it, use a fork.
Good to know
- Key-value state needs Laser Stack or LaserData Cloud. Check
capabilities.kv.available, andcapabilities.kv.casbefore you callcommit(). Without the capability, the call fails with an unsupported error. - Finish a guarded write with
.commit()and a plain write with.send(). Mixing them fails with an invalid error, so a precondition is never dropped. - Keys are at most 512 bytes, values at most 8 MiB, and namespaces at most 128 bytes. TypeScript keys are
Uint8Arrayvalues. - Namespaces are scoped to the default stream, so
laser.kv("config")addressesstream:<stream>/config. - Each entry from
get_entrycarriessource, the log record that wrote it.