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
 
# 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
  # 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 = build_listener(pg_connection, connection_factory)
  # 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.



107
108
109
110
111
112
113
114
115
116
117
118
 
# File 'lib/pgbus/web/streamer/instance.rb', line 107

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.



138
139
140
141
142
143
144
145
146
147
148
149
 
# File 'lib/pgbus/web/streamer/instance.rb', line 138

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 



83
84
85
86
87
88
89
90
91
92
93
94
 
# File 'lib/pgbus/web/streamer/instance.rb', line 83

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