Class: RubyReactor::AsyncWaiter

Inherits:
Object
  • Object
show all
Defined in:
lib/ruby_reactor/async_waiter.rb

Overview

The shared wait core behind result(:name) for an async_step or an async_reactor that has not finished yet. Both callers hand it a pub/sub channel and a callable that returns the terminal value (or nil while the work is still in flight), so there is exactly one implementation of the wait.

The contract is "durable record answers, signal only hurries":

* the completing side writes its durable outcome FIRST, then publishes;
* this side checks the durable target, then blocks until either the signal
arrives or a coarse fallback interval elapses, then re-checks.

Redis pub/sub is at-most-once and unpersisted, so a dropped signal must never cost correctness — only fallback latency. Every exit path here goes through the durable check, and the whole thing is bounded: it raises rather than hanging (SC-005).

Constant Summary collapse

FALLBACK_BOUNDS =

Latency backstop for a lost signal, not a tuning surface — derived from the timeout rather than configured (Principle V). The clamp guarantees ~10 re-checks inside any bound, so a dropped notification costs at most ~10% of the wait, and never re-checks hotter than once a second.

(1.0..5.0)

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(channel:, timeout: nil, &terminal_check) ⇒ AsyncWaiter

Returns a new instance of AsyncWaiter.

Parameters:

  • completion-signal channel to listen on

  • (defaults to: nil)

    seconds; defaults to async_wait_timeout

Yield Returns:

  • (Object, nil) —

    the terminal value, or nil while still pending



31
32
33
34
35
36
37
38
 
# File 'lib/ruby_reactor/async_waiter.rb', line 31

def initialize(channel:, timeout: nil, &terminal_check)
  @channel = channel
  @timeout = timeout || RubyReactor.configuration.async_wait_timeout
  @terminal_check = terminal_check
  @mutex = Mutex.new
  @condition = ConditionVariable.new
  @signalled = false
end

Instance Attribute Details

#channel ⇒ Object (readonly)

Returns the value of attribute channel.



26
27
28
 
# File 'lib/ruby_reactor/async_waiter.rb', line 26

def channel
  @channel
end

#timeout ⇒ Object (readonly)

Returns the value of attribute timeout.



26
27
28
 
# File 'lib/ruby_reactor/async_waiter.rb', line 26

def timeout
  @timeout
end

Instance Method Details

#wait ⇒ Object 



40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
 
# File 'lib/ruby_reactor/async_waiter.rb', line 40

def wait
  deadline = monotonic + @timeout
  subscriber = start_subscriber

  loop do
    value = @terminal_check.call
    return value unless value.nil?

    remaining = deadline - monotonic
    raise timeout_error if remaining <= 0

    block_until_signalled_or([fallback_interval, remaining].min)
  end
ensure
  # ponytail: killing the thread is enough — `subscribe`'s own ensure closes
  # the dedicated connection on the way out.
  subscriber&.kill
end