message-passing

194 notes in this subject, read out of this brain and free to use. This is page 2 of 4.

Signal handler error handling

When using Signal, the only exceptions that result from requests during its execution are the RPCErrors (UNAVAILABLE if client can't contact server, NOT_FOUND if workflow doesn't exist). For Queries and Updates, additional errors may occur during handler execution by the Worker.

Update error: no Workers polling Task Queue

When sending an Update and no Workflow Workers are polling the Task Queue, the request is retried indefinitely by the SDK Client. You can use asyncio.timeout to impose a timeout, which raises a temporalio.client.WorkflowUpdateRPCTimeoutOrCancelledError exception.

Update error: update failed - rejection by validator

You receive a temporalio.client.WorkflowUpdateFailedError exception when an Update is rejected by an Update validator defined in the Workflow. This indicates the Update was not accepted.

Update error: Workflow Task failure before acceptance

When an Update handler causes a Workflow Task to fail before the request is accepted, you receive a FAILED_PRECONDITION RPCError exception.

Update error: Workflow finished during handler execution

When a Workflow finishes while an Update handler execution is in progress, you receive a temporalio.service.RPCError exception with status RPCStatusCode.NOT_FOUND. This can happen if the Workflow was canceled or failed, completed normally, or continued-as-new without waiting for handlers to finish.

Query error: no Workers polling Task Queue

When sending a Query and there is no Workflow Worker polling the Task Queue, you receive a temporalio.service.RPCError exception where the status attribute is RPCStatusCode.FAILED_PRECONDITION.

Query error: query failed

You receive a temporalio.client.WorkflowQueryFailedError exception if something goes wrong during a Query execution. Any exception in a Query handler triggers this error. This differs from Signal and Update requests where exceptions can lead to Workflow Task Failure instead.

Query error: Workflow Task failure

If a Query handler blocks the thread for too long without yielding, it can cause a Workflow Task to fail.

Bind topic name to type with stream.topic()

Use self.stream.topic("name", type=Type) to bind a topic name to its event type once. This returns a handle that can call publish() on that topic. The handle records the per-stream binding from topic name to value type so call sites don't have to repeat the type on every publish and so subscribers reading the same handle decode to the matching type.

type= argument is optional in stream.topic()

The type= argument in stream.topic() is optional and defaults to Any. Pass it when you want the binding recorded so re-binding the same name to an unequal type raises an exception, or so subscribers can pick up the type from the same handle.

WorkflowStreamClient.create() for publishing or subscribing

Construct a WorkflowStreamClient with WorkflowStreamClient.create(client, workflow_id) from any process that has a Temporal Client and the target Workflow Id. This works for HTTP backends, starters, one-off scripts, other Workflows' Activities, and standalone Activities. Use it the same way you would the Workflow-side handle: bind a topic, publish through it, and let the async context manager flush on exit.

force_flush=True on publish for latency

Pass force_flush=True on a publish to wake the background flusher so the current buffer ships without waiting for the next interval. The flusher only runs while the Workflow Stream client is entered (async with client). Otherwise, force_flush=True queues the wake event, but nothing ships until you enter the context or call await client.flush(). The call returns immediately after appending to the buffer and signaling the flusher; it doesn't wait for delivery to the Workflow or to subscribers.

await client.flush() for mid-stream barrier

Use await client.flush() when you need a mid-stream barrier. Successful completion of the flush is proof that the Temporal server has received all prior publications, so subsequent work that depends on those events being durable can proceed. The client stays open for further publishing afterward. Exiting async with client already flushes on its way out, so the explicit call is only for barriers in the middle.

publish() is non-blocking and applies no backpressure

publish() is non-blocking and applies no backpressure. From an Activity or other client, it appends to the client's in-memory buffer and returns. From inside a Workflow, it appends synchronously to the in-memory log. Subscribers pull from the Workflow's log on their own schedule, so a slow subscriber doesn't slow down publishers. If a publisher emits faster than batches can ship to the server, the buffer grows: the process uses more memory, the stream falls further behind real time, and at the limit Signals can't keep up.

Subscribe using topic.subscribe() method

Once you have a WorkflowStreamClient, iterate a topic handle's subscribe(), the counterpart to publish(). The handle's bound type drives decoding, so each item.data arrives as T via the client's payload converter. The codec chain is applied once at the Update envelope, not per item.

Handle heterogeneous topics with stream.subscribe() and RawValue

To consume multiple topics whose payload types differ, call client.subscribe() directly with a list of names (or subscribe([]) for every topic) and pass result_type=temporalio.common.RawValue so each item arrives as the underlying Payload wrapped in a RawValue. Dispatch on item.topic and decode the wrapped payload with the client's payload converter.

RawValue avoids cancellation race in multi-topic subscriptions

A single iterator over multiple topics with RawValue avoids the cancellation race that two concurrent subscribers would create. RawValue is also the right shape when you want to forward the bytes through to another system without decoding them.

Default result_type behavior in subscribe()

Omitting result_type entirely or passing result_type=None decodes each item with the converter's default rules. For the stock JSON converter, that means a Python primitive, dict, or list. This works for fully homogeneous streams, but not for the dispatch-by-topic pattern where each topic has its own concrete dataclass.

Iterator behavior on workflow completion

subscribe() exits cleanly when the Workflow reaches COMPLETED, FAILED, CANCELED, TERMINATED, or TIMED_OUT, but does not distinguish among them. To know which status, call await temporal_client.get_workflow_handle(workflow_id).describe() after the loop returns to inspect the Workflow's status.

RPC timeout edge case in stream subscriber

An RPC timeout where Continue-As-New cannot be followed ends the iterator silently (no exception raised).

Validator rejection during Continue-As-New can surface as WorkflowUpdateFailedError

A validator rejection during a Continue-As-New handoff in workflow streams can surface as a WorkflowUpdateFailedError.

Sentinel event pattern for closing stream

A common pattern combines two pieces: (1) An in-band terminator where the Workflow or its Activity publishes a sentinel event the subscriber recognizes and breaks on. (2) A brief overlap before the Workflow returns. A poll Update that is still in flight when the Workflow returns surfaces to the client as AcceptedUpdateCompletedWorkflow, and no new polls can complete after that.

Fixed sleep approach for stream closing

Sleep between the terminator and the return so any in-flight poll has time to fetch the terminator before the Workflow exits. Example: after publishing StatusEvent(state="completed", progress=100), call await workflow.sleep(timedelta(seconds=30)) before returning.

Acknowledgment handshake approach for stream closing

The subscriber sends a Signal once it has the terminator; the Workflow waits up to a timeout, returning as soon as the ack arrives. Use @workflow.signal to define a handler for the ack, then await workflow.wait_condition() in @workflow.run to wait for the signal with a timeout fallback for when no subscriber is attached.

publisher_ttl deduplication window parameter

At each Continue-As-New, deduplicate entries whose last_seen is older than publisher_ttl are dropped. last_seen is updated on each successful publish, so a publisher that retries through a long partition without success can still time out. Tune upward if your publishers can be silent for extended windows using WorkflowStream.continue_as_new(publisher_ttl=...).

max_retry_duration parameter for WorkflowStreamClient

A WorkflowStreamClient retries a failed batch for up to max_retry_duration. If the duration elapses with the batch still pending, the client gives up, the pending batch is dropped, and a TimeoutError is raised. On timeout, the dropped batch is at-most-once: it may or may not have reached the Workflow.

TimeoutError in batch retry terminates flusher

When TimeoutError raises from a batch retry timeout in workflow streams, it raises from inside the background flusher task and terminates it. Until you call await client.flush() or exit the async with block, subsequent publishes accumulate in the buffer with no flusher to ship them.

Type bindings not shared across publishers

Each WorkflowStream and each WorkflowStreamClient records topic types only for its own instance. If two publishers bind the same topic name to different types, the mismatch is not caught at publish, and the subscriber gets a decode error when it processes events from the mismatched publisher.

batch_interval parameter for WorkflowStreamClient

When creating a WorkflowStreamClient, pass batch_interval=timedelta(...) to control how often the background flusher ships buffered events. Example: WorkflowStreamClient.create(temporal_client, workflow_id=workflow_id, batch_interval=timedelta(milliseconds=200)).

Iterator handles re-polling and Continue-As-New transparently

The subscribe() iterator handles re-polling, pagination when a poll response hits the ~1 MB cap, and Workflow-side log truncation transparently. Callers don't need to wrap the iterator for the common cases.

Ruby SDK Query handler definition and behavior

A Query is a synchronous operation that retrieves state from a Workflow Execution. Define a Query handler as a method on the Workflow class decorated with the `workflow_query` class method. A Query handler must not modify Workflow state and cannot perform async blocking operations such as executing an Activity. Query handlers can be defined as regular methods or as attribute readers using `workflow_query_attr_reader`. Sending a Query does not add events to a Workflow's Event History. You can send Queries to closed Workflow Executions within a Namespace's Workflow retention period, including Workflows that have completed, failed, or timed out. A Worker must be online and polling the Task Queue to process a Query.

Ruby SDK Signal handler definition and behavior

A Signal is an asynchronous message sent to a running Workflow Execution to change its state and control its flow. Define a Signal handler as a method on the Workflow class decorated with the `workflow_signal` class method. A Signal handler mutates the workflow state but cannot return a value. The response is sent immediately from the server without waiting for the Signal to be delivered to the Workflow Execution. Signal handlers can be asynchronous and blocking, allowing use of Activities, Child Workflows, durable Timers, and wait conditions. When a Signal is sent from a Client, the WorkflowExecutionSignaled Event appears in the Workflow's Event History.

Ruby SDK Update handler definition and behavior

An Update is a trackable synchronous request sent to a running Workflow Execution that can change the Workflow state, control its flow, and return a result. Define an Update handler as a method on the Workflow class decorated with the `workflow_update` class method. The sender must wait until the Worker accepts or rejects the Update. The sender may wait further to receive a returned value or an exception. Update handlers can be asynchronous and blocking, allowing use of Activities, Child Workflows, durable Timers, and wait conditions. When an Update is accepted, WorkflowExecutionUpdateAccepted is added to the Event History. When the Update completes, WorkflowExecutionUpdateCompleted is added to the Event History.

Ruby SDK Update validator definition

Define an Update validator with the `workflow_update_validator` class method invoked before defining the method. The first parameter when declaring the validator is the name of the Update handler method. The validator must accept the same argument types as the handler and should not return a value. Validators are always optional. Use validators to reject an Update before it is written to History. To reject an Update, raise an exception of any type in the validator. Without a validator, Updates are always accepted. When a Validator raises an error, the Update is rejected, the Update is not run, and WorkflowExecutionUpdateAccepted will not be added to the Event History. The caller receives an 'Update failed' error.

Ruby SDK sending Queries to Workflows

To send a Query to a Workflow Execution, call the Query method with `WorkflowHandle#query`, passing the Query class method and arguments. For example: `handle.query(MessagePassingSimple::GreetingWorkflow.languages, { include_unsupported: false })`. To check the argument types required when sending messages and the return type for Queries, refer to the corresponding handler method in the Workflow Definition.

Ruby SDK sending Signals from a Client

To send a Signal to a Workflow Execution from a Client, use `WorkflowHandle#signal` passing the Signal class method and arguments. For example: `handle.signal(MessagePassingSimple::GreetingWorkflow.approve, { name: 'John Q. Approver' })`. The call returns when the server accepts the Signal; it does not wait for the Signal to be delivered to the Workflow Execution.

Ruby SDK sending external Signals from a Workflow

A Workflow can send a Signal to another Workflow, known as an External Signal. Use `Temporalio::Workflow.external_workflow_handle`, passing a running Workflow Id, to retrieve a Workflow handle for the external Workflow. Then call `handle.signal()` on that handle. When an External Signal is sent, a SignalExternalWorkflowExecutionInitiated Event appears in the sender's Event History and a WorkflowExecutionSignaled Event appears in the recipient's Event History.

Ruby SDK Signal-With-Start

Signal-With-Start allows a Client to send a Signal to a Workflow Execution, starting the Execution if it is not already running. If there is a Workflow running with the given Workflow Id, it will be signaled. If there is not, a new Workflow will be started and immediately signaled. To use Signal-With-Start, call `signal_with_start_workflow` with a `WithStartWorkflowOperation` that defines the Workflow and its arguments.

Ruby SDK sending Updates to Workflows - execute_update

To send an Update to a Workflow Execution and wait for completion, call `execute_update` on the Workflow handle, passing the Update class method and arguments. For example: `prev_language = handle.execute_update(MessagePassingSimple::GreetingWorkflow.set_language, :chinese)`. This fetches the Update result and waits for the Update to complete. Updates cannot be sent directly from one Workflow to another. If you need to send Updates across Workflows, use an Activity.

Ruby SDK sending Updates to Workflows - start_update

To start an Update and receive a handle as soon as the Update is accepted, use `start_update` on the Workflow handle. This returns a `WorkflowUpdateHandle` that can be used later to fetch results. Use this when you want to wait until the Worker has accepted or rejected the Update but do not need to wait for all asynchronous operations to complete. For example: `update_handle = handle.start_update(MessagePassingSimple::GreetingWorkflow.apply_language_with_lookup, :arabic, wait_for_stage: Temporalio::Client::WorkflowUpdateWaitStage::ACCEPTED)` followed by `prev_language = update_handle.result`.

Ruby SDK Update-With-Start

Update-With-Start allows sending an Update that checks whether an already-running Workflow with that ID exists. If the Workflow exists, the Update is processed. If the Workflow does not exist, a new Workflow Execution is started with the given ID, and the Update is processed before the main Workflow method starts to execute. Use `execute_update_with_start_workflow` to start the Update and wait for the result in one go, or `start_update_with_start_workflow` to start the Update and receive a `WorkflowUpdateHandle` to retrieve the result later. You must provide a `WithStartWorkflowOperation` to define the Workflow that will be started if necessary and its arguments, and you must specify an `id_conflict_policy` when creating the `WithStartWorkflowOperation`. Requires Temporal Server version 1.26 or later.

Ruby SDK current_update_info

Use `Temporalio::Workflow.current_update_info` to obtain information about the current Update, including the Update ID. The Update ID can be useful for deduplication when using Continue-As-New.

Ruby SDK Continue-As-New restrictions with Updates

Temporal does not support Continue-As-New functionality within Update handlers. Complete all handlers before using Continue-As-New. Use Continue-As-New from your main Workflow Definition method, just as you would complete or fail a Workflow Execution.

Ruby SDK message handler parameter guidelines

Parameters and return values of handlers and the main Workflow function must be serializable. Prefer single hash/object input parameter to multiple input parameters, as hash/object parameters allow you to add fields without changing the calling signature.

Ruby SDK obtaining a Workflow handle

To obtain a Workflow handle for sending messages, use `Client#start_workflow` to start a Workflow and return its handle, or use the `Client#workflow_handle` method to retrieve a Workflow handle by its Workflow Id.

Ruby SDK wait_condition for message handlers

Use `Temporalio::Workflow.wait_condition` to set a function that prevents code from proceeding until the condition is truthy. This is useful for async Signal or Update handlers that need to meet certain conditions before they should continue. The condition state can be updated by any part of the Workflow code including the main Workflow method, other handlers, or child coroutines.

Ruby SDK wait_condition for finishing handlers before Workflow completes

Use `Temporalio::Workflow.wait_condition { Temporalio::Workflow.all_handlers_finished? }` in the main Workflow method to ensure all async Signal or Update handlers complete before the Workflow finishes. This prevents the Workflow from completing while a handler is still waiting on an async task, which could interrupt the handler and cause Client errors when trying to retrieve Update results. By default, the Worker logs a warning when allowing a Workflow Execution to finish with unfinished handler executions. Silence these warnings on a per-handler basis by passing the `unfinished_policy` argument to `workflow_signal` or `workflow_update` class methods with `Temporalio::Workflow::HandlerUnfinishedPolicy::ABANDON`.

Ruby SDK workflow_init for early input access

The `workflow_init` class method above `initialize` gives the constructor access to Workflow input. When you use `workflow_init` on your constructor, you give the constructor the same Workflow parameters as your `execute` method. The SDK ensures that your constructor receives the Workflow input arguments that the Client sent. The Workflow input arguments are also passed to your `execute` method whether or not you use the `workflow_init` class method. The constructor and `execute` must have the same parameters with the same types.

Ruby SDK using Mutex to prevent concurrent handler execution

Use `Mutex`, a mutual exclusion lock, to coordinate access in async handlers when multiple handler instances may execute concurrently. Locking makes sure that only one handler instance can execute a specific section of code at any given time. Initialize the mutex with `@mutex ||= Mutex.new` and wrap critical sections with `@mutex.synchronize do ... end`. This prevents race conditions where different handler instances could read or write state in an interleaved manner causing data inconsistency.

Ruby SDK error when Client cannot contact server with message

When sending a Signal, Update, or Query to a Workflow and the Client cannot contact the server, you will receive a `Temporalio::Error::RPCError` exception whose `code` is an `UNAVAILABLE` constant defined in `Code` after some retries.

Ruby SDK error when Workflow does not exist

When sending a Signal, Update, or Query to a Workflow that does not exist, you will receive a `Temporalio::Error::RPCError` exception whose `code` is a `NOT_FOUND` constant defined in `Code`.

Ruby SDK Update error when no Workers polling Task Queue

When sending an Update and no Workflow Workers are polling the Task Queue, the request will be retried by the SDK Client indefinitely. Use a `Cancellation` in your RPC options to cancel the Update, which raises a `WorkflowUpdateRPCTimeoutOrCanceledError` exception.

Ruby SDK WorkflowUpdateFailedError exception

When sending an Update, you will receive a `WorkflowUpdateFailedError` exception in two scenarios: (1) The Update was rejected by an Update validator defined in the Workflow alongside the Update handler. (2) The Update failed after having been accepted. Update failures are like Workflow failures. Issues that cause a Workflow failure in the main method also cause Update failures in the Update handler, including a failed Child Workflow, a failed Activity if retries are finite, the Workflow author raising `ApplicationError`, or any error listed in `workflow_failure_exception_types` on the Worker or `workflow_failure_exception_type` on the Workflow.

Ruby SDK Update error when Workflow Task fails

When an Update handler causes the Workflow Task to fail, a Workflow Task Failure causes the server to retry Workflow Tasks indefinitely. What happens to your Update request depends on its stage: If the request has not been accepted by the server, you receive a `FAILED_PRECONDITION` `Temporalio::Error::RPCError` exception. If the request has been accepted, it is durable. Once the Workflow is healthy again after a code deploy, use a `WorkflowUpdateHandle` to fetch the Update result.

Ruby SDK Update error when Workflow finished during handler execution

When the Workflow finishes while the Update handler execution is in progress, you will receive a `Temporalio::Error::RPCError` 'workflow execution already completed' exception. This happens if the Workflow was canceled or failed, or if the Workflow completed normally or continued-as-new and the Workflow author did not wait for handlers to be finished.

Ruby SDK Query error when no Worker polling Task Queue

When sending a Query and there is no Workflow Worker polling the Task Queue, you will receive a `Temporalio::Error::RPCError` exception whose `code` is a `FAILED_PRECONDITION` constant defined in `Code`.

Ruby SDK WorkflowQueryFailedError exception

When sending a Query, you will receive a `WorkflowQueryFailedError` exception if something goes wrong during the Query. Any exception in a Query handler will trigger this error. This differs from Signal and Update requests, where exceptions can lead to Workflow Task Failure instead.

Ruby SDK Dynamic Query handler

A Dynamic Query is a Query method that is invoked dynamically at runtime if no other Query with the same name is registered. Create a Dynamic Query by setting `dynamic: true` on the `workflow_query` class method. Only one Dynamic Query can be present on a Workflow. The Query Handler parameters must accept a string name as the first parameter. Often users set `raw_args: true` and set the second parameter as `*args` which will be an array of `Temporalio::Converters::RawValue`. Use `Temporalio::Workflow.payload_converter.from_payload()` to convert the raw value instances to proper types.

Ruby SDK Dynamic Signal handler

A Dynamic Signal is a Signal that is invoked dynamically at runtime if no other Signal with the same name is registered. Create a Dynamic Signal by setting `dynamic: true` on the `workflow_signal` class method. Only one Dynamic Signal can be present on a Workflow. The Signal Handler parameters must accept a string name as the first parameter. Often users set `raw_args: true` and set the second parameter as `*args` which will be an array of `Temporalio::Converters::RawValue`. Use `Temporalio::Workflow.payload_converter.from_payload()` to convert the raw value instances to proper types.

Ruby SDK Dynamic Update handler

A Dynamic Update is an Update that is invoked dynamically at runtime if no other Update with the same name is registered. Create a Dynamic Update by setting `dynamic: true` on the `workflow_update` class method. Only one Dynamic Update can be present on a Workflow. The Update Handler parameters must accept a string name as the first parameter. Often users set `raw_args: true` and set the second parameter as `*args` which will be an array of `Temporalio::Converters::RawValue`. Use `Temporalio::Workflow.payload_converter.from_payload()` to convert the raw value instances to proper types.

Give your agent this brain