fix(etl): skip errored state and interrupt flush wait on table-copy shutdown by abhizer · Pull Request #881 · supabase/etl

abhizer · 2026-07-07T14:39:45Z

Currently, during table-copy, if the destination drops the AsyncResult while the pipeline is shutting down, the table ends up in an Errored state. But dropping the AsyncResult during shutdown should be safe: on restart, we should just ensure the retry copies the table from the start.

Also make the copy loop stop waiting on the flush result once shutdown fires, instead of blocking on a destination that never completes it.

What kind of change does this PR introduce?

Bug fix

What is the current behavior?

During table-copy, if the destination drops the AsyncResult while the
pipeline is shutting down, the table ends up in an Errored state. The
copy loop also waits on the flush result unconditionally, so it can
hang if the destination never completes it.

What is the new behavior?

Dropping the AsyncResult during shutdown no longer persists an Errored
state; the table retries the copy from the start on restart. The copy
loop also stops waiting on the flush result once shutdown fires.

iambriccardo · 2026-07-10T22:44:31Z

Hi, thanks for this PR! I will get to it once I find the time.

abhizer · 2026-07-12T13:14:31Z

Thanks!

…hutdown Currently, during table-copy, if the destination drops the AsyncResult while the pipeline is shutting down, the table ends up in an Errored state. But dropping the AsyncResult during shutdown should be safe: on restart, we should just ensure the retry copies the table from the start. Also make the copy loop stop waiting on the flush result once shutdown fires, instead of blocking on a destination that never completes it. Signed-off-by: Abhinav Gyawali <22275402+abhizer@users.noreply.github.com>

bnjjj · 2026-07-15T08:45:22Z

Hey @abhizer I think I got this issue last week too. What was the error you got once it happened ?

coveralls · 2026-07-15T08:59:44Z

Coverage is 70.861% — abhizer:fix/table-copy-shutdown-write-race into supabase:main. No base build found for supabase:main.

iambriccardo · 2026-07-15T06:55:28Z

+        // a stored `Errored` state is never retried across restarts, so the
+        // table would otherwise stall on every later run. A dropped sender also
+        // counts as shutdown.
+        if shutdown_rx.has_changed().unwrap_or(true) {


This is not the correct solution for a concurrent system like ETL. Having the shutdown being true at this phase, doesn't guarantee us that the error will be caused by the shutdown procedure.

That's fair. We'd just like to be able to drop the acknowledgements once we've initiated the shutdown.

Can you give me more context on your ETL use case? So that I can see how we can better design shutdown in case.

The reason of why I am saying this is that, the shutdown on the destination is called on purpose after all of ETL is done, so that we can stop producing data on our end and then the destination can perform teardown.

The failure should not happen because a destination is technically unaware of shutdown until ETL won't be interested anymore about the result. And if the result is sent back and the channel is closed a warning will be raised.

if tx.send(result).is_err() { warn!("could not send async result because receiver was already closed"); }

Our use case is the Postgres CDC input connector for feldera.

We want the connector to snapshot, and then follow the cdc stream, this works.

But we'd also like to ensure fault tolerance and be able to reingest the table in cases of failure. In cases of failure, the connector sees that the pipeline is shutting down, and triggers the etl pipeline to shutdown as well. And in such cases, returns from write_table_rows without acknowledging the AsyncResult, which can lead to the table being in errored state.

Because we've already requested a shutdown here, we want this to be okay.

Related: The Accept / Durable api for table copy, currently on Accept can currently allow the table to transition to Ready, causing etl to skip reingesting on restart. I do not think that Accept should have this durable side effect.

Ah yeah, I get what you mean. The thing here is that we need to clarify the semantics. I feel like from ETL's side, if a channel is closed without a response, it's a problem. The destination should take care of sending back a response. Maybe we could classify a response as "gracefully stop". But from my idea, I would like the system to be like, if there is shutdown after the write_table_rows method, we immediately return. Would that work?

Sorry, I don't think I quite understand the immediately return model. Do you mean something like a biased tokio select, on shutdown.changed() and the write_table_rows method?

But yes, the gracefully stop idea, for shutdown during write_table_rows, returning an ErrorKind::DestinationShutdown should work well for us, and in other similar use cases. ETL could then treat it as a graceful cancellation, instead of an error.

One question I would have is. I assume your write_table_rows fails because you have an out-of-bound shutdown signal receiver in your destination which makes it stop and not return a result?

The reason I am asking is that ETL is designed in a way where the shutdown procedure of a destination should be made in the shutdown() method, so that teardown is properly controlled.

One question I would have is. I assume your write_table_rows fails because you have an out-of-bound shutdown signal receiver in your destination which makes it stop and not return a result?

Yes. As the connector is part of the pipeline, the shutdown procedure cannot quite be contained in the shutdown() method. We need to shutdown the etl pipeline safely, when shutting down the outer pipeline.

This would make the implementation quite a bit uglier, is this a problem also for the apply loop?

iambriccardo · 2026-07-15T06:55:50Z

                    .write_table_rows(&replicated_table_schema, table_rows, flush_result)
                    .await?;
-                let write_status = pending_flush_result.await.into_result()?;
+                let write_status = tokio::select! {


This is fair but I think it would be better if we make an abstraction which behind the scenes waits for shutdown on a result.

I am working now on a PR to do that.

#900 this is the PR in progress btw

Thanks for the fix!

abhizer · 2026-07-15T09:30:46Z

I think I got this issue last week too. What was the error you got once it happened ?

@bnjjj I don't exactly remember, but it was something to do with the table being in errored state, when we've tried to shutdown the etl pipeline, and then dropped the async result.

iambriccardo · 2026-07-15T09:33:45Z

I think I got this issue last week too. What was the error you got once it happened ?

@bnjjj I don't exactly remember, but it was something to do with the table being in errored state, when we've tried to shutdown the etl pipeline, and then dropped the async result.

This is interesting. The only way in which it could have happened is if the AsyncResult was dropped by the destination while it was waiting on it. At least this is what I am thinking of.

bnjjj · 2026-07-15T09:35:22Z

Yes probably something in the ducklake destination I'll check

The feldera/etl fork carried one commit, selecting the aws-lc-rs backend for sqlx TLS so etl does not link ring next to Feldera's aws-lc. Upstream made the backend selectable, so `default-features = false` plus `tls-rustls-aws-lc-rs` reproduces that change and the fork revision is no longer needed. feldera/etl main is synced to upstream 248fa407; the old pin stays reachable on the fork's `sqlx-aws-lc-rs` branch. This moves 175 commits, which brings in the rework that removes etl's SharedTableCache. That rework fixes the `Missing shared table state` flake in test_cdc_restart_resumes_from_slot, where the apply worker received a row event for a table whose schema the table-sync worker owned on the same connection. SyncDone now stores a durable decoder that relation-less DML restores. Upstream still persists an Errored table state on shutdown, so the connector-side rollback stays until supabase/etl#881 merges. Drop the etl-config and etl-postgres workspace entries: no crate in the workspace depends on them, and leaving them pinned at the old revision would put two etl source revisions in the lockfile. The connector does not build against this pin yet. Every etl import path moved, PipelineConfig gained three fields, and both Destination write methods changed signature. Porting cdc_input.rs follows in the next commit. Refs #7107

etl records the initial copy of a table as finished as soon as the connector acknowledges its rows, which is before Feldera has stepped them through the circuit, let alone checkpointed them. A checkpoint could therefore hold part of a copy, and a restart from it could hold none: etl streams from the replication slot once it considers the copy done, so the missing rows never arrive (#6121). The connector now holds etl's copy barrier until the circuit holds the copy. Upstream etl issues a terminal, empty `write_table_rows` once every copy worker of an attempt has finished, and records the copy as finished only when that call answers `Durable`. `FelderaDestination` answers each batch of copy rows with `Accepted`, counts the snapshot buffers it queues in a shared `CopyBarrier`, and parks the terminal call there until the reader's `Queue` flush has drained every one of them into the circuit. Streamed events share the input queue, so the barrier counts snapshot buffers rather than testing for an empty queue. While a copy is open the reader reports `Resume::Barrier`, so the controller defers a checkpoint or refuses a suspend instead of recording a state it could not finish on resume; once the copy is in the circuit it reports `Resume::Seek` with the pipeline id, because etl's `PostgresStore` holds the replication position. A copy that etl starts over, after a retry or a rollback, raises the barrier again from `drop_table_for_copy` or its first batch, before any row of it can reach the queue. A pending checkpoint stops the controller from stepping on its own, so the destination and the reader ask for the step that reports a lowered barrier. This is why the etl dependency moves in the same change from the feldera/etl fork's own revision to upstream supabase/etl commit 248fa407, still fetched through the feldera/etl mirror: the terminal write, `DestinationWriteStatus` and the batch id that tells a batch from the barrier exist only there. The fork carried one commit, which selected the aws-lc-rs backend for sqlx TLS so etl does not link ring next to Feldera's aws-lc; upstream made the backend selectable, so `default-features = false` plus `tls-rustls-aws-lc-rs` reproduces it, and the old pin stays reachable on the fork's `sqlx-aws-lc-rs` branch. The etl-config and etl-postgres workspace entries go: no crate depends on them, and pinning them at the old revision would put two etl source revisions in the lockfile. The bump moves 175 commits and brings three things the connector has to follow. etl no longer has a SharedTableCache: SyncDone stores a durable decoder that relation-less DML restores, which fixes the `Missing shared table state` flake in test_cdc_restart_resumes_from_slot where the apply worker received a row event for a table whose schema the table-sync worker owned. etl reads the source schema through helper functions its own source migrations install, so the connector runs them; the second migration extends the DDL event trigger to `ALTER PUBLICATION` and needs the same privileges as the first. Cells gained `TimeTz`, which the connector renders in the text form Postgres uses, offset included, because Feldera has no time-with-time-zone type. Startup reconciles etl's stored state with what the circuit holds. `reconcile_table_state` rolls back the two errors a stop leaves behind, the one the connector records when Feldera stops it mid-batch and the one etl records for a deferred result the connector dropped, along with timed retries, whose deadline etl does not honor across a restart. A table whose history offers nothing to roll back to is reset to `Init` and read again rather than left stalled. Under fault tolerance with no checkpoint the circuit starts empty while etl may consider the copy done, so a completed sync is reset to `Init` too; this is the one case where a resumed pipeline could otherwise miss the copy, and no resume metadata is needed for the others. The function returns whether the table is still to be copied, so the barrier starts in the right position before etl runs. Upstream still persists an Errored state on shutdown, so the rollback stays until supabase/etl#881 merges. `TableErrorMonitor` reports what would otherwise leave the pipeline silent: a non-retriable error on any table of the publication, and a source table etl has not moved in ten minutes while it had no other table left to copy, most often because the publication does not carry it. The stall clock counts queued and flushed snapshot buffers as progress, because etl records no state transition inside a copy. The `select!` arm awaiting termination kept the state watch's read guard alive while awaiting etl shutdown; the reader's `Drop` then queued a second write on the watch, etl's copy tasks blocked re-reading it, and stop hung. The guard is dropped before the await. The preceding commit resolves an unqualified `source_table` in `public`; this commit only adds the changelog entry for that rule. Tests: the four CDC tests that were red for this bug run in CI again (test_cdc_ft_mode_holds_slot, test_snapshot_replayed_when_stopped_before_checkpoint, and test_checkpoint_mid_snapshot_waits_for_the_whole_copy in its single- and multi-worker forms), joined by test_suspend_mid_copy_is_refused. Both mid-copy tests now measure a non-empty, strictly partial copy in the circuit while etl reports `data_sync` before they act, so a copy that finished early or a connector that died fails the test instead of passing it by accident. test_other_tables_in_publication_are_filtered asks etl whether both tables completed their sync, which catches a barrier answered wrongly for the table the connector does not read. Unit tests drive `CopyBarrier` through every call the destination and the reader make, `classify_copy_write`, the destination-to-barrier glue over a live queue, the two stop literals the rollback matches, and the `TimeTz` cells. The remaining CDC scenarios stay `#[ignore]`d, as on main. Docs: the connector page states when a table is read again, which starts keep the rows of the initial read, and what the connector installs in the source database and why that needs a superuser; the fault-tolerance table lists the PostgreSQL CDC input with checkpoint and resume and at- least-once fault tolerance but not exactly-once; the changelog records the barrier, the repeated read, the fixed flake and the second migration. Fixes #6121 Refs #7107

etl records the initial copy of a table as finished as soon as the connector acknowledges its rows, which is before Feldera has stepped them through the circuit, let alone checkpointed them. A checkpoint could therefore hold part of a copy, and a restart from it could hold none: etl streams from the replication slot once it considers the copy done, so the missing rows never arrive (#6121). The connector now holds etl's copy barrier until the circuit holds the copy. Upstream etl issues a terminal, empty `write_table_rows` once every copy worker of an attempt has finished, and records the copy as finished only when that call answers `Durable`. `FelderaDestination` answers each batch of copy rows with `Accepted`, counts the snapshot buffers it queues in a shared `CopyBarrier`, and parks the terminal call there until the reader's `Queue` flush has drained every one of them into the circuit. Streamed events share the input queue, so the barrier counts snapshot buffers rather than testing for an empty queue. While a copy is open the reader reports `Resume::Barrier`, so the controller defers a checkpoint or refuses a suspend instead of recording a state it could not finish on resume; once the copy is in the circuit it reports `Resume::Seek` with the pipeline id, because etl's `PostgresStore` holds the replication position. A copy that etl starts over, after a retry or a rollback, raises the barrier again from `drop_table_for_copy` or its first batch, before any row of it can reach the queue. A pending checkpoint stops the controller from stepping on its own, so the destination and the reader ask for the step that reports a lowered barrier. This is why the etl dependency moves in the same change from the feldera/etl fork to upstream supabase/etl at commit 248fa407: the terminal write, `DestinationWriteStatus` and the batch id that tells a batch from the barrier exist only there. The fork carried one commit, which selected the aws-lc-rs backend for sqlx TLS so etl does not link ring next to Feldera's aws-lc; upstream made the backend selectable, so `default-features = false` plus `tls-rustls-aws-lc-rs` reproduces it, and the old pin stays reachable on the fork's `sqlx-aws-lc-rs` branch. The etl-config and etl-postgres workspace entries go: no crate depends on them, and pinning them at the old revision would put two etl source revisions in the lockfile. The bump moves 175 commits and brings three things the connector has to follow. etl no longer has a SharedTableCache: SyncDone stores a durable decoder that relation-less DML restores, which fixes the `Missing shared table state` flake in test_cdc_restart_resumes_from_slot where the apply worker received a row event for a table whose schema the table-sync worker owned. etl reads the source schema through helper functions its own source migrations install, so the connector runs them; the second migration extends the DDL event trigger to `ALTER PUBLICATION` and needs the same privileges as the first. Cells gained `TimeTz`, which the connector renders in the text form Postgres uses, offset included, because Feldera has no time-with-time-zone type. etl's defaults also moved, and the connector follows them: copy batches of up to 32 MiB instead of 8, and four copy connections per table instead of two. Four more state-store migrations run on the source database; one retypes the snapshot_id columns, so an older Feldera cannot start against a database this version has migrated. Startup reconciles etl's stored state with what the circuit holds. `reconcile_table_state` rolls back the two errors a stop leaves behind, the one the connector records when Feldera stops it mid-batch and the one etl records for a deferred result the connector dropped, along with timed retries, whose deadline etl does not honor across a restart. A table whose history offers nothing to roll back to is reset to `Init` and read again rather than left stalled. Under fault tolerance with no checkpoint the circuit starts empty while etl may consider the copy done, so a completed sync is reset to `Init` too; this is the one case where a resumed pipeline could otherwise miss the copy, and no resume metadata is needed for the others. The function returns whether the table is still to be copied, so the barrier starts in the right position before etl runs. Upstream still persists an Errored state on shutdown, so the rollback stays until supabase/etl#881 merges. `TableErrorMonitor` reports what would otherwise leave the pipeline silent: a non-retriable error on any table of the publication, and a source table etl has not moved in ten minutes while it had no other table left to copy, most often because the publication does not carry it. The stall clock counts queued and flushed snapshot buffers as progress, because etl records no state transition inside a copy. The `select!` arm awaiting termination kept the state watch's read guard alive while awaiting etl shutdown; the reader's `Drop` then queued a second write on the watch, etl's copy tasks blocked re-reading it, and stop hung. The guard is dropped before the await. The preceding commit resolves an unqualified `source_table` in `public`; this commit only adds the changelog entry for that rule. Tests: the four CDC tests that were red for this bug run in CI again (test_cdc_ft_mode_holds_slot, test_snapshot_replayed_when_stopped_before_checkpoint, and test_checkpoint_mid_snapshot_waits_for_the_whole_copy in its single- and multi-worker forms), joined by test_suspend_mid_copy_is_refused, and test_cdc_restart_resumes_from_slot, the test that flaked on the fork revision, which passed ten consecutive runs against the new pin. Both mid-copy tests now measure a non-empty, strictly partial copy in the circuit while etl reports `data_sync` before they act, so a copy that finished early or a connector that died fails the test instead of passing it by accident. test_other_tables_in_publication_are_filtered asks etl whether both tables completed their sync, which catches a barrier answered wrongly for the table the connector does not read. Unit tests drive `CopyBarrier` through every call the destination and the reader make, `classify_copy_write`, the destination-to-barrier glue over a live queue, the two stop literals the rollback matches, and the `TimeTz` cells. The remaining CDC scenarios stay `#[ignore]`d, as on main. Docs: the connector page states when a table is read again, which starts keep the rows of the initial read, and what the connector installs in the source database and why that needs a superuser; the fault-tolerance table lists the PostgreSQL CDC input with checkpoint and resume and at- least-once fault tolerance but not exactly-once; the changelog records the barrier, the repeated read, the fixed flake and the second migration. Fixes #6121 Refs #7107

etl records the initial copy of a table as finished as soon as the connector acknowledges its rows, which is before Feldera has stepped them through the circuit, let alone checkpointed them. A checkpoint could therefore hold part of a copy, and a restart from it could hold none: etl streams from the replication slot once it considers the copy done, so the missing rows never arrive (#6121). The connector now holds etl's copy barrier until the circuit holds the copy. Upstream etl issues a terminal, empty `write_table_rows` once every copy worker of an attempt has finished, and records the copy as finished only when that call answers `Durable`. `FelderaDestination` answers each batch of copy rows with `Accepted`, counts the snapshot buffers it queues in a shared `CopyBarrier`, and parks the terminal call there until the reader's `Queue` flush has drained every one of them into the circuit. Streamed events share the input queue, so the barrier counts snapshot buffers rather than testing for an empty queue. While a copy is open the reader reports `Resume::Barrier`, so the controller defers a checkpoint or refuses a suspend instead of recording a state it could not finish on resume; once the copy is in the circuit it reports `Resume::Seek` with the pipeline id, because etl's `PostgresStore` holds the replication position. A copy that etl starts over, after a retry or a rollback, raises the barrier again from `drop_table_for_copy` or its first batch, before any row of it can reach the queue. A pending checkpoint stops the controller from stepping on its own, so the destination and the reader ask for the step that reports a lowered barrier. This is why the etl pin moves in the same change to 248fa407, an upstream supabase/etl commit synced into the feldera/etl fork, which keeps tracking upstream by hand: the terminal write, `DestinationWriteStatus` and the batch id that tells a batch from the barrier exist only there. The fork carried one commit, which selected the aws-lc-rs backend for sqlx TLS so etl does not link ring next to Feldera's aws-lc; upstream made the backend selectable, so `default- features = false` plus `tls-rustls-aws-lc-rs` reproduces it, and the old pin stays reachable on the fork's `sqlx-aws-lc-rs` branch. The etl- config and etl-postgres workspace entries go: no crate depends on them, and pinning them at the old revision would put two etl source revisions in the lockfile. The bump moves 175 commits and brings three things the connector has to follow. etl no longer has a SharedTableCache: SyncDone stores a durable decoder that relation-less DML restores, which fixes the `Missing shared table state` flake in test_cdc_restart_resumes_from_slot where the apply worker received a row event for a table whose schema the table-sync worker owned. etl reads the source schema through helper functions its own source migrations install, so the connector runs them; the second migration extends the DDL event trigger to `ALTER PUBLICATION` and needs the same privileges as the first. Cells gained `TimeTz`, which the connector renders in the text form Postgres uses, offset included, because Feldera has no time-with-time-zone type. etl's defaults also moved, and the connector follows them: copy batches of up to 32 MiB instead of 8, and four copy connections per table instead of two. Four more state-store migrations run on the source database; one retypes the snapshot_id columns, so an older Feldera cannot start against a database this version has migrated. Startup reconciles etl's stored state with what the circuit holds. `reconcile_table_state` rolls back the two errors a stop leaves behind, the one the connector records when Feldera stops it mid-batch and the one etl records for a deferred result the connector dropped, along with timed retries, whose deadline etl does not honor across a restart. A table whose history offers nothing to roll back to is reset to `Init` and read again rather than left stalled. Under fault tolerance with no checkpoint the circuit starts empty while etl may consider the copy done, so a completed sync is reset to `Init` too; this is the one case where a resumed pipeline could otherwise miss the copy, and no resume metadata is needed for the others. The function returns whether the table is still to be copied, so the barrier starts in the right position before etl runs. Upstream still persists an Errored state on shutdown, so the rollback stays until supabase/etl#881 merges. `TableErrorMonitor` reports what would otherwise leave the pipeline silent: a non-retriable error on any table of the publication, and a source table etl has not moved in ten minutes while it had no other table left to copy, most often because the publication does not carry it. The stall clock counts queued and flushed snapshot buffers as progress, because etl records no state transition inside a copy. The `select!` arm awaiting termination kept the state watch's read guard alive while awaiting etl shutdown; the reader's `Drop` then queued a second write on the watch, etl's copy tasks blocked re-reading it, and stop hung. The guard is dropped before the await. The preceding commit resolves an unqualified `source_table` in `public`; this commit only adds the changelog entry for that rule. Tests: the four CDC tests that were red for this bug run in CI again (test_cdc_ft_mode_holds_slot, test_snapshot_replayed_when_stopped_before_checkpoint, and test_checkpoint_mid_snapshot_waits_for_the_whole_copy in its single- and multi-worker forms), joined by test_suspend_mid_copy_is_refused, and test_cdc_restart_resumes_from_slot, the test that flaked on the fork revision, which passed ten consecutive runs against the new pin. Both mid-copy tests now measure a non-empty, strictly partial copy in the circuit while etl reports `data_sync` before they act, so a copy that finished early or a connector that died fails the test instead of passing it by accident. test_other_tables_in_publication_are_filtered asks etl whether both tables completed their sync, which catches a barrier answered wrongly for the table the connector does not read. Unit tests drive `CopyBarrier` through every call the destination and the reader make, `classify_copy_write`, the destination-to-barrier glue over a live queue, the two stop literals the rollback matches, and the `TimeTz` cells. The remaining CDC scenarios stay `#[ignore]`d, as on main. Docs: the connector page states when a table is read again, which starts keep the rows of the initial read, and what the connector installs in the source database and why that needs a superuser; the fault-tolerance table lists the PostgreSQL CDC input with checkpoint and resume and at- least-once fault tolerance but not exactly-once; the changelog records the barrier, the repeated read, the fixed flake and the second migration. Fixes #6121 Refs #7107

etl records the initial copy of a table as finished as soon as the connector acknowledges its rows, which is before Feldera has stepped them through the circuit, let alone checkpointed them. A checkpoint could therefore hold part of a copy, and a restart from it could hold none: etl streams from the replication slot once it considers the copy done, so the missing rows never arrive (#6121). The connector now holds etl's copy barrier until the circuit holds the copy. Upstream etl issues a terminal, empty `write_table_rows` once every copy worker of an attempt has finished, and records the copy as finished only when that call answers `Durable`. `FelderaDestination` answers each batch of copy rows with `Accepted`, counts the snapshot buffers it queues in a shared `CopyBarrier`, and parks the terminal call there until the reader's `Queue` flush has drained every one of them into the circuit. Streamed events share the input queue, so the barrier counts snapshot buffers rather than testing for an empty queue. While a copy is open the reader reports `Resume::Barrier`, so the controller defers a checkpoint or refuses a suspend instead of recording a state it could not finish on resume; once the copy is in the circuit it reports `Resume::Seek` with the pipeline id, because etl's `PostgresStore` holds the replication position. A copy that etl starts over, after a retry or a rollback, raises the barrier again from `drop_table_for_copy` or its first batch, before any row of it can reach the queue. A pending checkpoint stops the controller from stepping on its own, so the destination and the reader ask for the step that reports a lowered barrier. This is why the etl pin moves in the same change to 248fa407, an upstream supabase/etl commit synced into the feldera/etl fork, which keeps tracking upstream by hand: the terminal write, `DestinationWriteStatus` and the batch id that tells a batch from the barrier exist only there. The fork carried one commit, which selected the aws-lc-rs backend for sqlx TLS so etl does not link ring next to Feldera's aws-lc; upstream made the backend selectable, so `default- features = false` plus `tls-rustls-aws-lc-rs` reproduces it, and the old pin stays reachable on the fork's `sqlx-aws-lc-rs` branch. The etl- config and etl-postgres workspace entries go: no crate depends on them, and pinning them at the old revision would put two etl source revisions in the lockfile. The bump moves 175 commits and brings three things the connector has to follow. etl no longer has a SharedTableCache: SyncDone stores a durable decoder that relation-less DML restores, which fixes the `Missing shared table state` flake in test_cdc_restart_resumes_from_slot where the apply worker received a row event for a table whose schema the table-sync worker owned. etl reads the source schema through helper functions its own source migrations install, so the connector runs them; the second migration extends the DDL event trigger to `ALTER PUBLICATION` and needs the same privileges as the first. Cells gained `TimeTz`, which the connector renders in the text form Postgres uses, offset included, because Feldera has no time-with-time-zone type. etl's defaults also moved, and the connector follows them: copy batches of up to 32 MiB instead of 8, and four copy connections per table instead of two. Four more state-store migrations run on the source database; one retypes the snapshot_id columns, so an older Feldera cannot start against a database this version has migrated. Startup reconciles etl's stored state with what the circuit holds. `reconcile_table_state` rolls back the two errors a stop leaves behind, the one the connector records when Feldera stops it mid-batch and the one etl records for a deferred result the connector dropped, along with timed retries, whose deadline etl does not honor across a restart. A table whose history offers nothing to roll back to is reset to `Init` and read again rather than left stalled. Under fault tolerance with no checkpoint the circuit starts empty while etl may consider the copy done, so a completed sync is reset to `Init` too; this is the one case where a resumed pipeline could otherwise miss the copy, and no resume metadata is needed for the others. The function returns whether the table is still to be copied, so the barrier starts in the right position before etl runs. Upstream still persists an Errored state on shutdown, so the rollback stays until supabase/etl#881 merges. `TableErrorMonitor` reports what would otherwise leave the pipeline silent: a non-retriable error on any table of the publication, and a source table etl has not moved in ten minutes while it had no other table left to copy. That report is fatal only for a table etl never took up, most often because the publication does not carry it. A table etl is working on may stand still for as long as a transaction runs on the source, because creating the copy slot waits for every transaction that was running when the copy started, so the monitor only warns about it, every ten minutes. The stall clock counts queued and flushed snapshot buffers as progress, because etl records no state transition inside a copy. The `select!` arm awaiting termination kept the state watch's read guard alive while awaiting etl shutdown; the reader's `Drop` then queued a second write on the watch, etl's copy tasks blocked re-reading it, and stop hung. The guard is dropped before the await. The preceding commit resolves an unqualified `source_table` in `public`; this commit only adds the changelog entry for that rule. Tests: the four CDC tests that were red for this bug run in CI again (test_cdc_ft_mode_holds_slot, test_snapshot_replayed_when_stopped_before_checkpoint, and test_checkpoint_mid_snapshot_waits_for_the_whole_copy in its single- and multi-worker forms), joined by test_suspend_mid_copy_is_refused, and test_cdc_restart_resumes_from_slot, the test that flaked on the fork revision, which passed ten consecutive runs against the new pin. Both mid-copy tests now measure a non-empty, strictly partial copy in the circuit while etl reports `data_sync` before they act, so a copy that finished early or a connector that died fails the test instead of passing it by accident. test_other_tables_in_publication_are_filtered asks etl whether both tables completed their sync, which catches a barrier answered wrongly for the table the connector does not read. Unit tests drive `CopyBarrier` through every call the destination and the reader make, `classify_copy_write`, the destination-to-barrier glue over a live queue, the two stop literals the rollback matches, and the `TimeTz` cells. The remaining CDC scenarios stay `#[ignore]`d, as on main. Docs: the connector page states when a table is read again, which starts keep the rows of the initial read, and what the connector installs in the source database and why that needs a superuser; the fault-tolerance table lists the PostgreSQL CDC input with checkpoint and resume and at- least-once fault tolerance but not exactly-once; the changelog records the barrier, the repeated read, the fixed flake and the second migration. Fixes #6121 Refs #7107

etl records the initial copy of a table as finished as soon as the connector acknowledges its rows, which is before Feldera has stepped them through the circuit, let alone checkpointed them. A checkpoint could therefore hold part of a copy, and a restart from it could hold none: etl streams from the replication slot once it considers the copy done, so the missing rows never arrive (#6121). The connector now holds etl's copy barrier until the circuit holds the copy. Upstream etl issues a terminal, empty `write_table_rows` once every copy worker of an attempt has finished, and records the copy as finished only when that call answers `Durable`. `FelderaDestination` answers each batch of copy rows with `Accepted`, counts the snapshot buffers it queues in a shared `CopyBarrier`, and parks the terminal call there until the reader's `Queue` flush has drained every one of them into the circuit. Streamed events share the input queue, so the barrier counts snapshot buffers rather than testing for an empty queue. While a copy is open the reader reports `Resume::Barrier`, so the controller defers a checkpoint or refuses a suspend instead of recording a state it could not finish on resume; once the copy is in the circuit it reports `Resume::Seek` with the pipeline id, because etl's `PostgresStore` holds the replication position. A copy that etl starts over, after a retry or a rollback, raises the barrier again from `drop_table_for_copy` or its first batch, before any row of it can reach the queue. A pending checkpoint stops the controller from stepping on its own, so the destination and the reader ask for the step that reports a lowered barrier. This is why the etl pin moves in the same change to 248fa407, an upstream supabase/etl commit synced into the feldera/etl fork, which keeps tracking upstream by hand: the terminal write, `DestinationWriteStatus` and the batch id that tells a batch from the barrier exist only there. The fork carried one commit, which selected the aws-lc-rs backend for sqlx TLS so etl does not link ring next to Feldera's aws-lc; upstream made the backend selectable, so `default- features = false` plus `tls-rustls-aws-lc-rs` reproduces it, and the old pin stays reachable on the fork's `sqlx-aws-lc-rs` branch. The etl- config and etl-postgres workspace entries go: no crate depends on them, and pinning them at the old revision would put two etl source revisions in the lockfile. The bump moves 175 commits and brings three things the connector has to follow. etl no longer has a SharedTableCache: SyncDone stores a durable decoder that relation-less DML restores, which fixes the `Missing shared table state` flake in test_cdc_restart_resumes_from_slot where the apply worker received a row event for a table whose schema the table-sync worker owned. etl reads the source schema through helper functions its own source migrations install, so the connector runs them; the second migration extends the DDL event trigger to `ALTER PUBLICATION` and needs the same privileges as the first. Cells gained `TimeTz`, which the connector renders in the text form Postgres uses, offset included, because Feldera has no time-with-time-zone type. etl's defaults also moved, and the connector follows them: copy batches of up to 32 MiB instead of 8, and four copy connections per table instead of two. Four more state-store migrations run on the source database; one retypes the snapshot_id columns, so an older Feldera cannot start against a database this version has migrated. Startup reconciles etl's stored state with what the circuit holds. `reconcile_table_state` rolls back the two errors a stop leaves behind, the one the connector records when Feldera stops it mid-batch and the one etl records for a deferred result the connector dropped, along with timed retries, whose deadline etl does not honor across a restart. A table whose history offers nothing to roll back to is reset to `Init` and read again rather than left stalled. Under fault tolerance with no checkpoint the circuit starts empty while etl may consider the copy done, so a completed sync is reset to `Init` too; this is the one case where a resumed pipeline could otherwise miss the copy, and no resume metadata is needed for the others. The function returns whether the table is still to be copied, so the barrier starts in the right position before etl runs. Upstream still persists an Errored state on shutdown, so the rollback stays until supabase/etl#881 merges. `TableErrorMonitor` reports what would otherwise leave the pipeline silent: a non-retriable error on any table of the publication, and a source table etl has not moved in ten minutes while it had no other table left to copy. That report is fatal only for a table etl never took up, most often because the publication does not carry it. A table etl is working on may stand still for as long as a transaction runs on the source, because creating the copy slot waits for every transaction that was running when the copy started, so the monitor only warns about it, every ten minutes. The stall clock counts queued and flushed snapshot buffers as progress, because etl records no state transition inside a copy. The `select!` arm awaiting termination kept the state watch's read guard alive while awaiting etl shutdown; the reader's `Drop` then queued a second write on the watch, etl's copy tasks blocked re-reading it, and stop hung. The guard is dropped before the await. The preceding commit resolves an unqualified `source_table` in `public`; this commit only adds the changelog entry for that rule. Tests: the four CDC tests that were red for this bug run in CI again (test_cdc_ft_mode_holds_slot, test_snapshot_replayed_when_stopped_before_checkpoint, and test_checkpoint_mid_snapshot_waits_for_the_whole_copy in its single- and multi-worker forms), joined by test_suspend_mid_copy_is_refused, and test_cdc_restart_resumes_from_slot, the test that flaked on the fork revision, which passed ten consecutive runs against the new pin. Both mid-copy tests measure a non-empty, strictly partial copy in the circuit while etl reports `data_sync` before they act; a connector that died fails the test, and a runner that copied the whole table before the poll caught it makes the test retry with a table four times larger, up to three attempts, so the precondition is about the connector and not the machine. test_other_tables_in_publication_are_filtered asks etl whether both tables completed their sync, which catches a barrier answered wrongly for the table the connector does not read. Unit tests drive `CopyBarrier` through every call the destination and the reader make, `classify_copy_write`, the destination-to-barrier glue over a live queue, the two stop literals the rollback matches, and the `TimeTz` cells; the all-types table gained TIMETZ and TIMETZ[] columns so the new cells round-trip through Postgres on both the copy and the streaming path. The remaining CDC scenarios stay `#[ignore]`d, as on main. Docs: the connector page states when a table is read again, which starts keep the rows of the initial read, and what the connector installs in the source database and why that needs a superuser; the fault-tolerance table lists the PostgreSQL CDC input with checkpoint and resume and at- least-once fault tolerance but not exactly-once; the changelog records the barrier, the repeated read, the fixed flake and the second migration. Fixes #6121 Refs #7107

etl records the initial copy of a table as finished as soon as the connector acknowledges its rows, which is before Feldera has stepped them through the circuit, let alone checkpointed them. A checkpoint could therefore hold part of a copy, and a restart from it could hold none: etl streams from the replication slot once it considers the copy done, so the missing rows never arrive (#6121). The connector now holds etl's copy barrier until the circuit holds the copy. Upstream etl issues a terminal, empty `write_table_rows` once every copy worker of an attempt has finished, and records the copy as finished only when that call answers `Durable`. `FelderaDestination` answers each batch of copy rows with `Accepted`, counts the snapshot buffers it queues in a shared `CopyBarrier`, and parks the terminal call there until the reader's `Queue` flush has drained every one of them into the circuit. Streamed events share the input queue, so the barrier counts snapshot buffers rather than testing for an empty queue. While a copy is open the reader reports `Resume::Barrier`, so the controller defers a checkpoint or refuses a suspend instead of recording a state it could not finish on resume; once the copy is in the circuit it reports `Resume::Seek` with the pipeline id, because etl's `PostgresStore` holds the replication position. A copy that etl starts over, after a retry or a rollback, raises the barrier again from `drop_table_for_copy` or its first batch, before any row of it can reach the queue. A pending checkpoint stops the controller from stepping on its own, so the destination and the reader ask for the step that reports a lowered barrier. This is why the etl pin moves in the same change to 248fa407, an upstream supabase/etl commit synced into the feldera/etl fork, which keeps tracking upstream by hand: the terminal write, `DestinationWriteStatus` and the batch id that tells a batch from the barrier exist only there. The fork carried one commit, which selected the aws-lc-rs backend for sqlx TLS so etl does not link ring next to Feldera's aws-lc; upstream made the backend selectable, so `default- features = false` plus `tls-rustls-aws-lc-rs` reproduces it, and the old pin stays reachable on the fork's `sqlx-aws-lc-rs` branch. The etl- config and etl-postgres workspace entries go: no crate depends on them, and pinning them at the old revision would put two etl source revisions in the lockfile. The bump moves 175 commits and brings three things the connector has to follow. etl no longer has a SharedTableCache: SyncDone stores a durable decoder that relation-less DML restores, which fixes the `Missing shared table state` flake in test_cdc_restart_resumes_from_slot where the apply worker received a row event for a table whose schema the table-sync worker owned. etl reads the source schema through helper functions its own source migrations install, so the connector runs them; the second migration extends the DDL event trigger to `ALTER PUBLICATION` and needs the same privileges as the first. Cells gained `TimeTz`, which the connector renders in the text form Postgres uses, offset included, because Feldera has no time-with-time-zone type. etl's defaults also moved, and the connector follows them: copy batches of up to 32 MiB instead of 8, and four copy connections per table instead of two. Four more state-store migrations run on the source database; one retypes the snapshot_id columns, so an older Feldera cannot start against a database this version has migrated. Startup reconciles etl's stored state with what the circuit holds. `reconcile_table_state` rolls back the two errors a stop leaves behind, the one the connector records when Feldera stops it mid-batch and the one etl records for a deferred result the connector dropped, along with timed retries, whose deadline etl does not honor across a restart. A table whose history offers nothing to roll back to is reset to `Init` and read again rather than left stalled. Under fault tolerance with no checkpoint the circuit starts empty while etl may consider the copy done, so a completed sync is reset to `Init` too; this is the one case where a resumed pipeline could otherwise miss the copy, and no resume metadata is needed for the others. The function returns whether the table is still to be copied, so the barrier starts in the right position before etl runs. Upstream still persists an Errored state on shutdown, so the rollback stays until supabase/etl#881 merges. A resume from a checkpoint has one more case to guard. etl skips a `sync_done` table's events below its sync LSN, because the copy at that LSN is meant to hold them; a checkpoint taken before a later copy of the table finished holds neither that copy nor those events, so resuming from it would lose the copy's rows for good. After answering a copy's terminal barrier the reader therefore keeps reporting a barrier until the monitor has seen etl record the copy as `sync_done` or `ready`, records the `sync_done` LSN, and writes it into every step's resume metadata as `copy_sync_lsn`. Startup reads the table again when etl's LSN is newer than the checkpoint's, or when the checkpoint carries no LSN while etl reports `sync_done`. `Ready` needs no such check: etl reaches it only once its persisted progress, which under fault tolerance advances only past checkpointed steps, has passed the LSN. The cost is a checkpoint deferred by about one monitor poll after each copy, and one extra copy after an upgrade for a table still in `sync_done`. `TableErrorMonitor` reports what would otherwise leave the pipeline silent: a non-retriable error on any table of the publication, and a source table etl has not moved in ten minutes while it had no other table left to copy. That report is fatal only for a table etl never took up, most often because the publication does not carry it. A table etl is working on may stand still for as long as a transaction runs on the source, because creating the copy slot waits for every transaction that was running when the copy started, so the monitor only warns about it, every ten minutes. The stall clock counts queued and flushed snapshot buffers as progress, because etl records no state transition inside a copy. The `select!` arm awaiting termination kept the state watch's read guard alive while awaiting etl shutdown; the reader's `Drop` then queued a second write on the watch, etl's copy tasks blocked re-reading it, and stop hung. The guard is dropped before the await. The preceding commit resolves an unqualified `source_table` in `public`; this commit only adds the changelog entry for that rule. Tests: the four CDC tests that were red for this bug run in CI again (test_cdc_ft_mode_holds_slot, test_snapshot_replayed_when_stopped_before_checkpoint, and test_checkpoint_mid_snapshot_waits_for_the_whole_copy in its single- and multi-worker forms), joined by test_suspend_mid_copy_is_refused, and test_cdc_restart_resumes_from_slot, the test that flaked on the fork revision, which passed ten consecutive runs against the new pin. Both mid-copy tests measure a non-empty, strictly partial copy in the circuit while etl reports `data_sync` before they act; a connector that died fails the test, and a runner that copied the whole table before the poll caught it makes the test retry with a table four times larger, up to three attempts, so the precondition is about the connector and not the machine. test_other_tables_in_publication_are_filtered asks etl whether both tables completed their sync, which catches a barrier answered wrongly for the table the connector does not read. Unit tests drive `CopyBarrier` through every call the destination and the reader make, `classify_copy_write`, the destination-to-barrier glue over a live queue, the two stop literals the rollback matches, and the `TimeTz` cells; the all-types table gained TIMETZ and TIMETZ[] columns so the new cells round-trip through Postgres on both the copy and the streaming path. The remaining CDC scenarios stay `#[ignore]`d, as on main. Docs: the connector page states when a table is read again, which starts keep the rows of the initial read, and what the connector installs in the source database and why that needs a superuser; the fault-tolerance table lists the PostgreSQL CDC input with checkpoint and resume and at- least-once fault tolerance but not exactly-once; the changelog records the barrier, the repeated read, the fixed flake and the second migration. Fixes #6121 Refs #7107

etl records the initial copy of a table as finished as soon as the connector acknowledges its rows, which is before Feldera has stepped them through the circuit, let alone checkpointed them. A checkpoint could therefore hold part of a copy, and a restart from it could hold none: etl streams from the replication slot once it considers the copy done, so the missing rows never arrive (#6121). The connector now holds etl's copy barrier until the circuit holds the copy. Upstream etl issues a terminal, empty `write_table_rows` once every copy worker of an attempt has finished, and records the copy as finished only when that call answers `Durable`. `FelderaDestination` answers each batch of copy rows with `Accepted`, counts the snapshot buffers it queues in a shared `CopyBarrier`, and parks the terminal call there until the reader's `Queue` flush has drained every one of them into the circuit. Streamed events share the input queue, so the barrier counts snapshot buffers rather than testing for an empty queue. While a copy is open the reader reports `Resume::Barrier`, so the controller defers a checkpoint or refuses a suspend instead of recording a state it could not finish on resume; once the copy is in the circuit it reports `Resume::Seek` with the pipeline id, because etl's `PostgresStore` holds the replication position. A copy that etl starts over, after a retry or a rollback, raises the barrier again from `drop_table_for_copy` or its first batch, before any row of it can reach the queue. A pending checkpoint stops the controller from stepping on its own, so the destination and the reader ask for the step that reports a lowered barrier. This is why the etl pin moves in the same change to 248fa407, an upstream supabase/etl commit synced into the feldera/etl fork, which keeps tracking upstream by hand: the terminal write, `DestinationWriteStatus` and the batch id that tells a batch from the barrier exist only there. The fork carried one commit, which selected the aws-lc-rs backend for sqlx TLS so etl does not link ring next to Feldera's aws-lc; upstream made the backend selectable, so `default- features = false` plus `tls-rustls-aws-lc-rs` reproduces it, and the old pin stays reachable on the fork's `sqlx-aws-lc-rs` branch. The etl- config and etl-postgres workspace entries go: no crate depends on them, and pinning them at the old revision would put two etl source revisions in the lockfile. The bump moves 175 commits and brings three things the connector has to follow. etl no longer has a SharedTableCache: SyncDone stores a durable decoder that relation-less DML restores, which fixes the `Missing shared table state` flake in test_cdc_restart_resumes_from_slot where the apply worker received a row event for a table whose schema the table-sync worker owned. etl reads the source schema through helper functions its own source migrations install, so the connector runs them; the second migration extends the DDL event trigger to `ALTER PUBLICATION` and needs the same privileges as the first. Cells gained `TimeTz`, which the connector renders in the text form Postgres uses, offset included, because Feldera has no time-with-time-zone type. etl's defaults also moved, and the connector follows them: copy batches of up to 32 MiB instead of 8, and four copy connections per table instead of two. Four more state-store migrations run on the source database; one retypes the snapshot_id columns, so an older Feldera cannot start against a database this version has migrated. Startup reconciles etl's stored state with what the circuit holds. `reconcile_table_state` rolls back the two errors a stop leaves behind, the one the connector records when Feldera stops it mid-batch and the one etl records for a deferred result the connector dropped, along with timed retries, whose deadline etl does not honor across a restart. A table whose history offers nothing to roll back to is reset to `Init` and read again rather than left stalled. Under fault tolerance with no checkpoint the circuit starts empty while etl may consider the copy done, so a completed sync is reset to `Init` too; this is the one case where a resumed pipeline could otherwise miss the copy, and no resume metadata is needed for the others. The function returns whether the table is still to be copied, so the barrier starts in the right position before etl runs. Upstream still persists an Errored state on shutdown, so the rollback stays until supabase/etl#881 merges. A resume from a checkpoint has one more case to guard. etl skips a `sync_done` table's events below its sync LSN, because the copy at that LSN is meant to hold them; a checkpoint taken before a later copy of the table finished holds neither that copy nor those events, so resuming from it would lose the copy's rows for good. After answering a copy's terminal barrier the reader therefore keeps reporting a barrier until the monitor has seen etl record the copy as `sync_done` or `ready`, records the `sync_done` LSN, and writes it into every step's resume metadata as `copy_sync_lsn`. Startup reads the table again when etl's LSN is newer than the checkpoint's, or when the checkpoint carries no LSN while etl reports `sync_done`. `Ready` needs no such check: etl reaches it only once its persisted progress, which under fault tolerance advances only past checkpointed steps, has passed the LSN. The cost is a checkpoint deferred by about one monitor poll after each copy, and one extra copy after an upgrade for a table still in `sync_done`. `TableErrorMonitor` reports what would otherwise leave the pipeline silent: a non-retriable error on any table of the publication, and a source table etl has not moved in ten minutes while it had no other table left to copy. That report is fatal only for a table etl never took up, most often because the publication does not carry it. A table etl is working on may stand still for as long as a transaction runs on the source, because creating the copy slot waits for every transaction that was running when the copy started, so the monitor only warns about it, every ten minutes. The stall clock counts queued and flushed snapshot buffers as progress, because etl records no state transition inside a copy. Before etl starts, the connector lists the publication's tables from `pg_publication_tables` and refuses to start when the source table is not among them, naming the tables it found. An unqualified name that used to match a table outside `public` is the likeliest way to hit this, and a start-time refusal beats ten silent minutes and a stall report. The stall report stays as the backstop for a table etl never takes up. The `select!` arm awaiting termination kept the state watch's read guard alive while awaiting etl shutdown; the reader's `Drop` then queued a second write on the watch, etl's copy tasks blocked re-reading it, and stop hung. The guard is dropped before the await. The preceding commit resolves an unqualified `source_table` in `public`; this commit only adds the changelog entry for that rule. Tests: the four CDC tests that were red for this bug run in CI again (test_cdc_ft_mode_holds_slot, test_snapshot_replayed_when_stopped_before_checkpoint, and test_checkpoint_mid_snapshot_waits_for_the_whole_copy in its single- and multi-worker forms), joined by test_suspend_mid_copy_is_refused, and test_cdc_restart_resumes_from_slot, the test that flaked on the fork revision, which passed ten consecutive runs against the new pin. Both mid-copy tests measure a non-empty, strictly partial copy in the circuit while etl reports `data_sync` before they act; a connector that died fails the test, and a runner that copied the whole table before the poll caught it makes the test retry with a table four times larger, up to three attempts, so the precondition is about the connector and not the machine. test_other_tables_in_publication_are_filtered asks etl whether both tables completed their sync, which catches a barrier answered wrongly for the table the connector does not read. Unit tests drive `CopyBarrier` through every call the destination and the reader make, `classify_copy_write`, the destination-to-barrier glue over a live queue, the two stop literals the rollback matches, and the `TimeTz` cells; the all-types table gained TIMETZ and TIMETZ[] columns so the new cells round-trip through Postgres on both the copy and the streaming path. The remaining CDC scenarios stay `#[ignore]`d, as on main. Docs: the connector page states when a table is read again, which starts keep the rows of the initial read, and what the connector installs in the source database and why that needs a superuser; the fault-tolerance table lists the PostgreSQL CDC input with checkpoint and resume and at- least-once fault tolerance but not exactly-once; the changelog records the barrier, the repeated read, the fixed flake and the second migration. Fixes #6121 Refs #7107

Add `AsyncResult::shutdown`, which a destination calls instead of `send` for work it abandons because it is shutting down, for example together with the process that embeds ETL. ETL requests pipeline shutdown when it receives that outcome, so the pipeline stops through the same path an external `ShutdownTx::shutdown` takes and records no table error: the next start resumes the table from its persisted state. Neither side has to come first. `PendingAsyncResult::with_shutdown` already returned early on the pipeline's own signal; it now also converts a reported shutdown into the same result, and the apply loop does the same for the streaming write it was waiting on. To let a worker request what it observes, `ShutdownRx` becomes `Shutdown`, a handle that carries its transmitter alongside the receiver. Its `is_requested` replaces the private helper of the table copy. Before this, a destination could only fail such work, which raced the shutdown signal. When the failure won, the table sync worker stored an `Errored` state with a manual retry policy and the table stalled on every later start, although a restart could simply copy it again. Document the contract on `Destination` and in the custom destination guide, and cover the copy and streaming paths, plus the handle's own behavior, with tests. Continues supabase#881.

Add `AsyncResult::shutdown`, which a destination calls instead of `send` for work it abandons because it is shutting down, for example together with the process that embeds ETL. ETL requests pipeline shutdown when it receives that outcome, so the pipeline stops through the same path an external `ShutdownTx::shutdown` takes and records no table error: the next start resumes the table from its persisted state. Neither side has to come first. `PendingAsyncResult::with_shutdown` already returned early on the pipeline's own signal; it now also converts a reported shutdown into the same result, and the apply loop does the same for the streaming write it was waiting on. An abandoned streaming write also forces the loop to exit as paused. A completion intent is recorded before the write that has to make it durable, so an abandoned terminal durability barrier would otherwise complete a table sync short of `SyncDone`, and the worker would persist an error that only a manual retry clears. To let a worker request what it observes, `ShutdownRx` becomes `Shutdown`, a handle that carries its transmitter alongside the receiver. Its `is_requested` replaces the private helper of the table copy. Before this, a destination could only fail such work, which raced the shutdown signal. When the failure won, the table sync worker stored an `Errored` state with a manual retry policy and the table stalled on every later start, although a restart could simply copy it again. Document the contract on `Destination` and in the custom destination guide, and cover the copy, streaming, and durability-barrier paths, plus the handle's own behavior, with tests. Continues supabase#881.

Add `AsyncResult::shutdown`, which a destination calls instead of `send` for work it abandons because it is shutting down, for example together with the process that embeds ETL. ETL requests pipeline shutdown when it receives that outcome, so the pipeline stops through the same path an external `ShutdownTx::shutdown` takes and records no table error: the next start resumes the table from its persisted state. Neither side has to come first. `PendingAsyncResult::with_shutdown` already returned early on the pipeline's own signal; it now also converts a reported shutdown into the same result, and the apply loop does the same for the streaming write it was waiting on. An abandoned streaming write stops that loop for good. It drops the batch queued behind the write, because a durable result for that successor would report progress covering events the destination never received, and it pins the exit to paused, because a completion intent is recorded before the write that has to make it durable. Completing instead would end a table sync short of `SyncDone`, and the worker would persist an error that only a manual retry clears. To let a worker request what it observes, `ShutdownRx` becomes `Shutdown`, a handle that carries its transmitter alongside the receiver. Its `is_requested` replaces the private helper of the table copy. Because a handle owns a transmitter, the signal channel no longer loses its last one when the pipeline goes away, so the pipeline holds a guard that requests shutdown when it is dropped, taking over what closing the channel used to do for workers that only wait. Before this, a destination could only fail such work, which raced the shutdown signal. When the failure won, the table sync worker stored an `Errored` state with a manual retry policy and the table stalled on every later start, although a restart could simply copy it again. Document the contract on `Destination` and in the custom destination guide, and cover the copy, streaming and durability-barrier paths, the abandoned successor batch, and the shutdown handle itself, with tests. Continues supabase#881.

etl records the initial copy of a table as finished as soon as the connector acknowledges its rows, which is before Feldera has stepped them through the circuit, let alone checkpointed them. A checkpoint could therefore hold part of a copy, and a restart from it could hold none: etl streams from the replication slot once it considers the copy done, so the missing rows never arrive (feldera#6121). The connector now holds etl's copy barrier until the circuit holds the copy. Upstream etl issues a terminal, empty `write_table_rows` once every copy worker of an attempt has finished, and records the copy as finished only when that call answers `Durable`. `FelderaDestination` answers each batch of copy rows with `Accepted`, counts the snapshot buffers it queues in a shared `CopyBarrier`, and parks the terminal call there until the reader's `Queue` flush has drained every one of them into the circuit. Streamed events share the input queue, so the barrier counts snapshot buffers rather than testing for an empty queue. While a copy is open the reader reports `Resume::Barrier`, so the controller defers a checkpoint or refuses a suspend instead of recording a state it could not finish on resume; once the copy is in the circuit it reports `Resume::Seek` with the pipeline id, because etl's `PostgresStore` holds the replication position. A copy that etl starts over, after a retry or a rollback, raises the barrier again from `drop_table_for_copy` or its first batch, before any row of it can reach the queue. A pending checkpoint stops the controller from stepping on its own, so the destination and the reader ask for the step that reports a lowered barrier. This is why the etl pin moves in the same change to 248fa407, an upstream supabase/etl commit synced into the feldera/etl fork, which keeps tracking upstream by hand: the terminal write, `DestinationWriteStatus` and the batch id that tells a batch from the barrier exist only there. The fork carried one commit, which selected the aws-lc-rs backend for sqlx TLS so etl does not link ring next to Feldera's aws-lc; upstream made the backend selectable, so `default- features = false` plus `tls-rustls-aws-lc-rs` reproduces it, and the old pin stays reachable on the fork's `sqlx-aws-lc-rs` branch. The etl- config and etl-postgres workspace entries go: no crate depends on them, and pinning them at the old revision would put two etl source revisions in the lockfile. The bump moves 175 commits and brings three things the connector has to follow. etl no longer has a SharedTableCache: SyncDone stores a durable decoder that relation-less DML restores, which fixes the `Missing shared table state` flake in test_cdc_restart_resumes_from_slot where the apply worker received a row event for a table whose schema the table-sync worker owned. etl reads the source schema through helper functions its own source migrations install, so the connector runs them; the second migration extends the DDL event trigger to `ALTER PUBLICATION` and needs the same privileges as the first. Cells gained `TimeTz`, which the connector renders in the text form Postgres uses, offset included, because Feldera has no time-with-time-zone type. etl's defaults also moved, and the connector follows them: copy batches of up to 32 MiB instead of 8, and four copy connections per table instead of two. Four more state-store migrations run on the source database; one retypes the snapshot_id columns, so an older Feldera cannot start against a database this version has migrated. Startup reconciles etl's stored state with what the circuit holds. `reconcile_table_state` rolls back the two errors a stop leaves behind, the one the connector records when Feldera stops it mid-batch and the one etl records for a deferred result the connector dropped, along with timed retries, whose deadline etl does not honor across a restart. A table whose history offers nothing to roll back to is reset to `Init` and read again rather than left stalled. Under fault tolerance with no checkpoint the circuit starts empty while etl may consider the copy done, so a completed sync is reset to `Init` too; this is the one case where a resumed pipeline could otherwise miss the copy, and no resume metadata is needed for the others. The function returns whether the table is still to be copied, so the barrier starts in the right position before etl runs. Upstream still persists an Errored state on shutdown, so the rollback stays until supabase/etl#881 merges. A resume from a checkpoint has one more case to guard. etl skips a `sync_done` table's events below its sync LSN, because the copy at that LSN is meant to hold them; a checkpoint taken before a later copy of the table finished holds neither that copy nor those events, so resuming from it would lose the copy's rows for good. After answering a copy's terminal barrier the reader therefore keeps reporting a barrier until the monitor has seen etl record the copy as `sync_done` or `ready`, records the `sync_done` LSN, and writes it into every step's resume metadata as `copy_sync_lsn`. Startup reads the table again when etl's LSN is newer than the checkpoint's, or when the checkpoint carries no LSN while etl reports `sync_done`. `Ready` needs no such check: etl reaches it only once its persisted progress, which under fault tolerance advances only past checkpointed steps, has passed the LSN. The cost is a checkpoint deferred by about one monitor poll after each copy, and one extra copy after an upgrade for a table still in `sync_done`. `TableErrorMonitor` reports what would otherwise leave the pipeline silent: a non-retriable error on any table of the publication, and a source table etl has not moved in ten minutes while it had no other table left to copy. That report is fatal only for a table etl never took up, most often because the publication does not carry it. A table etl is working on may stand still for as long as a transaction runs on the source, because creating the copy slot waits for every transaction that was running when the copy started, so the monitor only warns about it, every ten minutes. The stall clock counts queued and flushed snapshot buffers as progress, because etl records no state transition inside a copy. Before etl starts, the connector lists the publication's tables from `pg_publication_tables` and refuses to start when the source table is not among them, naming the tables it found. An unqualified name that used to match a table outside `public` is the likeliest way to hit this, and a start-time refusal beats ten silent minutes and a stall report. The stall report stays as the backstop for a table etl never takes up. The `select!` arm awaiting termination kept the state watch's read guard alive while awaiting etl shutdown; the reader's `Drop` then queued a second write on the watch, etl's copy tasks blocked re-reading it, and stop hung. The guard is dropped before the await. The preceding commit resolves an unqualified `source_table` in `public`; this commit only adds the changelog entry for that rule. Tests: the four CDC tests that were red for this bug run in CI again (test_cdc_ft_mode_holds_slot, test_snapshot_replayed_when_stopped_before_checkpoint, and test_checkpoint_mid_snapshot_waits_for_the_whole_copy in its single- and multi-worker forms), joined by test_suspend_mid_copy_is_refused, and test_cdc_restart_resumes_from_slot, the test that flaked on the fork revision, which passed ten consecutive runs against the new pin. Both mid-copy tests measure a non-empty, strictly partial copy in the circuit while etl reports `data_sync` before they act; a connector that died fails the test, and a runner that copied the whole table before the poll caught it makes the test retry with a table four times larger, up to three attempts, so the precondition is about the connector and not the machine. test_other_tables_in_publication_are_filtered asks etl whether both tables completed their sync, which catches a barrier answered wrongly for the table the connector does not read. Unit tests drive `CopyBarrier` through every call the destination and the reader make, `classify_copy_write`, the destination-to-barrier glue over a live queue, the two stop literals the rollback matches, and the `TimeTz` cells; the all-types table gained TIMETZ and TIMETZ[] columns so the new cells round-trip through Postgres on both the copy and the streaming path. The remaining CDC scenarios stay `#[ignore]`d, as on main. Docs: the connector page states when a table is read again, which starts keep the rows of the initial read, and what the connector installs in the source database and why that needs a superuser; the fault-tolerance table lists the PostgreSQL CDC input with checkpoint and resume and at- least-once fault tolerance but not exactly-once; the changelog records the barrier, the repeated read, the fixed flake and the second migration. Fixes feldera#6121 Refs feldera#7107

abhizer requested a review from a team as a code owner July 7, 2026 14:39

abhizer force-pushed the fix/table-copy-shutdown-write-race branch from 5a9d35f to 6d33c24 Compare July 13, 2026 09:25

iambriccardo reviewed Jul 15, 2026

View reviewed changes

bestouff mentioned this pull request Sep 10, 2026

Postgres CDC: land upstream etl PR 881 and drop the shutdown-error workaround feldera/feldera#7108

Open

This was referenced Sep 15, 2026

Postgres CDC: startup rollback depends on etl's dropped-result message text feldera/feldera#7139

Open

feat(core): Let destinations report shutdown without table errors #1038

Open

Uh oh!

Conversation

abhizer commented Jul 7, 2026

What kind of change does this PR introduce?

What is the current behavior?

What is the new behavior?

Uh oh!

iambriccardo commented Jul 10, 2026

Uh oh!

abhizer commented Jul 12, 2026

Uh oh!

bnjjj commented Jul 15, 2026

Uh oh!

coveralls commented Jul 15, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

abhizer commented Jul 15, 2026

Uh oh!

iambriccardo commented Jul 15, 2026

Uh oh!

bnjjj commented Jul 15, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants