Class: Pgbus::Web::Streamer::Instance

Inherits:
Object
  • Object
show all
Defined in:
lib/pgbus/web/streamer/instance.rb

Overview

Composes the Streamer's three background threads (Listener, Dispatcher, Heartbeat) with the shared Registry and dispatch_queue. One Instance per Puma worker. Owns the lifecycle of all three threads and the dedicated PG::Connection for LISTEN.

Lifecycle:

Instance.new(...)  — allocates wiring, does NOT start threads
#start             — spawns listener/dispatcher/heartbeat in order
#register(conn)    — enqueues a ConnectMessage into the dispatch queue
#shutdown!         — sends shutdown sentinel to every connection and
                   stops all threads in reverse order

Dependency injection: every collaborator is constructor-injected so tests can swap in fakes without touching Pgbus.configuration. In production the module-level Streamer.current(...) builds all of the defaults from the configuration.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(client: Pgbus.client, config: Pgbus.configuration, pg_connection: nil, logger: Pgbus.logger, registry: nil, dispatch_queue: nil, connection_factory: nil) ⇒ Instance

Returns a new instance of Instance.



26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
 
# File 'lib/pgbus/web/streamer/instance.rb', line 26

def initialize(
  client: Pgbus.client,
  config: Pgbus.configuration,
  pg_connection: nil,
  logger: Pgbus.logger,
  registry: nil,
  dispatch_queue: nil,
  connection_factory: nil
)
  @client = client
  @config = config
  @logger = logger
  @registry = registry || Registry.new
  @dispatch_queue = dispatch_queue || Queue.new

  @stream_counter = StreamCounter.new
  @pg_connection = pg_connection || build_pg_connection
  # Self-tuning streams-pool autoscaler (issue #323). Opt-in; nil unless
  # enabled AND on the dedicated connection path (the shared-AR streams
  # pool aliases the non-thread-safe job pool and resize is a no-op there).
  # It's a pure decision object — no thread, no connection. It runs as a
  # throttled maintenance task on the Listener's idle LISTEN connection
  # (zero extra connections), querying live headroom there.
  @autoscaler =
    if @config.streams_pool_autoscale && !@client.shared_connection?
      Pgbus::Streams::PoolAutoscaler.new(client: @client, config: @config, logger: @logger)
    end
  @listener = Listener.new(
    pg_connection: @pg_connection,
    dispatch_queue: @dispatch_queue,
    health_check_ms: @config.streams_listen_health_check_ms,
    # Opt-in dispatch-queue backpressure (issue #315 item 3). 0 =
    # unbounded (default). The queue itself stays an unbounded
    # Queue.new so the request-thread Connect push and the dispatcher's
    # own prune_dead self-post never block.
    dispatch_queue_limit: @config.streams_dispatch_queue_limit,
    maintenance: build_autoscale_maintenance,
    logger: @logger,
    # On reconnect the Listener rebuilds its OWN connection via this
    # factory (fresh connect re-resolves DNS, converges on the promoted
    # primary after a failover) instead of resetting a possibly-dead
    # socket. Always provided — even when an initial pg_connection: is
    # injected, the reconnect path builds a fresh raw connection. A test
    # can inject its own factory to avoid touching real configuration.
    connection_factory: connection_factory || -> { build_raw_pg_connection }
  )
  # Off-thread durable fanout writer (issue #321). Built only when
  # streams_writer_threads > 0; nil means fanout writes stay inline on
  # the dispatcher thread (the default, pre-#321 behavior). The pump
  # reports write acks on @ack_queue (drained by the dispatcher) and
  # posts a DisconnectMessage via on_dead when a write fails, so the
  # dispatcher owns all cursor + registry cleanup on its own thread.
  @ack_queue = Queue.new
  @pump = build_pump
  @dispatcher = StreamEventDispatcher.new(
    client: @client,
    registry: @registry,
    listener: @listener,
    dispatch_queue: @dispatch_queue,
    logger: @logger,
    config: @config,
    stream_counter: @stream_counter,
    pump: @pump,
    ack_queue: @ack_queue
  )
  @heartbeat = Heartbeat.new(
    registry: @registry,
    dispatch_queue: @dispatch_queue,
    interval: @config.streams_heartbeat_interval,
    idle_timeout: @config.streams_idle_timeout,
    logger: @logger
  )
  @started = false
  @shutdown_mutex = Mutex.new
end

Instance Attribute Details

#autoscaler ⇒ Object (readonly)

Returns the value of attribute autoscaler.



23
24
25
 
# File 'lib/pgbus/web/streamer/instance.rb', line 23

def autoscaler
  @autoscaler
end

#dispatch_queue ⇒ Object (readonly)

Returns the value of attribute dispatch_queue.



23
24
25
 
# File 'lib/pgbus/web/streamer/instance.rb', line 23

def dispatch_queue
  @dispatch_queue
end

#dispatcher ⇒ Object (readonly)

Returns the value of attribute dispatcher.



23
24
25
 
# File 'lib/pgbus/web/streamer/instance.rb', line 23

def dispatcher
  @dispatcher
end

#heartbeat ⇒ Object (readonly)

Returns the value of attribute heartbeat.



23
24
25
 
# File 'lib/pgbus/web/streamer/instance.rb', line 23

def heartbeat
  @heartbeat
end

#listener ⇒ Object (readonly)

Returns the value of attribute listener.



23
24
25
 
# File 'lib/pgbus/web/streamer/instance.rb', line 23

def listener
  @listener
end

#pump ⇒ Object (readonly)

Returns the value of attribute pump.



23
24
25
 
# File 'lib/pgbus/web/streamer/instance.rb', line 23

def pump
  @pump
end

#registry ⇒ Object (readonly)

Returns the value of attribute registry.



23
24
25
 
# File 'lib/pgbus/web/streamer/instance.rb', line 23

def registry
  @registry
end

#stream_counter ⇒ Object (readonly)

Returns the value of attribute stream_counter.



23
24
25
 
# File 'lib/pgbus/web/streamer/instance.rb', line 23

def stream_counter
  @stream_counter
end

Instance Method Details

#register(connection) ⇒ Object

Enqueue a new SSE client. The dispatcher picks this up on its next iteration and runs the 5-step race-free replay sequence. The StreamApp calls this right after hijacking the socket.

Guarded against the worker-shutdown race: if the request thread arrives here after shutdown! has flipped @started, we mark the connection dead and bail out instead of enqueueing a ConnectMessage. Otherwise the message would land on a dispatch queue that no one is draining, leaving the socket outside the registry and outside close_all_connections — the client would never see the pgbus:shutdown sentinel.



126
127
128
129
130
131
132
133
134
135
136
137
 
# File 'lib/pgbus/web/streamer/instance.rb', line 126

def register(connection)
  return if connection.dead?

  @shutdown_mutex.synchronize do
    unless @started
      connection.mark_dead!
      return
    end

    @dispatch_queue << StreamEventDispatcher::ConnectMessage.new(connection: connection)
  end
end

#shutdown! ⇒ Object

Graceful shutdown for Puma worker restart. Order matters:

1. Heartbeat first (stop writing comments to connections we're
 about to close)
2. Listener next (stop accepting new NOTIFYs)
3. Dispatcher next (drain the queue; it's now finite because
 nothing else writes into it)
4. Writer pump next (issue #321): once the dispatcher — the pump's
 only producer — is stopped, each partition is finite, so the
 pump drains every accepted-but-unflushed durable frame and joins
 its workers before we touch the sockets. No-op when offload is
 off. Must come BEFORE close_all_connections so buffered frames
 flush before their sockets close.
5. Send pgbus:shutdown sentinel to every connection and close
 their sockets. We do this AFTER stopping the dispatcher AND the
 pump so nothing else is writing to these IOs concurrently.

Bounded by the configured write deadline per connection; a dead client drops instantly, a slow one stalls for at most write_deadline_ms.



157
158
159
160
161
162
163
164
165
166
167
168
 
# File 'lib/pgbus/web/streamer/instance.rb', line 157

def shutdown!
  @shutdown_mutex.synchronize do
    return unless @started

    @started = false
    safely { @heartbeat.stop }
    safely { @listener.stop }
    safely { @dispatcher.stop }
    safely { @pump&.stop }
    close_all_connections
  end
end

#start ⇒ Object 



102
103
104
105
106
107
108
109
110
111
112
113
 
# File 'lib/pgbus/web/streamer/instance.rb', line 102

def start
  return if @started

  @started = true
  # Pump first: it must be ready to accept writes before the dispatcher
  # can post any (issue #321). No-op when offload is off (@pump nil).
  @pump&.start
  @listener.start
  @dispatcher.start
  @heartbeat.start
  self
end