Allow Destination to control confirmed_flush_lsn

[no-flaks-given]

Problem

Currently, the ApplyLoop auto-advances flush_lsn after write_events() returns successfully. This works well for destinations that durably persist data synchronously in write_events() — by the time the method returns, the data is safe, so it's correct to tell Postgres it can discard that WAL.

But some destinations process data asynchronously. For example, Feldera ingests events via HTTP and processes them through a streaming pipeline before durably persisting results. When write_events() returns, the data has been accepted but not yet durably processed. If the destination crashes before processing completes and etl has already advanced flush_lsn, Postgres has discarded the WAL and those events are lost.

This makes it impossible to build at-least-once delivery guarantees for asynchronous destinations using etl today.

Related: feldera/feldera#5208 — native Postgres CDC connector for Feldera, which would use etl as its foundation.

Proposal

Add an optional method to the Destination trait that lets the destination report its own confirmed flush LSN:

/// Returns the LSN up to which the destination has durably processed data.
/// 
/// If `None` (the default), etl auto-advances flush_lsn after each
/// successful write_events() call — preserving current behavior.
/// If `Some(lsn)`, etl uses this value instead, letting the destination
/// control when Postgres is told it can discard WAL.
fn confirmed_flush_lsn(&self) -> Option<PgLsn> {
    None
}

In the ApplyLoop, when sending standby_status_update, check if the destination provides a confirmed LSN:

let flush_lsn = match destination.confirmed_flush_lsn() {
    Some(lsn) => lsn,
    None => self.progress.last_flush_lsn,  // current behavior
};

Why this approach

• Backwards-compatible: default returns None, all existing destinations behave identically
• Small change: one new trait method with a default impl, a few lines in the apply loop
• No new dependencies or concepts: just lets the destination participate in LSN tracking
• Correct: the destination knows best when it has durably processed data

Use case

We're building a native Postgres CDC connector for Feldera and want to use etl as the CDC source (snapshot + follow) feeding into Feldera's HTTP ingress. Without this change, we can ship a working connector but cannot offer fault tolerance guarantees — a crash after etl advances flush_lsn but before Feldera processes the data would silently lose events.

With this change, our Destination implementation would query Feldera for its latest checkpointed LSN and report it back, giving us at-least-once delivery end-to-end.

Alternatives considered

• Custom WAL loop: skip etl for the follow phase and use LogicalReplicationStream directly, calling standby_status_update ourselves. Works but duplicates significant logic that etl already handles well (batching, keepalives, table sync coordination).
• Re-snapshot on crash: our current plan for v1 — use etl as-is with no FT. Functional but not ideal for large tables where re-snapshot is expensive.

[iambriccardo]

Thanks for the suggestion.

This is something that could be implemented, but it requires some careful consideration. In particular, the destination would need to return a valid LSN to the engine, specifically the END LSN of the last commit it has applied. If the destination were to return an LSN from the middle of a transaction, PostgreSQL would still resend the entire transaction starting from its beginning.

Implementing this in the core engine would require touching a few additional parts of the codebase, but it should be feasible. One behavioral change is that when we flush a batch and perform table syncs, we should not update the flush LSN ourselves if the destination is responsible for reporting the flushed LSN.

// Take the last commit end LSN, which is the highest LSN from any commit message
// in this batch. Batches can flush mid-transaction, so this may refer to the
// previous transaction.
let Some(last_commit_end_lsn) = self.state.last_commit_end_lsn.take() else {
    return Ok(ApplyLoopAction::Continue);
};

// This would be removed if the destination reports its own progress.

// Update replication progress to notify PostgreSQL of durable flush. Only
// reports progress up to the last completed transaction, which may cause
// duplicates on restart for partial transactions. Destinations must handle
// at-least-once delivery semantics.
self.state
    .replication_progress
    .update_last_flush_lsn(last_commit_end_lsn);

Also for the idle situation (no data coming), it's a bit tricker, since we would need to choose whether to report the destination LSN or the write LSN (which is what we do right now but it becomes a bit more tricky). The reason for this is that suppose we are idle, in that case we might have some data currently being processed, so how do we report the LSN and when do we switch from reporting the destination flush LSN vs the write LSN? We might need the destination to tell us if there are writes in progress, this way we won't start reporting write LSN when in reality data was still not acked.

These are my high-level thoughts. If you would like to attempt an implementation, I would be happy to review a PR. At the moment I am focused on other improvements and would prefer to take the time to properly tackle this ticket in the future. Contributions are therefore welcome if you need it earlier.

[iambriccardo]

A possible solution could be to have the destination also return a boolean together with the LSN specifying if there is an in-flight write so this is not the most up to date LSN, in that case during idle time, ETL can say:

1. If you give me a LSN and you have no inflight data, I will then use write LSN directly.
2. If you give me a LSN and you have inflight data, i won't return write LSN since it will mess up the progress, so I will just keep on returning the LSN given by the destination.

Check out

etl/etl/src/replication/apply.rs

Line 414 in 865e721

fn effective_flush_lsn(&self) -> PgLsn {

if you want to understand a bit more the rationale of this handling. It's a bit tricky to grasp but the reasoning is that when we are idle we still need to progress LSN acks since the WAL growth is not only dependent on the replicated tables and if we don't ack back progress during idle, we will cause WAL buildup on the sender.

[no-flaks-given]

Hi @iambriccardo, thanks for the detailed guidance in your comments. I'd like to take a crack at implementing this. Here's my proposed plan based on your suggestions — would appreciate your feedback before I open a PR.

Proposed API

Add a new default method to the Destination trait:

/// Returns the confirmed flush LSN and in-flight write status for async destinations.
///
/// - `PgLsn`: END LSN of the last fully committed transaction durably processed
///   by the destination (must be a transaction boundary, never mid-transaction).
/// - `bool`: Whether the destination has in-flight (unconfirmed) writes.
///
/// Returns `None` by default, preserving the existing auto-advance behavior.
fn confirmed_flush_lsn(&self) -> Option<(PgLsn, bool)> {
    None
}

The (PgLsn, bool) tuple follows your suggestion from the second comment — the boolean distinguishes:

• (lsn, false) → no inflight data, safe to report last_received_lsn when idle (prevents WAL buildup)
• (lsn, true) → inflight data, keep reporting the destination's confirmed LSN only

None (default) means "I don't control flushing" → all existing destinations work unchanged.

Changes in apply.rs

1. process_syncing_tables_after_batch_flush()

When confirmed_flush_lsn() returns Some, use the destination's confirmed LSN instead of auto-advancing to last_commit_end_lsn:

match self.destination.confirmed_flush_lsn() {
    None => {
        // Legacy: auto-advance
        self.state.replication_progress.update_last_flush_lsn(last_commit_end_lsn);
    }
    Some((confirmed_lsn, _)) => {
        // Destination-controlled: only advance to what destination confirms
        let capped = std::cmp::min(confirmed_lsn, self.state.last_received_lsn());
        self.state.replication_progress.update_last_flush_lsn(capped);
    }
}

2. effective_flush_lsn() — move from ApplyLoopState to ApplyLoop

The current method on ApplyLoopState doesn't have access to the destination. I'll add a wrapper method on ApplyLoop<S, D> that incorporates destination awareness:

fn effective_flush_lsn(&self) -> PgLsn {
    let is_idle = !self.state.handling_transaction() && self.state.events_batch.is_empty();

    if !is_idle {
        return self.state.replication_progress.last_flush_lsn;
    }

    // Idle: check destination
    match self.destination.confirmed_flush_lsn() {
        None => self.state.replication_progress.last_received_lsn,
        Some((_, true)) => self.state.replication_progress.last_flush_lsn,  // inflight — don't advance
        Some((_, false)) => self.state.replication_progress.last_received_lsn,  // no inflight — safe
    }
}

Then update all call sites from self.state.effective_flush_lsn() to self.effective_flush_lsn().

3. PrimaryKeepAlive handler

Poll confirmed_flush_lsn() during keepalives to advance flush progress even when no new data arrives:

ReplicationMessage::PrimaryKeepAlive(message) => {
    // ... existing LSN update ...
    
    // Poll destination for confirmed flush progress
    if let Some((confirmed_lsn, _)) = self.destination.confirmed_flush_lsn() {
        let capped = std::cmp::min(confirmed_lsn, self.state.last_received_lsn());
        self.state.replication_progress.update_last_flush_lsn(capped);
    }

    // ... send_status_update using self.effective_flush_lsn() ...
}

4. initiate_graceful_shutdown()

Poll destination one last time before shutdown:

if let Some((confirmed_lsn, _)) = self.destination.confirmed_flush_lsn() {
    let capped = std::cmp::min(confirmed_lsn, self.state.last_received_lsn());
    self.state.replication_progress.update_last_flush_lsn(capped);
}
let flush_lsn = self.effective_flush_lsn();

Files NOT modified

• stream.rs — send_status_update() already accepts a flush_lsn parameter
• event.rs — event types already carry the correct LSN fields

Backward compatibility

• Default None return preserves exact current behavior on all code paths
• No breaking changes to existing trait methods
• Existing tests should pass unchanged

My use case

I'm building a Feldera CDC connector that uses etl for Postgres logical replication. Feldera processes events through a streaming pipeline before durable persistence, so the current auto-advance behavior risks data loss on crash. The Feldera destination would implement confirmed_flush_lsn() to report only durably checkpointed LSNs.

Let me know if this approach looks right or if you'd want any changes before I submit the PR.

[no-flaks-given]

Required vs Optional Changes

Following up on my previous comment — here's a breakdown of what's strictly required for correctness vs optional enhancements.

Required (prevents data loss)

These are the minimum changes needed to support async destinations safely:

1. confirmed_flush_lsn() on Destination trait — The core API. Option<(PgLsn, bool)> with None default. The bool is necessary to distinguish "destination finished everything" from "destination still processing" during idle (as you identified in your second comment).

2. process_syncing_tables_after_batch_flush() — When confirmed_flush_lsn() returns Some, use the destination's confirmed LSN instead of auto-advancing to last_commit_end_lsn. Without this, PostgreSQL discards WAL before the destination has durably processed it.

3. effective_flush_lsn() idle-state awareness — Move to ApplyLoop so it can access the destination. When idle with inflight writes (lsn, true), report last_flush_lsn instead of last_received_lsn. Without this, idle periods with pending data cause WAL discard → data loss.

Required but minor (completes the above)

4. PrimaryKeepAlive polling — Poll confirmed_flush_lsn() during keepalives to advance last_flush_lsn between batch flushes. Without this, flush progress stalls during idle and restart replays more data than necessary.

5. Shutdown polling — Poll confirmed_flush_lsn() one last time in initiate_graceful_shutdown(). Same rationale — reduces duplicate replay on restart.

Optional enhancements (can be follow-up PRs)

6. Replication lag metric — Emit a metric for the gap between last_received_lsn and last_flush_lsn. Lets operators alert on growing WAL retention caused by slow async destinations.

7. Staleness warning — Log a warning when the destination's confirmed LSN hasn't advanced for N keepalive cycles while new data keeps arriving. Catches stuck or deadlocked destinations early.

8. Graceful shutdown drain — Instead of polling the destination once during shutdown, optionally wait (with a configurable timeout) for inflight writes to complete before sending the final status update. Reduces duplicates on planned restarts at the cost of slower shutdown.

9. Batch LSN passthrough — Convenience method like write_events_with_commit_lsn(events, last_commit_end_lsn) so destinations don't have to parse Vec<Event> to find CommitEvent.end_lsn. Default impl falls back to write_events().

My plan is to submit a PR with items 1–5 and leave 6–9 as follow-ups. Let me know if you'd want any of the optionals included in the initial PR, or if the scope of 1–5 looks right.

[iambriccardo]

Your proposal looks sound, the only thing, we just need to be careful on which LSN we supply to the process_syncing_tables_after_batch_flush function. That LSN must always coincide with the end_lsn of the last flushed commit.

For this to work we might have to track all end_lsns that we encountered and whenever we update the last_flush_lsn we basically remove all end_lsn(s) that are < the last flush lsn communicated by the destination. And than we take the smallest one of the list and supply that to the process_syncing_tables_after_batch_flush.

I am worried though this adds a lot of complexity in the system but I don't see better ways to do this.

The other problem is that the progress acks will be based on how frequently the apply worker polls the destination, so we might need a reactive system to react quickly to it, otherwise it will be mostly driven by keep alives or batch flushes.

[iambriccardo]

What I recommend is, try to make a PR thinking about all of the inputs I gave you, then put it up for review, I will try to review it and also try the code in your own infra. It's very important that we make sure the pipeline never stalls progress during idle, nor that it misses status updates with correct values.

This is a very tricky change to test, so we have to be careful.

As a side note, I will be working on an async flushing system for ETL but it's complementary to this.

[no-flaks-given]

PR up: #626

Implements the destination-controlled flush approach we discussed. Key decisions:

• No tracking needed — the value chain is structurally guaranteed to be a commit boundary (CommitEvent.end_lsn → destination → confirmed_flush_lsn()), so tracking pending LSNs is redundant. Contract is enforced via doc comments, debug_assert!, and a std::cmp::min cap in release builds.
• confirmed_flush_lsn(&self) -> Option<(PgLsn, bool)> — sync method, returns None by default so all existing destinations are unaffected. The bool distinguishes idle-with-inflight from idle-with-nothing-pending for correct keepalive behavior.
• Reactive keepalive polling — poll_destination_flush() is called in the keepalive handler (before the existing process_syncing_tables_when_idle() at line 892), so progress advances during idle periods without stalling table transitions.
• 30 unit tests — first unit tests in apply.rs, covering all new methods, edge cases, and interactions.

[no-flaks-given]

@iambriccardo Thanks for the feedback on #626 and for sharing your experimental PR #628 — really helpful to see your thinking on this.

Why blocking is tricky for async destinations

For context on why I went the callback route: Feldera's input system is a fire-and-forget queue (InputQueue.push()). There's currently no API to block until data is checkpointed. Adding one would actually require changes to Feldera's core InputConsumer trait and pipeline controller — it's a bigger lift on the Feldera side than the ETL side.

There's also the concern that while write_events blocks waiting for a checkpoint, the apply loop's select! can't process PrimaryKeepAlive messages, which could trigger wal_sender_timeout.

Prior art

Looking at how other CDC systems handle this:

• Kafka Connect decoupled flush and offset commit via KIP-89 — the framework provides the acknowledgment mechanism, the connector controls when offsets are safe to commit
• Debezium Engine uses RecordCommitter.markBatchFinished() — same pattern, the consumer fires it when ready

Both put the mechanism in the framework and let the destination/consumer decide timing.

Path forward

Looking at #628, the core pattern aligns with this — BatchFlushResult<T> provides a framework-level acknowledgment mechanism that destinations control. The ETL-side changes to support this are relatively small (one select! branch, one state field), and existing destinations like BigQuery and Iceberg would just fire the confirmation at the end of write_events — one line change, same behavior as today.

Happy to help implement this in whatever form you think is best, or to stay out of the way if you're planning to revisit #628 yourself. I think it's worth getting right — Kafka Connect and Debezium both landed on this same pattern of framework-provided acknowledgment with destination-controlled timing. It'd be great to see ETL align with that model.

[iambriccardo]

@iambriccardo Thanks for the feedback on #626 and for sharing your experimental PR #628 — really helpful to see your thinking on this.

Why blocking is tricky for async destinations

For context on why I went the callback route: Feldera's input system is a fire-and-forget queue (InputQueue.push()). There's currently no API to block until data is checkpointed. Adding one would actually require changes to Feldera's core InputConsumer trait and pipeline controller — it's a bigger lift on the Feldera side than the ETL side.

There's also the concern that while write_events blocks waiting for a checkpoint, the apply loop's select! can't process PrimaryKeepAlive messages, which could trigger wal_sender_timeout.

Prior art

Looking at how other CDC systems handle this:

• Kafka Connect decoupled flush and offset commit via KIP-89 — the framework provides the acknowledgment mechanism, the connector controls when offsets are safe to commit
• Debezium Engine uses RecordCommitter.markBatchFinished() — same pattern, the consumer fires it when ready

Both put the mechanism in the framework and let the destination/consumer decide timing.

Path forward

Looking at #628, the core pattern aligns with this — BatchFlushResult<T> provides a framework-level acknowledgment mechanism that destinations control. The ETL-side changes to support this are relatively small (one select! branch, one state field), and existing destinations like BigQuery and Iceberg would just fire the confirmation at the end of write_events — one line change, same behavior as today.

Happy to help implement this in whatever form you think is best, or to stay out of the way if you're planning to revisit #628 yourself. I think it's worth getting right — Kafka Connect and Debezium both landed on this same pattern of framework-provided acknowledgment with destination-controlled timing. It'd be great to see ETL align with that model.

Hi, thanks for the clear answer!

The reason why the async system is not ideal is that even if we were to have it, we would still need to wait for the inflight batch to be resolved before sending the other one, to avoid making it harder for destinations to track the order of operations to apply them in the right order. Classic example is: we send a batch to your destination, while you are processing it, another one comes and you might send it too but then they arrive out of order, which breaks the whole consistency model of streaming.

What I am not fully understand is, how is the current system giving you feedback that data up to a certain amount was received and durably persisted? I assume you can block the write_events method on that callback.

If you want to take over my PR and improve it, feel free to do it! The thing is, the complexity increase is too big to justify at this point in time. Especially because the apply loop heavily relies on the blocking behavior of the write_events method to make the state machine simpler to understand.

It's a solvable problem and my PR proves it but there are so many edge cases that arise with inflight batches now (especially for controlled shutdown). Happy to support if you want to take over that PR.

Related to keepalives, this is indeed one of the biggest reasons for why someone would choose to use async flushing. However, if ETL keeps on moving batches without ever blocking on writes, it will accumulate memory and will block processing nonetheless (depending on the destination implementation).

[iambriccardo]

I feel like the biggest questions to answer are:

1. Do we want to allow back to back write_events calls with async ack? Or allow the ack to be done async but having at most one inflight write_events call and in the meanwhile a new batch is getting ready (like in my experimental PR)?
2. How do we want to deal with the shutdown semantics when there is inflight data? What if there is an active transaction?
3. How do we want to ack progress of multiple inflight calls? Do we want to ack it in strict order of submission (e.g., using FuturesOrdered) or in monotonically increasing order but with the possibility of skipping older acks?
4. How do we plan on enforcing ordering in a nice way on destinations that require it without having each destination implement their own barriers for blocking writes while another one is in flight?

[no-flaks-given]

@iambriccardo Great questions, thanks for digging into this.

The current state

To answer your question about durability feedback: we don't actually have that yet on the Feldera side — that's what motivated this issue in the first place. Right now our destination pushes events into Feldera's InputQueue and returns immediately, so ETL auto-advances the slot. If Feldera crashes before checkpointing, that data is lost.

There's no existing callback we can block on, and adding one to Feldera would mean going against how Feldera's other input sources work. Sources like Kafka already follow the pattern where the framework provides acknowledgment control and the connector decides when to fire it. Adding a blocking checkpoint API to Feldera's input system would be the odd one out. It makes more sense for ETL to align with that same model, so the Feldera CDC adapter works like every other Feldera input.

On the design questions

These are really good questions. I think the good news is that for the initial use case, scoping to one inflight batch at a time (which is what your #628 does) makes most of them straightforward:

1. Inflight batches: One at a time — write_events is still awaited inline, the only change is that slot advancement waits for the confirmation. This avoids all the ordering and concurrency complexity.

2. Shutdown semantics: With one inflight batch, we just check if the pending confirmation has arrived. If yes, advance and report. If not, report last_flush_lsn and accept replay on restart — consistent with at-least-once semantics.

3. Ack ordering: With one inflight batch, there's only ever one pending confirmation, so ordering is trivially maintained.

4. Destination ordering: Since write_events is still called sequentially, destinations see batches in order with no extra work needed.

Multiple inflight batches adds a lot of complexity and I don't think most destinations would benefit from it. For reference, Debezium is one of the most widely deployed CDC frameworks and its Postgres connector is single-threaded with single-inflight — one batch at a time, acknowledged via RecordCommitter.markBatchFinished() before the next batch is sent. It handles ~7,000 events/sec this way, and the bottleneck is almost always the source database or the destination, not the framework's event loop. Single-inflight solves the correctness problem cleanly without introducing ordering, shutdown, or ack-tracking concerns.

Taking over #628

I'd love to take over #628 if you're open to it — your BatchFlushResult<T> abstraction is really clean. Would you prefer I open a new PR based on that branch, or reopen #626 and adapt it to use your API shape?