the cost of saying safe

The slot is a server side bookmark for a stream you never see end. I wanted to know when it is actually safe to tell postgres that a byte of WAL can be forgotten. That sounds like a simple question until you have to prove what “safe” actually means when the process can die between receiving data and acknowledging it. So I built the connector around that boundary and started killing it at different points to see what actually happened.

Feldera already ships a postgres CDC input connector. I built the HTTP path anyway, because I wanted to make the delivery boundary explicit and be able to crash the process on purpose. That is the only way I found to see the delivery contract instead of assuming it.

the bookmark is not a file you keep

postgres' write-ahead log is an append-only stream of WAL records. A logical replication slot is a server side checkpoint for that stream. The slot's confirmed_flush_lsn records how far the consumer has confirmed receiving, while restart_lsn is the oldest WAL position the slot may still require. Moving the consumer position is done through a Standby Status Update. The position that matters there is flush_lsn.

I measured this on postgres 16.15 instead of trusting a comment. Holdingwrite_lsn and apply_lsn ahead while leavingflush_lsn behind moved confirmed_flush_lsn by nothing. Sendingflush_lsn moved it.

A lot of early advice, including my own first reading of the replay I was seeing, blamedSTART_REPLICATION … 0/0. That start position does not mean “from the beginning of time.” For a logical slot, postgres starts at the greater of the requested LSN and the slot'sconfirmed_flush_lsn. 0/0 therefore means “let the slot's position determine where to start.” After eighty inserts and an ack that left thirty of them unconfirmed, reconnecting at 0/0 replayed exactly those thirty. The other fifty were no longer available from the slot's confirmed position. A hard coded saved LSN would have introduced a second source of truth and could have left WAL pinned if it disagreed with the server side state. The slot is the checkpoint, and a second source of truth is the classic wrong turn.

Acknowledging also does not immediately mean the disk space disappears.confirmed_flush_lsn records how far the consumer has confirmed;restart_lsn determines how far back postgres may still need WAL. With a client acking every message, restart_lsn stayed frozen for fourteen seconds while pinned WAL climbed from about 1.3 MB to 9.1 MB. The next checkpoint allowed postgres to recycle WAL that was no longer required. checkpoint_timeout defaults to five minutes. Lag and WAL retention are therefore related but different measurements.

A slot that nobody consumes is worse than lag. Two hundred thousand inserts with no feedback grew retained WAL from 568 bytes to 26 MB. That is the production outage version of forgetting a slot. The first version of this connector never sent a Standby Status Update at all. Keepalives were printed. The bookmark never moved.

The timestamp on that status message is microseconds since postgres' 2000-01-01 epoch. That epoch belongs to the replication protocol. It is not how a timestamptz column arrives. I confused those once in an early writeup and had to correct it after looking at the wire.

I also had to pick which LSN a commit promises. commit_lsn is the commit record.end_lsn is the first byte after the transaction. Promisingcommit_lsn redelivers the last transaction on every restart, forever. The code acksend_lsn.

wal_end() on an XLogData message is the server's WAL head, not the record you just read. Using it as received_lsn claims you have seen bytes you have not. The stream path now takes wal_start() for received data. Keepalives are the other case, and they come later.

what the wire actually sends

pgoutput in protocol v1, without the binary option, sends column values using their textual output representation. A TIMESTAMPTZ is therefore a formatted value, not microseconds since 2000. The 2000 epoch is protocol metadata only. That correction changed how I thought about decoding for the rest of the project.

The first decoder had four real bugs, found by running the live connector against postgres 16.15 rather than by staring at types.

1. unknown type OIDs used continue — a column disappeared.
2. UnchangedToast fell into the same arm, so a toasted column could become a silent blank.
3. indexing relation.columns[index] could panic if someone ranALTER TABLE … ADD COLUMN mid stream.
4. an UPDATE that ignored the old image left a ghost row when the primary key changed — Feldera received the new identity and kept the old one.

The first ghost row fix compared old and new key values. Measurement killed it. UnderREPLICA IDENTITY DEFAULT, a key changing UPDATE does not necessarily send a full old tuple. It can send an identity tuple:

UPDATE id  K=['1', NULL, NULL]  new=['99', 'n2', 'small']

old_row is None, so “did the key change?” evaluates to false in the exact case the check was written for. Column flags cannot substitute. flags & 1means that a column belongs to the replica identity, not that it is the SQL primary key. UnderREPLICA IDENTITY FULL, every column belongs to the identity, so comparing the flags would turn every update into a delete plus insert.

The rule that survives is to read what the message actually carried.

I captured the rest of that matrix on the same cluster. UnchangedToast arrives in the new tuple, not the old one. A compressible value stays inline; I needed on the order of 200 KB of incompressible text to see the marker. REPLICA IDENTITY NOTHING on a published table makes postgres refuse the write. An earlier note that claimed a silent drop was wrong.

Text encoding follows the consuming session. boolout prints t andf. timestamptz follows TimeZone. jsonbcanonicalizes spaces and key order. bytea_output chooses hex versus escape. A decoder that parses text is therefore coupled to those GUCs unless the session pins them.SET works on a replication connection. I verified that. Protocol v1 is what I stayed on; postgres 16 supports newer protocol versions as well, but this connector deliberately stayed with v1.

“done” has six meanings and http 200 is the second

Once the bookmark moved after a successful POST, the next lie appeared. Feldera's/ingress returning 200 means the request was accepted by the HTTP input path. It does not by itself mean the pipeline has finished processing the input or that the resulting outputs have been written to all connected sinks.

I treated that as a barrier. FelderaSink::send returns acceptance.wait turns acceptance into a completion condition by polling/completion_status until the token reports complete. Only then doesflush_lsn move. I could observe acceptance and the later completion signal from the connector side. I did not dig into what happens inside Feldera between the two; the boundary I needed was only that the token exists and means the sinks wrote. The completion token documentation is what names those states.

A wait over five seconds logs, because unacked WAL is disk pressure on the primary. The overall timeout defaults to 60 seconds and is classified as unavailable, so the session can reconnect instead of acking a guess.

A completion token tells me that Feldera has processed the corresponding input and written the resulting outputs to its connected data sinks. It does not, by itself, mean that this connector and Feldera together provide exactly once recovery across every possible crash boundary.

Feldera's own fault-tolerance machinery is a separate layer. Current Feldera documentation describes checkpoint/resume, at least once, and exactly once fault tolerance as Enterprise features, and documents HTTP as supporting those modes while postgres is not currently listed as fault-tolerant. Without Feldera checkpointing and journaling, restarting a pipeline does not restore the previous materialized state. In my test setup, that meant a pipeline restart could lose the materialized table state. The connector's responsibility here is narrower: do not tell postgres to forget WAL that Feldera has only accepted into its input path.

That is why FELDERA_ACK_MODE=accepted still exists. It is the old unsafe behavior, kept behind a flag so the loss can be demonstrated. Against a stub that never completed, completed mode left the slot still and the connector exited so the transaction could replay. Accepted mode advanced the position without asking whether the data had been processed. The nastiness of acking on 200 is that a test suite without a crash in that window can still go green.

Exactly once is not a property this connector provides. There is always a crash window between delivering a batch and recording the postgres acknowledgement. Closing that window requires durable coordination beyond this connector's current design. At least once is the deliberate contract: anything not safely acknowledged can be replayed.

One more encoding bug sat in the token path. Completion tokens are base64. A raw +in a query string becomes a space, so I did not want the token to depend on implicit query string encoding. The URLs are built explicitly and the token is percent encoded.

Early on, the connector posted once per change. A five row transaction was five HTTP requests and five tokens. The unit postgres commits, and the unit a restart can replay, is the transaction. Batching one request per transaction was the first reduction. It was not the last.

replay is only harmless if feldera can see a key

At least once is the contract. A crash after Feldera has processed a batch and before postgres records the acknowledgement is the intended replay window, not an accident.

If Feldera counts row occurrences, a replayed insert can become weight 2. One laterDELETE brings it back to 1. The row survives forever. A delete of a key that is not present is ignored on a keyed table and cannot repair the duplicate.

That property is declared in someone else's SQL. I read Feldera's compiled schema at startup and, by default, refuse a table with no PRIMARY KEY before the stream opens.FELDERA_REQUIRE_PRIMARY_KEY=false is the escape hatch that makes the divergence reachable on purpose, the same way accepted mode exists.

I first enforced replica identity at the RELATION message: Feldera's key had to be a subset of postgres' identity, or the connector stopped. That wedged the process permanently. WAL recorded while the table had no usable identity replays forever. Widening the identity now cannot rewrite history. The blocked transaction in the case that taught me this was a plain INSERT, which never needed a pre image at all.

The replica identity only governs the old image of an UPDATE orDELETE. The connector now reports a coverage gap at RELATION and fails at the operation that actually loses correctness: a retraction that does not carry Feldera's key. Feldera would look up a key it was never given, match nothing, and ignore the event without an error. The row stays. No alarm fires.

A table mismatch is still fatal. There is no configuration that makes filing a row under the wrong name correct.

postgres folds unquoted names down. Feldera folds them up. An insert that usedid and a later /query that used ID have to collapse to the same retract key. Field names come from the contract. The unit test for that is{"id": 9050} and {"ID": 9050} becoming{"ID":9050}. That number showed up again later, in a debugging episode that was not a decoder bug.

crashes you can name

The argument so far is entirely about what survives a crash, and for a long time no crash had been taken on purpose. CRASH_AT names a boundary andCRASH_AFTER_BATCHES names which batch. The process callsstd::process::exit(70). A panic would unwind and run destructors. A forced process exit does not get that cleanup.

The boundaries are the steps the design actually orders.

A later change counted batches instead of commits, because once a transaction can split or several transactions can join, a commit is no longer the unit that is sent.

The oracle is boring on purpose. After a mixed workload of inserts, a primary key change, deletes, and a reinsert, the keys in Feldera have to match postgres.

staying up is not the same as making progress

The first supervised loop treated the end of the stream as success and returned Okfrom main. A walsender killed at the far end looked, to anything watching the exit code, like a clean finish.

The loop is now a session. Disconnect reconnects with exponential backoff and full jitter, because every connector attached to one primary can lose it at the same instant. Progress, not merely a socket that came up, resets the backoff. A session that reconnects, replays, fails at the same transaction, and drops again has achieved nothing. Counting that as recovery would disarm the only deadline still working.

Retrying forever looks safe and is not. The slot pins WAL. An infinite retry converts a Feldera outage into a full disk on the primary. After five minutes without progress the process exits nonzero.

Two liveness bugs fell out of the same restructuring.

The periodic status update had to live in a select! beside the Feldera wait. Awaiting the wait directly meant the connector stopped reporting for up to the completion timeout, which defaults to the same 60 seconds postgres uses, by default, aswal_sender_timeout. I then hardcoded a ten second report interval. That is only safe when the server's patience is longer, and this program does not own that setting.

With wal_sender_timeout = 4s, postgres closed the connection four seconds into a Feldera wait. The timer fired at ten seconds, found the socket already gone, reconnected, replayed the same transaction, and did it again.

The interval is now derived from the server: SHOW wal_sender_timeout, divide by three, clamp between 500 ms and 10 seconds. The third is headroom, not a calibration — I wanted margin over a server I do not configure, and the 4-second experiment was the only value I actually validated. The same 4 second experiment sent seventeen status updates across a twenty two second wait and finished in one session. Zero disables the timeout on the server; we still report at the ceiling, because the position has to reach postgres for the slot to release WAL.

The other hole was idle in a different sense. The flush position only moved on a commit of a replicated table. Writes to unpublished tables still move the WAL head. A caught up connector had nothing to apply and postgres retained every segment on its behalf. The connector had nothing to do and looked perfectly healthy doing it, which is exactly why it took an explicit test to notice — progress on the stream and progress on the slot turn out to be different clocks. I wrote 10,000 rows to a table the publication did not include and watched about 5223 kB stay retained and grow. After treating a keepalive as “everything below wal_end has already been read,” with a guard that an open transaction is still in flight, that figure went to zero.

A keepalive is the walsender saying it has nothing further to send. TCP delivers in order. The open transaction guard is what keeps us from acking past a BEGIN whoseCOMMIT we have not seen.

Shutdown has a measured definition. A graceful stop redelivers nothing. An abrupt one redelivers at least the last transaction. The difference is whether the final status update reached the server. The protocol has no acknowledgement of an acknowledgement. Sending one and exiting immediately dropped it: the slot advanced zero times out of eight. The process now sends, then reads confirmed_flush_lsn on a second connection, and waits up to five seconds. A longer wait is a hang. The next signal from an orchestrator is SIGKILL. If the slot did not move, restart redelivers and the primary key absorbs it.

Connections themselves needed a helper. tokio-postgres::connect returns aClient that queues work and a connection future that drives the socket. Nothing useful happens until that future is polled. Spawning it and discarding the join handle made the connection lifecycle easy to get wrong. An exported snapshot also lives only as long as the replication connection that created it. Abandoning that connection is how you get a snapshot name that fails SET TRANSACTION SNAPSHOT a moment later.

the stream is not an archive

A new slot only retains WAL written after it was created. Stream only on a non empty table leaves Feldera empty for every row that already existed. An UPDATE then arrives as a delete of a row the target never held plus an insert. A DELETE arrives as a retraction of a row that was never there. The gap does not close. It just stops being obvious.

CREATE_REPLICATION_SLOT … (SNAPSHOT 'export') is the meeting point. The slot's consistent point is both the snapshot the COPY reads and the LSN the stream resumes from. A row committed at or before that point is visible to COPY and is not replayed as a new change. A row committed after it is invisible to COPY and rides the stream. There is no third window.

export is a deliberate choice, not the absence of an option. On 16.15 the walsender held a non null backend_xmin for as long as that connection lived, which stops vacuum from cleaning up rows that are still visible to that snapshot.SNAPSHOT 'nothing' leaves no such exported snapshot holder. The option is always written out so a slot created for stream only does not accidentally pin an xmin nobody will use.

The obvious order, copy the table and then create the slot, loses every row committed in between. That mode is still in the binary as SNAPSHOT_MODE=inconsistent, so the loss can be measured instead of merely asserted. never is stream only, for a target you already believe is seeded. It records no state file, because recording “we snapped” when we did not would make a later initial run skip a snapshot it should take.

The slot cannot record whether the table was fully read, half read, or never read. A slot sitting at its consistent point looks the same in all three cases. That fact goes in a state file: temp file, fsync, rename, fsync the directory. A file that sayspending is the instruction to read the table again. A file that exists and cannot be parsed is an error, not an absence. Absence means start from scratch. Treating a typo as absence would recopy the table without anyone having decided to.

plan_startup looks at the pair (state file, slot exists) together, because the interesting cases are the disagreements. A slot with no state is not proof the target was seeded. A complete state with a missing slot means the WAL since the last acknowledgement may no longer be recoverable; the table has to be read again from a new slot.

COPY and pgoutput both render a value with the type's output function. Splitting the value decoder from the tuple decoder let both paths share it. If they ever disagree about what a value means, a restart can “change” rows that nobody updated.

memory has a workload shaped hole

A postgres transaction has no fixed size limit. Buffering every change untilCOMMIT has no fixed size limit either. On a two column table the old path cost 2063 bytes per row and held every row twice: once as a decoded event, once as the JSON built at commit. A million row transaction peaked at 1.98 GB. Under a 1 GiB cgroup cap the process was OOM-killed with zero events applied, then replayed the same transaction and was killed again, pinning WAL forever.

The discouraging part was the loop. Restart replayed the same million rows, rebuilt the same buffers, reached the same peak, and died the same way. The connector had turned one large transaction into a poison pill: it couldn't submit the batch, it couldn't acknowledge the transaction, and because it couldn't acknowledge the transaction, the slot kept the WAL that produced it.

Changes now convert on arrival and go into an outbox flushed at BATCH_MAX_ROWS(default 10,000). Peak memory became a function of the bound: about 31.5 MB at ten thousand rows and still about 31.5 MB at a million. Encoding the JSON into one buffer, instead ofmap(to_string).collect::<Vec<_>>().join("\n"), saved a real but smaller amount. On a 200,000-row transaction that extra copying was about 20 MB of a 410 MB peak. The event vectors were the dominant cost.

What this gives up is written at the flush site. Sending a transaction in pieces means Feldera can observe the first batch before the rest of the source transaction has arrived. That is a real weakening of atomicity. I do not have a satisfying way to keep both the hard memory ceiling and whole transaction atomicity. This connector picks the ceiling and lets the primary key reconcile the seam; if a case ever shows that seam from inside Feldera's SQL, I expect to revisit it.

One request per source transaction then showed a different ceiling, and this number is easy to misread. I measured it against the local HTTP double, not against Feldera. That double completes on a 20 ms timer, so two thousand single row commits were two thousand waits: about 49 rows per second. A 50 ms linger coalesced the same workload into one request, and the number that transfers is the shape — removing the round trip per commit is what jumped it to about 15k rows per second, on that double. Feldera's completion time is its own. The 15k figure is not a Feldera benchmark.

The linger is a deadline from the oldest buffered event, not the newest. A steady trickle would keep pushing a newest based deadline out of reach. The timer is sleep_until an absolute instant, so dropping and recreating it on every loop iteration does not reset the wait.BATCH_LINGER_MS=0 restores one request per transaction, which is how the crash point matrix stays pinned to individual commits.

A COMMIT now records how far the position could move and lets flush decide whether it does. A transaction still waiting for delivery is not acknowledged. A keepalive in that window is ignored for the same reason. Joining whole transactions costs no atomicity Feldera could have observed, because one request is one apply step. Splitting a transaction still does.

The send and the ack are one operation again, for a new reason. While a batch was always one transaction, the commit handler knew both that the batch was durable and which LSN that made safe. Once a batch can span several transactions, a COMMIT whose events are still in the buffer must not be acknowledged. The only place that knows the difference is flush. Emptying the outbox is the last thing it does, and only after durability. Everything else leaves the events where they are and ends the session, which is what makes a failed flush a redelivery.

types, identity, and the hole that is not null

The shared decoder still only knew int4 and text for a long time.bool, numeric, timestamptz, jsonb, andbytea were refused. An UPDATE that left a toasted column alone died even when REPLICA IDENTITY FULL had just sent the missing value in the old tuple.

That turned out not to be a type problem at all. It was the Unchanged hole — and seeing it at this level is what finally connected the identity story from the wire section to the values this section decodes. With the hole handled, the remaining work was stopping the decoder from guessing.

On 16.15, boolout prints t/f. The same timestamptz instant printed as 2024-01-01 07:00:00-05,2024-01-01 12:00:00+00, or 2024-01-01 17:30:00+05:30 depending only on TimeZone. The snapshot connection and the walsender do not share a session, so they do not share that GUC. The decoder shifts the instant to UTC with a +00offset. JSON has no NaN or infinity, so those become JSON null, matching the representation I use for Feldera. numeric stays decimal text. bytea is hex only (\x…). The escape format is refused, because silently accepting two encodings is how the snapshot and the stream disagree the moment bytea_output changes. BYTEA in JSON is an array of bytes, matching the Feldera JSON representation used here.

Unchanged is not a value. It is the hole pgoutput leaves when a toasted column was not rewritten. Under REPLICA IDENTITY FULL the old tuple carries it and the new row can be completed. Under a thinner identity there is nothing honest to put in the hole. Feldera's own connector may choose different semantics; this connector refuses rather than silently inventing a value. Repeating x stays inline. An 8000-byte incompressible payload is what actually produces the marker.

REPLICA IDENTITY NOTHING (or DEFAULT with no usable key) is refused when Feldera has a primary key, because a key changing update could insert the new identity and leave the old one in place, and a delete would carry no usable key.

User defined type OIDs are refused rather than guessed.

empty has to mean retract

A snapshot that only inserts is a delta pretending to be a replacement. Two holes follow.

If a snapshot crashes after some rows are durable, those rows sit in Feldera. The source may delete one of them before the next attempt. The new slot has no WAL for that delete. A secondCOPY of what exists now never mentions the deleted key. The leftover stays forever.

TRUNCATE is the same shape without a reload. postgres empties the table. If the connector ignores the message, Feldera keeps every row, and there is no later WAL that repeats the truncate. Upstream postgres-replication did not decode TRUNCATEwhen I needed it. The binary still builds against a local fork at /tmp/pgfork for that message.

Both holes need the live key set. This process does not hold the table. That was the point of the bounded outbox. So both paths ask Feldera /query for the columns a delete needs, send those deletes, and wait until they are durable.

The first snapshot retracts nothing. A resnapshot retracts the leftover.TRUNCATE retracts and does not COPY. Unsent events in the open transaction are discarded, because the source discarded them too and they were never durable.

A publication with pubtruncate = false is a warning. The message never arrives. Looking caught up would be a lie.

The first retract trusted /query alone. That was too optimistic. If the query returned empty, wrapped rows in a JSON array that we treated as one broken line, or returned 400 because the table was not materialized, the connector could have believed an empty retract succeeded.

The process now keeps enough local knowledge of the keys it has made durable to cross check the query result, and accepts both NDJSON and JSON array responses. A missingRELATION for a truncate is not treated as proof that nothing needs to be retracted. The connector refuses or falls back according to the route information it actually has rather than pretending an unknown relation is harmless.

I spent a while convinced DELETE FROM users was the same bug. It is not. That statement is per row WAL. If the connector wrote (9050, SYSTEM) and postgres deletes that row, Feldera empties. If 9050 exists only in Feldera,DELETE FROM users cannot retract it, because there is no corresponding WAL delete. The case that confused me was a leftover connector still holding p9_slot with a stale binary, plus max_replication_slots = 10 filling up with abandoned slots. Kill leftovers. Drop unused slots. Then believe the truncate tests.

one publication is not one ingress

postgres can put users and items in one publication. Feldera's HTTP API cannot. Each ingress URL names one table. A mixed body files every row under whichever table the connector chooses for that request.

TABLES=users,items is the common case. users:USERS orpublic.items:LINEITEMS is the override when the names actually differ.FELDERA_TABLE / PG_TABLE still mean one table. Duplicate sources (users,USERS) and duplicate destinations are rejected.

Each RELATION picks a route index. The outbox carries that index on every event from the moment of conversion. Flush groups by route and POSTs separately. Snapshot and truncate apply per table.

A table in TABLES that the publication does not list would snapshot and then never update. That is refused at startup. A table the publication lists that TABLES does not will arrive as a RELATION and stop the session. The lab publicationmy_pub also publishes typed. A single table run warns. A write totyped stops the connector. Restrict the publication or list every table.

I did not parallelize per table. Total commit order is a property of one stream; per table parallel ingress would let the sink see writes in a different order than the source committed them, and the entire correctness argument in this article assumes the stream is the order. Breaking that for throughput would be a different product.

src/main.rs was about four thousand lines. It is now a thin main. The rest sits in files named after the question they answer. Decode does not know about HTTP. The sink does not know about slots. The connector is still the largest file, because snapshot, stream, flush, and retract are one story.

What is still ugly is local: the rust-postgres fork until TRUNCATE lands upstream, and a lab publication that is wider than TABLES. Neither changes the delivery rule.

what i actually learned

The hard part of a CDC connector was not decoding an INSERT.

It was deciding when an event had crossed enough boundaries that the source could safely forget it, and then making every failure mode preserve that decision.

postgres owns the WAL history. Feldera owns the state after ingestion. The connector's only job is to never claim one side has reached a point the other side has not — that is the boundary I wanted to understand.

And that is the point at which this connector is finally allowed to say:

references