Run a Worker - Rust SDK

View Markdown

The Rust SDK is in Public Preview, and its API can change between releases. The code on this page is written against temporalio-sdk 0.5.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 = "0.5.0"
temporalio-client = "0.5.0"
temporalio-sdk-core = "0.5.0"
temporalio-common = "0.5.0"
temporalio-macros = "0.5.0"
temporalio-workflow = "0.5.0"
futures = "0.3"
tokio = { version = "1", features = ["full"] }
url = "2"

A Worker needs a CoreRuntime and a connected 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};
use temporalio_common::telemetry::TelemetryOptions;
use temporalio_sdk::{Worker, WorkerOptions};
use temporalio_sdk_core::{CoreRuntime, RuntimeOptions, Url};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let runtime = CoreRuntime::new_assume_tokio(
        RuntimeOptions::builder()
            .telemetry_options(TelemetryOptions::builder().build())
            .build()?,
    )?;

    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::{
    protos::temporal::api::enums::v1::VersioningBehavior,
    worker::{WorkerDeploymentOptions, WorkerDeploymentVersion},
};

let worker_options = WorkerOptions::new("my-task-queue")
    .deployment_options(WorkerDeploymentOptions {
        version: WorkerDeploymentVersion {
            deployment_name: "my-app".to_owned(),
            build_id: "1.0".to_owned(),
        },
        use_worker_versioning: true,
        default_versioning_behavior: Some(VersioningBehavior::Pinned),
    })
    .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 Some(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.