Class: Pgoutput::Client::SlotInspector

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

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

  • database_url: (String)
  • connection_factory: (Object) (defaults to: nil) 


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

#open_connection ⇒ Object

Returns:

  • (Object) 


62
63
64
65
66
67
 
# File 'lib/pgoutput/client/slot_inspector.rb', line 62

def open_connection
  return @connection_factory.call(database_url) if @connection_factory

  require "pg"
  PG.connect(database_url)
end

#parse_catalog(value) ⇒ Hash[(String | Symbol), untyped]

Parameters:

  • value (Object)

Returns:

  • (Hash[(String | Symbol), untyped]) 


69
70
71
72
73
 
# File 'lib/pgoutput/client/slot_inspector.rb', line 69

def parse_catalog(value)
  return value if value.is_a?(Hash)

  JSON.parse(value)
end

#pg_error_class ⇒ singleton(::PG::Error)

Returns:



75
76
77
78
 
# File 'lib/pgoutput/client/slot_inspector.rb', line 75

def pg_error_class
  require "pg"
  PG::Error
end