Repository navigation
Consumer-controlled start: initial ack position, start confirmation, application_name, gated sessions - #15
Merged
Merged
Conversation
vnvo
force-pushed
the
feedback-observer
branch
from
October 18, 2026 21:44
eb25c0f to
eae8c82
Compare
vnvo
force-pushed
the
consumer-control
branch
from
October 18, 2026 21:48
e7697a3 to
33ee48b
Compare
vnvo
force-pushed
the
feedback-observer
branch
from
October 18, 2026 21:58
eae8c82 to
2a38baf
Compare
ReplicationConfig::ack_lsn sets the LSN reported as flushed and applied before update_applied_lsn is called (default: start_lsn). A consumer that resumes reading ahead of its durable position keeps the slot at that position.
ReplicationClient::wait_started returns once the slot is held by the session, or the worker's error if it ended first.
ReplicationClient::connect_gated authenticates and waits: simple_query, identify_system and timeline_history run on that session until start sends START_REPLICATION.
A oneshot receiver for the start signal made ReplicationClient lose UnwindSafe and RefUnwindSafe; use a watch channel, like the stop signal.
A gated session's simple_query read rows with the default 1 GiB reader, so a configured limit did not apply to them. A larger row is now refused with MessageTooLarge and ends the session, like the replication stream.
vnvo
force-pushed
the
consumer-control
branch
from
October 19, 2026 00:43
33ee48b to
7828e8b
Compare
Merged
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
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.
Stacked on #14. Rebased onto main with #11: the new config fields sit alongside
max_message_size.Four additive features for consumers that manage their own durability and failover, and a documentation fix. No existing signature changes;
ReplicationConfigis#[non_exhaustive], so the new fields are not breaking.ack_lsn/with_ack_lsn: the LSN reported as flushed and applied until the firstupdate_applied_lsn(defaultstart_lsn, as before). A consumer that resumes reading ahead of its durable position keeps the slot at that position.ReplicationClient::wait_started: returns once the server acceptedSTART_REPLICATION, when the slot is held by the session, or the worker's error if it ended first.application_name/with_application_name: the startupapplication_name(defaultpgwire-replication), so a consumer can find its own walsender inpg_stat_activity.connect_gatedauthenticates and waits.simple_query,identify_systemandtimeline_historyrun on that session untilstartsendsSTART_REPLICATION, so what is read before the start describes the server the stream opens on. A server error leaves the session usable; dropping the client beforestartsends Terminate. Query rows are bounded bymax_message_size(feat: configurable maximum replication message size #11) like the replication stream: a larger row is refused withMessageTooLargeand ends the session.status_intervaldocumented a 1 second default; it is 10 seconds. The README describes change-driven feedback on an idle stream.Tests
Unit: the gate protocol (answers, a server error, a closed gate, a broken session, a malformed DataRow) and the startup
application_name.Integration (PostgreSQL 16):
postgres_ack_lsn_holds_the_slot, with a control showing the slot advances withoutack_lsn;postgres_wait_started_holds_the_slot, including a missing slot;postgres_gated_session.Each was checked to fail against a mutation of the feature.
cargo semver-checksagainst 0.4.1: 202 checks pass. It first found thatReplicationConfig(the observer from Observe how long acknowledged positions wait before the server hears of them #14) andReplicationClient(a oneshot start signal) had lostUnwindSafe/RefUnwindSafe; both are fixed, andtests/auto_traits.rsguards the public types' auto traits.fmt, clippy (default and all features, also on Rust 1.99), tests, doc tests, docs with
-D warnings, the loom model, andcargo auditon a fresh resolve.