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



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

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) 


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

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) 


169
170
171
 
# File 'lib/pgoutput/client/runner.rb', line 169

def connected?
  !@stream.nil?
end

#current_lsn_string(value) ⇒ Object

Parameters:

  • value (Object)

Returns:

  • (Object) 


277
278
279
280
281
 
# File 'lib/pgoutput/client/runner.rb', line 277

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) 


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

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:



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

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) 


273
274
275
 
# File 'lib/pgoutput/client/runner.rb', line 273

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) 


283
284
285
 
# File 'lib/pgoutput/client/runner.rb', line 283

def reconnect_backoff_for(attempts)
  DEFAULT_RECONNECT_BACKOFF * attempts
end

#replication_slot_already_exists?(error) ⇒ Boolean

Parameters:

  • error (Object)

Returns:

  • (Boolean) 


232
233
234
 
# File 'lib/pgoutput/client/runner.rb', line 232

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



147
148
149
150
 
# File 'lib/pgoutput/client/runner.rb', line 147

def restart(&)
  stop
  start(&)
end

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

Parameters:

  • configuration (Object)

Yields:

Yield Parameters:

  • arg0

Yield Returns:

  • (Object)

Returns:

  • (Object) 


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

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) 


155
156
157
 
# File 'lib/pgoutput/client/runner.rb', line 155

def running?
  @running
end

#setup_connection(connection) ⇒ Object

Parameters:

  • connection (Object)

Returns:

  • (Object) 


217
218
219
220
221
 
# File 'lib/pgoutput/client/runner.rb', line 217

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) 


287
288
289
 
# File 'lib/pgoutput/client/runner.rb', line 287

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:



211
212
213
 
# File 'lib/pgoutput/client/runner.rb', line 211

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
 
# 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
    result = run_stream_cycle(current_configuration, &block)
    break if result == :done

    @reconnect_attempts += 1
    raise @last_error if @reconnect_attempts > DEFAULT_RECONNECT_ATTEMPTS

    sleep(reconnect_backoff_for(@reconnect_attempts))
  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.



133
134
135
136
137
 
# File 'lib/pgoutput/client/runner.rb', line 133

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

#stopped? ⇒ Boolean

Whether the runner has stopped.

Returns:

  • (Boolean) 


162
163
164
 
# File 'lib/pgoutput/client/runner.rb', line 162

def stopped?
  !running?
end