Log

An event or command is still a message. The log records it once, then services can read it live, resume from an offset, or replay it from the beginning.

Built for

Event-driven services, audit trails, live feeds

How it works

The hierarchy is stream, topic, then partition. A stream groups related topics. A topic is a named log with no SDK-imposed schema. Partitions are the ordered logs that store its messages.

One connection can address every stream on the server. The full path is laser.stream("shop").topic("orders"). Set a default stream at connection time to omit .stream(..). See Connect.

Partitions decide ordering and parallelism. A topic splits into partitions - independent, strictly ordered logs. Order holds within a partition only, never across the whole topic. Three ways to route a published message:

• Balanced (default) - Iggy spreads messages across partitions for throughput. No ordering guarantee beyond what lands in the same partition by chance.
• Key - hash a non-empty key to one partition, so every message with the same key keeps its order relative to each other.
• Partition - target one partition number directly.

More partitions raises the ceiling on parallel consumption. Fewer partitions raises the ceiling on ordering. Most services want per-key ordering - one partition per customer or conversation - rather than either extreme.

Payloads are bytes. The raw primitive assumes no encoding. JSON, MessagePack, and Avro are convenience encoders. The examples use JSON because it is concise.

Reading it back is two different jobs:

• A named, resumable cursor. It decodes into your type and tracks per-partition offsets on the reader handle. Persist offsets() and restore them with from_offsets(saved) after a restart. A fresh reader without restored offsets starts at the beginning. The quick example uses this path.
• A live consumer. A single reader, or a consumer group splitting partitions across several processes (the server assigns partitions, one per member, and rebalances when members join or leave). Returns a stream you poll continuously. This is the production read path.

Delivery is at least once. A crash after delivery but before offset commit causes redelivery. Handlers must be idempotent. Each consumer chooses when to store offsets through its commit policy. See Consumers, offsets, and commit policies.

Quick example

const topic = laser.stream("shop").topic("orders")
await topic.ensure(2)

for (const order of ORDERS) {
  await topic.publish().json(order).send()
}

const replay = await topic
  .json(ORDER_CODEC)
  .records("log-example")

for (const result of await replay.poll()) {
  if (result.kind === "record") {
    console.log(result.record.value.total)
  }
}
// One poll reads at most one configured batch per partition, so a real
// reader drains until it reaches the current tail.

let topic = laser.stream("shop").topic("orders");
topic.ensure(2).await?;

for order in [Order { id: 1, total: 99 }, Order { id: 2, total: 42 }] {
    topic.publish().json(&order)?.send().await?;
}

let mut replay = topic.json::<Order>().records("log-example")?;
while let Some(next) = replay.next().await {
    println!("{}", next?.value.total);
}

topic = laser.stream("shop").topic("orders", cls=Order)
await topic.ensure(partitions=2)

for order in (Order(id=1, total=99), Order(id=2, total=42)):
    await topic.publish(order).send()

reader = topic.records("log-example")
while (record := await reader.next()) is not None:
    print(record.value.total)

The runnable examples define Order, ORDERS, and TypeScript's ORDER_CODEC. Production workloads should use a tuned producer and a consumer group with a commit policy.

Full runnable example: Rust · Python · TypeScript

Producing at volume

publish() is the one-shot path: build one message, send it, await the result. Past a demo you pick one of three heavier write paths, in increasing order of throughput:

• publish_batch() accumulates several messages client-side and sends them in one round trip. No background machinery, you control exactly when the batch goes.
• topic.producer() builds a long-lived, tuned producer. It coalesces sends into server batches and retries failures for you. This is the default production write path.
• Background mode (Rust only) switches the producer to Apache Iggy's buffered, sharded send pipeline for maximum throughput.

The producer builder's full surface, with defaults:

A tuned producer, from the native-streaming deep dive:

await using producer = topic.producer({
  retries: 3,
  retryIntervalMs: 1_000
})

await producer.send(utf8("message-0"), {
  key: utf8("account-42"),
  headers: { type: HeaderValue.uint16(7) }
})
await producer.sendBatch(payloads)

let producer = topic
    .producer()
    .batch_length(1_000)
    .linger(Duration::from_millis(5))
    .retries(Some(3), Some(Duration::from_secs(1)))
    .routing(Routing::Balanced)
    .build()
    .await?;

producer
    .send_keyed(
        ProducerMessage::new(b"message-0".as_slice())
            .header(HeaderKey::try_from("type")?, HeaderValue::from(7_u16)),
        b"account-42".to_vec(),
    )
    .await?;
producer.send_batch(batch).await?;

producer = topic.producer(
    batch_length=1000,
    linger_ms=5,
    retries=3,
    retry_interval_ms=1000,
)
await producer.init()

await producer.send(
    b"message-0",
    headers={"type": ("uint16", 7)},
    key=b"account-42",
)
await producer.send_batch(values)

Per-send routing overrides the builder default. send_keyed(msg, key) and send_to_partition(msg, n) in Rust, key= / partition= on Python's send, { key } / { partition } on TypeScript's send. A key and an explicit partition are mutually exclusive on one send.

Client differences:

• Rust background mode uses Apache Iggy's buffered and sharded pipeline. BackgroundConfig controls batching, shards, and backpressure. OrderedSharding preserves order. BalancedSharding favors throughput. Call shutdown() to flush buffered messages.
• Rust explicit batching is available through topic.batching(). It provides flush() and close() without a background task. Its defaults are 512 records, 1 MiB, and 5 ms.
• Python mirrors Rust's direct mode as keyword arguments (batch_length, linger_ms, retries, retry_interval_ms, plus the same create/partitions/expiry/size knobs). Call await producer.init() once before sending. No background mode.
• TypeScript's producer takes routing, retries, and retryIntervalMs only. There is no client-side batchLength/linger on the TS producer today, coalesce round trips with publishBatch()/sendBatch(payloads) instead.

Consumers, offsets, and commit policies

topic.consumer(name, partition) pins one named reader to one partition. topic.consumer_group(group) joins a named group. The server assigns one member to each partition and rebalances when membership changes.

Groups are created and joined automatically by default through create_group(true) and auto_join_group(true).

Offsets live on the server for each consumer name or group. A restarted process reconnects with the same name and resumes from its stored offset.

• start_at(..) sets the initial position when no relevant offset exists. Rust accepts First, Last, Next, Offset(n), or TimestampMicros(t). TypeScript uses { kind: "first" | "last" | "next" | "offset" | "timestamp" }. Python uses polling="first" | "last" | "next" with offset= or timestamp_micros=.
• allow_replay() permits a read at or below an existing stored offset. Set it when re-reading a group from First. Without it, the consumer skips records it has already consumed.
• commit(message) stores a handled record's offset explicitly. store_offset(offset, partition) and delete_offset(partition) are the raw forms, last_consumed_offset(p) / last_stored_offset(p) read the local bookkeeping back.

The commit policy decides when offsets are stored automatically. The full Rust enum, all server-side stores:

Every automatic policy trades a wider redelivery window for fewer round trips. Disabled plus commit-after-handled is crash-safe to the exact message and costs one store_offset round trip per message, measurably slower on a real network, by design. Both patterns, from the same deep dive:

await using consumer = await topic.consumerGroup("workers", {
  batchLength: 100,
  autoCommit: false,
  startFrom: { kind: "first" },
  pollIntervalMs: 5
})

for (;;) {
  const message = await consumer.nextWithin(5_000, { signal })
  if (message === null) break
  handle(message)
  await consumer.commit(message)
}

let mut consumer = topic
    .consumer_group("workers")
    .batch_length(100)
    .poll_interval(Duration::from_millis(5))
    .start_at(ConsumerStart::First)
    .allow_replay()
    .commit_policy(CommitPolicy::Disabled)
    .build()
    .await?;

while let Some(message) = consumer.next().await {
    let message = message?;
    handle(&message)?;
    consumer.commit(&message).await?;
}
consumer.shutdown().await?;

consumer = topic.consumer_group(
    "workers",
    batch_length=100,
    poll_interval_ms=5,
    polling="first",
    auto_commit="disabled",
    allow_replay=True,
)
await consumer.init()

async for message in consumer:
    handle(message)
    await consumer.commit(message)

await consumer.shutdown()

Swap the policy for the batched production default by dropping the manual commit and setting commit_policy(CommitPolicy::IntervalOrEach(Duration::from_secs(1))) in Rust or auto_commit="each", commit_interval_ms=1000 in Python.

Client differences:

• Rust exposes the full nine-variant CommitPolicy plus next_within(timeout) (a typed timeout error instead of hand-rolled tokio::time::timeout plumbing), polling_retry_interval, and init_retries.
• Python maps policies to auto_commit strings. commit_interval_ms adds the interval variant. commit_every sets the count for every. Call await consumer.init() before iterating.
• TypeScript uses autoCommit: true to store offsets on each poll. With false, call consumer.commit(message). It has no interval, each, or every policy. Its defaults are batchLength 100 and pollIntervalMs 250.

Shutdown matters. shutdown() stops polling, leaves the group, and flushes final offset state under the automatic policies. Disabled preserves exactly what you last committed, nothing more.

Apache Iggy access

The wrapper never locks you out of the substrate. topic.iggy_producer(), topic.iggy_consumer(..), and topic.iggy_consumer_group(group) (Rust) hand back Apache Iggy's own builders on the same VSR connection, for any knob the Laser surface doesn't re-export.

Key operations

Not an exhaustive list. See Views for the full field-extraction and projection-registration mechanics .index(..) feeds into.

Running it

Log runs on Laser Stack, LaserData Cloud, and standalone VSR-enabled Iggy. Context, folded Memory, and Fabric use the same streaming path. Views, Changes, State, and Graph require laser-plane. Delivery is at least once, so consumers must handle redelivery.

Related