Class: Pgbus::Web::Streamer::HubClient

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

Overview

Worker-side client for the MasterHub (issue #382). Presents the same surface the Dispatcher consumes from a Listener — synchronous ensure_listening (the no-lost-broadcast ack contract, now crossing the process boundary), async remove_listening — while wakes arrive as HubProtocol frames and are re-materialized into the worker's dispatch queue as WakeMessages.

Failure model: this class never retries. Connect refusal, an ack deadline, or transport EOF (master died / eviction) marks the client dead, fails every pending sub, and fires on_failure exactly once — the FailoverListener's cue to swap in a per-worker Listener. One-way: once a worker has fallen back it stays local until it recycles (settled on #382 — no flap-back complexity).

Defined Under Namespace

Classes: HubUnavailableError

Instance Method Summary collapse

Constructor Details

#initialize(socket_path:, dispatch_queue:, ack_timeout: 2.0, on_failure: nil, logger: Pgbus.logger) ⇒ HubClient

Optimistic before the first status broadcast, mirroring WakePipe / NotifyListener: a just-connected worker isn't treated as degraded before the hub has said anything.



27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
 
# File 'lib/pgbus/web/streamer/hub_client.rb', line 27

def initialize(socket_path:, dispatch_queue:, ack_timeout: 2.0,
               on_failure: nil, logger: Pgbus.logger)
  @socket_path = socket_path
  @dispatch_queue = dispatch_queue
  @ack_timeout = ack_timeout
  @on_failure = on_failure
  @logger = logger
  @write_mutex = Mutex.new
  @ack_mutex = Mutex.new
  @pending_acks = Hash.new { |h, k| h[k] = [] }
  @hub_healthy = true
  @dead = false
  @stopping = false
  @sock = nil
  @reader = nil
end

Instance Method Details

#connect ⇒ Object 



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

def connect
  @sock = UNIXSocket.new(@socket_path)
  @reader = Thread.new { reader_loop }
  self
rescue SystemCallError, IOError, ArgumentError, ThreadError => e
  # ArgumentError: a socket path over the platform sun_path limit;
  # IOError: a path that exists but is not a socket; ThreadError: the
  # reader thread could not spawn. All must fall back exactly like a
  # refused connect, never abort worker boot — and never leak the
  # half-opened socket.
  close_quietly(@sock)
  @sock = nil
  raise HubUnavailableError, "cannot reach master hub at #{@socket_path}: #{e.class}: #{e.message}"
end

#dead? ⇒ Boolean

Returns:

  • (Boolean) 


63
64
65
 
# File 'lib/pgbus/web/streamer/hub_client.rb', line 63

def dead?
  @dead
end

#ensure_listening(queue) ⇒ Object

Synchronous, bounded: returns :done once the master has confirmed LISTEN is active for queue. Raises HubUnavailableError on a dead transport or an expired ack deadline (which also kills the transport — a hub that can't ack in time can't be trusted with the no-lost-broadcast contract either).

Raises:



72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
 
# File 'lib/pgbus/web/streamer/hub_client.rb', line 72

def ensure_listening(queue)
  raise HubUnavailableError, "master hub transport is dead" if @dead

  waiter = Queue.new
  @ack_mutex.synchronize { @pending_acks[queue] << waiter }
  write_frame({ "t" => "sub", "q" => queue })

  result = waiter.pop(timeout: @ack_timeout)
  if result.nil?
    discard_waiter(queue, waiter)
    mark_dead("sub ack for #{queue} not received within #{@ack_timeout}s")
    raise HubUnavailableError, "master hub ack timeout for #{queue}"
  end
  raise HubUnavailableError, "master hub died while awaiting ack for #{queue}" if result == :dead

  :done
end

#hub_healthy? ⇒ Boolean

Returns:

  • (Boolean) 


59
60
61
 
# File 'lib/pgbus/web/streamer/hub_client.rb', line 59

def hub_healthy?
  @hub_healthy
end

#remove_listening(queue) ⇒ Object

Lazy GC, fire-and-forget — no correctness path waits on UNLISTEN (mirrors Listener#remove_listening). A dead transport is a no-op: the master's EOF cleanup already released this worker's refs.



93
94
95
96
97
98
99
 
# File 'lib/pgbus/web/streamer/hub_client.rb', line 93

def remove_listening(queue)
  return if @dead

  write_frame({ "t" => "unsub", "q" => queue })
rescue HubUnavailableError
  nil
end

#stop ⇒ Object 



101
102
103
104
105
106
107
 
# File 'lib/pgbus/web/streamer/hub_client.rb', line 101

def stop
  @stopping = true
  close_quietly(@sock)
  @reader&.join(2)
  @reader = nil
  self
end