Skip to content

Consumer-controlled start: initial ack position, start confirmation, application_name, gated sessions - #15

Merged
vnvo merged 7 commits into
mainfrom
consumer-control
Oct 19, 2026
Merged

vnvo merged 7 commits into
mainfrom
consumer-control

Conversation

@vnvo

@vnvo vnvo commented Oct 13, 2026 •

Copy link
Copy Markdown
Owner

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; ReplicationConfig is #[non_exhaustive], so the new fields are not breaking.

  • ack_lsn / with_ack_lsn: the LSN reported as flushed and applied until the first update_applied_lsn (default start_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 accepted START_REPLICATION, when the slot is held by the session, or the worker's error if it ended first.
  • application_name / with_application_name: the startup application_name (default pgwire-replication), so a consumer can find its own walsender in pg_stat_activity.
  • Gated sessions: connect_gated authenticates and waits. simple_query, identify_system and timeline_history run on that session until start sends START_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 before start sends Terminate. Query rows are bounded by max_message_size (feat: configurable maximum replication message size #11) like the replication stream: a larger row is refused with MessageTooLarge and ends the session.
  • Docs: status_interval documented 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 without ack_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-checks against 0.4.1: 202 checks pass. It first found that ReplicationConfig (the observer from Observe how long acknowledged positions wait before the server hears of them #14) and ReplicationClient (a oneshot start signal) had lost UnwindSafe/RefUnwindSafe; both are fixed, and tests/auto_traits.rs guards 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, and cargo audit on a fresh resolve.

@vnvo
vnvo force-pushed the feedback-observer branch from eb25c0f to eae8c82 Compare October 18, 2026 21:44
@vnvo
vnvo force-pushed the consumer-control branch from e7697a3 to 33ee48b Compare October 18, 2026 21:48
@vnvo
vnvo force-pushed the feedback-observer branch from eae8c82 to 2a38baf Compare October 18, 2026 21:58
vnvo added 7 commits October 19, 2026 03:41
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
vnvo changed the base branch from feedback-observer to main October 19, 2026 00:43
@vnvo
vnvo force-pushed the consumer-control branch from 33ee48b to 7828e8b Compare October 19, 2026 00:43
@vnvo
vnvo merged commit 10458d1 into main Oct 19, 2026
4 checks passed
@vnvo vnvo mentioned this pull request Oct 19, 2026
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant