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), projections, query, key-value state, forks, graph, and watch against LaserData Cloud or Laser Stack. Consumer-group policies and reads are part of streaming.
• Agentic (agent feature), sessions, 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.8.2" # typed streaming plus provenance
# Add only the layers the application uses:
laser-sdk = { version = "0.8.2", 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. A connection string is user:password@host[:port] or token@host[:port], optionally prefixed with iggy:// or iggy+tcp://, and the port defaults to 8090. Credentials are required. Other schemes, unknown or repeated options, and malformed credentials or ports return LaserError::Config before dialing.

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 a failed dial and reconnects a dropped connection, by default without limit at one-second intervals. The 30-second connect budget described under Connect and publish limits stops the retries of the initial connect. Use reconnection_retries=<count|unlimited> and reconnection_interval=<duration> to change the reconnect 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.

A connection string's own tls_ca_file=<path> turns TLS on without tls=true, and tls=false next to it is refused. Set LASER_TLS_CERT=<path> to use an explicit CA with any host. Set LASER_NO_TLS=1 to disable automatic TLS. An explicit tls=false does the same for one connection string. 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(name) selects a group. group.consumer().build().await starts 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();

The managed backend 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("readings_v1")
    .where_eq("host_id", "node-7")
    .filter_gte("cpu", 90_i64)
    .limit(100)
    .fetch()
    .await?;

for row in &result.rows {
    println!("{}", result.value_text(row, "cpu").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("host_id", "node-7")
    .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 Reading { host: String, cpu: u8 }
# async fn run(laser: Laser, reading: Reading) -> Result<(), LaserError> {
let readings = laser.stream("fleet").topic("readings").json::<Reading>(); // or .cbor::<Reading>()
readings.publish(&reading)?.send().await?;              // stamps agdx.ct, builder verbs still chain

let mut records = readings.records("metrics-export")?;  // named cursor, not consumer-group delivery
while let Some(next) = records.next().await {
    let reading: Reading = next?.value;
}
# Ok(()) }

laser.stream("fleet").topic("readings").schema::<Reading>(id).await? uses a registered schema. This form requires schema-codecs and the registry served by the managed backend. 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"rotate the storage credentials".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. Intent builders reject invalid configuration. Call validate before publishing a modified or independently decoded intent. Voting and folding validate the intent again. 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.

Sessions

A session is one conversation with a recorded lifecycle. Its id is the conversation id, so every conversation read finds it. Bootstrap the agent topics with a retention for agent.sessions, then begin a session as an agent:

use laser_sdk::agent::{ModelRequest, ModelResponse, TopicRetention};
use std::time::Duration;

laser.sessions().bootstrap(4, TopicRetention::expire_after(Duration::from_secs(86_400))).await?;
let (session, lease) = laser.sessions().create("ticket-42").agent("triage".parse::<AgentId>()?).begin().await?;
let answer = session
    .run(lease, |session| async move {
        let call = session.model(ModelRequest::new("gpt-4o", prompt), None).await?;
        let body = provider.complete(&call).await?; // your model client
        call.complete(ModelResponse { body: body.clone(), ..Default::default() }).await?;
        session.state().set("status", serde_json::json!("triaged")).await?;
        Ok(body)
    })
    .await?;

run ends the session as completed on success and as failed on an error or a panic, then re-raises the panic. Lifecycle and state records ride the session's partition of agent.sessions. Every record there carries its addressee in agdx.to, with * for every agent. SessionConfig::layout picks a per-agent partition, per-agent topic, or single-partition layout instead of the shared default, and Agents, groups, and layouts explains the choice. laser.sessions().submit(agent, input).from(me).send() hands a new session to an agent, whose handler reaches it through ctx.session(). laser.sessions().control(stream, id).as_operator(op).cancel() writes on agent.control, which only operators can send to. pause() names the participants that must acknowledge (participants(..) sets them), agents hold work that arrives while the session is paused and replay it after resume(), and session.parked() lists held work that was never handled. session.over_budget() reports whether the session's usage passed the budget in its start record. On a deployment that indexes sessions, an agent with an id fails an over-budget session once with reason budget instead of calling its handler, and a workflow stops at the next step boundary with LaserError::BudgetExceeded and ends the run session failed with reason budget. On a deployment that announces sessions, laser.sessions().list(), get(id), events(id), state(id, history_limit), links(id, surface), sources(id), changes(after, limit), and watch(poll_every) read the managed session index, and laser.read_at(source) fetches the one record a timeline row points at. The client behavior guide and the AGDX specification describe the details.

Stream-scoped names

A Laser with a default stream scopes every managed resource name it sends to that stream, stream:<stream>/<name>. That covers KV and memory namespaces, lease and fence namespaces, the key registry, graph names, projection and index ids, query indexes, fork ids, and the change-feed index filter. A name that already starts with stream: is sent as is, listings return your local names, and schema requests carry the stream. laser.resource_name(name), Kv::resource_namespace(), and ForkHandle::resource_id() show the scoped name. LaserBuilder::resource_naming(ResourceNaming::Bare) or laser.with_resource_naming(ResourceNaming::Bare) sends names exactly as written, which is how 0.6 named them. capabilities().stream_tenancy reports a deployment that enforces the scoping.

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.

Consumer filters

Configure the group once, then consume using only its name or ID. The filter runs on the server. Matching records retain their original payloads, headers and offsets. An unbound group receives all records through the same consumer API:

use laser_sdk::filters::{ConsumerFilter, FilterExpr};
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 topic = laser.stream("agent-telemetry").topic("inferences");
    let setup = topic.consumer_group("on-call-review");
    let created = setup.create().filter(slow_or_failed).build().await?;

    let group = topic.consumer_group_id(u64::from(created.id));
    let mut consumer = group.consumer()
        .commit_policy(CommitPolicy::Disabled)
        .batch_length(100)
        .build().await?;
    let record = consumer.next_within(std::time::Duration::from_secs(10)).await?;
    println!("offset {}: {:?}", record.position.offset, record.payload);
    consumer.commit(&record).await?;
    consumer.shutdown().await?;
    Ok(())
}

Use group.create().build().await for a plain group, and group.filter().configure(filter).await to configure it afterward. The advanced group.reader()?.build().await also resolves the group's policy and accepts an unbound group. next_page and ack_page handle its batches, while ack_through stores a checkpoint inside a page. CompiledFilter, with the optional filters feature, runs the evaluator locally. Fault, foreign-codec and type-mismatch policies decide whether an unevaluable record stops the read, is dropped or is delivered with evaluated == false.

A normal consumer's batch_length(100) examines at most 100 source records per partition request and returns zero to 100 matches. The iterator continues through empty selections without sleeping until it reaches visible data's end. The advanced reader has separate count and max_examined limits for match-oriented scans.

Consumer filter groups and offsets

All consumers in the same group share its policy and divide the partitions. Use separate groups for A/B policies and independent offsets. Drafting a revision does not activate it. Configuring another digest on an already-used group returns a conflict. A fresh group using Next starts at the first retained record.

Group consumers acknowledge through the fenced group contract. Automatic policies treat the previous delivery as handled when the application requests another record or shuts down cleanly. Use CommitPolicy::Disabled and commit(&record) for an explicit processing checkpoint. On original Apache Iggy, native consumers retain their existing commit behavior. Advanced group readers acknowledge explicitly and keep at most 1024 unacknowledged record-bearing pages per partition by default, configured with max_unacked_pages. 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.

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, sessions, 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, 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(), then .commit(). .send() is unconditional and refuses a builder that carries a precondition with LaserError::Invalid. copy_to and move_to use one transaction. get_many uses a mixed batch. The managed backend provides storage.
• streaming includes consumer-group policies, group-aware consumers, explicit acknowledgment readers and group filter administration. The supporting server selects matching records for a bound group and returns all records for an unbound group. Configuration needs a ready catalog. filters adds only the local evaluator and the advanced reader's optional local_guard check. A reader joins over its own coordinator connection, and a partition it gains on a rebalance resumes after the group's stored offset.
• 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
• a2a-http / mcp-http, the ready-made axum router() for each bridge
• 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.

Connect and publish limits

Laser::connect gives up after 30 seconds. The budget covers the TCP dial, the TLS handshake, the login, and the capability probe, and an expired budget returns LaserError::Timeout naming the stage that stalled. Each publish attempt times out after 60 seconds and is retried three times, with delays that start at 250 milliseconds, double after each failure, and stop growing at 30 seconds. LaserBuilder::connect_timeout, publish_timeout, publish_max_retries, and publish_retry_backoff change these limits and override LASER_CONNECT_TIMEOUT_MS and the LASER_PUBLISH_* variables. A publish that gives up returns LaserError::PublishFailed with the committed ranges and the unconfirmed records. stream(name).delete() removes a stream you no longer need, and Laser::close ends the shared connection. See connect timeout and cleanup and publish recovery.

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.