Class: Octo::Server::HttpServer::WebSocketConnection
- Inherits:
-
Object
- Object
- Octo::Server::HttpServer::WebSocketConnection
- Defined in:
- lib/octo/server/http_server.rb
Overview
Wraps a raw TCP socket, providing thread-safe WebSocket frame sending.
IMPORTANT: send_raw is called from the Agent thread via broadcast() →send_json(). A blocking socket write with no deadline can pin the Agent thread indefinitely when the client’s receive buffer fills up (silent disconnects such as Wi-Fi handoff or NAT timeout, where TCP keepalive defaults are measured in hours). Thread#raise on blocking native socket writes is best-effort and unreliable, so instead we bound every write with an explicit deadline using IO.select + write_nonblock and declare the connection dead on timeout.
Constant Summary collapse
- SEND_DEADLINE =
Maximum time a single send_raw call is allowed to spend writing. 5 seconds is generous for healthy LAN/Internet clients and short enough that a stuck Agent becomes responsive again quickly.
5.0- SEND_SLOW_WARN =
Warn threshold — any individual send_raw that exceeds this is logged so we can spot sluggish clients before they fully hang.
1.0
Instance Attribute Summary collapse
-
#session_id ⇒ Object
Returns the value of attribute session_id.
Class Method Summary collapse
-
.apply_keepalive(socket) ⇒ Object
Enable TCP keepalive on the underlying socket so silently dead peers are detected in minutes instead of the OS default of hours.
Instance Method Summary collapse
-
#closed? ⇒ Boolean
Returns true if the underlying socket has been detected as dead.
-
#force_close! ⇒ Object
Force-close the connection (used by the interrupt watchdog when an Agent thread is stuck on an unresponsive socket write).
-
#initialize(socket, version) ⇒ WebSocketConnection
constructor
A new instance of WebSocketConnection.
-
#send_json(data) ⇒ Object
Send a JSON-serializable object over the WebSocket.
-
#send_raw(type, data) ⇒ Object
Send a raw WebSocket frame.
Constructor Details
#initialize(socket, version) ⇒ WebSocketConnection
Returns a new instance of WebSocketConnection.
3404 3405 3406 3407 3408 3409 3410 |
# File 'lib/octo/server/http_server.rb', line 3404 def initialize(socket, version) @socket = socket @version = version @send_mutex = Mutex.new @closed = false WebSocketConnection.apply_keepalive(socket) end |
Instance Attribute Details
#session_id ⇒ Object
Returns the value of attribute session_id.
3393 3394 3395 |
# File 'lib/octo/server/http_server.rb', line 3393 def session_id @session_id end |
Class Method Details
.apply_keepalive(socket) ⇒ Object
Enable TCP keepalive on the underlying socket so silently dead peers are detected in minutes instead of the OS default of hours. Best-effort: any failure is logged at debug level and ignored.
3524 3525 3526 3527 3528 3529 3530 3531 3532 3533 3534 3535 3536 3537 3538 3539 3540 3541 3542 3543 3544 3545 3546 3547 3548 3549 3550 |
# File 'lib/octo/server/http_server.rb', line 3524 def self.apply_keepalive(socket) return unless socket.respond_to?(:setsockopt) socket.setsockopt(Socket::SOL_SOCKET, Socket::SO_KEEPALIVE, true) # TCP-level keepalive tuning — constants vary by platform and are # only set when available. Values chosen to detect dead peers in # roughly 60-90 seconds total. if defined?(Socket::IPPROTO_TCP) # Idle time before first probe (Linux: TCP_KEEPIDLE, macOS: TCP_KEEPALIVE) idle_const = if Socket.const_defined?(:TCP_KEEPIDLE) Socket::TCP_KEEPIDLE elsif Socket.const_defined?(:TCP_KEEPALIVE) Socket::TCP_KEEPALIVE end socket.setsockopt(Socket::IPPROTO_TCP, idle_const, 60) if idle_const if Socket.const_defined?(:TCP_KEEPINTVL) socket.setsockopt(Socket::IPPROTO_TCP, Socket::TCP_KEEPINTVL, 10) end if Socket.const_defined?(:TCP_KEEPCNT) socket.setsockopt(Socket::IPPROTO_TCP, Socket::TCP_KEEPCNT, 3) end end rescue StandardError => e Octo::Logger.debug("[WS] failed to set keepalive: #{e.class}: #{e.}") end |
Instance Method Details
#closed? ⇒ Boolean
Returns true if the underlying socket has been detected as dead.
3413 3414 3415 |
# File 'lib/octo/server/http_server.rb', line 3413 def closed? @closed end |
#force_close! ⇒ Object
Force-close the connection (used by the interrupt watchdog when an Agent thread is stuck on an unresponsive socket write).
3419 3420 3421 3422 3423 3424 |
# File 'lib/octo/server/http_server.rb', line 3419 def force_close! @closed = true @socket.close rescue StandardError # best effort end |
#send_json(data) ⇒ Object
Send a JSON-serializable object over the WebSocket. Returns true on success, false if the connection is dead.
3428 3429 3430 3431 3432 3433 |
# File 'lib/octo/server/http_server.rb', line 3428 def send_json(data) send_raw(:text, JSON.generate(data)) rescue => e Octo::Logger.debug("WS send error (connection dead): #{e.}") false end |
#send_raw(type, data) ⇒ Object
Send a raw WebSocket frame. Returns true on success, false on broken/closed/sluggish socket.
Uses write_nonblock with an overall deadline so the caller (typically the Agent thread) never blocks longer than SEND_DEADLINE, even if the client silently stopped reading.
3441 3442 3443 3444 3445 3446 3447 3448 3449 3450 3451 3452 3453 3454 3455 3456 3457 3458 3459 3460 3461 3462 3463 3464 3465 3466 3467 3468 3469 3470 3471 3472 3473 3474 3475 3476 3477 3478 3479 3480 3481 3482 3483 3484 3485 3486 |
# File 'lib/octo/server/http_server.rb', line 3441 def send_raw(type, data) started_at = Process.clock_gettime(Process::CLOCK_MONOTONIC) @send_mutex.synchronize do return false if @closed outgoing = WebSocket::Frame::Outgoing::Server.new( version: @version, data: data, type: type ) bytes = outgoing.to_s unless write_with_deadline(bytes, SEND_DEADLINE) # Deadline exceeded — treat as a dead connection so broadcast # purges it and the Agent thread is freed immediately. @closed = true begin @socket.close rescue StandardError # ignore end Octo::Logger.warn( "[WS] send_raw deadline exceeded — closing sluggish connection " \ "(bytes=#{bytes.bytesize}, deadline=#{SEND_DEADLINE}s)" ) return false end end elapsed = Process.clock_gettime(Process::CLOCK_MONOTONIC) - started_at if elapsed > SEND_SLOW_WARN Octo::Logger.warn( "[WS] send_raw slow: #{elapsed.round(2)}s (type=#{type})" ) end true rescue Errno::EPIPE, Errno::ECONNRESET, IOError, Errno::EBADF => e @closed = true Octo::Logger.debug("WS send_raw error (client disconnected): #{e.}") false rescue => e @closed = true Octo::Logger.debug("WS send_raw unexpected error: #{e.}") false end |