Class: Pgoutput::Client::SlotInspector

Inherits:
Object
  • Object
show all
Defined in:
lib/pgoutput/client/slot_inspector.rb

Overview

Reads PostgreSQL replication-slot catalog state.

Inspection uses a short-lived ordinary database connection because catalog queries are separate from the long-lived replication protocol connection. Querying the whole catalog row through to_jsonb keeps the result compatible with PostgreSQL versions that expose different optional slot-health columns.

Constant Summary collapse

QUERY =

Version-tolerant catalog query for one replication slot.

Returns:

  • (String) 
<<~SQL
  SELECT
    to_jsonb(slot) || jsonb_build_object(
      'retained_wal_bytes',
      CASE
        WHEN slot.restart_lsn IS NULL THEN NULL
        ELSE pg_wal_lsn_diff(pg_current_wal_lsn(), slot.restart_lsn)::bigint
      END
    ) AS slot
  FROM pg_catalog.pg_replication_slots AS slot
  WHERE slot.slot_name = $1
SQL

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(database_url:, connection_factory: nil) ⇒ SlotInspector

Returns a new instance of SlotInspector.

Parameters:

  • database_url (String) —

    PostgreSQL connection URL

  • connection_factory (#call, nil) (defaults to: nil) —

    optional connection factory for tests



38
39
40
41
 
# File 'lib/pgoutput/client/slot_inspector.rb', line 38

def initialize(database_url:, connection_factory: nil)
  @database_url = database_url
  @connection_factory = connection_factory
end

Instance Attribute Details

#database_url ⇒ String (readonly)

Returns PostgreSQL connection URL.

Returns:

  • (String) —

    PostgreSQL connection URL



34
35
36
 
# File 'lib/pgoutput/client/slot_inspector.rb', line 34

def database_url
  @database_url
end

Instance Method Details

#fetch(slot_name) ⇒ SlotStatus?

Fetch the configured slot's current catalog state.

Parameters:

  • slot_name (String) —

    replication slot name

Returns:

  • (SlotStatus, nil) —

    snapshot, or nil when the slot is missing

Raises:



48
49
50
51
52
53
54
55
56
57
58
 
# File 'lib/pgoutput/client/slot_inspector.rb', line 48

def fetch(slot_name)
  connection = open_connection
  row = connection.exec_params(QUERY, [String(slot_name)]).first
  return nil unless row

  SlotStatus.from_catalog(parse_catalog(row.fetch("slot")))
rescue pg_error_class => e
  raise ConnectionError, e.message
ensure
  connection&.close unless connection&.finished?
end