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.

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

Uses V1::V1Dispatcher for registering and listening for durable events via bidirectional gRPC streaming.

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.



31
32
33
 
# File 'lib/hatchet/durable_context.rb', line 31

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:



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

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.



42
43
44
 
# File 'lib/hatchet/durable_context.rb', line 42

def engine_version
  @engine_version
end

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

Returns:



27
28
29
 
# File 'lib/hatchet/durable_context.rb', line 27

def eviction_manager
  @eviction_manager
end

#invocation_count ⇒ Integer

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

Returns:

  • (Integer) —

    Durable-task invocation count (>= 1).



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

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:



236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
 
# File 'lib/hatchet/durable_context.rb', line 236

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:



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
223
224
225
226
227
228
229
 
# File 'lib/hatchet/durable_context.rb', line 184

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



264
265
266
267
268
269
270
271
272
273
274
275
276
277
 
# File 'lib/hatchet/durable_context.rb', line 264

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.

Delegates to #wait_for with a SleepCondition so that both sleeps and event waits share a single registration / eviction path.

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



53
54
55
56
57
58
59
60
 
# File 'lib/hatchet/durable_context.rb', line 53

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) 


169
170
171
172
173
174
175
 
# File 'lib/hatchet/durable_context.rb', line 169

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.

Register the durable wait with send_event first, then start eviction tracking only while blocked on wait_for_callback.

Parameters:

  • key (String) —

    A unique key for this wait operation

  • condition (Object) —

    The condition to wait for (UserEventCondition, SleepCondition, Hash, etc.)

  • 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



72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
 
# File 'lib/hatchet/durable_context.rb', line 72

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