Class: SolidObjects::WakeUpAdapters::Postgresql

Inherits:
Object
  • Object
show all
Defined in:
lib/solid_objects/wake_up_adapters/postgresql.rb,
sig/generated/lib/solid_objects/wake_up_adapters/postgresql.rbs

Overview

Wakes runtime roles across processes using PostgreSQL notifications.

The in-process wake-up cannot reach another process, so a commit in a web process leaves a worker waiting out its polling interval. This adapter keeps that polling interval as the upper bound and delivers a notification when one is available, so a missed or failed notification costs latency rather than correctness.

Constant Summary collapse

CHANNEL =

Returns:

  • (::String) 
"solid_objects_wake_up"
FAILED_WAIT_INTERVAL =

Returns:

  • (::Float) 
0.05

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(channel: CHANNEL) ⇒ Postgresql

Returns a new instance of Postgresql.

RBS:

  • (?channel: String) -> void

Parameters:

  • channel: (String) (defaults to: CHANNEL) 


23
24
25
26
27
 
# File 'lib/solid_objects/wake_up_adapters/postgresql.rb', line 23

def initialize(channel: CHANNEL)
  @channel = channel
  @mutex = Thread::Mutex.new
  @connections = []
end

Instance Attribute Details

#channel ⇒ Object (readonly)

RBS:

  • @channel: String

  • @mutex: Thread::Mutex

  • @connections: Array[untyped]

Returns:

  • (Object) 


20
21
22
 
# File 'lib/solid_objects/wake_up_adapters/postgresql.rb', line 20

def channel
  @channel
end

#connections ⇒ Object (readonly)

Returns the value of attribute connections.

Returns:

  • (Object) 


73
74
75
 
# File 'lib/solid_objects/wake_up_adapters/postgresql.rb', line 73

def connections
  @connections
end

#mutex ⇒ Object (readonly)

Returns the value of attribute mutex.

Returns:

  • (Object) 


73
74
75
 
# File 'lib/solid_objects/wake_up_adapters/postgresql.rb', line 73

def mutex
  @mutex
end

Instance Method Details

#disconnect(connection) ⇒ void

This method returns an undefined value.

RBS:

  • (untyped) -> void

Parameters:

  • (Object) 


110
111
112
113
114
 
# File 'lib/solid_objects/wake_up_adapters/postgresql.rb', line 110

def disconnect(connection)
  connection.disconnect!
rescue
  nil
end

#instrument_failure(operation, error) ⇒ void

This method returns an undefined value.

RBS:

  • (Symbol, Exception) -> void

Parameters:

  • (Symbol)
  • (Exception) 


125
126
127
128
129
130
131
132
133
 
# File 'lib/solid_objects/wake_up_adapters/postgresql.rb', line 125

def instrument_failure(operation, error)
  SolidObjects.instrument(
    :"wake_up.failed",
    adapter: "postgresql",
    operation: operation.to_s,
    error_class: error.class.name,
    error_message: error.message
  )
end

#listen ⇒ Boolean

Starts listening before a caller blocks, so a notification sent between startup and the first wait is not missed.

RBS:

  • () -> bool

Returns:

  • (Boolean) 


51
52
53
54
55
56
57
 
# File 'lib/solid_objects/wake_up_adapters/postgresql.rb', line 51

def listen
  listening_connection
  true
rescue => error
  instrument_failure(:listen, error)
  false
end

#listening_connection ⇒ Object

A listening connection is dedicated and per thread. LISTEN is per connection, a blocking wait must not hold a connection the rest of the runtime needs, and one connection cannot serve concurrent waiters: the supervisor shares one adapter across roles, and a notification consumed by one waiter would leave the others asleep until their poll expired.

RBS:

  • () -> untyped

Returns:

  • (Object) 


88
89
90
91
92
93
 
# File 'lib/solid_objects/wake_up_adapters/postgresql.rb', line 88

def listening_connection
  connection = Thread.current[thread_key]
  return connection if connection&.active?

  open_listening_connection
end

#notify_channel ⇒ void

This method returns an undefined value.

RBS:

  • () -> void



76
77
78
79
80
 
# File 'lib/solid_objects/wake_up_adapters/postgresql.rb', line 76

def notify_channel
  Record.connection_pool.with_connection do |connection|
    connection.execute("NOTIFY #{connection.quote_table_name(channel)}")
  end
end

#open_listening_connection ⇒ Object

RBS:

  • () -> untyped

Returns:

  • (Object) 


96
97
98
99
100
101
102
 
# File 'lib/solid_objects/wake_up_adapters/postgresql.rb', line 96

def open_listening_connection
  connection = Record.connection_pool.send(:new_connection)
  connection.execute("LISTEN #{connection.quote_table_name(channel)}")
  Thread.current[thread_key] = connection
  mutex.synchronize { connections << connection }
  connection
end

#pace_after_failure(timeout) ⇒ void

This method returns an undefined value.

RBS:

  • (Numeric) -> void

Parameters:

  • (Numeric) 


117
118
119
120
121
122
 
# File 'lib/solid_objects/wake_up_adapters/postgresql.rb', line 117

def pace_after_failure(timeout)
  interval = [ timeout.to_f, FAILED_WAIT_INTERVAL ].min
  return unless interval.positive?

  sleep interval
end

#signal ⇒ Boolean

RBS:

  • () -> bool

Returns:

  • (Boolean) 


30
31
32
33
34
35
36
 
# File 'lib/solid_objects/wake_up_adapters/postgresql.rb', line 30

def signal
  notify_channel
  true
rescue => error
  instrument_failure(:signal, error)
  false
end

#stop ⇒ Boolean

RBS:

  • () -> bool

Returns:

  • (Boolean) 


60
61
62
63
64
65
66
67
68
69
 
# File 'lib/solid_objects/wake_up_adapters/postgresql.rb', line 60

def stop
  open = mutex.synchronize do
    listening = connections.dup
    connections.clear
    listening
  end
  Thread.current[thread_key] = nil
  open.each { |connection| disconnect(connection) }
  open.any?
end

#thread_key ⇒ Symbol

RBS:

  • () -> Symbol

Returns:

  • (Symbol) 


105
106
107
 
# File 'lib/solid_objects/wake_up_adapters/postgresql.rb', line 105

def thread_key
  :"solid_objects_wake_up_#{object_id}"
end

#wait(timeout:) ⇒ Boolean

RBS:

  • (timeout: Numeric) -> bool

Parameters:

  • timeout: (Numeric)

Returns:

  • (Boolean) 


39
40
41
42
43
44
45
46
 
# File 'lib/solid_objects/wake_up_adapters/postgresql.rb', line 39

def wait(timeout:)
  connection = listening_connection
  !connection.raw_connection.wait_for_notify(timeout.to_f).nil?
rescue => error
  instrument_failure(:wait, error)
  pace_after_failure(timeout)
  false
end