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 need catch-up after a disconnect. At-least-once processing depends on when the binding saves its offset; see Processing guarantees below. Use pub/sub when you only care about live signals and don’t need history.

The on-disk stream event format is shared 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})

.NET

using Honker;

using var db = Database.Open("app.db");
var stream = db.Stream("orders");
stream.Publish(new { id = 42, amount = 9900 });

Java

import dev.honker.*;

try (Database db = Honker.open("app.db")) {
    Stream stream = db.stream("orders");
    stream.publish("{\"id\":42,\"amount\":9900}");
}

Kotlin

import dev.honker.kotlin.honker

honker("app.db").use { db ->
    db.stream("orders").publish("""{"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}');

Consume and checkpoint

A named consumer resumes where it left off. Most bindings provide a subscription that checkpoints on a cadence; Ruby and Elixir expose the same behavior as an explicit read/process/save loop.

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

let mut subscription = s.subscribe("email-worker")?;
for event in &mut subscription {
    let event = event?;
    send_order_email(&event.payload)?;
}

Go

ctx := context.Background()
for event := range s.Subscribe(ctx, "email-worker") {
    sendOrderEmail(event.Payload)
}

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)

.NET

await foreach (var eventRow in stream.Subscribe(consumer: "email-worker"))
{
    await SendOrderEmail(eventRow.Payload);
    // Offset auto-saves every 1000 events or 1 second by default.
}

Java

import dev.honker.*;

try (StreamHandle subscription = stream.subscribe(event -> {
    sendOrderEmail(event.payloadJson());
}, SubscribeOptions.builder().consumer("email-worker").build())) {
    Thread.currentThread().join();
}

Kotlin

import dev.honker.kotlin.asFlow
import dev.honker.kotlin.subscribeOptions

stream.asFlow(subscribeOptions { consumer("email-worker") })
    .collect { event -> sendOrderEmail(event.payloadJson()) }

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.)

Binding	Transaction-aware offset save
Node, Rust, Go, Ruby, Bun, Elixir, .NET, C++	Typed transaction-aware method
Python, Java, Kotlin	Run honker_stream_save_offset through the open transaction

The extension-level SQL participates in whichever transaction owns that connection.

Replay semantics

Binding	Saved consumer offset	Explicit starting offset	Consumer optional
Python	yes	from_offset	yes
.NET	yes	fromOffset	yes
Java, Kotlin	yes	SubscribeOptions.fromOffset	yes
Node, Rust, Go, Bun, C++	yes	no	no
Ruby, Elixir	manual read/save APIs	pass offset to read API	n/a

For Python, .NET, Java, and Kotlin, an explicit starting offset takes precedence over a saved consumer offset. Omitting both starts after offset 0, which replays the whole topic. None of these APIs provides a special live-tail-from-now mode; query the current maximum offset first if that is the behavior you need.

Node named-consumer interoperability

The current Node wrapper passes its stream topic and consumer name to the extension in reverse order when saving and loading offsets. Node-to-Node resume is internally consistent, but another binding using the same consumer name will not see that checkpoint. Event publishing and reading remain interoperable. Fix the wrapper before relying on a shared named-consumer checkpoint across Node and another binding.

Processing guarantees

Durable storage does not by itself mean that every high-level subscriber waits for your application work before saving its offset.

• Python and .NET save a previous event only when the consumer asks for the next event. Java advances after its callback returns. These shapes replay an event when its handler fails before that point.
• Node, Bun, Rust, Go, and C++ may persist an offset on a cadence or when the subscription closes without knowing whether downstream application work committed. Kotlin Flow uses the Java callback to hand an event into a coroutine channel, so the Java runtime can consider delivery complete before the collector finishes processing it.
• Ruby and Elixir expose manual read/save loops, so the application chooses when to advance.

For a strict at-least-once processing boundary, the Python/.NET iterator shape and Java callback save only after successful processing as described above. In the other bindings, use the manual read/save APIs and advance only after successful work; disabling the periodic cadence alone is not sufficient when a subscription final-flushes its current offset. For exactly-once-within-one SQLite transaction, save the offset alongside the downstream write as shown above.

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.