Class: Restate::Server::Context

Inherits:
Object
  • Object
show all
Includes:
WorkflowContext, WorkflowSharedContext
Defined in:
lib/restate/server/context.rb,
sig/restate.rbs

Overview

The core execution context for a Restate handler invocation. Implements the progress loop and all context API methods (state, run, sleep, call, send).

Concurrency model:

- The handler runs inside a Fiber managed by Falcon/Async.
- `run` blocks spawn child Async tasks.
- When the progress loop needs input, it dequeues from @input_queue, yielding the Fiber.
- The HTTP input reader (a separate Async task) feeds chunks into @input_queue.
- Output chunks are written directly to the streaming response body.

Defined Under Namespace

Modules: BackgroundPool

Constant Summary collapse

LOGGER = 
Logger.new($stdout, progname: 'Restate::Server::Context')

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(vm:, handler:, invocation:, send_output:, input_queue:, middleware: [], outbound_middleware: []) ⇒ Context

Returns a new instance of Context.



27
28
29
30
31
32
33
34
35
36
37
38
 
# File 'lib/restate/server/context.rb', line 27

def initialize(vm:, handler:, invocation:, send_output:, input_queue:, middleware: [],
               outbound_middleware: [])
  @vm = vm
  @handler = handler
  @invocation = invocation
  @send_output = send_output
  @input_queue = input_queue
  @run_coros_to_execute = {}
  @attempt_finished_event = AttemptFinishedEvent.new
  @middleware = middleware
  @outbound_middleware = outbound_middleware
end

Instance Attribute Details

#invocation ⇒ Object (readonly)

Returns the value of attribute invocation.



25
26
27
 
# File 'lib/restate/server/context.rb', line 25

def invocation
  @invocation
end

#vm ⇒ Object (readonly)

Returns the value of attribute vm.



25
26
27
 
# File 'lib/restate/server/context.rb', line 25

def vm
  @vm
end

Instance Method Details

#all(*futures) ⇒ Object

Build a lazy combinator over the given futures and return it as a CombinedFuture. Nothing blocks until .await is called, so the combinators compose: Restate.race(Restate.all(a, b), c).await.

Semantics match JS Promise.all — when awaited, returns the values in input order, short-circuiting on the first TerminalError.



178
179
180
181
 
# File 'lib/restate/server/context.rb', line 178

def all(*futures)
  futures = futures.first if futures.length == 1 && futures.first.is_a?(Array)
  CombinedFuture.new(self, :all_succeeded_or_first_failed, futures)
end

#all_settled(*futures) ⇒ Object

Lazy combinator. Awaiting waits for every future to settle and returns an Array of { status: :fulfilled, value: ... } or { status: :rejected, reason: TerminalError } entries, in input order. Matches JS Promise.allSettled.



205
206
207
208
 
# File 'lib/restate/server/context.rb', line 205

def all_settled(*futures)
  futures = futures.first if futures.length == 1 && futures.first.is_a?(Array)
  CombinedFuture.new(self, :all_completed, futures)
end

#any(*futures) ⇒ Object

Lazy combinator. Awaiting returns the value of the first successful future; raises only if every future fails terminally. Matches JS Promise.any.

Raises:

  • (ArgumentError) 


194
195
196
197
198
199
 
# File 'lib/restate/server/context.rb', line 194

def any(*futures)
  futures = futures.first if futures.length == 1 && futures.first.is_a?(Array)
  raise ArgumentError, 'any requires at least one future' if futures.empty?

  CombinedFuture.new(self, :first_succeeded_or_all_failed, futures)
end

#awakeable(serde: JsonSerde) ⇒ Object

Creates an awakeable and returns [awakeable_id, DurableFuture].



355
356
357
358
 
# File 'lib/restate/server/context.rb', line 355

def awakeable(serde: JsonSerde)
  id, handle = @vm.sys_awakeable
  [id, DurableFuture.new(self, handle, serde: serde)]
end

#cancel_invocation(invocation_id) ⇒ Object

Requests cancellation of another invocation by its id.



426
427
428
 
# File 'lib/restate/server/context.rb', line 426

def cancel_invocation(invocation_id)
  @vm.sys_cancel_invocation(invocation_id)
end

#clear(name) ⇒ Object

Durably removes a single state entry by name.



114
115
116
 
# File 'lib/restate/server/context.rb', line 114

def clear(name)
  @vm.sys_clear_state(name)
end

#clear_all ⇒ Object

Durably removes all state entries for this virtual object or workflow.



119
120
121
 
# File 'lib/restate/server/context.rb', line 119

def clear_all
  @vm.sys_clear_all_state
end

#completed?(handle) ⇒ Boolean

Check if a handle is completed (non-blocking).

Returns:

  • (Boolean) 


155
156
157
 
# File 'lib/restate/server/context.rb', line 155

def completed?(handle)
  @vm.is_completed(handle)
end

#enter ⇒ Object

Runs the handler to completion, writing the output (or failure) to the journal.



43
44
45
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
76
77
78
79
80
81
82
83
84
85
86
 
# File 'lib/restate/server/context.rb', line 43

def enter
  Thread.current[:restate_context] = self
  Thread.current[:restate_service_kind] = @handler.service_tag.kind
  Thread.current[:restate_handler_kind] = @handler.kind
  in_buffer = @invocation.input_buffer
  out_buffer = Restate.invoke_handler(handler: @handler, ctx: self, in_buffer: in_buffer,
                                      middleware: @middleware)
  @vm.sys_write_output_success(out_buffer)
  @vm.sys_end
rescue TerminalError => e
  failure = Failure.new(code: e.status_code, message: e.message, metadata: e.metadata)
  @vm.sys_write_output_failure(failure)
  @vm.sys_end
rescue SuspendedError, InternalError
  # These are expected internal control flow exceptions; do nothing.
rescue DisconnectedError
  raise
rescue StandardError => e
  # Walk the cause chain for TerminalError or internal exceptions
  cause = e
  handled = false
  while cause
    if cause.is_a?(TerminalError)
      f = Failure.new(code: cause.status_code, message: cause.message, metadata: cause.metadata)
      @vm.sys_write_output_failure(f)
      @vm.sys_end
      handled = true
      break
    elsif cause.is_a?(SuspendedError) || cause.is_a?(InternalError)
      handled = true
      break
    end
    cause = cause.cause
  end
  unless handled
    @vm.notify_error(e.inspect, e.backtrace&.join("\n"))
    raise
  end
ensure
  @run_coros_to_execute.clear
  Thread.current[:restate_context] = nil
  Thread.current[:restate_service_kind] = nil
  Thread.current[:restate_handler_kind] = nil
end

#generic_call(service, handler, arg, key: nil, idempotency_key: nil, headers: nil, scope: nil, limit_key: nil) ⇒ Object

Durably calls a handler using raw bytes (no serialization). Useful for proxying.

Optionally accepts a scope to route the call within (see #scope) and a limit_key to enforce hierarchical concurrency limits within that scope.



436
437
438
439
440
441
442
443
444
445
446
 
# File 'lib/restate/server/context.rb', line 436

def generic_call(service, handler, arg, key: nil, idempotency_key: nil, headers: nil,
                 scope: nil, limit_key: nil)
  with_outbound_middleware(service, handler, headers) do |hdrs|
    call_handle = @vm.sys_call(
      service: service, handler: handler, parameter: arg,
      key: key, idempotency_key: idempotency_key, headers: hdrs, scope: scope, limit_key: limit_key
    )
    DurableCallFuture.new(self, call_handle.result_handle, call_handle.invocation_id_handle,
                          output_serde: nil)
  end
end

#generic_send(service, handler, arg, key: nil, delay: nil, idempotency_key: nil, headers: nil, scope: nil, limit_key: nil) ⇒ Object

Sends a one-way invocation using raw bytes (no serialization). Useful for proxying.

Optionally accepts a scope to route the send within (see #scope) and a limit_key to enforce hierarchical concurrency limits within that scope.



452
453
454
455
456
457
458
459
460
461
462
463
 
# File 'lib/restate/server/context.rb', line 452

def generic_send(service, handler, arg, key: nil, delay: nil, idempotency_key: nil, headers: nil,
                 scope: nil, limit_key: nil)
  delay_ms = delay ? (delay * 1000).to_i : nil
  with_outbound_middleware(service, handler, headers) do |hdrs|
    invocation_id_handle = @vm.sys_send(
      service: service, handler: handler, parameter: arg,
      key: key, delay: delay_ms, idempotency_key: idempotency_key, headers: hdrs,
      scope: scope, limit_key: limit_key
    )
    SendHandle.new(self, invocation_id_handle)
  end
end

#get(name, serde: JsonSerde) ⇒ Object

Durably retrieves a state entry by name. Returns nil if unset.



98
99
100
 
# File 'lib/restate/server/context.rb', line 98

def get(name, serde: JsonSerde)
  get_async(name, serde: serde).await
end

#get_async(name, serde: JsonSerde) ⇒ Object

Returns a DurableFuture for a state entry. Resolves to nil if unset.



103
104
105
106
 
# File 'lib/restate/server/context.rb', line 103

def get_async(name, serde: JsonSerde)
  handle = @vm.sys_get_state(name)
  DurableFuture.new(self, handle, serde: serde)
end

#key ⇒ Object

Returns the key for this virtual object or workflow invocation.



489
490
491
 
# File 'lib/restate/server/context.rb', line 489

def key
  @invocation.key
end

#object_call(service, handler, key, arg, idempotency_key: nil, headers: nil, input_serde: NOT_SET, output_serde: NOT_SET, scope: nil, limit_key: nil) ⇒ Object

Durably calls a handler on a Restate virtual object, keyed by key.

scope and limit_key are internal parameters set by ScopedContext (via ctx.scope(...)); prefer that API over passing them directly.



289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
 
# File 'lib/restate/server/context.rb', line 289

def object_call(service, handler, key, arg, idempotency_key: nil, headers: nil,
                input_serde: NOT_SET, output_serde: NOT_SET, scope: nil, limit_key: nil)
  svc_name, handler_name, handler_meta = resolve_call_target(service, handler)
  in_serde = resolve_serde(input_serde, handler_meta, :input_serde)
  out_serde = resolve_serde(output_serde, handler_meta, :output_serde)
  parameter = in_serde.serialize(arg)
  with_outbound_middleware(svc_name, handler_name, headers, handler_meta: handler_meta) do |hdrs|
    call_handle = @vm.sys_call(
      service: svc_name, handler: handler_name, parameter: parameter,
      key: key, idempotency_key: idempotency_key, headers: hdrs, scope: scope, limit_key: limit_key
    )
    DurableCallFuture.new(self, call_handle.result_handle, call_handle.invocation_id_handle,
                          output_serde: out_serde)
  end
end

#object_send(service, handler, key, arg, delay: nil, idempotency_key: nil, headers: nil, input_serde: NOT_SET, scope: nil, limit_key: nil) ⇒ Object

Sends a one-way invocation to a Restate virtual object handler (fire-and-forget).

scope and limit_key are internal parameters set by ScopedContext (via ctx.scope(...)); prefer that API over passing them directly.



309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
 
# File 'lib/restate/server/context.rb', line 309

def object_send(service, handler, key, arg, delay: nil, idempotency_key: nil, headers: nil,
                input_serde: NOT_SET, scope: nil, limit_key: nil)
  svc_name, handler_name, handler_meta = resolve_call_target(service, handler)
  in_serde = resolve_serde(input_serde, handler_meta, :input_serde)
  parameter = in_serde.serialize(arg)
  delay_ms = delay ? (delay * 1000).to_i : nil
  with_outbound_middleware(svc_name, handler_name, headers, handler_meta: handler_meta) do |hdrs|
    invocation_id_handle = @vm.sys_send(
      service: svc_name, handler: handler_name, parameter: parameter,
      key: key, delay: delay_ms, idempotency_key: idempotency_key, headers: hdrs,
      scope: scope, limit_key: limit_key
    )
    SendHandle.new(self, invocation_id_handle)
  end
end

#on_attempt_finished ⇒ Object

Called by the server when the attempt ends (handler completed, disconnected, or transient error). Signals the attempt_finished_event so that user code and background pool jobs can clean up.



91
92
93
 
# File 'lib/restate/server/context.rb', line 91

def on_attempt_finished
  @attempt_finished_event.set!
end

#peek_promise(name, serde: JsonSerde) ⇒ Object

Peeks at a durable promise value without blocking. Returns nil if not yet resolved.



401
402
403
404
405
406
 
# File 'lib/restate/server/context.rb', line 401

def peek_promise(name, serde: JsonSerde)
  handle = @vm.sys_peek_promise(name)
  poll_and_take(handle) do |raw|
    raw.nil? ? nil : serde.deserialize(raw)
  end
end

#promise(name, serde: JsonSerde) ⇒ Object

Gets a durable promise value, blocking until resolved.



393
394
395
396
397
398
 
# File 'lib/restate/server/context.rb', line 393

def promise(name, serde: JsonSerde)
  handle = @vm.sys_get_promise(name)
  poll_and_take(handle) do |raw|
    raw.nil? ? nil : serde.deserialize(raw)
  end
end

#race(*futures) ⇒ Object

Lazy combinator. Awaiting returns the value of the first future to settle, or raises if it settled with a TerminalError. Matches JS Promise.race.

Raises:

  • (ArgumentError) 


185
186
187
188
189
190
 
# File 'lib/restate/server/context.rb', line 185

def race(*futures)
  futures = futures.first if futures.length == 1 && futures.first.is_a?(Array)
  raise ArgumentError, 'race requires at least one future' if futures.empty?

  CombinedFuture.new(self, :first_completed, futures)
end

#reject_awakeable(awakeable_id, message, code: 500) ⇒ Object

Rejects an awakeable with a terminal failure.



366
367
368
369
 
# File 'lib/restate/server/context.rb', line 366

def reject_awakeable(awakeable_id, message, code: 500)
  failure = Failure.new(code: code, message: message)
  @vm.sys_complete_awakeable_failure(awakeable_id, failure)
end

#reject_promise(name, message, code: 500) ⇒ Object

Rejects a durable promise with a terminal failure.



416
417
418
419
420
421
 
# File 'lib/restate/server/context.rb', line 416

def reject_promise(name, message, code: 500)
  failure = Failure.new(code: code, message: message)
  handle = @vm.sys_complete_promise_failure(name, failure)
  poll_and_take(handle)
  nil
end

#reject_signal(invocation_id, name, message, code: 500) ⇒ Object

Send a terminal failure to a named signal on another invocation.



385
386
387
388
 
# File 'lib/restate/server/context.rb', line 385

def reject_signal(invocation_id, name, message, code: 500)
  failure = Failure.new(code: code, message: message)
  @vm.sys_complete_signal_failure(invocation_id, name, failure)
end

#request ⇒ Object

Returns metadata about the current invocation (id, headers, raw body, and — when present — the scope, limit_key, and idempotency_key it was invoked with).



476
477
478
479
480
481
482
483
484
485
486
 
# File 'lib/restate/server/context.rb', line 476

def request
  @request ||= Request.new(
    id: @invocation.invocation_id,
    headers: @invocation.headers.to_h,
    body: @invocation.input_buffer,
    attempt_finished_event: @attempt_finished_event,
    scope: @invocation.scope,
    limit_key: @invocation.limit_key,
    idempotency_key: @invocation.idempotency_key
  )
end

#resolve_awakeable(awakeable_id, payload, serde: JsonSerde) ⇒ Object

Resolves an awakeable with a success value.



361
362
363
 
# File 'lib/restate/server/context.rb', line 361

def resolve_awakeable(awakeable_id, payload, serde: JsonSerde)
  @vm.sys_complete_awakeable_success(awakeable_id, serde.serialize(payload))
end

#resolve_handle(handle) ⇒ Object

Block until a previously created handle completes. Returns the value.



145
146
147
 
# File 'lib/restate/server/context.rb', line 145

def resolve_handle(handle)
  poll_and_take(handle)
end

#resolve_promise(name, payload, serde: JsonSerde) ⇒ Object

Resolves a durable promise with a success value.



409
410
411
412
413
 
# File 'lib/restate/server/context.rb', line 409

def resolve_promise(name, payload, serde: JsonSerde)
  handle = @vm.sys_complete_promise_success(name, serde.serialize(payload))
  poll_and_take(handle)
  nil
end

#resolve_signal(invocation_id, name, payload, serde: JsonSerde) ⇒ Object

Send a success value to a named signal on another invocation.



380
381
382
 
# File 'lib/restate/server/context.rb', line 380

def resolve_signal(invocation_id, name, payload, serde: JsonSerde)
  @vm.sys_complete_signal_success(invocation_id, name, serde.serialize(payload))
end

#run(name, serde: JsonSerde, retry_policy: nil, background: false, &action) ⇒ Object

Executes a durable side effect. The block runs at most once; its result is journaled and replayed on retries. Returns a DurableFuture for the result.

Pass background: true to run the block in a real OS Thread, keeping the fiber event loop responsive for other concurrent handlers. Use this for CPU-intensive work.



218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
 
# File 'lib/restate/server/context.rb', line 218

def run(name, serde: JsonSerde, retry_policy: nil, background: false, &action)
  run = @vm.sys_run(name)
  handle = run.handle

  # Schedule the run closure only if the run wasn't replayed.
  unless run.replayed
    @run_coros_to_execute[handle] =
      if background
        -> { execute_run_threaded(handle, action, serde, retry_policy) }
      else
        -> { execute_run(handle, action, serde, retry_policy) }
      end
  end

  DurableFuture.new(self, handle, serde: serde)
end

#run_sync(name, serde: JsonSerde, retry_policy: nil, background: false, &action) ⇒ Object

Convenience shortcut for run(...).await — executes the durable side effect and returns the result directly.

Accepts all the same options as run, including background: true.



239
240
241
 
# File 'lib/restate/server/context.rb', line 239

def run_sync(name, serde: JsonSerde, retry_policy: nil, background: false, &action)
  run(name, serde: serde, retry_policy: retry_policy, background: background, &action).await
end

#scope(scope) ⇒ Object

Returns a ScopedContext that routes all outgoing calls within the given scope. See Context#scope for the full semantics and constraints.



467
468
469
 
# File 'lib/restate/server/context.rb', line 467

def scope(scope)
  ScopedContext.new(self, scope)
end

#service_call(service, handler, arg, key: nil, idempotency_key: nil, headers: nil, input_serde: NOT_SET, output_serde: NOT_SET, scope: nil, limit_key: nil) ⇒ Object

Durably calls a handler on a Restate service and returns a future for its result.

scope and limit_key are internal parameters set by ScopedContext (via ctx.scope(...)); prefer that API over passing them directly.



249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
 
# File 'lib/restate/server/context.rb', line 249

def service_call(service, handler, arg, key: nil, idempotency_key: nil, headers: nil,
                 input_serde: NOT_SET, output_serde: NOT_SET, scope: nil, limit_key: nil)
  svc_name, handler_name, handler_meta = resolve_call_target(service, handler)
  in_serde = resolve_serde(input_serde, handler_meta, :input_serde)
  out_serde = resolve_serde(output_serde, handler_meta, :output_serde)
  parameter = in_serde.serialize(arg)
  with_outbound_middleware(svc_name, handler_name, headers, handler_meta: handler_meta) do |hdrs|
    call_handle = @vm.sys_call(
      service: svc_name, handler: handler_name, parameter: parameter,
      key: key, idempotency_key: idempotency_key, headers: hdrs, scope: scope, limit_key: limit_key
    )
    DurableCallFuture.new(self, call_handle.result_handle, call_handle.invocation_id_handle,
                          output_serde: out_serde)
  end
end

#service_send(service, handler, arg, key: nil, delay: nil, idempotency_key: nil, headers: nil, input_serde: NOT_SET, scope: nil, limit_key: nil) ⇒ Object

Sends a one-way invocation to a Restate service handler (fire-and-forget).

scope and limit_key are internal parameters set by ScopedContext (via ctx.scope(...)); prefer that API over passing them directly.



269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
 
# File 'lib/restate/server/context.rb', line 269

def service_send(service, handler, arg, key: nil, delay: nil, idempotency_key: nil, headers: nil,
                 input_serde: NOT_SET, scope: nil, limit_key: nil)
  svc_name, handler_name, handler_meta = resolve_call_target(service, handler)
  in_serde = resolve_serde(input_serde, handler_meta, :input_serde)
  parameter = in_serde.serialize(arg)
  delay_ms = delay ? (delay * 1000).to_i : nil
  with_outbound_middleware(svc_name, handler_name, headers, handler_meta: handler_meta) do |hdrs|
    invocation_id_handle = @vm.sys_send(
      service: svc_name, handler: handler_name, parameter: parameter,
      key: key, delay: delay_ms, idempotency_key: idempotency_key, headers: hdrs,
      scope: scope, limit_key: limit_key
    )
    SendHandle.new(self, invocation_id_handle)
  end
end

#set(name, value, serde: JsonSerde) ⇒ Object

Durably sets a state entry. The value is serialized via serde.



109
110
111
 
# File 'lib/restate/server/context.rb', line 109

def set(name, value, serde: JsonSerde)
  @vm.sys_set_state(name, serde.serialize(value))
end

#signal(name, serde: JsonSerde) ⇒ Object

Wait for a named signal addressed to this invocation. Returns a DurableFuture.



374
375
376
377
 
# File 'lib/restate/server/context.rb', line 374

def signal(name, serde: JsonSerde)
  handle = @vm.sys_signal(name)
  DurableFuture.new(self, handle, serde: serde)
end

#sleep(seconds) ⇒ Object

Returns a durable future that completes after the given duration. The timer survives handler restarts.



138
139
140
141
142
 
# File 'lib/restate/server/context.rb', line 138

def sleep(seconds)
  millis = (seconds * 1000).to_i
  handle = @vm.sys_sleep(millis)
  DurableFuture.new(self, handle)
end

#state_keys ⇒ Object

Returns the list of all state entry names for this virtual object or workflow.



124
125
126
 
# File 'lib/restate/server/context.rb', line 124

def state_keys
  state_keys_async.await
end

#state_keys_async ⇒ Object

Returns a DurableFuture for the list of all state entry names.



129
130
131
132
 
# File 'lib/restate/server/context.rb', line 129

def state_keys_async
  handle = @vm.sys_get_state_keys
  DurableFuture.new(self, handle)
end

#take_completed(handle) ⇒ Object

Take a completed handle's notification, returning the value. Raises TerminalError if the handle resolved to a failure.



161
162
163
 
# File 'lib/restate/server/context.rb', line 161

def take_completed(handle)
  must_take_notification(handle)
end

#wait_any(*futures) ⇒ Object

Wait until any of the given futures completes. Returns [completed, remaining].



166
167
168
169
170
 
# File 'lib/restate/server/context.rb', line 166

def wait_any(*futures)
  handles = futures.map(&:handle)
  wait_any_handle(handles)
  futures.partition(&:completed?)
end

#wait_any_handle(handles) ⇒ Object

Wait until any of the given handles completes. Does not take notifications.



150
151
152
 
# File 'lib/restate/server/context.rb', line 150

def wait_any_handle(handles)
  poll_or_cancel(handles) unless handles.any? { |h| @vm.is_completed(h) }
end

#wait_combined(future_tree) ⇒ Object

Drive progress over a combinator tree. Returns when the combinator logically completes (the shared-core decides based on the tree shape). future_tree follows the encoding documented in lib/restate/vm.rb#do_await. Public so CombinedFuture#await can drive it via @ctx.wait_combined.



497
498
499
 
# File 'lib/restate/server/context.rb', line 497

def wait_combined(future_tree)
  progress_loop { @vm.do_await(future_tree) }
end

#workflow_call(service, handler, key, arg, idempotency_key: nil, headers: nil, input_serde: NOT_SET, output_serde: NOT_SET, scope: nil, limit_key: nil) ⇒ Object

Durably calls a handler on a Restate workflow, keyed by key.

scope and limit_key are internal parameters set by ScopedContext (via ctx.scope(...)); prefer that API over passing them directly.



329
330
331
332
333
334
335
336
337
 
# File 'lib/restate/server/context.rb', line 329

def workflow_call(service, handler, key, arg, idempotency_key: nil, headers: nil,
                  input_serde: NOT_SET, output_serde: NOT_SET, scope: nil, limit_key: nil)
  object_call(
    service, handler, key, arg,
    idempotency_key: idempotency_key, headers: headers,
    input_serde: input_serde, output_serde: output_serde,
    scope: scope, limit_key: limit_key
  )
end

#workflow_send(service, handler, key, arg, delay: nil, idempotency_key: nil, headers: nil, input_serde: NOT_SET, scope: nil, limit_key: nil) ⇒ Object

Sends a one-way invocation to a Restate workflow handler (fire-and-forget).

scope and limit_key are internal parameters set by ScopedContext (via ctx.scope(...)); prefer that API over passing them directly.



343
344
345
346
347
348
349
350
 
# File 'lib/restate/server/context.rb', line 343

def workflow_send(service, handler, key, arg, delay: nil, idempotency_key: nil, headers: nil,
                  input_serde: NOT_SET, scope: nil, limit_key: nil)
  object_send(
    service, handler, key, arg,
    delay: delay, idempotency_key: idempotency_key, headers: headers,
    input_serde: input_serde, scope: scope, limit_key: limit_key
  )
end