Metadata-Version: 2.5
Name: durastream
Version: 0.0.1a1
Summary: Tiny durable streaming on local disk
Requires-Python: >=3.12
Description-Content-Type: text/markdown

# 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](https://picomq.com), 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

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

Requires Python 3.12+. No runtime dependencies.

## Quick start

```python
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
`fsync`s 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.

```python
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):

```python
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:

```bash
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

```python
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:

```python
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:

| DS HTTP | durastream |
|---|---|
| `PUT /ds/{name}` (Content-Type) | `store.create(name, content_type)` |
| `POST /ds/{name}` body | `stream.append(body)` → `Stream-Next-Offset: to_token(next_offset)` |
| `GET /ds/{name}?offset=` | `stream.read(from_token(offset, next_offset))` |
| `GET ...&live=long-poll\|sse` | `stream.subscribe(...)` |
| `HEAD /ds/{name}` | `content_type`, `next_offset`, `closed` |
| `POST ... Stream-Closed: true` | `stream.close()` |
| `DELETE /ds/{name}` | `store.delete(name)` |

## 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

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