postgres replication stream does not get refresh

[Sh4d1]

Hey!

It seems that in etl/src/replication/apply.rs, in start_apply_loop, when we use the biased tokio select, that the replication stream never gets update if there is a continuous stream of replication events, as the priority 4 branch will never be executed, and then it fails on a postgres error (I hit the wal timeout in my case).

I added a manual send_status_update (if last one was sent after the REFRESH_INTERVAL) in the priority 2 branch, and it seems to fix the issue for me.

[iambriccardo]

Thanks for reporting this!

Postgres is generally expected to send status update messages along the main replication stream. It’s quite confusing that it isn’t doing so in your case. Could you add debug statements to the logs to check whether Postgres is actually sending status update messages in the main stream?

[Sh4d1]

It’s quite confusing that it isn’t doing so in your case

It's quite logic when you see the code! Since it uses the tokio biased select, if there is always a message to read on the replication stream here

            Some(message) = logical_replication_stream.next() => {

then the branch below:

            _ = tokio::time::sleep(REFRESH_INTERVAL) => {

will never gets executed, and never sending updates to the stream

Or am i missing something here?

[Sh4d1]

Sh4d1@899aa3c here is my fix that seems to fix the issue for me

[iambriccardo]

This is the line I was talking about:

etl/etl/src/replication/apply.rs

Line 915 in 51a5a24

ReplicationMessage::PrimaryKeepAlive(message) => {

[iambriccardo]

Postgres should request status updates as part of the logical replication stream. The last branch in the select exists to handle cases when the system is idle.

[Sh4d1]

Aaah yes ok! So it would seem it only receives the XLogData events and not the PrimaryKeepAlive 🤔 I'll try to add some more logs and reproduce the issue!

[Sh4d1]

There could be the case where a log of XLogData are processed before any keepalive message then? Anyway I'll try to gather more info, thanks!

[iambriccardo]

I need to look at the Postgres source code, will report back!

[iambriccardo]

@Sh4d1 what would be really helpful, as I mentioned above, is to have a log showing when keep-alive messages are returned. My only guess is that there is too much time between the keep-alive messages, even in the main message stream.

Could you also share the error you’re getting?

[Sh4d1]

Yep I'm currently deploying a version with some logs (would need to start again from scratch, initial sync takes ~ 45min), so I might have an updated later tonight!

As for the error, here's what I get:

{"type": "errored", "reason": "[SourceConnectionFailed] Postgres connection failed @ ./external/+crates_server+crates_server__etl-0.1.0/src/replication/stream.rs:170:9\n  Detail:\n    connection closed\n  Backtrace:\n
   0: <etl::error::EtlError as core::convert::From<tokio_postgres::error::Error>>::from\n       1: <etl::replication::stream::EventsStream>::send_status_update::{closure#0}\n       2: <etl::workers::table_sync::TableSyncWorker<etl::store::both::postgres::PostgresStore
, postgres_etl::destination::redpanda::RedpandaDestination>>::run_table_sync_worker::{closure#0}\n       3: <etl::workers::table_sync::TableSyncWorker<etl::store::both::postgres::PostgresStore, postgres_etl::destination::redpanda::RedpandaDestination>>::guarded_run_ta
ble_sync_worker::{closure#0}\n       4: <tracing::instrument::Instrumented<<etl::workers::table_sync::TableSyncWorker<etl::store::both::postgres::PostgresStore, postgres_etl::destination::redpanda::RedpandaDestination>>::guarded_run_table_sync_worker::{closure#0}> as
core::future::future::Future>::poll\n       5: <tracing::instrument::Instrumented<tracing::instrument::Instrumented<<etl::workers::table_sync::TableSyncWorker<etl::store::both::postgres::PostgresStore, postgres_etl::destination::redpanda::RedpandaDestination>>::guarde
d_run_table_sync_worker::{closure#0}>> as core::future::future::Future>::poll\n       6: <tokio::runtime::task::harness::Harness<tracing::instrument::Instrumented<tracing::instrument::Instrumented<<etl::workers::table_sync::TableSyncWorker<etl::store::both::postgres::
PostgresStore, postgres_etl::destination::redpanda::RedpandaDestination>>::guarded_run_table_sync_worker::{closure#0}>>, alloc::sync::Arc<tokio::runtime::scheduler::multi_thread::handle::Handle>>>::poll\n       7: <tokio::runtime::scheduler::multi_thread::worker::Cont
ext>::run_task\n       8: <std::thread::local::LocalKey<tokio::runtime::context::Context>>::with::<tokio::runtime::context::set_scheduler<(), tokio::runtime::scheduler::multi_thread::worker::run::{closure#0}::{closure#0}>::{closure#0}, ()>\n       9: tokio::runtime::c
ontext::runtime::enter_runtime::<tokio::runtime::scheduler::multi_thread::worker::run::{closure#0}, ()>\n      10: tokio::runtime::scheduler::multi_thread::worker::run\n      11: <tracing::instrument::Instrumented<tokio::runtime::blocking::task::BlockingTask<<tokio::r
untime::scheduler::multi_thread::worker::Launch>::launch::{closure#0}>> as core::future::future::Future>::poll\n      12: <tokio::runtime::task::core::Core<tracing::instrument::Instrumented<tokio::runtime::blocking::task::BlockingTask<<tokio::runtime::scheduler::multi
_thread::worker::Launch>::launch::{closure#0}>>, tokio::runtime::blocking::schedule::BlockingSchedule>>::poll\n      13: <tokio::runtime::task::harness::Harness<tracing::instrument::Instrumented<tokio::runtime::blocking::task::BlockingTask<<tokio::runtime::scheduler::
multi_thread::worker::Launch>::launch::{closure#0}>>, tokio::runtime::blocking::schedule::BlockingSchedule>>::poll\n      14: <tokio::runtime::task::UnownedTask<tokio::runtime::blocking::schedule::BlockingSchedule>>::run\n      15: <tokio::runtime::blocking::pool::Inn
er>::run\n      16: std::sys::backtrace::__rust_begin_short_backtrace::<<tokio::runtime::blocking::pool::Spawner>::spawn_thread::{closure#0}, ()>\n      17: <<std::thread::Builder>::spawn_unchecked_<<tokio::runtime::blocking::pool::Spawner>::spawn_thread::{closure#0},
 ()>::{closure#1} as core::ops::function::FnOnce<()>>::call_once::{shim:vtable#0}\n      18: std::sys::pal::unix::thread::Thread::new::thread_start\n      19: <unknown>\n      20: <unknown>", "solution": null, "retry_policy": {"type": "timed_retry", "next_retry": "202
5-10-14T14:49:57.905927034Z"}}

What's weird, is that I got it after the initial replication, and when it happens, it just stops, the next retry is in the past, and it only gets written to the etl.replication_state when I shutdown the pod

[iambriccardo]

If we realize that this is the problem, we can implement a solution similar to what you suggested. Another possibility is that the batch takes so long to process that we handle many elements and time out before processing the keepalive from PostgreSQL.

Could you show me your config params:

• wal_sender_timeout
• wal_receiver_timeout

[Sh4d1]

Both are set to 1min for me!

[iambriccardo]

Great, these are the defaults then.

What is your batch config?