Keep the replication connection single-threaded#92
Open
lukashes wants to merge 3 commits into
Open
Conversation
The flush/commit worker ran on its own thread and called both producer.flush() (Kafka) and source.sendFeedback() (libpq) on the same PGconn the main receive loop reads from. libpq forbids using one connection from two threads; under load the race wedged the connection, so a Kafka queue-full stall hung the graceful teardown in PQflush until SIGINT (EINTR) broke it loose, instead of failing fast. Keep the worker for the slow Kafka flush only: it records how far it has flushed in an atomic flushed_lsn and never touches libpq. The main thread sends standby feedback -- receiveBatch now takes the confirmed LSN and reports it before reading more WAL, so feedback rides the same single-threaded replication loop as the read, like the native walreceiver. Graceful shutdown confirms the worker's final flush. The flush interval becomes a Processor field (default from the constant). Add an e2e test asserting confirmed_flush_lsn advances when a receive finds no new changes.
Contributor
📊 Benchmark ResultsCurrent run is the minimum over 3 passes, compared against the base branch (
Summary: ➡️ 11 neutral · ⚪ 2 ignored (sub-μs) Thresholds: <1μs ignore · 1–20μs 15% · 20–50μs 10% · ≥50μs 5%. Measured on a shared CI runner — treat small deltas as noise. Informational only; this check never fails the build. |
Owner
Author
|
/bench |
Two failures surfaced under load: Postgres killed the walsender with 'terminating walsender due to replication timeout', and outboxx then hung in teardown instead of exiting, so it never restarted. Standby feedback only went out from receiveBatch when flushed_lsn advanced, and keepalives were never answered. During a startup backlog, where the worker's first successful flush + feedback did not land within wal_sender_timeout (60s), no status update was sent and Postgres dropped the connection. Now reply to a keepalive's reply_requested with the last durably-confirmed position (last_feedback_lsn, never the received LSN, so at-least-once holds), which resets the timeout even when flushed_lsn is not advancing yet. The teardown hung because errdefer flush_future.cancel awaited the worker's final producer.flush() into a wedged/backlogged broker. Drop the worker's final flush so cancel returns promptly; do the final flush on the main thread in the graceful path and confirm pending_lsn (durable after a full flush). On a fatal error the process now exits fast instead of hanging forever.
Owner
Author
|
/bench |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Problem
The flush/commit worker ran on its own thread (
io.concurrentunderstd.Io.Threaded) and called bothproducer.flush()(Kafka) andsource.sendFeedback()(libpq) on the samePGconnthe main receive loop reads from viaPQgetCopyData. libpq forbids using one connection from two threads.Under load the race wedged the connection. When Kafka's local queue filled (
Local: Queue full), the produce error propagated into the graceful teardown (errdefer flush_future.cancel(io)), which awaits the worker's finalsendFeedback— aPQflushthat blocked inselect()on the corrupted connection with no timeout. The process hung instead of failing fast; onlySIGINT(EINTR) unblocked it, which is why a manual restart was needed and noexit(1)fired on its own.Solution
Keep all libpq access on the receive thread:
flushed_lsn. It never touches libpq.receiveBatchnow takes the confirmed LSN and reports it to Postgres before reading more WAL, so feedback rides the same single-threaded replication loop as the read — the way the native walreceiver works.commitFlushedis gone.Processorfield (default from the constant).Test
New e2e test: after a batch is received and flushed, a subsequent receive with no new changes still advances the slot's
confirmed_flush_lsn— the idle-commit guarantee.Closes #81