module std.topic

std.topic — topics: a durable, partitioned, replicated log with broadcast subscribers and consumer groups (§10.5, §11.2, D-4900..D-4919).

@retain(7d)
topic OrderEvents: OrderEvent sharded by .userId

emit OrderEvents(e)                                              // OrderEvents.emit(e): panics when the topic is unreachable
OrderEvents.publish(e)!                                          // returns the error instead
for await e in OrderEvents.subscribe(from: .Latest) { … }        // broadcast: every subscriber sees every element
for await e in OrderEvents.subscribeKey(uid) { … }               // only the elements with this sharding key
for await e in OrderEvents.subscribe(group: "indexer") { … }     // a queue: one member of the group gets each element
atomic(db) { db.insert(order); emit OrderEvents(e) }             // the event goes out after the commit (outbox)

The compiler turns every topic declaration into a static of a generated type with these methods (compiler/src/sema/TopicSema.cpp); sharded by .field gives the partition key of publish/emit. The functions of this module are what the generated methods call. Elements are stored with std.ser (Format.Oxl). A partition is the key prefix m/{topic#p}/ in the node's std.kv store __topics (one shard, so offsets, cursors and leases of a partition change in one Raft batch); hash(key) % partitions picks the partition (one partition without sharded by), offsets are assigned by the shard, per partition, gapless and monotonic. Order is guaranteed per key. Delivery is at-least-once.

A group member holds an element for lease (30 s): the element is acknowledged when the loop asks for the NEXT one, so a body that finishes normally acks, a body that panics, returns or is cancelled does not and the element is redelivered after the lease. Runtime: runtime/src/topic/topic_api.cpp over topic_store.cpp. TopicError and TopicStart (the from: values) are prelude types (std/core/topic.oxl).

Example:

data Job { id: i64 }

topic Jobs: Job

cluster demo {
  role store { journal }
}

fn main() -> () | TopicError io, db {
  Jobs.publish(Job{ id: 1 })?
  for await j in Jobs.subscribe(from: .Earliest) {
    print("job {j.id}")
    break
  }
}

job 1

Functions