Class: Hatchet::WorkerRuntime::DurableEventListener

Inherits:
Object
  • Object
show all
Defined in:
lib/hatchet/worker/durable_event_listener.rb,
sig/hatchet/worker/durable_event_listener.rbs

Overview

Thread-safe multiplexer over the V1Dispatcher.DurableTask bidirectional gRPC stream.

A single stream is shared across all durable task invocations running on the worker; callers send send_event / wait_for_callback / send_evict_invocation requests and block on per-call Queues until the response-dispatch thread routes the matching DurableTaskResponse back.

Examples:

listener = DurableEventListener.new(config: config, channel: channel, logger: logger)
listener.start("worker-id-123")
ack = listener.send_event(task_id, invocation_count, wait_for_event)
result = listener.wait_for_callback(task_id, invocation_count, branch_id, node_id)

Defined Under Namespace

Classes: MemoEvent, WaitForEvent

Constant Summary collapse

DEFAULT_RECONNECT_INTERVAL =

seconds

Returns:

  • (Integer) 
3
EVICTION_ACK_TIMEOUT_SECONDS =

Returns:

  • (Float) 
30.0
REGISTER_WORKER_ACK_TIMEOUT_SECONDS = 
10.0

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(config:, channel:, logger:, on_server_evict: nil) ⇒ DurableEventListener

Returns a new instance of DurableEventListener.

Parameters:

  • config (Hatchet::Config)
  • channel (GRPC::Core::Channel)
  • logger (Logger)
  • on_server_evict (Proc, nil) (defaults to: nil) —

    Called with (durable_task_external_id, invocation_count) when the server notifies about a stale invocation.

  • config: (Config)
  • channel: (Object)
  • logger: (Logger)
  • on_server_evict: (Proc, nil) (defaults to: nil) 


46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
 
# File 'lib/hatchet/worker/durable_event_listener.rb', line 46

def initialize(config:, channel:, logger:, on_server_evict: nil)
  @config = config
  @channel = channel
  @logger = logger
  @on_server_evict = on_server_evict

  @worker_id = nil
  @stub = nil
  @request_queue = nil

  @mu = Monitor.new

  # (task_external_id, invocation_count) => Queue (push [:ok, ack] or [:err, exc])
  @pending_event_acks = {}
  # (task_external_id, invocation_count) => Queue (push [:ok, nil] or [:err, exc])
  @pending_eviction_acks = {}
  # (task_external_id, invocation_count, branch_id, node_id) => Queue
  @pending_callbacks = {}
  # key -> [inserted_at, result] (rudimentary TTL cache)
  @buffered_completions = {}

  @running = false
  @start_mu = Mutex.new
  @registration_mu = Mutex.new
  @registration_cv = ConditionVariable.new
  @worker_registered = false

  @receive_thread = nil
  @send_thread = nil
end

Instance Attribute Details

#worker_id ⇒ String? (readonly)

Returns:

  • (String, nil) 


39
40
41
 
# File 'lib/hatchet/worker/durable_event_listener.rb', line 39

def worker_id
  @worker_id
end

Instance Method Details

#cleanup_task_state(durable_task_external_id, invocation_count) ⇒ void

This method returns an undefined value.

Drop pending callbacks / acks / buffered completions whose invocation count is <= invocation_count for the given task id.

Parameters:

  • durable_task_external_id (String)
  • invocation_count (Integer) 


239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
 
# File 'lib/hatchet/worker/durable_event_listener.rb', line 239

def cleanup_task_state(durable_task_external_id, invocation_count)
  @mu.synchronize do
    @pending_callbacks.each_key do |k|
      next unless k[0] == durable_task_external_id && k[1] <= invocation_count

      @pending_callbacks.delete(k)&.close
    end

    @pending_event_acks.each_key do |k|
      next unless k[0] == durable_task_external_id && k[1] <= invocation_count

      @pending_event_acks.delete(k)&.close
    end

    @buffered_completions.each_key do |k|
      next unless k[0] == durable_task_external_id && k[1] <= invocation_count

      @buffered_completions.delete(k)
    end
  end
end

#ensure_started(worker_id) ⇒ void

This method returns an undefined value.

Start the listener if not already running.

Parameters:

  • worker_id (String) 


97
98
99
 
# File 'lib/hatchet/worker/durable_event_listener.rb', line 97

def ensure_started(worker_id)
  start(worker_id) unless @running
end

#handle_response_for_test(response) ⇒ void

This method returns an undefined value.

Hook for tests: handle a single response message (bypassing the network).

Parameters:

  • response (Object) 


262
263
264
 
# File 'lib/hatchet/worker/durable_event_listener.rb', line 262

def handle_response_for_test(response)
  handle_response(response)
end

#send_event(durable_task_external_id, invocation_count, event) ⇒ Object

Send a DurableTask message and block for its ack.

Parameters:

  • durable_task_external_id (String)
  • invocation_count (Integer)
  • event (WaitForEvent, MemoEvent) —

    The event to send

Returns:

  • (Object) —

    The parsed ack body (a simple Hash describing the ack)

Raises:

  • (Hatchet::Error) —

    on server-reported errors or listener disconnection



119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
 
# File 'lib/hatchet/worker/durable_event_listener.rb', line 119

def send_event(durable_task_external_id, invocation_count, event)
  raise Hatchet::Error, "DurableEventListener not started" unless @request_queue

  key = [durable_task_external_id, invocation_count]
  queue = Queue.new

  @mu.synchronize { @pending_event_acks[key] = queue }

  request = build_event_request(durable_task_external_id, invocation_count, event)
  @logger&.debug(
    "durable event listener send_event: task=#{durable_task_external_id} " \
    "invocation=#{invocation_count} event=#{event.class}",
  )
  @request_queue << request

  ack = await_queue(queue)
  @logger&.debug(
    "durable event listener send_event ack: task=#{durable_task_external_id} " \
    "invocation=#{invocation_count} ack_type=#{ack[:ack_type]} " \
    "branch_id=#{ack[:branch_id]} node_id=#{ack[:node_id]}",
  )
  ack
end

#send_evict_invocation(durable_task_external_id, invocation_count, reason: nil) ⇒ void

This method returns an undefined value.

Request eviction of a stale invocation from the server and block until ack.

Parameters:

  • durable_task_external_id (String)
  • invocation_count (Integer)
  • reason (String, nil) (defaults to: nil) —

    Optional human-readable reason.

  • reason: (String, nil) (defaults to: nil)

Raises:



186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
 
# File 'lib/hatchet/worker/durable_event_listener.rb', line 186

def send_evict_invocation(durable_task_external_id, invocation_count, reason: nil)
  raise Hatchet::Error, "DurableEventListener not started" unless @request_queue

  key = [durable_task_external_id, invocation_count]
  queue = Queue.new
  @mu.synchronize { @pending_eviction_acks[key] = queue }

  args = {
    durable_task_external_id: durable_task_external_id,
    invocation_count: invocation_count,
  }
  args[:reason] = reason if reason
  req = ::V1::DurableTaskEvictInvocationRequest.new(**args)

  @logger&.debug(
    "durable event listener send_evict_invocation: task=#{durable_task_external_id} " \
    "invocation=#{invocation_count} reason=#{reason}",
  )
  @request_queue << ::V1::DurableTaskRequest.new(evict_invocation: req)

  await_queue(queue, timeout: EVICTION_ACK_TIMEOUT_SECONDS)
  @logger&.debug(
    "durable event listener send_evict_invocation ack: task=#{durable_task_external_id} " \
    "invocation=#{invocation_count}",
  )
rescue Timeout::Error
  @mu.synchronize { @pending_eviction_acks.delete(key) }
  raise Hatchet::Error,
        "Eviction ack timed out after #{EVICTION_ACK_TIMEOUT_SECONDS.to_i}s " \
        "for task #{durable_task_external_id} invocation #{invocation_count}"
end

#send_memo_completed_notification(durable_task_external_id:, node_id:, branch_id:, invocation_count:, memo_key:, memo_result_payload:) ⇒ void

This method returns an undefined value.

Fire-and-forget complete_memo notification.

Parameters:

  • durable_task_external_id: (String)
  • node_id: (Integer)
  • branch_id: (Integer)
  • invocation_count: (Integer)
  • memo_key: (String)
  • memo_result_payload: (String)

Raises:



219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
 
# File 'lib/hatchet/worker/durable_event_listener.rb', line 219

def send_memo_completed_notification(durable_task_external_id:, node_id:, branch_id:, invocation_count:, memo_key:,
                                     memo_result_payload:)
  raise Hatchet::Error, "DurableEventListener not started" unless @request_queue

  ref = ::V1::DurableEventLogEntryRef.new(
    durable_task_external_id: durable_task_external_id,
    node_id: node_id,
    invocation_count: invocation_count,
    branch_id: branch_id,
  )
  complete = ::V1::DurableTaskCompleteMemoRequest.new(
    ref: ref,
    memo_key: memo_key,
    payload: memo_result_payload,
  )
  @request_queue << ::V1::DurableTaskRequest.new(complete_memo: complete)
end

#start(worker_id) ⇒ void

This method returns an undefined value.

Start the listener if not already running. Idempotent.

Parameters:

  • worker_id (String) 


80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
 
# File 'lib/hatchet/worker/durable_event_listener.rb', line 80

def start(worker_id)
  @start_mu.synchronize do
    return if @running

    @worker_id = worker_id
    @running = true
    @registration_mu.synchronize { @worker_registered = false }

    connect

    @receive_thread = Thread.new { receive_loop }
    @send_thread = Thread.new { send_loop }
    wait_for_register_worker_ack
  end
end

#stop ⇒ void

This method returns an undefined value.

Stop the listener and release resources.



102
103
104
105
106
107
108
109
110
 
# File 'lib/hatchet/worker/durable_event_listener.rb', line 102

def stop
  @running = false

  fail_all_pending(Hatchet::Error.new("DurableListener stopped"))

  @request_queue&.close
  rescue_thread(@receive_thread)
  rescue_thread(@send_thread)
end

#wait_for_callback(durable_task_external_id, invocation_count, branch_id, node_id) ⇒ Hash

Block until the server delivers an entry_completed (or error) for this durable task / invocation / branch / node id tuple.

Parameters:

  • durable_task_external_id (String)
  • invocation_count (Integer)
  • branch_id (Integer)
  • node_id (Integer)

Returns:

  • (Hash) —

    { durable_task_external_id:, node_id:, payload: }



147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
 
# File 'lib/hatchet/worker/durable_event_listener.rb', line 147

def wait_for_callback(durable_task_external_id, invocation_count, branch_id, node_id)
  key = [durable_task_external_id, invocation_count, branch_id, node_id]

  buffered = @mu.synchronize { @buffered_completions.delete(key) }
  if buffered
    @logger&.debug(
      "durable event listener wait_for_callback: buffered completion hit " \
      "task=#{durable_task_external_id} invocation=#{invocation_count} " \
      "branch_id=#{branch_id} node_id=#{node_id}",
    )
    return buffered[1]
  end

  queue = @mu.synchronize do
    @pending_callbacks[key] ||= Queue.new
  end

  @logger&.debug(
    "durable event listener wait_for_callback: waiting " \
    "task=#{durable_task_external_id} invocation=#{invocation_count} " \
    "branch_id=#{branch_id} node_id=#{node_id}",
  )
  poll_worker_status

  result = await_queue(queue)
  @logger&.debug(
    "durable event listener wait_for_callback: completed " \
    "task=#{durable_task_external_id} invocation=#{invocation_count} " \
    "branch_id=#{branch_id} node_id=#{node_id}",
  )
  result
end