Class: Hatchet::DurableContext

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

Overview

Extended context for durable tasks that supports sleep and event-waiting across task suspensions. All methods and attributes of Context are also available.

Durable tasks can be suspended and resumed by the Hatchet engine, allowing long-running workflows that survive process restarts.

Examples:

Sleep for a duration

hatchet.durable_task(name: "my_task") do |input, ctx|
  ctx.sleep_for(duration: 60) # sleep for 60 seconds
end

Wait for an event

hatchet.durable_task(name: "my_task") do |input, ctx|
  result = ctx.wait_for("event", Hatchet::UserEventCondition.new(event_key: "user:update"))
end

Instance Attribute Summary collapse

Attributes inherited from Context

#additional_metadata, #attempt_number, #deps, #filter_payload, #priority, #retry_count, #step_run_id, #worker_id, #workflow_run_id

Instance Method Summary collapse

Methods inherited from Context

#cancel, #cancelled?, #get_task_run_error, #initialize, #log, #put_stream, #refresh_timeout, #release_slot, #task_output, #task_run_errors, #was_skipped?, #worker

Constructor Details

This class inherits a constructor from Hatchet::Context

Instance Attribute Details

#action_key ⇒ String?

Returns The action key used by the eviction manager to identify this run invocation.

Returns:

  • (String, nil) —

    The action key used by the eviction manager to identify this run invocation.



29
30
31
 
# File 'lib/hatchet/durable_context.rb', line 29

def action_key
  @action_key
end

#durable_event_listener ⇒ Hatchet::WorkerRuntime::DurableEventListener?

Returns New-style bidi listener. When set the context delegates through it instead of the legacy RegisterDurableEvent/ListenForDurableEvent path.

Returns:



34
35
36
 
# File 'lib/hatchet/durable_context.rb', line 34

def durable_event_listener
  @durable_event_listener
end

#engine_version ⇒ String?

Returns Engine version string advertised via GetVersion.

Returns:

  • (String, nil) —

    Engine version string advertised via GetVersion.



40
41
42
 
# File 'lib/hatchet/durable_context.rb', line 40

def engine_version
  @engine_version
end

#eviction_manager ⇒ Hatchet::WorkerRuntime::DurableEviction::DurableEvictionManager?

Returns:



25
26
27
 
# File 'lib/hatchet/durable_context.rb', line 25

def eviction_manager
  @eviction_manager
end

#invocation_count ⇒ Integer

Returns Durable-task invocation count (>= 1).

Returns:

  • (Integer) —

    Durable-task invocation count (>= 1).



37
38
39
 
# File 'lib/hatchet/durable_context.rb', line 37

def invocation_count
  @invocation_count
end

Instance Method Details

#build_durable_conditions(key, condition) ⇒ V1::DurableEventListenerConditions

Build DurableEventListenerConditions from a condition object.

Parameters:

  • key (String) —

    The signal key

  • condition (Object) —

    The condition (UserEventCondition, SleepCondition, OrCondition, Hash, etc.)

Returns:



229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
 
# File 'lib/hatchet/durable_context.rb', line 229

def build_durable_conditions(key, condition)
  sleep_conditions = []
  user_event_conditions = []

  if condition.is_a?(Hatchet::OrCondition)
    # All conditions in an OR group share the same or_group_id
    or_group_id = SecureRandom.uuid
    condition.conditions.each do |cond|
      process_durable_condition(key, cond, or_group_id, sleep_conditions, user_event_conditions)
    end
  else
    process_durable_condition(key, condition, SecureRandom.uuid, sleep_conditions, user_event_conditions)
  end

  ::V1::DurableEventListenerConditions.new(
    sleep_conditions: sleep_conditions,
    user_event_conditions: user_event_conditions,
  )
end

#listen_for_event(signal_key) ⇒ V1::DurableEvent?

Listen for a durable event using bidirectional streaming.

In Ruby's grpc gem, bidi streams use an Enumerator for requests and return an Enumerator for responses.

Parameters:

  • signal_key (String) —

    The signal key to listen for

Returns:



177
178
179
180
181
182
183
184
185
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
217
218
219
220
221
222
 
# File 'lib/hatchet/durable_context.rb', line 177

def listen_for_event(signal_key)
  # Create a request enumerator for the bidi stream
  request_queue = Queue.new
  request_enum = Enumerator.new do |yielder|
    # Send initial request
    yielder << ::V1::ListenForDurableEventRequest.new(
      task_id: @step_run_id,
      signal_key: signal_key,
    )

    # Keep the stream alive until we get a response
    loop do
      msg = request_queue.pop
      break if msg == :done

      yielder << msg
    end
  end

  # Start the bidi stream
  response_stream = v1_dispatcher_stub.listen_for_durable_event(
    request_enum,
    metadata: @client.config.auth_metadata,
  )

  # Wait for the first matching response
  result = nil
  response_stream.each do |event|
    if event.signal_key == signal_key
      result = event
      break
    end
  end

  # Signal the request stream to close
  request_queue << :done

  result
rescue StandardError => e
  begin
    request_queue << :done
  rescue StandardError
    nil
  end
  raise e
end

#process_durable_condition(key, condition, or_group_id, sleep_conditions, user_event_conditions) ⇒ void

This method returns an undefined value.

Process a single condition into the appropriate proto lists. Delegates to ConditionConverter for shared logic.

Parameters:

  • key (String) —

    The signal key

  • condition (Object) —

    The condition to process

  • or_group_id (String) —

    The OR group ID for this condition

  • sleep_conditions (Array) —

    Accumulator for sleep conditions

  • user_event_conditions (Array) —

    Accumulator for user event conditions



257
258
259
260
261
262
263
264
265
266
267
268
269
270
 
# File 'lib/hatchet/durable_context.rb', line 257

def process_durable_condition(key, condition, or_group_id, sleep_conditions, user_event_conditions)
  ConditionConverter.convert_condition(
    condition,
    # Do not force base.action. Leaving it unset keeps protobuf default semantics on the server path.
    action: nil,
    sleep_conditions: sleep_conditions,
    user_event_conditions: user_event_conditions,
    or_group_id: or_group_id,
    readable_data_key: key,
    proto_method: :to_durable_proto,
    proto_arg: key,
    config: @client&.config,
  )
end

#sleep_for(duration:, label: nil) ⇒ Hash?

Sleep for a specified duration. The task is suspended and resumed by the engine after the duration expires, so no worker slot is blocked while sleeping (subject to the task's eviction policy).

Parameters:

  • duration (Integer, String) —

    Duration in seconds, or a duration string (e.g. "60s")

  • label (String, nil) (defaults to: nil) —

    Optional wait label shown in durable event logs.

  • duration: (Integer, String)
  • label: (String, nil) (defaults to: nil)

Returns:

  • (Hash, nil) —

    Result from the sleep event



49
50
51
52
53
54
55
56
 
# File 'lib/hatchet/durable_context.rb', line 49

def sleep_for(duration:, label: nil)
  duration_str = duration.is_a?(String) ? duration : "#{duration}s"
  duration_value = duration.is_a?(String) ? duration : duration.to_i
  wait_index = increment_wait_index
  signal_key = "sleep:#{duration_str}-#{wait_index}"

  wait_for(signal_key, Hatchet::SleepCondition.new(duration_value), label: label)
end

#v1_dispatcher_stub ⇒ Object

Get or create the V1::V1Dispatcher::Stub for durable events.

Returns:

  • (Object) 


162
163
164
165
166
167
168
 
# File 'lib/hatchet/durable_context.rb', line 162

def v1_dispatcher_stub
  @v1_dispatcher_stub ||= ::V1::V1Dispatcher::Stub.new(
    @client.config.host_port,
    nil,
    channel_override: @client.channel,
  )
end

#wait_for(key, condition, label: nil) ⇒ Hash

Wait for a condition to be met (event or sleep). The task is suspended and resumed when the condition is satisfied.

Parameters:

  • key (String) —

    A unique key for this wait operation

  • condition (UserEventCondition, SleepCondition, OrCondition) —

    The condition to wait for

  • label (String, nil) (defaults to: nil) —

    Optional wait label shown in durable event logs.

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

Returns:

  • (Hash) —

    Result from the wait, including which condition was satisfied



65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
 
# File 'lib/hatchet/durable_context.rb', line 65

def wait_for(key, condition, label: nil)
  conditions = build_durable_conditions(key, condition)

  if supports_durable_eviction?
    invocation = @invocation_count || 1

    event = Hatchet::WorkerRuntime::DurableEventListener::WaitForEvent.new(
      wait_for_conditions: conditions,
      label: label,
    )

    ack = @durable_event_listener.send_event(@step_run_id, invocation, event)

    with_eviction_wait(wait_kind: "wait_for", resource_id: key) do
      result = @durable_event_listener.wait_for_callback(
        @step_run_id,
        invocation,
        ack[:branch_id],
        ack[:node_id],
      )

      raise Hatchet::TaskRunError, result[:error_message] || "child task failed" if result[:is_failure]

      result[:payload] || {}
    end
  else
    with_eviction_wait(wait_kind: "wait_for", resource_id: key) do
      legacy_wait_for(key, conditions)
    end
  end
end