How It Works

~26 min read

🧁 WAL Cake: Ordered ACKs from Parallel S3 Uploads

WAL Cake is a Go service that copies row changes from Postgres to Parquet files in S3, one folder per day, to build a data lake. Postgres sends those changes as one ordered stream. The consumer confirms its progress with a single position, which means “everything before this is safe”. The slow work in between, encoding Parquet and uploading to S3, can run in parallel. This post shows how WAL Cake does that while it still reads and acknowledges the WAL in order, how it is tested, and where it can still replay or stall.

I wrote most of WAL Cake, and we ran it in production at a large Indian fintech company. Everything in this post refers to commit 300e7ab. The benchmarks ran on an Apple M3 Max with Go 1.26.5, and the integration tests against Postgres 18.

Terms #

If you already know Postgres logical replication, skip to the rule.

The figures use one colour code: black is structure, brown is data, green is durable, red is a problem, purple is the part to look at, blue is an option, and orange is the option we chose.

The rule #

WAL Cake builds the raw layer of a data lake. It reads committed row changes, writes them to S3 as Parquet files, and tells Postgres which WAL it can delete. Each file holds rows from one UTC commit day, under a YYYY/MM/DD prefix. A query or a daily job for one day reads one prefix and filters out nothing.

A row is safe in Postgres after commit. A file is safe in S3 after PutObject returns. Between the two, WAL Cake holds rows only in memory. If it crashes, Postgres must still have every row that is not yet in S3. So the consumer follows one rule:

Advance the slot only past rows that are already in S3.

Existing options #

Fig. 1Options for copying Postgres changes to S3
Four options for copying Postgres changes to S3 as Parquet, with benefits, costs, and the one WAL Cake chose

Every option follows the same rule. They differ in which system tracks the slot position, and in how much control you get over the files.

We chose direct pgoutput because we wanted day-split Parquet files from one Go binary, without a broker or a replication instance, and with control over batch size and the ACK. In exchange, we are responsible for reconnects, backpressure, and the ACK. This post is about the ACK.

Why one upload at a time is too slow #

The simplest correct consumer does this in a loop:

  1. Encode a batch of rows as Parquet.
  2. Upload it to S3 and wait.
  3. Send the batch’s last LSN to Postgres.

This follows the rule, because nothing is confirmed before it is in S3. But the worker is idle during step 2, so S3 latency limits the throughput.

An S3 PUT from inside the same region usually takes tens to a few hundred milliseconds. One public benchmark of 500 KB PUTs measured a median of 70 ms and a p99 of 137 ms4. The estimate below uses 100 ms:

B = events per batch             1,000    (default)
E = Parquet encode time          0.0012 s (benchmark below)
P = S3 PUT time                  0.100 s  (assumed)
R = incoming events per second   10,000

one worker handles   B / (E + P)       = 1,000 / 0.1012      ≈ 9,881 events/s
workers needed       ceil(R × (E+P)/B) = ceil(10,000 × 0.1012 / 1,000) = 2

At 100 ms, one worker handles about 9,900 events/s, so at 10,000 events/s the backlog grows by about 120 events every second and never shrinks. At 200 ms, three workers are needed. The worker count is a setting; the default is four5. Use the figure below to try other values.

Fig. 2How many workers?
Workers needed across S3 PUT latency and input rate
Each grey band is one worker count from ceil(R × (E + P) / B). The orange curve is the capacity of the workers you run. Move the cursor to evaluate any point, or change B and E to move the bands.

This is an upper bound: it assumes full batches and no retries. More workers do not make one upload faster. They let other uploads run while one waits.

Which steps can run in parallel #

The WAL must be read in order, and the ACK must be sent in order. Each step of the pipeline, in order:

So WAL Cake keeps one reader and one ACK, and runs only encode and upload in parallel:

one reader  →  cut into segments  →  N workers: encode + upload  →  one ACK
(in order)                           (finish in any order)           (in order)

Amdahl’s law gives the limit. Per 1,000 rows, the serial part takes about 0.5 ms and the parallel part about 101 ms, so the serial part is about 0.5% of the work. In theory, the single reader becomes the limit only at around 200 workers. In practice, other limits come first: memory per worker, and the walsender, the single Postgres process that decodes the slot. I have not measured the walsender’s ceiling. The 0.5 µs is for a short five-column row; wider rows take longer.

The last arrow is the difficult step, because results that finish in any order must become one ordered ACK.

Parallel uploads break the ACK #

Take four batches that start uploading in WAL order S1, S2, S3, S4, and finish in the order S2, S4, S1, S3.

The ACK is a single LSN, and it confirms everything before it7. Postgres cannot accept “S2 and S4 are done, S1 and S3 are not.” If the consumer sends the end of S4, it also confirms S1 and S3. If the process crashes at that moment, Postgres will not send S1 and S3 again, and S3 does not have them. Those rows are lost.

ACK only a contiguous prefix #

Keep a cursor at the oldest batch that is not yet in S3. When a batch finishes, mark it done. If it is the batch at the cursor, move the cursor forward over it and every done batch directly after it. Stop at the first batch that is not done. Send the LSN at the new cursor.

With the example above:

finishes   done so far        cursor stops before   ACK covers
S2         S2                 S1                    nothing new
S4         S2, S4             S1                    nothing new
S1         S1, S2, S4         S3                    S1, S2
S3         S1, S2, S3, S4     (end)                 S3, S4

To do this, the consumer needs three things:

  1. A place to keep each row after a worker takes it, until all earlier batches are in S3. The ACK needs the LSN of the last row in the finished run.
  2. A list of every batch boundary, written before any worker starts. Without it, the consumer cannot tell a slow batch from a missing one.
  3. A cursor that only moves forward and never skips a batch that is not done.

A Go channel cannot do (1), because it does not keep an item after it is received. WAL Cake uses a fixed-size ring of event pointers for (1), a small map of batch ranges for (2), and an atomic index for (3). A ring works here because its slots are freed in the same order that the cursor moves.

The figure below follows one INSERT through the whole system, from commit to the ACK that lets Postgres delete its WAL.

Fig. 3Follow one row
WAL Cake system map: data path from Postgres to S3 and the ACK path back
One INSERT from commit to confirmed_flush_lsn. Hex bytes follow the pgoutput format. LSNs, keys, and times are sample values.

The components are connected in cmd/cake/main.go:

// cmd/cake/main.go (abridged)
eventsCh := make(chan *model.CDCEvent, cfg.BatchSize*cfg.Concurrency)
acked := &ack.Position{} // highest LSN that is durable in S3

repl := replication.NewPGReplicator(cfg)
processor := buffer.NewParquetBatchProcessor(transformer, uploader, bpCfg)
rb := buffer.NewRingBuffer(cfg.BatchSize, cfg.Concurrency,
    cfg.FlushInterval, processor, acked)

replCtx, stopRepl := context.WithCancel(context.Background())
go repl.Start(replCtx, eventsCh, acked)

rb.Start(ctx, eventsCh) // until SIGTERM, then drains
stopRepl()              // the replicator sends the final ACK

This gives at-least-once delivery. Exactly-once delivery is not possible here, because no transaction covers both Postgres and S3. If the process crashes after an upload and before the ACK, the file is in S3 but the slot has not moved, so Postgres sends those rows again. Figure 6 shows what that does to the bucket.

Reading pgoutput #

System map with Postgres, the replicator, and eventsCh highlighted

The replicator opens a ?replication=database connection and streams from the slot’s confirmed_flush_lsn8. A short-lived normal connection runs the publication and slot queries first. Each XLogData frame becomes at most one event:

// internal/replication/pg_replicator.go (abridged)
xld, err := pglogrepl.ParseXLogData(data[1:])
if err != nil {
    return fmt.Errorf("parse XLogData: %w", err)
}
logicalMsg, err := pglogrepl.Parse(xld.WALData)
if err != nil {
    return fmt.Errorf("parse logical replication message at %s: %w", xld.WALStart, err)
}
return r.proccessLogicalMsg(ctx, logicalMsg, xld.WALStart, ch)

Every event has an LSN, and the ring may later send any of them as the ACK. So each one must be safe to confirm once the event and everything before it is in S3:

WALStart + len(WALData) is a correct ACK position in physical replication, where WALData is WAL bytes. In logical replication WALData holds pgoutput bytes, so the sum is not a WAL position. For a commit it points 26 bytes past the end of the record. When the next transaction commits right after, that position is inside its commit record, and Postgres would skip that transaction on restart.

Commit events go into the ring, so a finished run can end on a commit. The Parquet writer skips them.

A few pgoutput details9:

Any error ends the session: a lost connection, a server error, a failed status update, or a message that does not parse. The replicator reconnects with backoff from 1 to 30 seconds and resumes from confirmed_flush_lsn, so nothing is skipped. The readiness check passes only while a session is streaming and event delivery has not been blocked for more than a minute.

The ring buffer #

Two types move through the pipeline:

// internal/model/cdc_event.go
type CDCEvent struct {
    Table      string         `json:"table"`
    Operation  Operation      `json:"op"`
    Before     map[string]any `json:"before,omitempty"`
    After      map[string]any `json:"after,omitempty"`
    Timestamp  time.Time      `json:"timestamp"`   // decode time
    CommitTime time.Time      `json:"commit_time"` // from BEGIN
    LSN        uint64         `json:"lsn"`
}

// internal/buffer/ring_buffer.go
type Segment struct {
    StartIdx int64 // Start index in the ring buffer
    EndIdx   int64 // End index in the ring buffer
    done     bool
}

A CDCEvent is one row change and its LSN. A Segment is a batch: the range [StartIdx, EndIdx) of positions in the ring. The rest of the post uses “segment” for a batch.

The ring has a fixed array, three counters, a tracker, and three channels:

buffer      fixed array of *CDCEvent
writeIdx    next position to write
lastSegIdx  end of the last segment sent to workers
readIdx     end of the finished prefix (the cursor)
tracker     map: StartIdx → segment, guarded by a mutex
segments    channel: segments going to workers
ackSeg      channel: finished segments coming back
space       channel: wakes the writer when readIdx moves

The three counters only increase. Only the array index wraps, using index % size. So the ring is full exactly when writeIdx - readIdx >= size.

At any moment, the positions fall into four ranges:

[0, readIdx)             done; slots can be reused
[readIdx, lastSegIdx)    sent to workers: waiting, running, or done behind a gap
[lastSegIdx, writeIdx)   written, not yet in a segment
[writeIdx, readIdx+size) free

Writing to the ring #

System map with eventsCh and the ring highlighted

One goroutine reads eventsCh and calls Add. It is the only writer, so it does not need a mutex:

// internal/buffer/ring_buffer.go
func (rb *RingBuffer) Add(event *model.CDCEvent) bool {
    w := rb.writeIdx.Load()
    if w-rb.readIdx.Load() >= rb.size {
        return false
    }
    rb.buffer[w%rb.size] = event
    rb.writeIdx.Add(1)
    return true
}

If the ring is full, Add returns false and does not overwrite anything. The goroutine waits on space until the walker moves readIdx. While it waits, eventsCh fills up and the replicator blocks. No data is lost, but replication pauses until uploads finish.

The ring holds 2 × workers × batch events: 8,000 pointers, or 64 KB, with the defaults. Half of that is what the workers hold at once. The other half lets the writer fill the next segments while every worker is busy. This limits the number of events, not their size: one large JSONB value can still use a lot of memory.

Sending segments to workers #

System map with the ring, the segments channel, and the workers highlighted

The same goroutine creates a segment when 1,000 events are waiting. When traffic is low, a 30 s timer creates a smaller one, so a few events do not wait forever and hold WAL in the slot. The segment is added to tracker before it is sent:

// internal/buffer/ring_buffer.go (logging elided)
segment := Segment{StartIdx: lastSegPos, EndIdx: writePos}
rb.lastSegIdx.Store(writePos)
rb.tracker.Set(segment.StartIdx, &segment)
rb.segments <- segment

Because every segment is in tracker before any worker starts, the walker can see a gap: a segment that exists but is not done yet.

A channel send happens before the matching receive in the Go memory model10, so a worker sees the events written before its segment was sent. Only one goroutine, the walker, changes done and readIdx.

Moving the cursor #

When a worker finishes, it sends the segment back on ackSeg. The walker marks it done. If the segment starts at readIdx, the walker moves forward over every done segment:

// internal/buffer/ring_buffer.go
func (rb *RingBuffer) findHighestContiguous(start, read int64) int64 {
    s, ok := rb.tracker.Get(start)
    if !ok { return read }
    s.done = true

    if s.StartIdx == read {
        cur := s
        for cur.done {
            rb.tracker.Del(cur.StartIdx)
            read = cur.EndIdx
            next, ok := rb.tracker.Get(cur.EndIdx)
            if !ok { break }
            cur = next
        }
    }
    return read
}

Then it publishes the new prefix:

// internal/buffer/ring_buffer.go (handleSegmentAck, abridged)
lastEvent := rb.buffer[(highContiguous-1)%rb.size]
rb.readIdx.Store(highContiguous)
select {
case rb.space <- struct{}{}: // wake the writer if it is waiting
default:
}
rb.acked.Advance(lastEvent.LSN)

It reads the last event before it stores readIdx, because the writer may reuse that slot as soon as readIdx moves. acked is an ack.Position: an atomic register that only moves forward. The replicator reads it on each loop iteration, which is at least once a second while it is receiving, and sends a status update when it grows. A register cannot drop an ACK, and it never holds a stale one, because only the highest value matters.

The walk touches each segment three times: added, marked done, and deleted. So the total work is O(N) for N segments.

The figure below runs the S1–S4 example. Watch tracker and readIdx.

Fig. 4The contiguous walk, one step at a time
An eight-slot ring: out-of-order completions and the contiguous readIdx walk
Four workers, two events per batch, eight slots. Use ← and → to step.

At step 10, event e9 is written to buffer[0], the slot e1 used before. Its position is 8, and 8 % 8 = 0.

The ring at production size #

The next figure uses the defaults: 4 workers, 1,000 events per segment, and 8,000 slots. Slow PUT delays one upload, so later segments finish but wait behind it. S3 stall stops all uploads, so the ring fills and the replicator blocks.

Fig. 5The ring at production size
Ring buffer simulator with 8,000 slots and four workers
The figure is a simulation. PUT times are sampled from a lognormal with the chosen p50 and an assumed p99 of 2 × p50. Encode time is 1.2 ms. The receiver and walker logic match the source code.

TestRingCapacity runs the WAL Cake ring with a stand-in uploader that sleeps for 7 ms plus a PUT time from the same lognormal (p50 100 ms). With four workers it sustains about 35,500 events/s, against an ideal of about 36,000. A variant with three workers, a p50 of 200 ms, and a 1.2 ms encode kept up with 14,000 events/s.

What “mutex-free” means #

Add uses no mutex. It has one writer and atomic counters, only the writer fills slots, and only workers and the walker read them. The tracker takes a mutex a few times per segment, so a few times per 1,000 events. Go channels lock internally, and a full ring blocks the replicator. So the system is not lock-free. This does not limit throughput, because Add takes about 11 ns. The purpose of the ring is to keep the ACK in order.

Parquet: small files #

Each Parquet file has one row group and seven columns: table, operation, timestamp, lsn, before, after, and commit_time. before and after hold the old and new row as JSON, and are null when that image does not exist: an insert has no before, and a delete has no after. NUMERIC values are written as JSON numbers with their exact digits, so 9999999999999999.99 stays 9999999999999999.99. All columns use ZSTD compression at level 3.

The test fixture has 1,000 events across 12 tables, a third each inserts, updates, and deletes. Its file is 13,465 bytes, and the two JSON columns are 90% of that.

Benchmarks of the WAL Cake code, three to five runs each:

Step Time Allocated
Ring Add about 11 ns per event 0 B
Parse and decode one insert (five columns) about 0.5 µs per event 760 B
Write 1,000 events to Parquet 1.17–1.20 ms per batch 2.7–2.9 MB

The Parquet library, arrow-go v18.2.0, creates a new ZSTD encoder for every page it compresses, and each encoder allocates about 18 MB. A 1,000-event file has one page per column plus two dictionary pages, so with the library’s own codec the fixture takes about 6.5 ms and 170 MB per batch. WAL Cake registers a ZSTD codec that reuses encoders from a sync.Pool. It writes exactly the same bytes, and a test checks that.

The S3 upload and its key #

A segment can hold rows from two days around midnight. The processor checks every event’s commit date and makes one PutObject call per run of equal dates. The key is:

namespace/YYYY/MM/DD/<decode-time-µs>-<seq>.ZSTD.parquet

After a crash, Postgres sends the same rows again. They decode at a new time, so they get a new key, and S3 now has two files with the same rows. Step through it in the figure, then switch to an LSN-based key.

Fig. 6Crash after PUT, then replay
A crash after a successful PUT replays the segment; compare decode-time keys with LSN-range keys
Both schemes keep at-least-once delivery. The LSN-range key makes the second PUT a no-op.

An LSN-range key with an If-None-Match conditional write11 would make a replayed segment a no-op, but only when the replay cuts exactly the same segments. It often does not. A segment cut by the 30 s timer can have different boundaries after a restart. A segment that ends inside a transaction makes Postgres resend the whole transaction, which shifts every later boundary. And a COPY writes up to 1,000 rows in one WAL record, so they share an LSN, and an LSN range does not name a unique set of rows. A table format such as Iceberg handles deduplication better, so WAL Cake leaves it to the reader.

If processing fails, the worker retries the whole segment, up to three attempts, waiting 2 s and then 4 s. It reports the segment as done only after an attempt succeeds.

Shutdown #

On SIGTERM, the ring stops taking events. It sends the last partial segment, lets the workers finish their uploads, and moves the cursor over everything that finished. Workers use a context that shutdown does not cancel, so an upload is never cut off halfway. Then main stops the replicator, which sends one final status update with the latest acked position before it closes the connection.

After a deploy, Postgres resends at most the transaction that the last segment ended in. A crash, such as an OOM kill or a lost node, replays everything after the last status update.

How it is tested #

The tests are in the repository. The ones that need Postgres run when WALCAKE_TEST_PG is set.

To check that the pipeline test can fail, I pointed commit events 26 bytes past the end of the commit record. The state machine found a lost insert. The current code passes 25 random histories, and 5 more under the race detector.

What can still go wrong #

Fig. 7Where each fault happens
System map with eight numbered markers for replays, stalls, and changes WAL Cake does not capture
Numbers match the list below.

None of these lose a committed row.

Replays #

  1. Crash after a successful upload. The slot has not moved, so Postgres resends the rows and a second file holds them.
  2. Segment ends inside a transaction. The ACK is a change LSN, so after a restart Postgres resends the whole transaction, including rows already in S3.
  3. Upload fails three times. log.Fatal exits the process, and the slot replays everything after the last ACK.

Readers of the lake should expect duplicates, and deduplicate on table, lsn, operation, and the row’s key.

Stalls and reconnects #

  1. The ring is full. The receiver waits for free space, eventsCh fills, and the replicator blocks. After a minute, the readiness check fails.
  2. The replicator is blocked for a long time. It cannot send status updates while it is blocked, so Postgres drops the connection after wal_sender_timeout (60 s by default). When the ring drains, WAL Cake reconnects from the slot.
  3. The walsender dies or the primary fails over. The session ends, and WAL Cake reconnects with backoff. The readiness check fails until it streams again.

The buffers are large enough only for short delays. At 10,000 events/s with the defaults, the ring holds 0.8 s of events and eventsCh another 0.4 s. Part of the ring is always taken by in-flight segments, so a longer S3 outage pauses replication, and WAL builds up on the primary.

Changes not captured #

  1. TRUNCATE is logged and not written to the lake.
  2. An unchanged TOASTed value is stored as the string "<TOAST>", because pgoutput does not send it.

Takeaways #

  1. https://debezium.io/documentation/reference/stable/connectors/postgresql.html ↩︎

  2. https://debezium.io/documentation/reference/stable/operations/debezium-server.html ↩︎

  3. https://docs.aws.amazon.com/dms/latest/userguide/CHAP_Target.S3.html ↩︎

  4. S3 PUT latency benchmark, 500 KB objects, eu-north-1, 100 samples ↩︎

  5. WAL Cake config defaults ↩︎

  6. Postgres logical decoding: replication slots ↩︎

  7. Postgres streaming replication protocol: standby status update ↩︎

  8. https://www.postgresql.org/docs/current/view-pg-replication-slots.html ↩︎

  9. https://www.postgresql.org/docs/current/protocol-logicalrep-message-formats.html ↩︎

  10. https://go.dev/ref/mem ↩︎

  11. https://docs.aws.amazon.com/AmazonS3/latest/userguide/conditional-writes.html ↩︎

Tags:
Share: twitter/x linkedin