LaserData - Laser SDK

LaserData, Inc. maintains Laser SDK for Apache Iggy. This crate defines the Rust reference implementation. Python calls it, and TypeScript implements the same contract in Node. The clients use shared reference data and behavior tests. Streaming records form the base for projections, queries, key-value state, forks, and the optional AGDX agent runtime.

Laser SDK ships in independently adoptable layers:

• Streaming (streaming feature, default), streams, topics, raw and typed publish, batches, resumable cursors, and JSON/CBOR/MessagePack codecs on Apache Iggy.
• Managed platform (managed feature), consumer filters, projections, query, key-value state, forks, graph, watch, and the run registry against LaserData Cloud or Laser Stack.
• Agentic (agent feature), reliable consumer + DLQ, conversation and causality, request/reply, routing, memory, handlers, typed AGDX verbs, workflows, effect governance, and durable intent records.
• Edges, the optional A2A, MCP, and AG-UI adapters.

The SDK carries gen_ai.* provenance describing model calls but never makes them. It moves and coordinates messages only.

The laser-wire crate defines encoded messages, schemas, command codes, limits, and reference test data. It supports WebAssembly and does not require an asynchronous runtime. Laser SDK exposes it as laser_sdk::wire and through the existing module paths.

Filter before the network. Consumer filters select records on the server so each reader receives only its matching subset of a topic and its partitions. The shared CDC example delivers 4 of 240 records and saves 98.5% of payload transfer. It preserves original payloads and offsets, supports exact-width typed headers, and acknowledges only completed work. See the Consumer Filters guide and the three-language examples.

Install

[dependencies]
laser-sdk = "0.5.0" # typed streaming plus provenance
# Add only the layers the application uses:
laser-sdk = { version = "0.5.0", features = ["agent", "managed"] }

Quick example

use laser_sdk::prelude::*;
use serde::{Deserialize, Serialize};

#[derive(Debug, Serialize, Deserialize)]
struct ApiCall {
    endpoint:   String,
    status:     u16,
    latency_ms: u32,
    user_id:    String,
}

# async fn run() -> Result<(), LaserError> {
let laser = Laser::connect("iggy:iggy@127.0.0.1:8090").await?;
let api_calls = laser
    .stream("api-metrics")
    .topic("api_calls")
    .json::<ApiCall>();
api_calls.topic().ensure(4).await?;

let committed = api_calls.publish(&ApiCall {
        endpoint:   "/v1/items".into(),
        status:     200,
        latency_ms: 42,
        user_id:    "alice".into(),
    })?.send().await?;
println!("commit confirmations: {}", committed.confirmations.len());

let mut records = api_calls.records("latency-dashboard")?; // your reader's name, offsets stay caller-owned
while let Some(record) = records.next().await {
    let call: ApiCall = record?.value;
    println!("{} returned {}", call.endpoint, call.status);
}
# Ok(()) }

This example uses only the default streaming and provenance features and runs against Apache Iggy. One connection addresses every stream on the server. Laser::connect_with_stream only pins a default stream so laser.topic(name) can be used as a shortcut. It does not limit the connection to that stream. Laser::connect_env() reads LASER_CONNECTION_STRING and the optional LASER_STREAM, and Laser::local() targets local Apache Iggy.

Direct producers, topic sends, and publish builders return SendMessagesResponse. Each confirmation identifies the stream, topic, partition, and first offset in a batch. A server that does not report offsets returns an empty list. Completion follows the topic durability policy. An offset identifies a position only within its stream, topic, and partition.

Apache Iggy retries initial TCP connections and reconnects dropped connections. The default is unlimited retries at one-second intervals. Use reconnection_retries=<count|unlimited> and reconnection_interval=<duration> to change this behavior. After reconnecting, the client reapplies the credentials from the connection string.

For *.laserdata.cloud and *.laserdata.com, connect and connect_with_stream enable TLS with the bundled LaserData root CA. A CA is a certificate authority used to establish trust. The SDK stores this certificate in a directory that only the current user can access. It reuses the file only when its bytes match the bundled certificate.

Set LASER_TLS_CERT=<path> to use an explicit CA with any host. Set LASER_NO_TLS=1 to disable automatic TLS. Values 0 and false do not disable it. Other hosts retain their connection-string TLS configuration when neither variable is set.

Batch and any payload

publish_batch groups typed records for sending. A service can use topic.producer() for batching, delays, retries, and routing. .background(BackgroundConfig::builder()..) selects buffered sends through Apache Iggy. Before dropping a background producer, call Producer::shutdown() to flush its messages. Otherwise, unflushed messages can be lost. topic.consumer_group() provides continuous reads with offsets stored on the server.

A replay cursor reads a bounded set of records and keeps its offsets in the client. Save those offsets to resume later. Failed or canceled polls do not advance the saved offsets. Each partition read returns at most 10,000 messages, even when the configured request batch is larger.

The message body contains bytes in the format that the application selects. json, msgpack, add_json, and add_msgpack provide encoding helpers. add_payload sends raw bytes without inspecting their format. Use raw_bytes(bytes, ContentType::Avro) for encoded data or add_avro to encode with a schema. Protobuf, compressed data, and application-defined formats can use the same path.

# use laser_sdk::prelude::*;
# use serde::Serialize;
# #[derive(Serialize)] struct ApiCall { status: u16 }
# async fn run(laser: Laser, window: Vec<ApiCall>) -> Result<(), LaserError> {
let api_calls = laser.stream("api-metrics").topic("api_calls");
let mut batch = api_calls.publish_batch();
for call in &window {
    batch = batch.add_json(call)?;        // or .add_payload(raw_bytes) for any format
}
let committed = batch.send().await?;      // the whole window, one round-trip
println!("commit confirmations: {}", committed.confirmations.len());
# Ok(()) }

A projection describes how a service reads and indexes message fields. Declare a Projection through the control API. The declaration selects fields, their positions in the payload, and whether the result retains the body:

use laser_sdk::query::{Projection, ProjectionBinding};
use laser_sdk::stream::ContentType;

let api_call_v1 = Projection::builder("api.call.v1")
    .name("api.call").version(1)
    .content_type(ContentType::Json)
    .fields(["endpoint", "status", "latency_ms", "user_id"])
    .build();

let binding = ProjectionBinding::builder()
    .source("api-metrics", "api_calls")     // (data stream, topic)
    .allow("api.call.v1")
    .default_projection("api.call.v1")
    .build();

laser-plane runs the projector in Laser Stack and LaserData Cloud. Against Apache Iggy without a managed backend, projection and query calls return LaserError::Unsupported. The open streaming path above continues to run unchanged.

Typed operational and lakehouse queries

Query results carry one ordered logical field schema and positional tagged values. Read values through the result so field lookup and value typing remain explicit:

let result = laser
    .query("orders_v1")
    .where_eq("customer_id", "alice")
    .filter_gte("total", 100_i64)
    .limit(100)
    .fetch()
    .await?;

for row in &result.rows {
    println!("{}", result.value_text(row, "total").unwrap_or_default());
}

result.fields gives the ordered schema for row.values. Tagged values preserve integer widths, decimal precision, timestamps, UUIDs, bytes, structs, lists, maps, and nullability. Each page contains at most 1000 rows. Use the server-provided next_cursor to continue. has_more is true exactly when that cursor is present.

Lakehouse queries name one destination generation and can select a retained snapshot:

let historical = laser
    .query_lakehouse(destination_id, destination_generation)
    .at_snapshot(snapshot_id)?
    .filter_eq("customer_id", "alice")
    .limit(100)
    .fetch()
    .await?;

println!("{:?}", historical.context.resolved_target);

The result context identifies the selected engine and target. Lakehouse pages also identify destination and backend generations, the table UUID, snapshot, schema, and partition-spec IDs. They include the materialization boundary, checkpoint revision, and global state revision.

Each query builder keeps one execution identity. Use execution_id() to obtain it. The status() and cancel() commands require the corresponding deployment capabilities. The SDK rejects unsupported calls before sending them. fetch_all() and bounded row iteration follow server-issued cursors.

Destinations and Arrow IPC

Destination and explicit query-route state is available through laser.destinations(). Reads choose checkpoint consistency and remain bounded:

use laser_sdk::wire::checkpoint::{CheckpointReadConsistency, DestinationListFilter};

let page = laser
    .destinations()
    .list(
        DestinationListFilter::default(),
        None,
        50,
        CheckpointReadConsistency::Linearizable,
    )
    .await?;
println!("{} destination(s) at revision {}", page.destinations.len(), page.global_state_revision);

Registration and desired-state changes are revision-guarded. register takes one complete destination declaration. set_desired_state compares both global state and destination definition revisions before applying a change.

For analytical batches, publish one complete self-contained Arrow IPC stream per Apache Iggy message:

use laser_sdk::wire::arrow::{ArrowIpcMessageMetadata, ARROW_IPC_CONTRACT_VERSION};
use laser_sdk::wire::schema::SchemaFingerprint;

let metadata = ArrowIpcMessageMetadata {
    contract_version: ARROW_IPC_CONTRACT_VERSION,
    schema_fingerprint: SchemaFingerprint::new(schema_fingerprint),
    encoded_bytes: arrow_stream.len() as u64,
    field_count: 8,
    record_batch_count: 1,
    row_count: 10_000,
    dictionary_count: 0,
};
topic.publish().arrow_ipc(arrow_stream, metadata)?.send().await?;

Before sending, the SDK checks the metadata and exact payload length. Managed ingestion requires a self-contained Arrow stream with microsecond timestamps and stable dictionaries. Decimal widths cannot exceed 128 bits. Unions and extension types are not supported.

Typed topics

A typed handle binds a topic to one body type. The serde forms encode published values and decode received values. They do not need a schema registry. Decoded records include their log positions.

# use laser_sdk::prelude::*;
# use serde::{Deserialize, Serialize};
# #[derive(Serialize, Deserialize)] struct Order { customer: String, amount: i64 }
# async fn run(laser: Laser, order: Order) -> Result<(), LaserError> {
let orders = laser.stream("commerce").topic("orders").json::<Order>(); // or .cbor::<Order>()
orders.publish(&order)?.send().await?;              // stamps agdx.ct, builder verbs still chain

let mut records = orders.records("billing")?;       // named cursor, not consumer-group delivery
while let Some(next) = records.next().await {
    let order: Order = next?.value;
}
# Ok(()) }

laser.stream("commerce").topic("orders").schema::<Order>(id).await? uses a registered schema. This form requires schema-codecs and the registry served by laser-plane. The handle resolves and compiles the schema once, checks each body, and adds agdx.ct and agdx.sid. A body that does not match fails before publication.

The records reader reports TypedDecodeError { position, source } for a record that does not decode, then continues. An Agent handler can decode through the same API. Its existing dead-letter policy handles invalid records.

Durable approval records

With the agent feature, effects that need asynchronous approval use ordinary typed topics. The SDK validates the intent and ballots, but your application owns the topic layout, replay cursor, and final effect:

use laser_sdk::intent::{decide, Intent, IntentPolicy, Vote, VoteChoice};
use laser_sdk::prelude::*;
use std::time::{Duration, SystemTime, UNIX_EPOCH};

# async fn run(laser: Laser) -> Result<(), Box<dyn std::error::Error>> {
let deadline = (SystemTime::now() + Duration::from_secs(30))
    .duration_since(UNIX_EPOCH)?.as_micros() as u64;
let intent = Intent::builder()
    .conversation(ConversationId::new())
    .proposer("planner".parse()?)
    .body(b"reserve inventory".to_vec())
    .eligible_voters(vec!["safety".parse()?])
    .policy(IntentPolicy::All)
    .policy_version(7)
    .deadline_micros(deadline)
    .build()?;

laser.stream("governance").topic("intents").json::<Intent>().publish(&intent)?.send().await?;
let vote = Vote::cast(&intent, "safety".parse()?, VoteChoice::Allow)?;
let now = SystemTime::now().duration_since(UNIX_EPOCH)?.as_micros() as u64;
if let Some(decision) = decide(&intent, &[vote], now)? {
    if decision.authorizes(&intent)? {
        // Apply the idempotent, fenced effect, then persist the decision.
    }
}
# Ok(()) }

Intent, Vote, and Decision are SDK record conventions, not AGDX envelope types. Invalid configuration fails before publish, and deserialized intents are validated again before voting or folding. A voter name is still a record claim. Use signed principals or topic ACL isolation when authorship must be trusted.

Addressing

Every primitive is an accessor on the connected client. The accessor is free and synchronous, IO happens at the terminal verb.

AgentId identifies an agent for routing and attribution. ConsumerGroupName selects the Apache Iggy consumer group that shares work. PrincipalId identifies the authenticated user for access decisions. Agent::builder derives the default group name from the agent ID. .consumer_group(..) changes the group without changing the agent identity.

One connection can advertise one agent. A second advertisement receives LaserError::PresenceConflict. Router::to_principal(..) and CapabilitySelector::principal(..) require live presence to match the authenticated PrincipalId. A mismatch returns RoutePrincipalMismatch. With a verifier, contract replies require a valid signature from the selected identity. AgentMessage::verified_principal records the responder, including each ScatterReport branch.

A lease gives one holder temporary permission to coordinate an operation. A connection-backed Laser acquires it through a dedicated coordination connection. An application-supplied IggyClient uses an explicit FencedLeaseClient with its own transport. A timed-out attempt retires that connection. If the outcome is unknown, the call waits through the requested lifetime before returning an error.

Requested lifetimes must fall within MIN_LEASE_TTL_MICROS ..= MAX_LEASE_TTL_MICROS, from 1 second to 5 minutes. The SDK rejects other values before sending. The store can grant less time, but never more. A holder can renew before expiry. Exclusive workflows keep the lease through verification and the completion journal write. They release it after that record is durable.

FencedLeaseClient accepts a ManagedKvTransport. For a transport selected at runtime, use SharedKvTransport, which is Arc<dyn DynManagedKvTransport>. Each ManagedKvTransport implements the object-safe interface. The Arc also implements ManagedKvTransport. FencedLeaseClient::new therefore uses the same envelopes, retries, and decoding with either form.

One connection can address every stream that its user can access. connect_with_stream selects an optional default for laser.topic(name). Without a default, this shortcut returns NoStream. Apache Iggy controls access to streams and topics. Use is_permission_denied() and is_stream_or_topic_not_found() to identify access failures.

Managed read models can retain the originating conversation from gen_ai.conversation.id. Use laser.query(index).conversation(id) or laser.graph(name).conversation(id).neighbors(..) to narrow reads. A memory-view namespace also supports laser.kv(ns).scan().conversation(id) and .delete_many().conversation(id). These filters use record metadata and do not create an access boundary. A key-value entry without conversation metadata is excluded from a conversation-filtered scan.

The read ladder

Choose the read API that provides the behavior your application needs.

Use topic.send(..) for raw publication. Use publish() and publish_batch() for typed builders. Use topic.producer() for a persistent producer with batching, retries, topology discovery, and routing. topic.batching() adds governed batches that flush by size or time. contract(..) sends a directed task, and workflow(name) runs steps in dependency order. Apache Iggy builders and the client remain available for detailed configuration.

Features

• default = ["streaming", "provenance"]
• streaming, the open Apache Iggy foundation: Laser, streams, topics, direct producers, live partition and consumer-group streams, server offsets, raw and typed publish, batches, explicit-offset cursors, and JSON/CBOR/MessagePack codecs. The SDK uses Iggy's native VSR transport. Managed reads use the non-replicated extension path, and managed authorization writes use dedicated replicated operation codes.
• provenance, wire contract + provenance encoding/decoding
• agent, reliable consumer, Agent::builder, context, memory, state, contracts, workflows, and the ActionGovernor effect-boundary policy hook
• query, the managed materialized-view query client, including read_your_writes consistency and the unified ResultCode via LaserError::code()
• managed enables destinations, filters, fork, graph, kv, projections, query, rbac, runs, and watch. Each can also be selected separately. Streaming and agents remain available on Apache Iggy. Managed operations require reported deployment capabilities.
• kv provides managed key-value reads, writes, scans, expiry, and compare-and-swap through AGDX_KV. Conditional writes use .expect_version or .expect_absent().commit(). copy_to and move_to use one transaction. get_many uses a mixed batch. laser-plane provides storage.
• filters provides server-side consumer filters over the AGDX_FILTER* commands: laser.filters().reader(stream, topic) reads one partition set or a consumer group through the partition primary, stores progress only through fenced acknowledgments of completed pages, and reports fault and oversized stops as typed errors. A group reader joins over its own coordinator connection, and a partition it gains on a rebalance resumes after the group's stored offset. MatchedRecord::json::<T>() decodes a delivered record into your type. preview, test, and validate run without progress. The catalog verbs register, revise, bind, unbind, archive, and delete wait for the authoritative outcome of each operation id, and apply_as sends a mutation under an id the caller recorded first. Filtered reads need a server that advertises consumer_filters, and the catalog also needs laser-plane.
• capability RBAC over the managed surfaces (rbac feature, sdk/src/rbac/): laser.whoami() + list_roles/get_role/get_bindings/define_role/delete_role/bind_roles/bind_roles_expect_revision/authz_history, plus the pure grants_allow / delegated_allow decision helpers. Grants are effect feature:action [on resource-pattern] assembled through roles bound to the server-stamped user (deny-wins, default-deny), gated on the authz capability. Role names pass the wire-owned validate_role_name (64-byte charset safelist) before any round-trip. The layer is orthogonal to Iggy's own permissions and enforced at the streaming edge.
• a2a-bridge, A2A v1.0 JSON-RPC bridge over the agent topology (SendMessage + streaming, GetTask + CancelTask, the supportedInterfaces Agent Card)
• mcp-bridge, MCP JSON-RPC bridge (initialize, tools, resources, prompts) mapping tool calls onto AGDX
• agui, AG-UI state sync and event rendering over the log
• sign provides Ed25519 signing and verification. Agent::builder().signing_key(..) signs pickup and terminal replies, including respond_input. Agent::builder().verifier(..) rejects unsigned or invalid records before handling. LaserBuilder::verifier(..) applies the same requirement to correlated reply waits.

Signatures bind observed headers and are evaluated at the server-recorded timestamp. Signed quarantine and unquarantine facts require operator keys. A2aBridge::signed_card and sign::verify_card support detached JWS over the canonical card. KvKeyRegistry stores versioned keys in the managed platform.

Observability

The SDK creates tracing spans under the laser target. Publication, polls, and managed calls use debug. Connections, agent startup, workflows, and contracts use info. Fields include conversation, correlation, agent, topic, index, operation, and managed command code. The AGDX specification defines their mapping to record headers. An OpenTelemetry subscriber can connect client spans with traces derived from records.

The SDK supplies spans through tracing. Your application supplies the subscriber and exporter. The following example uses tracing-opentelemetry to show the connection. It is illustrative and does not compile as part of this crate:

use tracing_subscriber::layer::SubscriberExt;

let tracer = opentelemetry_otlp::new_pipeline().tracing().install_simple()?;
tracing::subscriber::set_global_default(
    tracing_subscriber::registry().with(tracing_opentelemetry::layer().with_tracer(tracer)),
)?;

Prelude

use laser_sdk::prelude::* imports the common accessors and types, about 70 items. use laser_sdk::prelude::full::* also imports bridge types, extension traits, projection types, and memory configuration. Prefer the smaller set and explicit imports for application code.

Documentation

The repository README links to docs/tutorial.md. The tutorial covers publication, projections, queries, batches, codecs, stream isolation, and agents. The API reference is on docs.rs. The protocol home is agdxprotocol.ai.

License

Apache-2.0. Copyright LaserData, Inc.

Apache and Apache Iggy are trademarks of the Apache Software Foundation. Use of these marks does not imply endorsement by the Apache Software Foundation.

Connect timeout

Connecting gives up after 30 seconds. Set another budget with the Rust connect_timeout builder method, the Python connect_timeout_ms connect argument, or the TypeScript connectTimeout builder method. The environment variable is LASER_CONNECT_TIMEOUT_MS, and explicit configuration overrides it. The budget covers the TCP dial, the TLS handshake, the login, and the capability probe. An expired budget returns a timeout error that says whether the server never accepted the connection or never answered the login. stream(name).delete() removes a stream you no longer need, and close() ends the shared connection.

See connect timeout and cleanup.

Publish recovery

Publish attempts default to 60 seconds with three retries. Retry delays start at 250 milliseconds, double after each failure, and stop increasing at 30 seconds. Configure these values through the client builder or connect arguments. The corresponding environment variables are LASER_PUBLISH_TIMEOUT_MS, LASER_PUBLISH_MAX_RETRIES, and LASER_PUBLISH_RETRY_BACKOFF_MS. Explicit configuration overrides these variables. Exhausted retries return an error for the application to handle.

See publish recovery and outage handling.

Consumer filters

A consumer filter runs on the streaming server, so a reader receives only the records it selects, with their original offsets:

use laser_sdk::filters::{ConsumerFilter, FilterExpr, FilteredStart};
use laser_sdk::prelude::*;
use laser_sdk::query::CmpOp;

#[tokio::main]
async fn main() -> Result<(), LaserError> {
    let laser = Laser::connect("iggy:iggy@127.0.0.1").await?;
    let slow_or_failed = ConsumerFilter::json(FilterExpr::any([
        FilterExpr::pred("outcome", CmpOp::Eq, "error"),
        FilterExpr::pred("latency_ms", CmpOp::Gt, 2_000),
    ]));
    let mut reader = laser
        .filters()
        .reader("agent-telemetry", "inferences")
        .consumer("on-call-review")
        .inline(slow_or_failed)
        .start(FilteredStart::First)
        .build()
        .await?;
    let record = reader.next_record().await?;
    println!("offset {}: {:?}", record.offset, record.message.payload);
    reader.ack(&record).await?;
    reader.close().await?;
    Ok(())
}

next_page and ack_page handle batches, ack_through stores an application checkpoint inside a page, and CompiledFilter from laser_sdk::filters runs the same evaluator locally. with_fault_policy, with_foreign_policy, and with_mismatch_policy decide what a record that does not decode, has another codec, or has a field of an unexpected type does: stop or drop, or reach the reader marked unevaluated.

Consumer filter groups and offsets

Provision a filtered group once, then consume by its ID. The filter API supports create-and-bind setup, numeric group selection, and revision pause/resume. Use separate groups for A/B revisions so their offsets remain independent. A fresh named consumer using Next starts at the first retained record. Ordinary consumers auto-commit each polled batch before delivery by default, so a later consumer can resume after records the application did not process. Disable auto-commit and commit after successful processing when that matters. Filtered readers use explicit acknowledgments, and keep at most 1024 unacknowledged record-bearing pages per partition by default, set with max_unacked_pages, so a reader that never acknowledges stops at that bound instead of growing without limit. Use a stable name to resume durable progress. See the Consumer Filters guide.

Filter fields inside JSON, CBOR, Avro, and Protobuf payloads on the server. Avro and Protobuf use registered writer schemas, immutable schema IDs in the filter, and the agdx.sid header on each record. Headers-only filters work with any payload format. Filtering preserves original bytes and offsets. See the Consumer Filters guide for codec profiles and examples.

For an application checkpoint inside a filtered page, call reader.ack_through(record) after persisting the checkpoint and processing all preceding records on that partition. Later records in the same page stay pending. Use ack_page when the whole page is complete.