Skip to main content

Run a Worker - Rust SDK

View Markdown

The code on this page is written against temporalio-sdk 1.0.0.

Create and run a Worker​

Start with these dependencies. The #[workflow] and #[activities] macros expand to code that refers to the temporalio-workflow and futures crates, so both must be direct dependencies of your crate even though your own code never names them:

[dependencies]
temporalio-sdk = "1.0.0"
temporalio-client = "1.0.0"
temporalio-common = "1.0.0"
temporalio-macros = "1.0.0"
temporalio-workflow = "1.0.0"
futures = "0.3"
tokio = { version = "1", features = ["full"] }

A Worker needs a Runtime and a connected Temporal Client. Build WorkerOptions with the Task Queue to poll, register the Workflows and Activities the Worker can execute, then call run():

use std::str::FromStr;

use temporalio_client::{Client, ClientOptions, Connection, ConnectionOptions, Url};
use temporalio_sdk::{Runtime, Worker, WorkerOptions};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let runtime = Runtime::from_current_tokio(Default::default())?;

    let connection_options =
        ConnectionOptions::new(Url::from_str("http://localhost:7233")?).build();
    let connection = Connection::connect(connection_options).await?;
    let client = Client::new(connection, ClientOptions::new("default").build())?;

    let worker_options = WorkerOptions::new("my-task-queue")
        .register_workflow::<GreetingWorkflow>()?
        .register_activities(GreetingActivities)
        .build();

    let mut worker = Worker::new(&runtime, client, worker_options)?;
    worker.run().await?;

    Ok(())
}

run() polls until the Worker shuts down. ClientOptions::new() takes the Namespace, and Connection::connect() takes the server address.

Register Workflows and Activities​

All Workers polling the same Task Queue must register the same Workflow Types and Activity Types. A Task Queue does not route by type, so any Worker polling it can receive any Task on that queue. A Worker that receives a Task for a type it did not register fails that Task.

Register each Workflow with register_workflow::<T>() and each Activity implementer with register_activities(). register_workflow returns a Result, so it needs ? or other error handling:

let worker_options = WorkerOptions::new("my-task-queue")
    .register_workflow::<GreetingWorkflow>()?
    .register_workflow::<OrderWorkflow>()?
    .register_activities(GreetingActivities)
    .build();

register_activities() takes an instance, so Activities can share state such as a database client through the fields of that value.

Connect to Temporal Cloud​

To run a Worker against Temporal Cloud, configure the Client connection with your Namespace address and authentication credentials. See Connect to Temporal Cloud for setup instructions.

Configure Worker options​

WorkerOptions controls the Task Queue, cache size, poller behavior, and slot allocation, including max_cached_workflows, workflow_task_poller_behavior, and tuner. Set client_identity_override to give the Worker an identity that is more useful than the default of {pid}@{hostname}. The defaults work for most cases.

To tune these values against real load, see Worker performance and the Worker tuning reference.

Run a versioned Worker​

Set a Worker Deployment Version and enable versioning in deployment_options, then set the default versioning behavior for the Workflows on the Worker.

use temporalio_common::worker::{
    VersioningBehavior, WorkerDeploymentOptions, WorkerDeploymentVersion,
};

let worker_options = WorkerOptions::new("my-task-queue")
    .deployment_options(
        WorkerDeploymentOptions::new(
            WorkerDeploymentVersion::builder()
                .deployment_name("my-app")
                .build_id("1.0")
                .build(),
        )
        .use_worker_versioning(true)
        .default_versioning_behavior(VersioningBehavior::Pinned)
        .build(),
    )
    .register_workflow::<GreetingWorkflow>()?
    .register_activities(GreetingActivities)
    .build();

The Rust SDK sets the versioning behavior on the Worker, not per Workflow, so default_versioning_behavior covers every Workflow the Worker registers. Setting it to VersioningBehavior::Unspecified is an error at startup.

See Worker Versioning for the available versioning behaviors and how new versions roll out.

Shut down a Worker​

run() borrows the Worker mutably, so call shutdown_handle() before starting the Worker. Calling the handle it returns initiates shutdown, which stops polling for new Tasks and lets in-flight Tasks finish.

let mut worker = Worker::new(&runtime, client, worker_options)?;
let shutdown = worker.shutdown_handle();

tokio::spawn(async move {
    tokio::signal::ctrl_c().await.expect("failed to listen for ctrl-c");
    shutdown();
});

worker.run().await?;

See Worker shutdown for what happens to in-flight Workflow Tasks and Activities.