Skip to main content

Workflow message passing

View Markdown

A Workflow can act like a stateful web service that receives messages: Queries, Signals, and Updates. The Workflow implementation defines these endpoints via handler methods that can react to incoming messages and return values.

Temporal Clients use messages to read Workflow state and control its execution. See Workflow message passing for a general overview of this topic. This page introduces these features for the Temporal Rust SDK.

Write message handlers​

Follow these guidelines when writing your message handlers:

  • Message handlers are defined as methods on your Workflow struct and registered with the Workflow runtime.
  • The parameters and return values of handlers and the main Workflow function must be serializable.
  • Prefer structs to multiple input parameters to allow for forward-compatible changes.

Query handlers​

A Query is a synchronous operation that retrieves state from a Workflow Execution:

rust-sandbox/workflows/message-passing/src/workflows.rs

use std::{collections::HashMap, time::Duration};

use temporalio_macros::{workflow, workflow_methods};
use temporalio_sdk::{
    ActivityOptions, SyncWorkflowContext, WorkflowContext, WorkflowContextView, WorkflowResult,
};

use crate::activities::GreetingActivities;

#[derive(serde::Serialize, serde::Deserialize, Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum Language {
    Arabic,
    Chinese,
    English,
    French,
    Hindi,
    Spanish,
}

#[derive(serde::Serialize, serde::Deserialize)]
pub struct GetLanguagesInput {
    pub include_unsupported: bool,
}
// ...
#[workflow]
pub struct GreetingsWorkflow {
    greetings: HashMap<Language, String>,
    language: Language,
    approved_for_release: bool,
    approver_name: Option<String>,
    greeting_service_busy: bool,
}

#[workflow_methods]
impl GreetingsWorkflow {
    #[init]
    fn new(_ctx: &WorkflowContextView) -> Self {
        let greetings = HashMap::from([
            (Language::Chinese, "你好，世界".to_string()),
            (Language::English, "Hello, world".to_string()),
        ]);

        Self {
            greetings,
            language: Language::English,
            approved_for_release: false,
            approver_name: None,
            greeting_service_busy: false,
        }
    }

    #[run]
    pub async fn run(ctx: &mut WorkflowContext<Self>) -> WorkflowResult<String> {
        // Wait for the approve Signal.
        ctx.wait_condition(|s| s.approved_for_release).await?;

        // Wait for running Signal and Update handlers to finish.
        ctx.wait_condition(|_| ctx.all_handlers_finished()).await?;

        Ok(ctx.state(|s| s.greetings[&s.language].clone()))
    }

    #[query]
    pub fn get_languages(
        &self,
        _ctx: &WorkflowContextView,
        input: GetLanguagesInput,
    ) -> Vec<Language> {
        if input.include_unsupported {
            vec![
                Language::Arabic,
                Language::Chinese,
                Language::English,
                Language::French,
                Language::Hindi,
                Language::Spanish,
            ]
        } else {
            let mut languages: Vec<Language> = self.greetings.keys().copied().collect();
            languages.sort_by_key(|language| format!("{language:?}"));
            languages
        }
    }
// ...
}

rust-sandbox/workflows/message-passing/src/client.rs

let wf_handle = client
    .start_workflow(
        GreetingsWorkflow::run,
        (),
        WorkflowStartOptions::new("my-task-queue", "greetings-workflow-10").build(),
    )
    .await?;

rust-sandbox/workflows/message-passing/src/client.rs

let supported_languages = wf_handle
    .query(
        GreetingsWorkflow::get_languages,
        GetLanguagesInput { include_unsupported: true },
        WorkflowQueryOptions::default(),
    )
    .await?;
  • Query handlers can't mutate Workflow state.
  • Query handlers can't perform async operations, like executing Activities.

Signal handlers​

A Signal is an asynchronous message sent to a running Workflow Execution to change its state and control its flow:

rust-sandbox/workflows/message-passing/src/workflows.rs

// ...
#[derive(serde::Serialize, serde::Deserialize)]
pub struct ApproveInput {
    pub name: String,
}
// ...
#[workflow_methods]
impl GreetingsWorkflow {
// ...
    #[signal]
    pub fn approve(&mut self, _ctx: &mut SyncWorkflowContext<Self>, input: ApproveInput) {
        self.approved_for_release = true;
        self.approver_name = Some(input.name);
    }
// ...
}

rust-sandbox/workflows/message-passing/src/client.rs

wf_handle
    .signal(
        GreetingsWorkflow::approve,
        ApproveInput { name: "Ziggy".to_string() },
        WorkflowSignalOptions::default(),
    )
    .await?;
  • Signal handlers do not return values.
  • Signal handlers can be async fn and use Activities, Timers, and Child Workflows. See Async handlers.

Update handlers and validators​

An Update is a trackable synchronous request sent to a running Workflow Execution. It can change the Workflow state, control its flow, and return a result. 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 if something goes wrong:

rust-sandbox/workflows/message-passing/src/workflows.rs

// ...
#[derive(serde::Serialize, serde::Deserialize)]
pub struct SetLanguageInput {
    pub language: Language,
}
// ...
#[workflow_methods]
impl GreetingsWorkflow {
// ...
    #[update]
    pub fn set_language(
        &mut self,
        _ctx: &mut SyncWorkflowContext<Self>,
        input: SetLanguageInput,
    ) -> Language {
        let previous_language = self.language;
        self.language = input.language;

        previous_language
    }

    #[update_validator(set_language)]
    fn validate_set_language(
        &self,
        _ctx: &WorkflowContextView,
        input: &SetLanguageInput,
    ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
        if !self.greetings.contains_key(&input.language) {
            Err("Not a valid language".into())
        } else {
            Ok(())
        }
    }
// ...
}
  • About validators:
    • Use validators to reject an Update before it is written to History. Validators are always optional. If you don't need to reject Updates, you can skip them.
    • If you choose to create validators for your Updates, they will reject Updates before they're applied.
  • Accepting and rejecting Updates with validators:
    • To reject an Update, return an Err from the validator.
    • Without a validator, Updates are always accepted.
  • Validators and Event History:
    • The WorkflowExecutionUpdateAccepted event is written into the History whether the acceptance was automatic or programmatic.
    • When a validator returns an error, the Update is rejected and WorkflowExecutionUpdateAccepted won't be added to the Event History. The caller receives the error.
  • Update and Signal handlers can be async, letting you use Activities, Child Workflows, and more. See Async handlers for safe usage guidelines.

Send messages​

To send Queries, Signals, or Updates, use a Workflow handle from the client:

rust-sandbox/workflows/message-passing/src/client.rs

let wf_handle = client
    .start_workflow(
        GreetingsWorkflow::run,
        (),
        WorkflowStartOptions::new("my-task-queue", "greetings-workflow-10").build(),
    )
    .await?;

Send a Query​

rust-sandbox/workflows/message-passing/src/client.rs

let supported_languages = wf_handle
    .query(
        GreetingsWorkflow::get_languages,
        GetLanguagesInput { include_unsupported: true },
        WorkflowQueryOptions::default(),
    )
    .await?;
  • Sending a Query doesn’t add events to a Workflow's Event History.
  • You can send Queries to closed Workflow Executions within a Namespace's Workflow retention period. This includes Workflows that have completed, failed, or timed out. Querying terminated Workflows is not safe and, therefore, not supported.
  • A Worker must be online and polling the Task Queue to process a Query.

Send a Signal​

You can send a Signal to a Workflow Execution from a Temporal Client or from another Workflow Execution. However, you can only send Signals to Workflow Executions that haven’t closed.

From a Client​

rust-sandbox/workflows/message-passing/src/client.rs

wf_handle
    .signal(
        GreetingsWorkflow::approve,
        ApproveInput { name: "Ziggy".to_string() },
        WorkflowSignalOptions::default(),
    )
    .await?;
  • The call returns when the server accepts the Signal; it does not wait for the Signal to be delivered to the Workflow Execution.

From a Workflow​

rust-sandbox/workflows/message-passing/src/approver.rs

// Pass a Run ID instead of None to target a specific run.
ctx.external_workflow(workflow_id, None)
    .signal(
        GreetingsWorkflow::approve,
        ApproveInput { name: "Ziggy".to_string() },
        SignalWorkflowOptions::default(),
    )
    .await?;

Signal-With-Start​

rust-sandbox/workflows/message-passing/src/client.rs

let wf_handle = client
    .signal_with_start_workflow(
        GreetingsWorkflow::run,
        (),
        GreetingsWorkflow::approve,
        ApproveInput { name: "Ziggy".to_string() },
        WorkflowStartOptions::new("my-task-queue", "greetings-workflow-11").build(),
    )
    .await?;

Send an Update​

An Update is a synchronous, blocking call that can change Workflow state, control its flow, and return a result.

A client sending an Update must wait until the Server delivers the Update to a Worker. Workers must be available and responsive. If you need a response as soon as the Server receives the request, use a Signal instead. You can't send Updates directly from one Workflow to another. If you need to send Updates across Workflows, like to Child Workflows, use an Activity.

  • WorkflowExecutionUpdateAccepted is added to the Event History when the Worker confirms that the Update passed validation.
  • WorkflowExecutionUpdateCompleted is added to the Event History when the Worker confirms that the Update has finished.

To send an Update to a Workflow Execution, you can call execute_update and wait for the Update to complete.

This code fetches an Update result:

rust-sandbox/workflows/message-passing/src/client.rs

let previous_language = wf_handle
    .execute_update(
        GreetingsWorkflow::set_language,
        SetLanguageInput { language: Language::Chinese },
        WorkflowExecuteUpdateOptions::default(),
    )
    .await?;

Start an Update​

You can also send start_update to receive an UpdateHandle as soon as the Update is accepted.

  • Use this UpdateHandle later to fetch your results.
  • Async Update handlers normally perform long-running asynchronous operations, such as executing an Activity.
  • start_update only waits until the Worker has accepted or rejected the Update, not until all asynchronous operations are complete.

For example:

rust-sandbox/workflows/message-passing/src/client.rs

let update_handle = wf_handle
    .start_update(
        GreetingsWorkflow::set_language_activity,
        SetLanguageInput { language: Language::French },
        WorkflowStartUpdateOptions::default(),
    )
    .await?;

// Wait for the Update's Activity to finish and fetch the result.
let previous_language_from_activity_update =
    update_handle.get_result(RpcOptions::default()).await?;
  • Updates are synchronous and return results.
  • Worker must accept the Update before it proceeds.

Update-With-Start​

Use execute_update_with_start_workflow to start a Workflow and send it an Update in one call. You must choose a Workflow ID conflict policy. With WorkflowIdConflictPolicy::UseExisting, the Update goes to the running Workflow if one already has this Workflow ID:

rust-sandbox/workflows/message-passing/src/client.rs

let previous_language = client
    .execute_update_with_start_workflow(
        GreetingsWorkflow::run,
        (),
        GreetingsWorkflow::set_language,
        SetLanguageInput { language: Language::Chinese },
        WorkflowUpdateWithStartOptions::new(
            "my-task-queue",
            "greetings-workflow-12",
            WorkflowIdConflictPolicy::UseExisting,
        )
        .build(),
    )
    .await?;

Message handler patterns​

Async handlers​

Signal and Update handlers can be async fn as well as fn. Using async fn allows you to use await with Activities, Child Workflows, Timers, and so on. This expands the possibilities for what can be done by a handler, but it also means that handler executions and your main Workflow method are all running concurrently, with switching occurring between them at await calls.

To use async fn handlers safely, understand what can go wrong. See Workflow message passing for guidance on safe usage of async Signal and Update handlers, the Safe message handlers sample and the sections below.

The following code executes an Activity that makes a network call to a remote service:

rust-sandbox/workflows/message-passing/src/activities.rs

use std::{collections::HashMap, time::Duration};

use temporalio_macros::activities;
use temporalio_sdk::activities::{ActivityContext, ActivityError};
use tokio::time::sleep;

use crate::workflows::Language;

pub struct GreetingActivities;

#[activities]
impl GreetingActivities {
    #[activity]
    pub async fn call_greeting_service(
        _ctx: ActivityContext,
        to_language: Language,
    ) -> Result<Option<String>, ActivityError> {
        // Pretend that we are calling a remote service.
        sleep(Duration::from_millis(200)).await;

        let greetings = HashMap::from([
            (Language::Arabic, "مرحبا بالعالم".to_string()),
            (Language::Chinese, "你好，世界".to_string()),
            (Language::English, "Hello, world".to_string()),
            (Language::French, "Bonjour, monde".to_string()),
            (Language::Hindi, "नमस्ते दुनिया".to_string()),
            (Language::Spanish, "Hola mundo".to_string()),
        ]);

        Ok(greetings.get(&to_language).cloned())
    }
}

After updating the code to use an async fn, your Update handler can schedule an Activity and await the result. Although an async fn Signal handler can also execute an Activity, using an Update handler allows the client to receive a result or error once the Activity completes.

This lets your client track the progress of asynchronous work performed by the Update's Activities, Child Workflows, and so on. Here's how you could start that Activity from within your Workflow. The greeting_service_busy flag lets one handler call the greeting service at a time:

rust-sandbox/workflows/message-passing/src/workflows.rs

// ...
#[workflow_methods]
impl GreetingsWorkflow {
// ...
    #[update]
    pub async fn set_language_activity(
        ctx: &mut WorkflowContext<Self>,
        input: SetLanguageInput,
    ) -> Result<Language, Box<dyn std::error::Error + Send + Sync>> {
        let language = input.language;
        let needs_greeting = ctx.state(|s| !s.greetings.contains_key(&language));

        if needs_greeting {
            // Let one handler call the greeting service at a time.
            ctx.wait_condition(|s| !s.greeting_service_busy).await?;
            ctx.state_mut(|s| s.greeting_service_busy = true);

            let greeting = ctx
                .execute_activity(
                    GreetingActivities::call_greeting_service,
                    language,
                    ActivityOptions::start_to_close_timeout(Duration::from_secs(10)),
                )
                .await;

            ctx.state_mut(|s| s.greeting_service_busy = false);

            let Some(greeting) = greeting? else {
                return Err("The greeting service doesn't support this language".into());
            };
            ctx.state_mut(|s| {
                s.greetings.insert(language, greeting);
            });
        }

        let previous_language = ctx.state(|s| s.language);
        ctx.state_mut(|s| s.language = language);

        Ok(previous_language)
    }
}

Add wait conditions to block​

Sometimes, async Signal or Update handlers need to meet certain conditions before they should continue. You can use ctx.wait_condition to prevent the code from proceeding until a condition is true. You specify the condition by passing a function that returns a boolean and you can optionally set a timeout. This is an important feature that helps you control your handler logic.

Here are three important use cases for ctx.wait_condition:

  • Wait for a Signal or Update to arrive.
  • Wait in a handler until it's appropriate to continue.
  • Wait in the main Workflow until all active handlers have finished.

It's common to use ctx.wait_condition to wait for a particular Signal or Update to be sent by a client. This Workflow waits for the approve Signal:

rust-sandbox/workflows/message-passing/src/workflows.rs

// ...
#[workflow_methods]
impl GreetingsWorkflow {
// ...
    #[run]
    pub async fn run(ctx: &mut WorkflowContext<Self>) -> WorkflowResult<String> {
        // Wait for the approve Signal.
        ctx.wait_condition(|s| s.approved_for_release).await?;

        // Wait for running Signal and Update handlers to finish.
        ctx.wait_condition(|_| ctx.all_handlers_finished()).await?;

        Ok(ctx.state(|s| s.greetings[&s.language].clone()))
    }
// ...
}

Ensure handlers finish before completion​

Workflow wait conditions can ensure your handler completes before a Workflow finishes. When your Workflow uses async Signal or Update handlers, your main Workflow method can return or continue-as-new while a handler is still waiting on an async task, such as an Activity result.

The Workflow completing may interrupt the handler before it finishes crucial work and cause client errors when trying retrieve Update results. Use ctx.wait_condition with ctx.all_handlers_finished() to address this problem and allow your Workflow to end smoothly. The run method in the previous example does this before it returns.