durastream

Minimal durable streaming on local disk. Append-only, crash-safe, tailable streams you import into any Python app — no server, no dependencies (stdlib only).

A tiny reimplementation of the durable-stream idea behind PicoMQ, stripped to a single machine: one append-only log file per stream (fsync = durable) plus a SQLite index. Offset and metadata semantics mirror PicoMQ's Durable Streams (DS) HTTP protocol, so a DS-compatible HTTP layer can be bolted on later without touching the core.

Install

uv pip install -e .        # or: pip install -e .

Requires Python 3.12+. No runtime dependencies.

Quick start

from durastream import Store

store = Store("./data")  # creates ./data/{streams,meta.db}
s = store.create("orders", content_type="text/plain")  # idempotent, like DS PUT
# content_type is optional; defaults to "application/octet-stream"

s.append(b"order-1")  # -> 1  (new next_offset)
s.append(b"order-2")  # -> 2
s.append_many([b"o-3", b"o-4"])  # -> 4  batch: one fsync for the whole list

s.read(0)  # [b"order-1", b"order-2", b"o-3", b"o-4"]  all records from offset 0
s.read(1)  # [b"order-2", b"o-3", b"o-4"]              from offset 1 to tail
s.read(0, 1)  # [b"order-1"]                           half-open [start, end)

s.next_offset  # 4                                     == record count
s.content_type  # "text/plain"

Durability is per-flush: append() writes a length+CRC-framed record and fsyncs before returning; append_many() writes the whole list in one fsync (much faster for bulk ingest — same durability guarantee once it returns). After a crash, reopening the store rebuilds state by scanning the log — a torn or corrupt tail record is dropped, the intact prefix survives.

store2 = Store("./data")  # fresh process, same disk
s = store2.open("orders")
s.read(0)  # [b"order-1", b"order-2"]  — recovered from the log

Tailing (tail -f)

subscribe() yields existing records from an offset, then blocks and yields new ones as they're appended (same process):

import threading


def worker():
    for record in s.subscribe(0):  # replays history, then follows the tail
        print("got", record)


threading.Thread(target=worker, daemon=True).start()
s.append(b"live-1")  # worker prints it

The generator returns once the stream is closed and the consumer has caught up.

Demo

demos/bulk_stream.py bulk-streams 100k JSON readings through one stream while a second thread tails them live, then reopens the store from disk to prove the data survived a restart:

make demo        # or: python3 demos/bulk_stream.py

ingesting 100,000 readings in batches of 1,000 ...
  tailed  10,000     0.5 MB     409,514 rec/s
  ...
ingested 100,000 readings (5.1 MB) in 0.22s  ->  447,142 rec/s, 23 MB/s
reopened from disk: next_offset=100,000  (DS token 00000000000000100000)
resumed read at offset 50,000: [b'{"id": 50000, ...}', b'{"id": 50001, ...}']
durable OK — data survived the restart.

demos/append_vs_batch.py (make demo-bench) contrasts append() (one fsync per record) with append_many() (one fsync per batch) — same durability, ~27x faster here (more on platforms with a costlier fsync).

Closing & deleting

s.close()  # no more appends; reads still work
s.append(b"x")  # raises StreamClosed
s.closed  # True (persisted)

store.delete("orders")  # removes the log file + metadata row
store.list()  # ["other-stream", ...]

Offsets

Offset = logical record index (0-based). next_offset is the record count and the position the next append lands at. Helpers convert to/from the DS wire token format:

from durastream import to_token, from_token

to_token(1)  # "00000000000000000001"
from_token("-1", next_offset)  # 0            (start of stream)
from_token("now", next_offset)  # next_offset  (current tail)
from_token("00000000000000000003", next_offset)  # 3

On-disk layout

data/
  meta.db                 SQLite: name, content_type, closed, created_at
  streams/
    orders.log            append-only frames: [u32 len][u32 crc32][payload]...

CRC is zlib.crc32 (CRC-32/ISO-HDLC, the same algorithm PicoMQ uses). One writer per stream is serialized by an in-process lock; SQLite runs in WAL mode.

Mapping to the Durable Streams HTTP protocol

The library covers the DS semantics; wiring it behind HTTP is mechanical:

Scope

Single-node, single-process durability. Deliberately not included (add if you need them): HTTP server, content-type shaping (text concat / JSON arrays), producer fencing, S3/object-store tier, cross-process tailing, TTL/ETag. These are the upgrade paths from PicoMQ's full design.

Develop

make test        # dep-free self-check (python3 tests/test_durastream.py)
make lint        # ruff format + ty typecheck