Harden replication transport lifecycle and restart recovery

- add socket-aware stream polling
- normalize idle get_copy_data responses
- improve reconnect and recovery behavior
- add PostgreSQL restart recovery E2E coverage
- fix replication timeout recovery edge cases
- improve E2E startup readiness checks
- harden replication slot lifecycle handling
- expand YARD documentation coverage

Author: kanutocd

CHANGELOG.md

@@ -2,6 +2,35 @@
 
 ## Unreleased
 
+## 0.2.4 - 2026-06-17
+
+### Added
+
+- Added socket-aware replication stream polling.
+- Added E2E coverage for PostgreSQL restart and replication stream recovery.
+- Added connection readiness checks to E2E infrastructure.
+
+### Changed
+
+- Improved replication stream handling during idle periods.
+- Improved reconnect behavior after PostgreSQL restarts.
+- Improved standby feedback reliability during long-running streams.
+- Improved E2E test stability across PostgreSQL startup and restart scenarios.
+- Normalized `PG#get_copy_data` idle responses to simplify stream processing.
+
+### Fixed
+
+- Fixed replication stream recovery after PostgreSQL restart.
+- Fixed handling of idle COPY stream reads.
+- Fixed reconnect loops triggered by PostgreSQL replication timeouts.
+- Fixed E2E race conditions during PostgreSQL initialization.
+- Fixed replication slot creation behavior when a slot already exists.
+- Fixed several edge cases uncovered by Mammoth integration testing.
+
+### Documentation
+
+- Expanded YARD documentation coverage to 98.95%.
+
 ## 0.2.3 - 2026-06-17
 
 ### Fixed

README.md

@@ -363,6 +363,57 @@ Equivalent Rake task:
 bundle exec rake e2e:run
 ```
 
+## Transport lifecycle behavior
+
+`pgoutput-client` owns PostgreSQL logical replication transport and lifecycle
+management. It opens the replication connection, optionally creates the logical
+replication slot, starts streaming, sends standby status feedback, and retries
+reconnectable failures.
+
+### Idle standby feedback
+
+Long-running replication streams can be quiet for long periods when no WAL
+changes are produced. During those idle periods the client wakes periodically
+and sends standby status feedback so PostgreSQL does not terminate the walsender
+for replication timeout.
+
+Control the feedback cadence with `feedback_interval`:
+
+```ruby
+runner = Pgoutput::Client::Runner.new(
+  database_url: ENV.fetch("DATABASE_URL"),
+  slot_name: "mammoth_live",
+  publication_names: ["mammoth_publication"],
+  feedback_interval: 10.0
+)
+```
+
+### Idempotent automatic slot creation
+
+When `auto_create_slot` is enabled, the client treats slot creation as
+"ensure this slot exists". Missing slots are created before streaming; existing
+slots are reused and do not cause startup failure.
+
+```ruby
+runner = Pgoutput::Client::Runner.new(
+  database_url: ENV.fetch("DATABASE_URL"),
+  slot_name: "mammoth_live",
+  publication_names: ["mammoth_publication"],
+  auto_create_slot: true,
+  temporary_slot: false
+)
+```
+
+Publication creation remains outside this gem. Create publications through
+application migrations, database bootstrap SQL, or infrastructure tooling.
+
+### Restart recovery
+
+After a stream has connected successfully, transient PostgreSQL outages are
+retried through the reconnect lifecycle. This includes ordinary container or
+process restart windows where PostgreSQL temporarily refuses connections or
+reports that the database system is starting up.
+
 ---
 
 ## License

lib/pgoutput/client.rb

@@ -13,6 +13,13 @@
 require_relative "client/stream"
 require_relative "client/runner"
 
+# Namespace for PostgreSQL pgoutput logical replication components.
+#
+# The top-level namespace is shared by pgoutput ecosystem gems. This gem
+# defines only the `Pgoutput::Client` transport namespace and leaves protocol
+# parsing, value decoding, and CDC normalization to sibling libraries.
+#
+# @api public
 module Pgoutput
   # Namespace for PostgreSQL logical replication transport support.
   #

lib/pgoutput/client/connection.rb

@@ -76,14 +76,19 @@ def start_replication
 
       # Receive one CopyData payload from the server.
       #
-      # The call is non-blocking because the underlying `pg` call receives
-      # `false` for its blocking argument. `nil` means no complete CopyData
-      # payload is currently available.
+      # The stream must not block forever while PostgreSQL is idle, because the
+      # caller needs opportunities to send periodic standby feedback. Wait
+      # briefly for socket readability, then use the pg driver's blocking
+      # CopyData read only when data is available. `nil` means the stream is
+      # currently idle.
       #
       # @return [String, nil] raw CopyData payload or `nil`
       # @raise [ConnectionError] if receiving fails
       def get_copy_data # rubocop:disable Naming/AccessorMethodName
-        @pg_connection.get_copy_data(false)
+        return nil unless copy_data_readable?
+
+        copy_data = @pg_connection.get_copy_data(false)
+        copy_data == false ? nil : copy_data
       rescue PG::Error => e
         raise ConnectionError, e.message
       end
@@ -110,6 +115,15 @@ def close
 
       private
 
+      def copy_data_readable?
+        return true unless @pg_connection.respond_to?(:socket_io)
+
+        socket = @pg_connection.socket_io
+        return true unless socket
+
+        !!IO.select([socket], nil, nil, 0.1)
+      end
+
       def exec(sql)
         @pg_connection.exec(sql)
       rescue PG::Error => e

lib/pgoutput/client/feedback.rb

@@ -2,6 +2,9 @@
 
 module Pgoutput
   module Client
+    # Internal immutable base class generated by `Data.define` for {Feedback}.
+    #
+    # @api private
     FeedbackData = Data.define(:received_lsn, :flushed_lsn, :applied_lsn, :client_clock, :reply_requested)
 
     # Standby status feedback message builder.

lib/pgoutput/client/keepalive.rb

@@ -2,6 +2,9 @@
 
 module Pgoutput
   module Client
+    # Internal immutable base class generated by `Data.define` for {Keepalive}.
+    #
+    # @api private
     KeepaliveData = Data.define(:wal_end, :server_clock, :reply_requested)
 
     # Immutable primary keepalive replication message.

lib/pgoutput/client/runner.rb

@@ -37,7 +37,17 @@ module Client
     # @see Stream
     # @api public
     class Runner
+      # Default number of reconnect attempts after a previously healthy stream
+      # fails. The default is intentionally large enough to survive ordinary
+      # PostgreSQL restart windows.
+      #
+      # @return [Integer]
       DEFAULT_RECONNECT_ATTEMPTS = 30
+
+      # Base reconnect backoff, in seconds. Attempt `n` sleeps for
+      # `n * DEFAULT_RECONNECT_BACKOFF`.
+      #
+      # @return [Float]
       DEFAULT_RECONNECT_BACKOFF = 0.5
 
       # Configuration used by this runner.

lib/pgoutput/client/state.rb

@@ -2,6 +2,9 @@
 
 module Pgoutput
   module Client
+    # Internal immutable base class generated by `Data.define` for {RunnerState}.
+    #
+    # @api private
     RunnerStateData = Data.define(
       :running,
       :stop_requested,

lib/pgoutput/client/version.rb

@@ -5,6 +5,6 @@ module Client
     # Current pgoutput-client gem version.
     #
     # @return [String]
-    VERSION = "0.2.3"
+    VERSION = "0.2.4"
   end
 end

lib/pgoutput/client/xlog_data.rb

@@ -2,6 +2,9 @@
 
 module Pgoutput
   module Client
+    # Internal immutable base class generated by `Data.define` for {XLogData}.
+    #
+    # @api private
     ReplicationXLogData = Data.define(:wal_start, :wal_end, :server_clock, :payload)
 
     # Immutable XLogData replication envelope.