PyStream

PyStream is a learning-oriented project that aims to become a lightweight, Kafka-inspired distributed event streaming system implemented from first principles in Python.

It is not Apache Kafka, is not Kafka protocol compatible, and is not intended as a drop-in replacement for a production message broker.

Motivation

The project exists to explore how event streaming systems work by building the core ideas directly rather than consuming them through a client library:

• distributed messaging
• append-only logs
• partitioning
• offsets
• replication
• failure recovery

All of these exist today in some form (see Current capabilities), within the limits described under Limitations — most importantly that brokers are logical nodes inside one process rather than machines on a network.

Current capabilities

Phase 0 — application foundation

• a FastAPI application skeleton
• environment-based configuration via pydantic-settings
• centralized standard-library logging
• a single health endpoint: GET /api/health
• tests, linting, and static type checking

Phase 1 — persistent append-only log

• a durable, append-only log backed by a single file
• monotonically increasing integer offsets starting at 0
• explicit binary record framing (standard library only)
• restart recovery that derives the next offset from the log itself
• corruption detection for truncated, malformed, and non-contiguous records
• thread-safe appends within a single process

Phase 2 — topics and partitions

• topics with a fixed, positive partition count
• one independent append-only log per partition
• partition-local offsets, each partition starting at 0
• deterministic keyed routing (SHA-256), stable across processes and restarts
• round-robin routing for keyless records
• persistent topic metadata with atomic writes and validated recovery
• thread-safe routing and appends within a process

Phase 3 — producers, consumers, and durable offsets

• a producer that sends raw byte payloads and reports where they landed
• a consumer bound to one explicitly assigned topic partition
• a transient fetch position, separate from durable committed progress
• explicit commits — nothing is ever committed automatically
• committed offsets persisted atomically and recovered after restart
• explicit seek for replay, which never rewinds durable progress
• at-least-once delivery semantics

Phase 4 — multi-broker replication

• multiple brokers, each with a completely independent storage root
• one leader and zero or more follower replicas per partition
• deterministic, static replica placement across brokers
• writes accepted only through a partition's leader
• synchronous replication: every replica stores the leader's exact record
• replication status with per-follower log ends and lag
• explicit follower catch-up with prefix validation and divergence detection
• persistent cluster metadata recovered on restart

Phase 5 — failover and replica recovery

• explicit leader failover when a broker becomes unavailable
• deterministic, safety-checked promotion of a follower
• an active replica set, so writes continue with reduced redundancy
• recovery of a returning broker as a follower, never as leader again
• controlled truncation of a recovered replica's unacknowledged tail
• reconciliation to an exact copy of the current leader before reactivation
• failover and recovery state persisted and restored across restart

All five layers are standalone libraries. None imports FastAPI, none is reachable over HTTP, and GET /api/health remains the only endpoint.

Limitations

There is no consensus protocol. A single in-process Cluster object decides promotions under a lock. That is enough to keep this model consistent, and it is not Raft, Paxos, or any quorum protocol — it offers none of their guarantees once more than one process or a real network is involved.

There is no failure detector. A broker is unavailable when its instance is closed. There are no heartbeats, leases, timeouts, or health-check threads, and nothing fails over on its own: failover and recovery are operations you call.

There is no networking. Brokers are logical storage nodes inside one process. No sockets, no RPC, no network partitions.

There is no Kafka ISR. The active replica set changes only through failover and recovery; replicas are never ejected for lagging.

Also absent: consumer groups, group coordination, heartbeats, and rebalancing — a consumer_id is a durable progress identity, not a group id. Neither do retention, compaction, segment rolling, batching, compression, checksums, transactions, nor exactly-once semantics exist. Partition counts and replica placement cannot be changed after creation. PyStream is not Kafka compatible.

Requirements

• Python 3.12+
• uv for dependency management

Local setup

uv sync

Optionally, copy the example configuration and adjust it:

cp .env.example .env

Configuration is read from environment variables prefixed with PYSTREAM_:

Variable	Default	Description
PYSTREAM_APP_NAME	pystream	Application name reported by the API
PYSTREAM_HOST	127.0.0.1	Bind address
PYSTREAM_PORT	8000	Bind port
PYSTREAM_LOG_LEVEL	INFO	DEBUG, INFO, WARNING, ERROR, or CRITICAL

Running the application

uv run uvicorn app.main:app --reload

Or run the module directly, which honours PYSTREAM_HOST and PYSTREAM_PORT:

uv run python -m app.main

Then check the health endpoint:

curl http://127.0.0.1:8000/api/health

{ "status": "ok", "service": "pystream" }

Interactive API documentation is available at http://127.0.0.1:8000/docs.

Storage: the append-only log

The log is framework independent and usable on its own:

from app.storage import AppendOnlyLog

with AppendOnlyLog("data/pystream.log") as log:
    record = log.append(b"payload")   # -> Record(offset=0, timestamp=..., payload=b"payload")
    records = log.read(0, max_records=10)

Payloads are opaque bytes; the log never inspects or interprets them.

Record format

Each record is one self-describing frame of big-endian fixed-width integers followed by the raw payload:

field	type	meaning
record_length	uint32	bytes following this field
offset	uint64	offset assigned by the log
timestamp	int64	append time, nanoseconds since the epoch
payload_length	uint32	length of the payload
payload	bytes	opaque caller-supplied bytes

record_length is deliberately redundant with payload_length: disagreement between the two is what makes malformed framing detectable rather than silently accepted.

Durability and recovery

Every append is flushed and fsync-ed before the offset is returned, so an acknowledged record survives a crash. Correctness is favoured over throughput.

On open, the log scans the file to rebuild an in-memory offset index and derive the next offset. No next-offset metadata is persisted separately — the log file is the only source of truth, and the index is a cache. A file that is not a contiguous run of well-formed records starting at offset 0 raises LogCorruptionError; Phase 1 never repairs or truncates a damaged log.

Topics and partitions

A topic is a named set of partitions with a fixed partition count. Each partition is an independent append-only log with its own offsets:

Topic: events

Partition 0: offsets 0,1,2
Partition 1: offsets 0,1
Partition 2: offsets 0,1,2,3

Offsets are partition-local. partition 0 / offset 5 and partition 1 / offset 5 are different records; there is no global topic offset, and no topic-wide ordered read.

from app.stream import Topic

with Topic(path="data", name="events", partition_count=3) as topic:
    result = topic.append(b"payload", key=b"user-123")
    records = topic.read(result.partition_id, start_offset=0, max_records=100)

append returns an AppendResult carrying the partition_id, the partition-local offset, and the stored record.

Ordering guarantee

PyStream guarantees ordering within an individual partition, not across a topic. Records that a partition accepts keep the order it accepted them in, and their offsets increase by one. Nothing is promised about the relative order of records in different partitions.

Routing a stream of records under the same key is what keeps them ordered together, since a key always resolves to one partition.

Routing

key	behaviour
b"user-123"	sha256(key) as a big-endian integer, modulo the partition count
b""	a real key — routed deterministically, exactly like any other key
None	no key — round robin across partitions

SHA-256 comes from the standard library and depends only on the key bytes, so a key maps to the same partition in every process and after every restart. Python's built-in hash() is deliberately avoided because it is randomized per interpreter. Distinct keys may share a partition; such collisions are expected.

The round-robin cursor lives in memory only. After a restart it begins again at partition 0, which avoids a metadata write per keyless append and costs nothing in correctness.

On-disk layout

data/
└── events/
    ├── topic.meta
    ├── partition-000000/log.bin
    ├── partition-000001/log.bin
    └── partition-000002/log.bin

topic.meta is a small JSON document — version, name, partition_count — written to a temporary file, fsynced, then atomically renamed into place. It is control-plane metadata only; records never go through JSON.

Reopening a topic validates that metadata before touching any partition: bad JSON, missing fields, an unsafe stored name, or a partition count that disagrees with the request all raise TopicMetadataError. Damaged metadata is reported, never silently repaired or overwritten.

Topic names may contain only letters, digits, ., _, and -, up to 249 characters. That rules out path separators, .., absolute paths, and whitespace, so a name can never escape its root directory.

Producing and consuming

Producer

A producer sends raw byte payloads to a topic and reports where each record landed. It delegates routing, offset assignment, and durability to the topic below it — it adds no encoding of its own:

from app.client import Producer

producer = Producer(topic)
result = producer.send(b"payload", key=b"user-123")
# result.topic, result.partition_id, result.offset, result.timestamp

Payloads are bytes and stay bytes: nothing is JSON-encoded, UTF-8-encoded, or pickled on your behalf. Key semantics are exactly the topic's — a key routes deterministically, None uses round robin, b"" is a real key. A producer borrows its topic and never closes it.

Consumer

A consumer is assigned one partition explicitly and keeps a durable progress identity:

from app.client import Consumer, OffsetStore

store = OffsetStore("offsets")
consumer = Consumer(
    consumer_id="analytics-worker",
    topic=topic,
    partition_id=0,
    offset_store=store,
)

records = consumer.fetch(max_records=100)
# ... process the records ...
consumer.commit()

The consumer_id is a durable progress identity only — it is not a consumer group id, and no coordination or rebalancing exists.

Position vs committed offset

These are deliberately different things:

	position	committed_offset
lives	in memory	on disk
means	next offset this consumer will fetch	next offset to fetch after a restart
changes on	fetch, seek	commit only

fetch offsets 0-4   → position becomes 5, committed offset still unset
commit()            → committed offset becomes 5

A new consumer starts at its committed offset, or at 0 when nothing is committed. committed_offset is None when this consumer has never committed, which is distinct from a committed 0.

At-least-once delivery

The intended cycle is fetch → process → commit. Because commits are explicit, a consumer that fetches and processes records but crashes before committing resumes from the old committed offset and sees those records again. That replay is intentional and is what makes delivery at-least-once. Exactly-once semantics are not implemented.

Seek and replay

seek(offset) moves the transient position only — it never changes durable progress, so you can replay already-committed records:

consumer.seek(0)          # replay from the beginning
consumer.fetch()          # committed offset is untouched

Committing from a position below what is already stored raises OffsetRegressionError, so durable progress cannot be accidentally rewound. Seeking past the end of a partition is allowed and simply yields empty fetches.

Offset storage

Committed offsets live outside the topic data, one file per consumer, topic, and partition:

offsets/
└── analytics-worker/
    └── orders/
        ├── partition-000000.offset
        └── partition-000001.offset

Each file holds {"version": 1, "offset": 123}, written to a temporary file, fsynced, then atomically renamed. A commit therefore leaves either the old valid offset or the new one, never a partial value. Malformed offset state — bad JSON, a missing field, a wrong type, a negative offset, an unknown version — raises OffsetCorruptionError and is never silently repaired or overwritten.

Brokers and replication

A cluster is a set of brokers — logical storage nodes inside one process, each owning a directory it alone reads and writes. There are no sockets and no network protocol; what makes brokers real is that isolation.

Each topic partition has one leader replica and zero or more follower replicas, placed deterministically across brokers:

3 brokers, 3 partitions, replication factor 2

partition 0 -> leader broker 1, follower broker 2
partition 1 -> leader broker 2, follower broker 3
partition 2 -> leader broker 3, follower broker 1

from app.cluster import Broker, Cluster, ClusterProducer

cluster = Cluster(path="cluster-data")
for broker_id in (1, 2, 3):
    cluster.register_broker(Broker(broker_id, "cluster-data"))

cluster.create_topic("orders", partition_count=3, replication_factor=2)

producer = ClusterProducer(cluster, "orders")
result = producer.send(b"payload", key=b"user-123")
# result.partition_id, result.offset, result.leader_broker_id

Placement is static: it is computed once, persisted, and never recomputed.

Leader writes and synchronous replication

Writes are accepted only through a partition's leader. The leader assigns the offset and timestamp, appends durably, then every follower stores that exact record — same offset, same timestamp, same payload. A follower never invents an offset or timestamp of its own.

A write succeeds only once all configured replicas hold the record. This is intentionally stricter and simpler than Kafka's ISR and quorum model: there are no acknowledgement modes to configure. The tradeoff is deliberate — every write costs a durable append on every replica, and one unavailable follower fails the write, so throughput and availability are traded for a guarantee that is easy to state and easy to verify.

Replica lag

Each replica has a log end — the offset its next record will occupy. A follower's lag is how far its log end trails the leader's:

leader   next offset = 100
follower next offset = 98   -> lag 2

cluster.status(topic, partition_id) reports this per replica. An unavailable replica reports None rather than zero, because its progress is unknown rather than known to be nothing. This is an observation, not ISR membership — there is no dynamic replica set.

Failure semantics

If the leader append succeeds but a follower cannot store the record:

• the caller receives a ReplicationError naming the failed brokers — the write is not reported as successful
• the leader's record stays durable and is never rolled back, because rolling back an append-only log is worse than being ahead
• the leader is therefore left ahead of that follower, which is visible as lag
• the follower is repaired later by explicit catch-up

cluster.catch_up("orders", partition_id=0, follower_broker_id=2)

Catch-up reads the leader's missing records and replicates them in order. Before copying anything it validates that the follower's existing records are an exact prefix of the leader's — equal offsets alone are not treated as proof of equal data, so every shared record is compared on offset, timestamp, and payload. A mismatch, or a follower whose log end is ahead of the leader, raises ReplicaDivergenceError. Divergence is reported, never repaired: nothing is truncated or overwritten.

If a leader becomes unavailable, its writes fail until you fail over. Nothing is promoted automatically; see Failover and recovery.

On-disk layout

cluster-data/
├── cluster.meta
├── broker-000001/topics/orders/partition-000000/log.bin
├── broker-000002/topics/orders/partition-000000/log.bin
└── broker-000003/topics/orders/partition-000001/log.bin

Every replica is an ordinary append-only log, so framing, fsync durability, offset recovery, and corruption detection are exactly the Phase 1 behaviour. cluster.meta is JSON — brokers, topics, partition counts, replication factors, and each partition's leader, followers, and active replicas — written atomically and fully validated on load. Every leadership change is committed by writing it, so metadata and memory never disagree: if the write fails, the failover is not reported and the old leader stays recorded. Malformed metadata raises ClusterMetadataError and is never silently rewritten.

Failover and recovery

Leader failover

Failure is explicit: a broker is unavailable when its instance is closed. Nothing polls, and nothing promotes itself. When a leader is gone, you ask the cluster to replace it:

result = cluster.failover("orders", partition_id=0)
# result.old_leader_broker_id, result.new_leader_broker_id, result.log_end_offset

Failover refuses to run while the leader is still usable (LeaderStillAvailableError) — it replaces a failed leader rather than moving leadership around on request.

Selection is deterministic: among the active followers whose brokers are reachable, the one with the most surviving history wins, and ties go to the lowest broker id. Nothing is random, and no election protocol is involved.

Promotion safety

Candidates are validated against each other before anything is promoted. Every candidate must agree with the chosen one on every shared offset — offset, timestamp, and payload. If two followers hold different records at the same offset, failover fails with ReplicaDivergenceError instead of guessing which history is real. If no candidate is safe or available, NoEligibleReplicaError is raised and the partition keeps its recorded leader; no empty or leaderless partition is ever created.

Unacknowledged records can be lost, by design. A write is reported successful only after every active replica persists it, so any record that reached only the failed leader was never acknowledged to anyone:

failed leader:  0 1 2 3      <- offset 3 was never acknowledged
promoted:       0 1 2

after failover, the next write gets offset 3 again

The promoted replica's log end becomes the authoritative history. Missing records are never invented from an unreachable broker, and the offsets they occupied are reused.

Active replicas

Each partition records its intended assignment (leader plus followers) and, separately, the active replicas — the subset a successful write must reach. A broker dropped by failover stays assigned but inactive, which is what lets writes continue once a broker is lost:

before:  leader 1, followers [2, 3], active [1, 2, 3]
after:   leader 2, followers [1, 3], active [2, 3]

Redundancy is genuinely reduced until the missing replica returns — two copies instead of three in that example. Replicas are only ever deactivated by broker unavailability handled through failover, never for lagging, so this is not ISR.

Old leader recovery

A returning broker rejoins as a follower. It never reclaims leadership just because it used to be leader, and that survives further restarts. Reconciling it is explicit:

cluster.recover_replica("orders", partition_id=0, broker_id=1)

Recovery finds the longest common prefix with the current leader, drops everything past it from the recovering follower only, replicates the leader's missing records, verifies the two are now identical, and only then adds the replica back to the active set. A replica is never activated before it matches the leader.

That truncation is destructive to the recovered follower, and deliberately so: the records it loses are exactly those that were never acknowledged cluster-wide. The current leader is authoritative after a failover and is never truncated. Structural corruption is not guessed through — a damaged log raises LogCorruptionError rather than being rebuilt from a guess.

Testing

uv run pytest

Linting

uv run ruff check .

Type checking

uv run mypy app

Roadmap

The phases below describe intended direction, not existing functionality.

Phase	Focus	Status
0	Project foundation: config, logging, health endpoint, tooling	Complete
1	Persistent append-only log: offsets, framing, restart recovery	Complete
2	Topics, partitions, and deterministic routing	Complete
3	Producer, consumer, and durable consumer offsets	Complete
4	Multiple brokers and leader/follower replication	Complete
5	Leader failover and replica recovery	Complete

Project layout

app/
├── main.py          # FastAPI application and health endpoint
├── core/
│   ├── config.py    # Environment-driven settings
│   └── logging.py   # Standard-library logging setup
├── storage/
│   ├── record.py    # Record model and binary framing
│   ├── log.py       # Persistent append-only log
│   └── errors.py    # Storage exception hierarchy
├── stream/
│   ├── topic.py     # Topic, metadata, and append routing
│   ├── partition.py # One partition over one append-only log
│   ├── routing.py   # Keyed hashing and round-robin selection
│   └── errors.py    # Topic exception hierarchy
├── client/
│   ├── producer.py  # Sends payloads to a topic
│   ├── consumer.py  # Fetches, tracks position, commits progress
│   ├── offsets.py   # Durable consumer offset store
│   └── errors.py    # Client exception hierarchy
└── cluster/
    ├── cluster.py     # Brokers, placement, writes, failover, recovery
    ├── broker.py      # One storage node and its replicas
    ├── replica.py     # Leader/follower replica over a partition
    ├── replication.py # Status, prefix validation, catch-up
    ├── failover.py    # Candidate selection and promotion result
    ├── recovery.py    # Common-prefix reconciliation of a returning replica
    ├── metadata.py    # Cluster metadata and placement planning
    ├── producer.py    # Producer that writes through leaders
    └── errors.py      # Cluster exception hierarchy
tests/
├── unit/            # Config, logging, storage, topics, client, cluster
└── integration/     # HTTP endpoint and every persistence/recovery path