Class: Pgoutput::Client::Runner

Inherits:
Object
  • Object
show all
Defined in:
lib/pgoutput/client/runner.rb,
sig/pgoutput/client/runner.rbs

Overview

High-level logical replication client facade.

Runner is the simplest public entry point for consumers that want to stream raw pgoutput payloads without manually managing a replication connection. It owns the connection lifecycle for one logical replication stream:

  1. Build an immutable Configuration from keyword arguments.
  2. Open a PostgreSQL replication connection.
  3. Optionally create the configured replication slot.
  4. Start logical replication.
  5. Yield raw pgoutput payload bytes and XLogData metadata.
  6. Close the connection when streaming exits.

If a live stream loses its connection, the runner retries a small number of times with a backoff and resumes from the latest confirmed WAL position. Replay, checkpointing, and deduplication are not owned here; those concerns belong to the downstream CDC runtime and sink layer.

Examples:

Stream raw pgoutput messages

runner = Pgoutput::Client::Runner.new(
  database_url: "postgres://localhost/app",
  slot_name: "cdc_slot",
  publication_names: ["app_publication"]
)

runner.start do |payload, metadata|
  puts "received #{payload.bytesize} bytes at #{metadata.wal_end_lsn}"
end

See Also:

Constant Summary collapse

DEFAULT_RECONNECT_ATTEMPTS =

Default number of reconnect attempts after a previously healthy stream fails. The default is intentionally large enough to survive ordinary PostgreSQL restart windows.

Returns:

  • (Integer) 
30
DEFAULT_RECONNECT_BACKOFF =

Base reconnect backoff, in seconds. Attempt n sleeps for n * DEFAULT_RECONNECT_BACKOFF.

Returns:

  • (Float) 
0.5

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(**options) ⇒ Runner

Build a runner from configuration keyword arguments.

The accepted keywords are the same as Configuration#initialize. The resulting configuration object is immutable and reused for the lifetime of this runner.

rubocop:disable Style/ArgumentsForwarding -- YARD needs a named parameter

Parameters:

Raises:



74
75
76
77
78
79
80
81
82
83
84
85
86
87
 
# File 'lib/pgoutput/client/runner.rb', line 74

def initialize(**options)
  @configuration = Configuration.new(
    **options # : untyped
  )
  @stopped = true
  @running = false
  @stream = nil
  @resume_lsn = configuration.start_lsn
  @acked_lsn = configuration.start_lsn
  @slot_created = false
  @connected_once = false
  @last_error = nil
  @reconnect_attempts = 0
end

Instance Attribute Details

#configuration ⇒ Configuration (readonly)

Configuration used by this runner.

Returns:



56
57
58
 
# File 'lib/pgoutput/client/runner.rb', line 56

def configuration
  @configuration
end

#last_error ⇒ Exception? (readonly)

Last transport error seen by the runner.

Returns:

  • (Exception, nil) 


61
62
63
 
# File 'lib/pgoutput/client/runner.rb', line 61

def last_error
  @last_error
end

Instance Method Details

#ack(lsn) ⇒ Integer

Mark a WAL position as durably handled by downstream code.

This does not checkpoint or persist anything. It only updates transport feedback state so future standby status updates can distinguish received WAL from downstream-acknowledged WAL.

Parameters:

  • lsn (String, Integer) —

    WAL position acknowledged by downstream code

Returns:

  • (Integer) —

    normalized acknowledged WAL position



183
184
185
186
187
188
 
# File 'lib/pgoutput/client/runner.rb', line 183

def ack(lsn)
  parsed = normalize_lsn_value(lsn)
  @acked_lsn = [@acked_lsn ? normalize_lsn_value(@acked_lsn) : 0, parsed].max
  @stream&.ack(@acked_lsn)
  @acked_lsn
end

#configuration_for_resume ⇒ Object

Returns:

  • (Object) 


258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
 
# File 'lib/pgoutput/client/runner.rb', line 258

def configuration_for_resume
  return configuration if @resume_lsn.nil?

  Configuration.new(
    database_url: configuration.database_url,
    slot_name: configuration.slot_name,
    publication_names: configuration.publication_names,
    start_lsn: @resume_lsn,
    proto_version: configuration.proto_version,
    binary: configuration.binary,
    messages: configuration.messages,
    auto_create_slot: configuration.auto_create_slot,
    temporary_slot: configuration.temporary_slot,
    feedback_interval: configuration.feedback_interval
  )
end

#connected? ⇒ Boolean

Whether an active replication stream exists.

Returns:

  • (Boolean) 


171
172
173
 
# File 'lib/pgoutput/client/runner.rb', line 171

def connected?
  !@stream.nil?
end

#current_lsn_string(value) ⇒ Object

Parameters:

  • value (Object)

Returns:

  • (Object) 


279
280
281
282
283
 
# File 'lib/pgoutput/client/runner.rb', line 279

def current_lsn_string(value)
  return nil if value.nil?

  value.is_a?(String) ? value : LSN.format(value)
end

#ensure_replication_slot(connection) ⇒ Object

Parameters:

  • connection (Object)

Returns:

  • (Object) 


225
226
227
228
229
230
231
232
 
# File 'lib/pgoutput/client/runner.rb', line 225

def ensure_replication_slot(connection)
  connection.create_replication_slot
  @slot_created = true
rescue ConnectionError => e
  raise unless replication_slot_already_exists?(e)

  @slot_created = true
end

#monitor ⇒ RunnerState

Return an immutable transport status snapshot.

Returns:



193
194
195
196
197
198
199
200
201
202
203
204
 
# File 'lib/pgoutput/client/runner.rb', line 193

def monitor
  RunnerState.new(
    running: running?,
    stop_requested: @stopped,
    connected: connected?,
    last_received_lsn: current_lsn_string(@stream&.latest_lsn || @resume_lsn),
    last_feedback_lsn: current_lsn_string(@stream&.acked_lsn || @acked_lsn),
    last_keepalive_at: @stream&.last_keepalive_at,
    last_error: @last_error&.message,
    reconnect_attempts: @reconnect_attempts
  )
end

#normalize_lsn_value(value) ⇒ Object

Parameters:

  • value (Object)

Returns:

  • (Object) 


275
276
277
 
# File 'lib/pgoutput/client/runner.rb', line 275

def normalize_lsn_value(value)
  value.is_a?(String) ? LSN.parse(value) : LSN.parse(LSN.format(value))
end

#reconnect_backoff_for(attempts) ⇒ Object

Parameters:

  • attempts (Object)

Returns:

  • (Object) 


285
286
287
 
# File 'lib/pgoutput/client/runner.rb', line 285

def reconnect_backoff_for(attempts)
  DEFAULT_RECONNECT_BACKOFF * attempts
end

#replication_slot_already_exists?(error) ⇒ Boolean

Parameters:

  • error (Object)

Returns:

  • (Boolean) 


234
235
236
 
# File 'lib/pgoutput/client/runner.rb', line 234

def replication_slot_already_exists?(error)
  error.message.match?(/replication slot .* already exists/i)
end

#restart {|payload, metadata| ... } ⇒ void

This method returns an undefined value.

Stop the active stream and start again with the same block.

The runner is synchronous, so this helper is primarily useful for supervisors that call it instead of manually calling #stop followed by #start.

Yields:

  • (payload, metadata) —

    called once for each XLogData payload



149
150
151
152
 
# File 'lib/pgoutput/client/runner.rb', line 149

def restart(&)
  stop
  start(&)
end

#run_stream_cycle(configuration) {|arg0| ... } ⇒ Object

Parameters:

  • configuration (Object)

Yields:

Yield Parameters:

  • arg0

Yield Returns:

  • (Object)

Returns:

  • (Object) 


238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
 
# File 'lib/pgoutput/client/runner.rb', line 238

def run_stream_cycle(configuration, &)
  connection = Connection.open(configuration)
  setup_connection(connection)
  @connected_once = true
  @stream = Stream.new(connection:, configuration:, acked_lsn: @acked_lsn)
  @stream.start(&)
  :done
rescue ConnectionError => e
  @last_error = e
  raise if @stopped
  raise if @stream.nil? && !@connected_once

  @resume_lsn = @stream&.latest_lsn || @resume_lsn
  @acked_lsn = @stream&.acked_lsn || @acked_lsn
  :retry
ensure
  @stream = nil
  connection&.close
end

#running? ⇒ Boolean

Whether the runner is currently inside its streaming loop.

Returns:

  • (Boolean) 


157
158
159
 
# File 'lib/pgoutput/client/runner.rb', line 157

def running?
  @running
end

#setup_connection(connection) ⇒ Object

Parameters:

  • connection (Object)

Returns:

  • (Object) 


219
220
221
222
223
 
# File 'lib/pgoutput/client/runner.rb', line 219

def setup_connection(connection)
  ensure_replication_slot(connection) if configuration.auto_create_slot && !@slot_created

  connection.start_replication
end

#sleep(duration) ⇒ Object

Parameters:

  • duration (Object)

Returns:

  • (Object) 


289
290
291
 
# File 'lib/pgoutput/client/runner.rb', line 289

def sleep(duration)
  Kernel.sleep(duration)
end

#slot_status ⇒ SlotStatus?

Inspect the configured replication slot through PostgreSQL's catalog.

This reports transport state without applying downstream checkpoint or recovery policy.

Returns:

  • (SlotStatus, nil) —

    current slot state, or nil when missing

Raises:



213
214
215
 
# File 'lib/pgoutput/client/runner.rb', line 213

def slot_status
  SlotInspector.new(database_url: configuration.database_url).fetch(configuration.slot_name)
end

#start {|payload, metadata| ... } ⇒ void

This method returns an undefined value.

Start streaming raw pgoutput payloads.

This method blocks until the stream stops or the underlying connection raises an error. The yielded payload is the raw logical decoding plugin payload contained inside PostgreSQL's XLogData envelope; callers normally pass this payload to a pgoutput parser.

Yields:

  • (payload, metadata) —

    called once for each XLogData payload

Yield Parameters:

  • payload (String) —

    frozen raw pgoutput payload bytes

  • metadata (XLogData) —

    WAL envelope metadata for the payload

Raises:

  • (ArgumentError) —

    if no block is provided

  • (ConnectionError) —

    if a PostgreSQL connection or command fails

  • (ProtocolError) —

    if an invalid replication message is received



104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
 
# File 'lib/pgoutput/client/runner.rb', line 104

def start(&block)
  raise ArgumentError, "block required" unless block

  @stopped = false
  @running = true
  @last_error = nil
  @reconnect_attempts = 0

  loop do
    current_configuration = configuration_for_resume
    case run_stream_cycle(current_configuration, &block)
    when :done
      break
    when :retry
      @reconnect_attempts += 1
      raise @last_error if @reconnect_attempts > DEFAULT_RECONNECT_ATTEMPTS

      sleep(reconnect_backoff_for(@reconnect_attempts))
    end
  end
ensure
  @running = false
  @stopped = true
end

#stop ⇒ void

This method returns an undefined value.

Request graceful stop.

This method records the caller's intent to stop and asks the active Stream, if any, to stop after its current iteration.



135
136
137
138
139
 
# File 'lib/pgoutput/client/runner.rb', line 135

def stop
  @stopped = true
  @stream&.stop
  nil
end

#stopped? ⇒ Boolean

Whether the runner has stopped.

Returns:

  • (Boolean) 


164
165
166
 
# File 'lib/pgoutput/client/runner.rb', line 164

def stopped?
  !running?
end