Streams

Streams are the durable cousin of pub/sub. Every event is a row in _honker_stream, consumers track offsets in _honker_stream_consumers, and replay-on-reconnect is built in. Use streams when you want at-least-once delivery or catch-up after a disconnect. Use pub/sub when you only care about live signals and don’t need history.

The on-disk format is identical across bindings: a Go publisher and a Python consumer can run against the same .db file.

Publish

Python

import honker

db = honker.open("app.db")
s = db.stream("orders")

s.publish({"id": 42, "amount": 9900})

# Atomic with a business write
with db.transaction() as tx:
    tx.execute("INSERT INTO orders VALUES (?, ?)", [42, 9900])
    s.publish({"id": 42, "amount": 9900}, tx=tx)

Node

const { open } = require('@russellthehippo/honker-node');
const db = open('app.db');
const s = db.stream('orders');

s.publish({ id: 42, amount: 9900 });

// Atomic with a business write
const tx = db.transaction();
tx.execute("INSERT INTO orders (id, amount) VALUES (?, ?)", [42, 9900]);
s.publishTx(tx, { id: 42, amount: 9900 });
tx.commit();

Rust

use honker::Database;
use serde_json::json;

let db = Database::open("app.db")?;
let s = db.stream("orders");

s.publish(&json!({"id": 42, "amount": 9900}))?;

Go

import honker "github.com/russellromney/honker-go"

db, _ := honker.Open("app.db", "./libhonker_ext.dylib")
defer db.Close()

s := db.Stream("orders")
s.Publish(map[string]any{"id": 42, "amount": 9900})

Ruby

require "honker"

db = Honker::Database.new("app.db", extension_path: "./libhonker_ext.dylib")
s = db.stream("orders")

s.publish({id: 42, amount: 9900})

Bun

import { open } from "@russellthehippo/honker-bun";

const db = open("app.db", "./libhonker_ext.dylib");
const s = db.stream("orders");

s.publish({ id: 42, amount: 9900 });

Elixir

{:ok, db} = Honker.open("app.db", extension_path: "./libhonker_ext.dylib")

Honker.Stream.publish(db, "orders", %{id: 42, amount: 9900})

C++

#include "honker.hpp"

int main() {
    honker::Database db{"app.db", "./libhonker_ext.dylib"};
    auto s = db.stream("orders");

    s.publish(R"({"id":42,"amount":9900})");

    honker::Transaction tx{db.raw()};
    sqlite3_exec(tx.raw(),
        "INSERT INTO orders (id, amount) VALUES (42, 9900)",
        nullptr, nullptr, nullptr);
    s.publish_tx(tx, R"({"id":42,"amount":9900})");
    tx.commit();
}

SQL (extension)

.load ./libhonker_ext
SELECT honker_bootstrap();

-- Returns the assigned offset (monotonic within a topic).
SELECT honker_stream_publish('orders', NULL, '{"id":42,"amount":9900}');

Subscribe with auto-offset-save

A named consumer resumes where it left off. The iterator amortizes offset saves across events so the single-writer slot doesn’t thrash.

Python

async for event in s.subscribe(consumer="email-worker"):
    await send_order_email(event.payload)
    # Offset auto-saves on a cadence (default: every 1000 events or 1 second).

Node

// Async iterator wakes on any commit to the db, auto-saves offset.
for await (const event of s.subscribe('email-worker')) {
  await sendOrderEmail(event.payload);
}

Rust

loop {
    let batch = s.read_from_consumer("email-worker", 100)?;
    if batch.is_empty() {
        std::thread::sleep(std::time::Duration::from_millis(100));
        continue;
    }
    for event in &batch {
        send_order_email(&event.payload)?;
    }
    s.save_offset("email-worker", batch.last().unwrap().offset)?;
}

Go

for {
    batch, _ := s.ReadFromConsumer("email-worker", 100)
    if len(batch) == 0 {
        time.Sleep(100 * time.Millisecond)
        continue
    }
    for _, event := range batch {
        sendOrderEmail(event.Payload)
    }
    s.SaveOffset("email-worker", batch[len(batch)-1].Offset)
}

Ruby

loop do
  batch = s.read_from_consumer("email-worker", 100)
  if batch.empty?
    sleep 0.1
    next
  end
  batch.each { |event| send_order_email(event.payload) }
  s.save_offset("email-worker", batch.last.offset)
end

Bun

// Async iterator; auto-saves offset every 1000 events.
for await (const event of s.subscribe("email-worker")) {
  await sendOrderEmail(event.payload);
}

Elixir

loop = fn loop ->
  case Honker.Stream.read_from_consumer(db, "orders", "email-worker", 100) do
    {:ok, []} ->
      :timer.sleep(100)
      loop.(loop)
    {:ok, batch} ->
      Enum.each(batch, fn e -> send_order_email(e.payload) end)
      last = List.last(batch)
      Honker.Stream.save_offset(db, "email-worker", "orders", last.offset)
      loop.(loop)
  end
end
loop.(loop)

C++

#include "honker.hpp"

#include <chrono>
#include <thread>

int main() {
    honker::Database db{"app.db", "./libhonker_ext.dylib"};
    auto s = db.stream("orders");
    auto sub = s.subscribe("email-worker", 1000, std::chrono::milliseconds(100));

    for (;;) {
        if (auto event = sub.next()) {
            send_order_email(event->payload());
        }
    }
}

SQL (extension)

-- Read up to 100 events after the consumer's saved offset.
-- Returns a JSON array: [{"offset": ..., "payload": ..., "created_at": ...}, ...]
SELECT honker_stream_read_since(
  'orders',
  honker_stream_get_offset('email-worker', 'orders'),
  100
);

-- Save the consumer's offset after processing.
SELECT honker_stream_save_offset('email-worker', 'orders', 42);

Manual offset control

For exactly-once-within-a-business-transaction semantics, save the offset in the same transaction as the downstream write. If the transaction rolls back, the offset doesn’t advance and the event replays on the next read.

# Python
async for event in s.subscribe(
    consumer="invariant-writer",
    save_every_n=0,  # disable auto-save
    save_every_s=0,
):
    with db.transaction() as tx:
        apply_to_read_model(event.payload, tx=tx)
        # Stream.save_offset has no tx parameter; use the extension
        # function directly so the offset lands in the same transaction.
        tx.query(
            "SELECT honker_stream_save_offset(?, ?, ?)",
            ["invariant-writer", s.name, event.offset],
        )

(Calling s.save_offset("invariant-writer", event.offset) without tx= also works, but it commits outside your transaction and the atomicity is lost.)

Other bindings pass the tx handle into their own save-offset call, or run honker_stream_save_offset inside the open transaction the same way — the extension-level SQL runs inside whatever transaction is open.

Replay semantics

These options are Python-only — Stream.subscribe(consumer=..., from_offset=...). In Node and Bun the signature is subscribe(consumer, opts); calling it with no consumer crashes, and there is no from_offset argument.

• subscribe(consumer=name) starts from the saved offset for this consumer (0 if never saved).
• subscribe(from_offset=N) starts after offset N, regardless of any saved offset.
• subscribe() with no arguments replays from offset 0 — a full replay of the topic. There is no live-tail-from-now mode in Python.

Where it lives

• _honker_stream is the append-only event log. A (topic, offset) index keeps reads O(log n) even with millions of historical events.
• _honker_stream_consumers stores (name, topic) -> offset. One row per consumer per topic.