Class: Quonfig::SSEConfigClient::ReadDeadlineWatchdog

Inherits:
Object
  • Object
show all
Defined in:
lib/quonfig/sse_config_client.rb

Overview

Background watchdog that interrupts the worker thread if no chunk arrives within deadline_s seconds. Uses Thread#raise — the only reliable cross-platform way to unblock a Ruby thread blocked in +Net::HTTP+'s body-read on macOS. (Closing or shutting down the underlying socket from another thread does NOT wake the reader on macOS; the kernel discards future reads but the in-flight syscall stays blocked until something else trips. sdk-go and sdk-node solve the equivalent problem with context cancellation / AbortController, which Ruby lacks at the IO layer.) Thread#raise is essentially what Timeout.timeout does internally; using it directly avoids Timeout.timeout's sketch reputation around ensure blocks.

Constant Summary collapse

POLL_INTERVAL = 
0.25

Instance Method Summary collapse

Constructor Details

#initialize(worker:, deadline_s:, stopped:, logger:) ⇒ ReadDeadlineWatchdog

Returns a new instance of ReadDeadlineWatchdog.



466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
 
# File 'lib/quonfig/sse_config_client.rb', line 466

def initialize(worker:, deadline_s:, stopped:, logger:)
  @worker = worker
  @deadline_s = deadline_s
  @stopped = stopped
  @logger = logger
  @active = true
  # Mutex covers @active AND the decision to fire Thread#raise. stop()
  # holds the mutex when flipping @active false, so a +stop+ that
  # arrives mid-deadline-check cannot lose the race against the
  # watchdog's @worker.raise call (which would inject a spurious
  # SSEReadDeadlineExceeded into the worker thread right after a
  # clean read_body return).
  @mutex = Mutex.new
  @last_read_at = Concurrent::AtomicReference.new(Process.clock_gettime(Process::CLOCK_MONOTONIC))
end

Instance Method Details

#reset! ⇒ Object 



486
487
488
 
# File 'lib/quonfig/sse_config_client.rb', line 486

def reset!
  @last_read_at.set(Process.clock_gettime(Process::CLOCK_MONOTONIC))
end

#start ⇒ Object 



482
483
484
 
# File 'lib/quonfig/sse_config_client.rb', line 482

def start
  @thread = Thread.new { watch }
end

#stop ⇒ Object 



490
491
492
493
494
 
# File 'lib/quonfig/sse_config_client.rb', line 490

def stop
  @mutex.synchronize { @active = false }
  @thread&.join(1)
  @thread = nil
end