Class: Alplus::Worker

Inherits:
Object
  • Object
show all
Defined in:
lib/alplus/worker.rb

Overview

Background-thread sender with a bounded queue: #enqueue never blocks the caller. When the queue is full, the event is dropped (non- authoritative, matching the JS SDK's own batching drop behavior) rather than applying backpressure to the request thread (issue #14 story 7/8).

The worker thread is lazily started on first enqueue and is never joined/killed explicitly: Ruby terminates all non-main threads when the process exits, so there is nothing to supervise for a short-lived script or a Rack::Handler process to leak.

One Worker instance is ONE independent delivery lane: its own SizedQueue and its own background thread, fixed to one kind: (:error or :session) for its whole lifetime (issue #12 fix). A PRIOR version routed both kinds through a single shared queue/thread — a stalled or slow /e/errors POST (up to ~20s across Retry::MAX_ATTEMPTS retries) head-of-line-blocked every queued session behind it, and a full queue during an error storm silently dropped the next session, exactly when crash-free data matters most. Client now owns two Workers so error backpressure can never delay or drop session delivery, or vice versa.

Instance Method Summary collapse

Constructor Details

#initialize(config, transport, kind: :error) ⇒ Worker

Returns a new instance of Worker.



27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
 
# File 'lib/alplus/worker.rb', line 27

def initialize(config, transport, kind: :error)
  @config = config
  @transport = transport
  @kind = kind
  @queue = SizedQueue.new(config.max_queue_size)
  @mutex = Mutex.new
  @thread = nil
  # Outstanding work count: incremented (under `@mutex`) the moment an
  # envelope is successfully pushed, decremented only after
  # `Transport#send_envelope` returns. `SizedQueue#pop` removes an item
  # from the queue *before* the worker thread finishes sending it, so
  # `@queue.empty?` alone goes true while a send is still in flight —
  # a concurrent `#flush`/`#close` reading only queue emptiness would
  # return early (TOCTOU). Counting outstanding work instead of
  # sampling a flag set after pop closes that window: the increment
  # happens atomically with the enqueue that a caller already observed
  # succeeding, not after some later point the reader could race.
  @outstanding = 0
end

Instance Method Details

#enqueue(envelope) ⇒ Object

Enqueues an envelope for background delivery on THIS worker's own lane (its kind:, fixed at construction). Returns true if queued, false if dropped because the queue is full. Never raises.



50
51
52
53
54
55
56
57
58
59
60
 
# File 'lib/alplus/worker.rb', line 50

def enqueue(envelope)
  @mutex.synchronize do
    @queue.push(envelope, true)
    @outstanding += 1
  end
  ensure_thread_started
  true
rescue ThreadError
  @config.logger&.warn("[alplus] #{@kind} queue full (max #{@config.max_queue_size}); dropping event")
  false
end

#flush(timeout: 2) ⇒ Object

Blocks up to timeout seconds for the queue to drain and any in-flight send to finish. Returns true if it drained in time, false on timeout. Never raises.



65
66
67
68
69
 
# File 'lib/alplus/worker.rb', line 65

def flush(timeout: 2)
  deadline = Time.now + timeout
  sleep(0.01) while !idle? && Time.now < deadline
  idle?
end

#queue_size ⇒ Object 



71
72
73
 
# File 'lib/alplus/worker.rb', line 71

def queue_size
  @queue.size
end