Class: Pgbus::Web::Streamer::OutboundPump

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

Overview

Off-thread durable-stream fanout writer (issue #321).

The dispatcher must never block on a slow client's socket write. When streams_writer_threads > 0, StreamEventDispatcher#handle_durable_wake hands each (connection, filtered envelopes, batch_max) to this pump instead of writing inline. The pump owns N worker threads; each connection is pinned to ONE worker by id.hash % N, so a connection's frames stay strictly ordered and its per-io mutex is never contended across workers.

The pump calls the UNCHANGED Connection#enqueue on a worker thread (the blocking write, last_msg_id_sent advance, and mark_dead! all move off the dispatcher), then reports back:

- success  → a WriteAckMessage(connection, accepted_max) onto ack_queue.
           accepted_max is the highest msg_id actually written (or
           the batch_max for a fully-filtered empty batch, so the
           dispatcher's scan cursor still advances past the hidden
           window). The dispatcher — the SOLE owner of @scanned_cursor
           — applies it on its own thread. This keeps the lockless
           single-owner invariant intact: the writer only reports a
           value, it never mutates dispatcher state.
- failure  → the injected on_dead callback (which posts a
           DisconnectMessage), NOT an ack. A failed write must never
           advance the cursor past a frame that never reached the
           socket (issue #321 B2). The dispatcher then scrubs the
           connection's state deterministically (B4), even on an
           otherwise-quiet stream.

EPHEMERAL frames are NEVER routed here — they have no archive to replay, so an async drop would be unrecoverable (issue #321 B1). #post raises ArgumentError on a negative msg_id as a defense-in-depth guard against a future refactor accidentally offloading one.

Constant Summary collapse

DRAIN = 
:__drain__

Instance Method Summary collapse

Constructor Details

#initialize(threads:, ack_queue:, on_dead:, buffer_limit: 0, logger: Pgbus.logger) ⇒ OutboundPump

Returns a new instance of OutboundPump.

Raises:

  • (ArgumentError) 


44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
 
# File 'lib/pgbus/web/streamer/outbound_pump.rb', line 44

def initialize(threads:, ack_queue:, on_dead:, buffer_limit: 0, logger: Pgbus.logger)
  raise ArgumentError, "threads must be positive" unless threads.positive?

  @ack_queue = ack_queue
  @on_dead = on_dead
  @buffer_limit = buffer_limit
  @logger = logger
  # One partition per worker. Each is a Partition wrapping a bounded
  # per-connection buffer so the drop-oldest policy is per connection,
  # not per partition (a fast connection can't be starved by a slow one
  # sharing its worker beyond ordering).
  @partitions = Array.new(threads) { Partition.new(buffer_limit, @logger) }
  @threads = []
  @started = false
end

Instance Method Details

#alive? ⇒ Boolean

True while any writer thread is still alive. Lets callers assert the pump's OWN threads stopped after #stop without inspecting the global Thread.list (which is noisy and can't distinguish a pump leak from unrelated thread churn).

Returns:

  • (Boolean) 


74
75
76
 
# File 'lib/pgbus/web/streamer/outbound_pump.rb', line 74

def alive?
  @threads.any?(&:alive?)
end

#post(connection, envelopes, batch_max, deadline_ms:) ⇒ Object

Hand a durable fanout write to the pump. Returns immediately — the dispatcher does not block. Raises on a negative (ephemeral) msg_id.



85
86
87
88
89
90
91
92
93
94
95
96
 
# File 'lib/pgbus/web/streamer/outbound_pump.rb', line 85

def post(connection, envelopes, batch_max, deadline_ms:)
  if envelopes.any? { |e| e.msg_id.negative? }
    raise ArgumentError,
          "OutboundPump received an ephemeral (negative msg_id) envelope; " \
          "ephemeral fanout must stay inline on the dispatcher thread (issue #321 B1)"
  end

  partition_for(connection).push(
    WriteJob.new(connection: connection, envelopes: envelopes,
                 batch_max: batch_max, deadline_ms: deadline_ms)
  )
end

#start ⇒ Object 



60
61
62
63
64
65
66
67
68
 
# File 'lib/pgbus/web/streamer/outbound_pump.rb', line 60

def start
  return self if @started

  @started = true
  @partitions.each do |partition|
    @threads << Thread.new { run_worker(partition) }
  end
  self
end

#stop ⇒ Object

Graceful drain: signal every partition to flush what it holds, then join each worker bounded by the write deadline. Never Thread#kill — a kill mid write_nonblock corrupts IO state (mirrors StreamEventDispatcher#stop). Idempotent.



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

def stop
  return self unless @started

  @started = false
  @partitions.each { |p| p.push(DRAIN) }
  @threads.each do |t|
    next if t.join(join_timeout_seconds)

    @logger.warn { "[Pgbus::Streamer::OutboundPump] writer thread did not drain within #{join_timeout_seconds}s" }
  end
  @threads.clear
  self
end

#worker_threads ⇒ Object

Snapshot of the pump's live writer threads — for test introspection.



79
80
81
 
# File 'lib/pgbus/web/streamer/outbound_pump.rb', line 79

def worker_threads
  @threads.dup
end