Skip to main content

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​

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.

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"

Register types​

All Workers listening to the same Task Queue must be registered to handle the same Workflow Types and Activity Types. If a Worker polls a Task for a type it does not know about, the Task fails. The Workflow Execution itself does not fail.

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 take a shutdown handle before starting it. Calling the handle 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.