objwal

README

objwal

A replication write-ahead log built on object-storage queue primitives. The log lives entirely in a bucket: segment objects hold the framed records, and one manifest object, mutated with conditional puts (CAS), is the broker-less ordering authority.

There is no broker, no consensus service, and no shared block device - just an object store that supports conditional writes (S3 If-Match/If-None-Match).

Lineage: from the opendata buffer

objwal descends from a Go port of the Rust opendata-buffer. All credit should go to the team at opendata.

A stateless, broker-less queue over object storage, where producers append batches and a single fenced consumer destructively dequeues them. We kept the hard-won primitives (the CAS manifest, ULID-named data objects, epoch fencing, optimistic-concurrency append) and reshaped the queue into a non-destruct log:

opendata buffer (queue)	objwal (replication log)
Consumer dequeues (AckThrough deletes entries)	Replicas tail read-only; entries are retained. Manifest.TruncateThrough can drop a snapshot-superseded prefix, but nothing in this package calls it yet - there is no automatic retention-window GC, and uploaded-then-uncommitted segments (from a halted flush) are never deleted either. Bring your own driver, or expect the bucket to grow without bound
N stateless producers contend on the manifest	One epoch-fenced primary writes; failover bumps the epoch
Consumer is the manifest owner	Many replicas read the same manifest concurrently (readers aren't fenced)
Batch = opaque entries for a worker to drain	Segment = framed records a replica applies via an Applier
Manifest footer v1	Footer v2 adds a snapshot pointer (v1 still parses)

The wire formats are deliberately byte-compatible: a segment is the buffer's batch-v1 record block, and the v2 manifest footer is a strict superset of v1, so a replica can even bootstrap from a plain upstream buffer manifest.

Topology

        PRIMARY (1, epoch-fenced)                 Object store bucket                 REPLICAS (N, read-only)
  ┌──────────────────────────────┐        ┌────────────────────────────────┐      ┌────────────────────────────┐
  │ local store  ◄── double-write│        │ wal/seg/<run>/0000…0  (segment)│ poll │ Replica.Poll/Run           │
  │                              │ upload │ wal/seg/<run>/0000…1  (segment)│ ───► │  → Applier.Apply(record)   │
  │ Producer.Append() ───────────┼──────► │ …                              │ get  │  → local store (full copy) │ ──► reads
  │           └─ group-commit ───┼─ CAS ► │ wal/manifest  ◄── ordering log │ ───► │  → advance cursor          │
  └──────────────────────────────┘        └────────────────────────────────┘      └────────────────────────────┘

Replicas materialize a full local copy and serve reads locally, lagging the primary by roughly one flush interval.

Timeline of writes and commits

A write travels through Append → buffer → flush (seal + upload) → commit (manifest CAS), and only then becomes visible to replicas. Uploads parallelize within a flush; the manifest commit is strictly serial and is what assigns the total order.

PRIMARY
  Append(r0) ─┐
  Append(r1) ─┼─► [ pending buffer ]
  Append(r2) ─┘        │ flush trigger: FlushInterval elapsed OR FlushBytes reached
                       ▼
                 plan segments (pack groups ≤ SegmentMaxBytes, reserve ordinals)
                       │
        ┌──────────────┴───────────────┐   uploads run concurrently (UploadConcurrency)
        ▼              ▼                ▼
     PUT seg0       PUT seg1        PUT seg2        ← object writes, unordered, idempotent
        └──────────────┬───────────────┘
                       ▼
        commit in ORDINAL order  ── manifest CAS ──►  seq0, seq1, seq2 assigned   ← single linearization point
                       │                              (coalesce ≤ ManifestAppendBatchSize per CAS)
                       ▼
              Durability.Wait() returns seq   (record is now durable + ordered)

REPLICA  (independently, every PollInterval)
  load manifest ─► EntriesFrom(cursor) ─► GET segN ─► decode ─► Apply each record ─► cursor = seqN + 1

Two things set the throughput/latency envelope:

• Per-flush parallelism is real: segment uploads overlap (UploadConcurrency) and multiple entries coalesce into one manifest CAS (ManifestAppendBatchSize).
• Flush boundaries are a serialization point: the flush loop runs one flush at a time, so flush N's manifest commit does not overlap flush N+1's uploads. Overlapping them (a persistent cross-flush committer) is the main remaining write-throughput lever; the replica tailer fetching segments sequentially is the read-side one. Both are tracked in the architecture doc.

Order is defined by the manifest commit, never by upload completion. Replicas only ever read a segment after its manifest entry exists, so out-of-order or early uploads are invisible until committed, the log a replica sees is always linear.

Usage

Producer (the primary)

import (
 "github.com/JayJamieson/objwal/objectstore"
 "github.com/JayJamieson/objwal/wal"
)

store := objectstore.NewInMemory() // or objectstore.NewS3(client, bucket)

p, err := wal.NewProducer(ctx, store, wal.ProducerConfig{
 ManifestPath:  "wal/manifest",
 SegmentPrefix: "wal/seg",
 FlushInterval: 50 * time.Millisecond,
 FlushBytes:    8 << 20, // seal a segment early at 8 MiB buffered
})
if err != nil { ... } // wal.ErrFenced => a newer primary already claimed the log

// Records are opaque framed bytes (the frame format is your own).
// meta is an optional payload attached to this Append group. Append BLOCKS once
// the in-flight byte budget is exhausted, that is the backpressure signal.
d, err := p.Append(ctx, [][]byte{frame0, frame1}, []byte("optional-meta"))
if err != nil { ... }

seq, err := d.Wait(ctx) // blocks until the group is durably committed
if err != nil { ... }   // wal.ErrFenced / wal.ErrHalted on failover or fatal error

p.Close(ctx) // flush remaining records and halt

Multiple concurrent Appends coalesce into segments and commit together. NewProducer claims the log by bumping the manifest epoch; if a newer primary later claims it, this one's next commit detects the epoch change and halts with ErrFenced. Failover is just constructing a fresh producer.

Replica (a reader)

The replica is system-agnostic: it hands each framed record to an Applier. The recommended TypedApplier splits that into a decoder (bytes → your op) and a handler (apply the op to your system):

applier := wal.TypedApplier(
 decodeOp, // func([]byte) (Op, error) mirrors however you framed the record
 func(ctx context.Context, seq uint64, op Op) error {
  return db.Apply(op) // your Bitcask / SQLite / cache / KV store
 },
)

r := wal.NewReplica(store, applier, wal.ReplicaConfig{
 ManifestPath: "wal/manifest",
 Cursor:       wal.NewFileCursorStore("/var/lib/app/wal.cursor"), // resume across restarts
})

// Tail until ctx is cancelled (or call r.Poll(ctx) yourself to step it).
if err := r.Run(ctx); err != nil { ... }

Apply must be idempotent: produce is at-least-once and the tailer re-applies a whole segment if an earlier record in it failed or the cursor wasn't persisted before a crash, so the same record may arrive more than once. Idempotent put/delete satisfies this with no extra state. For a quick in-line applier without the typed split, use wal.ApplyFunc.

Logging and metrics

Both ProducerConfig and ReplicaConfig take a Logger *slog.Logger and a Meter metric.Meter (OpenTelemetry). Neither is required.

p, err := wal.NewProducer(ctx, store, wal.ProducerConfig{
 ManifestPath:  "wal/manifest",
 SegmentPrefix: "wal/seg",
 Logger:        slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelInfo})),
 Meter:         otel.Meter("myapp"),
})

Logging. Left unset, both default to a text logger on stderr filtered to Warn - a healthy producer or replica is silent. Pass your own *slog.Logger (any handler, any level) to change verbosity or destination; Info/Debug cover routine flow (claims, flush/poll completion), Warn/Error cover retries, fencing, and halts.

Metrics. Left unset, both resolve against otel.GetMeterProvider(), which is a no-op until the process calls otel.SetMeterProvider - so metrics cost nothing unless wired up, here or globally. Instruments (see wal/metrics.go for exact names/units):

Producer	Replica
append.records, append.bytes - what came in	replica.applied - records applied
batch.records, batch.bytes - segment sizing	replica.poll.duration - Poll latency
commit.count (by outcome), commit.duration
commit.retries (by reason: load / cas_conflict / write_conflict / unknown)
claim.count (by outcome), claim.retries
fenced, halted (by outcome), upload.failures

All names are prefixed objwal.wal..

Intended use & use cases

objwal is the replication substrate for a single-writer storage engine that wants cheap, durable, broker-less read replicas you bring the engine and the record frame; objwal owns ordering, durability, and fan-out through a bucket.

• Read replicas for an embedded KV / log-structured store (Bitcask, an LSM, a custom append-only engine): the primary double-writes, replicas rebuild a full local copy and serve low-latency local reads.
• Cross-AZ / cross-region durability without inter-AZ traffic: data flows through the object store, not across zones, and survives a primary crash.
• Decoupling writes from slow readers: a lagging or restarting replica just resumes from its cursor; it never back-pressures the primary beyond the bucket's own throughput.
• Disaster recovery / point-in-time rebuild: a replica can bootstrap from a base snapshot (manifest footer v2) and tail forward.

It is not a general message queue, not multi-writer, and not exactly-once on the producer side. One log = one writer; run many independent logs for aggregate throughput.

Status

Probably don't use in production. There are tests, they could be wrong.

Wire formats

All integers little-endian, all footers read from the end of the object. Both formats are versioned and every version this package has ever written is still read back correctly; a re-commit upgrades an object to the current version.

Segment (<prefix>/<runID>/<ordinal:016x>):

record block (optionally zstd):  [len u32][data] ...
footer v2 (11 B, written by default): compression u8 | record_count u32 | crc32c u32 | version u16 (=2)
footer v1 (7 B, still read; written when ProducerConfig.LegacySegmentFormat is
           set, for consumers that read segments as upstream buffer batches):
           compression u8 | record_count u32 | version u16 (=1)

The crc32c in a v2 footer covers the record block plus the compression and count fields, so a corrupt object is rejected (wal.ErrCorrupt) before any zstd work runs. v1 segments carry no checksum.

Manifest (the snapshot block is omitted in v1; the per-entry Count field was added in v3; the crc32c was added in v4):

entry: [body_len u32][sequence u64][count u32 (v3+)][loc_len u16][loc][md_count u32]
       per md: [start_index u32][ingestion_ms i64][payload_len u32][payload]
snapshot block (v2+): [loc bytes][loc_len u16][through_seq u64][created_ms i64]
footer v4 (26 B, written by default): entries_count u32 | next_sequence u64 | epoch u64 | crc32c u32 | version u16 (=4)
footer v1-v3 (22 B, still read): entries_count u32 | next_sequence u64 | epoch u64 | version u16 (<=3)

The v4 crc32c covers the whole object ahead of it including the entry count that drives the decode loop so it's verified before any entry is decoded. The manifest is the ordering authority, so this is the higher-stakes checksum of the two. A silent flip here misdirects the whole log, not one segment.

The in-memory Manifest keeps appends O(1) (side buffer merged at serialize time). TruncateThrough(throughSeq) drops the prefix fully superseded by a snapshot covering up to throughSeq - but see the GC caveat above, nothing calls it automatically yet.

Benchmarks & tests

go test -race ./...                 # unit + integration, race-clean
golangci-lint run ./...             # same lint set CI runs (see .golangci.yml)
./scripts/test.sh                   # end-to-end WAL throughput over local MinIO
BENCH_LATENCY=true ./scripts/test.sh   # per-record commit latency (p50/p90/p99)

cmd/s3bench drives the real producer/replica path (not raw objectstore PUT/GET) and reports write/read throughput in Mbps or commit-latency percentiles; scripts/test.sh documents the tuning knobs and recipes for trading throughput against latency.