Module: Pgbus::Web::Streamer::IoWriter

Defined in:
lib/pgbus/web/streamer/io_writer.rb

Overview

Non-blocking IO writer with a per-call deadline, serialised through the connection's own mutex. This is the bug-fix for puma/puma#576: a naive io.write(bytes) on a dead or slow SSE client deadlocks the dispatcher thread until the OS closes the socket (which can take minutes under a TCP keepalive). The message_bus gem hit this in production; we copy the pattern.

The write loop uses write_nonblock + IO.select so a slow client at most stalls the CALLING thread's mutex-protected write for deadline_ms. During fanout the dispatcher IS that caller, writing to each connection serially, so K slow-but-not-yet-dead clients stack the deadline (~K * deadline_ms) before each is marked dead — the head-of-line block. Fanout writes therefore pass the SHORT streams_fanout_write_deadline_ms (not streams_write_deadline_ms) to bound that stall (issue #315 item 3). When the deadline expires with bytes still pending, we return :blocked; the caller (Connection#enqueue or Connection#write_comment) translates that into mark_dead!, and the heartbeat sweep unregisters the connection — the client then reconnects and replays the gap from the durable archive.

Returns:

:ok       — all bytes written
:closed   — peer gone (EPIPE / ECONNRESET / IOError on closed IO)
:blocked  — deadline hit before all bytes could be written

Class Method Summary collapse

Class Method Details

.write(connection, bytes, deadline_ms:) ⇒ Object 



31
32
33
34
35
 
# File 'lib/pgbus/web/streamer/io_writer.rb', line 31

def self.write(connection, bytes, deadline_ms:)
  connection.mutex.synchronize do
    write_all(connection.io, bytes, deadline_ms)
  end
end