Underway

To perform side effects that are replay-safe and durable, implement the Activity trait. Activities are registered with the workflow via .activity(). Inside a workflow step, you can interact with activities using:

• InvokeActivity::call: Executes an activity and waits for its result.
• InvokeActivity::emit: Triggers an activity (fire-and-forget style) without waiting for a return value.

Activities must define a NAME, an Input type, and an Output type. Errors within an activity can be categorized as retryable using ActivityError::retryable to signal to the runtime that the task should be attempted again.

use serde::{Deserialize, Serialize};
use sqlx::PgPool;
use underway::{Activity, ActivityError, InvokeActivity, Transition, Workflow};

#[derive(Clone)]
struct LookupEmail {
    pool: PgPool,
}

impl Activity for LookupEmail {
    const NAME: &'static str = "lookup-email";

    type Input = i64;
    type Output = String;

    async fn execute(&self, user_id: Self::Input) -> underway::activity::Result<Self::Output> {
        let email = sqlx::query_scalar::<_, String>("select concat('user-', $1::text, '@example.com')")
            .bind(user_id)
            .fetch_one(&self.pool)
            .await
            .map_err(|err| ActivityError::retryable("db_error", err.to_string()))?;

        Ok(email)
    }
}

struct TrackSignupMetric;

impl Activity for TrackSignupMetric {
    const NAME: &'static str = "track-signup-metric";

    type Input = String;
    type Output = ();

    async fn execute(&self, email: Self::Input) -> underway::activity::Result<Self::Output> {
        println!("tracking signup metric for {email}");
        Ok(())
    }
}

#[derive(Deserialize, Serialize)]
struct Signup {
    user_id: i64,
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let pool = PgPool::connect(&std::env::var("DATABASE_URL")?).await?;
    underway::run_migrations(&pool).await?;

    let workflow = Workflow::builder()
        .activity(LookupEmail { pool: pool.clone() })
        .activity(TrackSignupMetric)
        .step(|mut cx, Signup { user_id }| async move {
            let email: String = LookupEmail::call(&mut cx, &user_id).await?;
            TrackSignupMetric::emit(&mut cx, &email).await?;
            Transition::complete()
        })
        .name("signup-side-effects")
        .pool(pool)
        .build()
        .await?;

    workflow.enqueue(&Signup { user_id: 42 }).await?;
    workflow.runtime().run().await?;
    Ok()
}

You can define a multi-step workflow using the Workflow::builder(). Each step is defined using .step(), which takes a closure that receives a context and an input type, returning a Transition. Transitions can move to the next step using Transition::next(input) or finish the workflow using Transition::complete(). Once built, you can enqueue an initial input using .enqueue() and start the execution engine with .runtime().

use serde::{Deserialize, Serialize};
use sqlx::PgPool;
use underway::{Transition, Workflow};

#[derive(Deserialize, Serialize)]
struct ResizeImage {
    asset_id: i64,
}

#[derive(Deserialize, Serialize)]
struct PublishImage {
    object_key: String,
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let pool = PgPool::connect(&std::env::var("DATABASE_URL")?).await?;
    underway::run_migrations(&pool).await?;

    let workflow = Workflow::builder()
        .step(|_cx, ResizeImage { asset_id }| async move {
            let object_key = format!("images/{asset_id}.webp");
            Transition::next(PublishImage { object_key })
        })
        .step(|_cx, PublishImage { object_key }| async move {
            println!("Publishing {object_key}");
            Transition::complete()
        })
        .name("image-pipeline")
        .pool(pool)
        .build()
        .await?;

    workflow.enqueue(&ResizeImage { asset_id: 42 }).await?;

    let runtime_handle = workflow.runtime().start();
    runtime_handle.shutdown().await?;
    Ok()
}

Underway allows you to enqueue or schedule workflows within an existing SQL transaction using the *_using API. This ensures that your business logic updates (e.g., updating a database row) and the workflow trigger (e.g., enqueuing a cleanup task) succeed or fail together.

• enqueue_using(&mut *tx, &input): Adds a workflow to the queue within the transaction.
• schedule_using(&mut *tx, &cron_expression, &input): Schedules a workflow to run at a specific time (using cron syntax) within the transaction.

use serde::{Deserialize, Serialize};
use sqlx::PgPool;
use underway::{Transition, Workflow};

#[derive(Deserialize, Serialize)]
struct TenantCleanup {
    tenant_id: i64,
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let pool = PgPool::connect(&std::env::var("DATABASE_URL")?).await?;
    underway::run_migrations(&pool).await?;

    let workflow = Workflow::builder()
        .step(|_cx, TenantCleanup { tenant_id }| async move {
            println!("Running cleanup for tenant {tenant_id}");
            Transition::complete()
        })
        .name("tenant-cleanup")
        .pool(pool.clone())
        .build()
        .await?;

    let nightly = "0 2 * * *[UTC]".parse()?;
    let tenant_id = 7;

    let mut tx = pool.begin().await?;

    sqlx::query("update app_tenant set cleanup_enabled = true where id = $1")
        .bind(tenant_id)
        .execute(&mut *tx)
        .await?;

    let input = TenantCleanup { tenant_id };
    workflow.enqueue_using(&mut *tx, &input).await?;
    workflow.schedule_using(&mut *tx, &nightly, &input).await?;

    tx.commit().await?;
    Ok()
}

For side effects that need to be durable and replay-safe (like sending emails or calling external APIs), use Activities.

1. Implement the Activity trait: Define a NAME, Input type, and Output type. The execute method handles the logic and returns an underway::activity::Result.
2. Register the activity: Add it to the Workflow::builder using .activity(YourActivity).
3. Invoke the activity: Inside a workflow step, use InvokeActivity::call (request/response, suspends step until done) or InvokeActivity::emit (fire-and-forget).

Activities allow you to handle errors specifically as retryable or fatal.

use serde::{Deserialize, Serialize};
use sqlx::PgPool;
use underway::{Activity, ActivityError, InvokeActivity, Transition, Workflow};

struct ResolveEmail;

impl Activity for ResolveEmail {
    const NAME: &'static str = "resolve-email";
    type Input = i64;
    type Output = String;

    async fn execute(&self, user_id: Self::Input) -> underway::activity::Result<Self::Output> {
        if user_id <= 0 {
            return Err(ActivityError::fatal("invalid_user", "user_id must be positive"));
        }
        Ok(format!("{user_id}@example.com"))
    }
}

#[derive(Deserialize, Serialize)]
struct Input { user_id: i64 }

# async fn run_with_activity(pool: PgPool) -> Result<(), Box<dyn std::error::Error>> {
let workflow = Workflow::builder()
    .activity(ResolveEmail)
    .step(|mut cx, Input { user_id }| async move {
        let _email: String = ResolveEmail::call(&mut cx, &user_id).await?;
        Transition::complete()
    })
    .name("resolve-email-workflow")
    .pool(pool)
    .build()
    .await?;

workflow.runtime().run().await?;
# Ok(())
//}

The Runtime is the high-level entrypoint for executing workflows. It manages the coordination between three critical components:

1. Workflow Worker: Processes workflow steps.
2. Workflow Scheduler: Manages the scheduling of workflow tasks.
3. Activity Worker: Handles the execution of activities called by or emitted from workflows.

This orchestration is necessary because workflow steps often suspend while waiting for an activity to complete. The Runtime ensures that both the workflow execution and the activity execution run concurrently so that suspended steps can be resumed once their requested activities finish.

use serde::{Deserialize, Serialize};
use sqlx::PgPool;
use underway::{Activity, ActivityError, InvokeActivity, Transition, Workflow};

#[derive(Deserialize, Serialize)]
struct FetchUser {
    user_id: i64,
}

struct LookupEmail;

impl Activity for LookupEmail {
    const NAME: &'static str = "lookup-email";

    type Input = i64;
    type Output = String;

    async fn execute(&self, user_id: Self::Input) -> underway::activity::Result<Self::Output> {
        if user_id <= 0 {
            return Err(ActivityError::fatal(
                "invalid_user",
                "user_id must be positive",
            ));
        }

        Ok(format!("user-{user_id}@example.com"))
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let pool = PgPool::connect(&std::env::var("DATABASE_URL")?).await?;
    let workflow = Workflow::builder()
        .activity(LookupEmail)
        .step(|mut cx, FetchUser { user_id }| async move {
            let email: String = LookupEmail::call(&mut cx, &user_id).await?;
            println!("Got email {email}");
            Transition::complete()
        })
        .name("lookup-email-workflow")
        .pool(pool)
        .build()
        .await?;

    workflow.runtime().run().await?;
    Ok()
}

A Workflow is a series of sequential steps. You can build one using Workflow::builder(), defining steps that receive input and return a Transition.

• Transition::complete() signals the end of the workflow.
• Transition::next(input) signals the workflow should proceed to the next step using the provided input.

Once built, you can enqueue an initial input to start the workflow and use the runtime() to execute it.

use serde::{Deserialize, Serialize};
use sqlx::PgPool;
use underway::{Transition, Workflow};

#[derive(Deserialize, Serialize)]
struct ResizeImage { asset_id: i64 }

#[derive(Deserialize, Serialize)]
struct PublishImage { object_key: String }

# async fn run_workflow(pool: PgPool) -> Result<(), Box<dyn std::error::Error>> {
let workflow = Workflow::builder()
    .step(|_cx, ResizeImage { asset_id }| async move {
        let object_key = format!("images/{asset_id}.webp");
        Transition::next(PublishImage { object_key })
    })
    .step(|_cx, PublishImage { object_key }| async move {
        println!("Publishing {object_key}");
        Transition::complete()
    })
    .name("image-pipeline")
    .pool(pool)
    .build()
    .await?;

workflow.enqueue(&ResizeImage { asset_id: 42 }).await?;
workflow.runtime().run().await?;
# Ok(())
//}

To ensure atomicity between your application state and your background workflows, Underway provides *_using methods. This allows you to enqueue a task or schedule a cron job as part of a single Postgres transaction.

• workflow.enqueue_using(&mut transaction, &input): Adds a task to the queue immediately upon commit.
• workflow.schedule_using(&mut transaction, &cron_string, &input): Schedules a task to run at a specific time (using cron syntax) upon commit.

use serde::{Deserialize, Serialize};
use sqlx::PgPool;
use underway::{Transition, Workflow};

#[derive(Deserialize, Serialize)]
struct TenantCleanup { tenant_id: i64 }

# async fn transactional_enqueue(pool: PgPool) -> Result<(), Box<dyn std::error::Error>> {
let workflow = Workflow::builder()
    .step(|_cx, TenantCleanup { tenant_id }| async move {
        println!("Running cleanup for tenant {tenant_id}");
        Transition::complete()
    })
    .name("tenant-cleanup")
    .pool(pool.clone())
    .build()
    .await?;

let nightly = "0 2 * * *[UTC]".parse()?;
let tenant_id = 7;
let input = TenantCleanup { tenant_id };

let mut tx = pool.begin().await?;

// Perform app logic
sqlx::query("update app_tenant set cleanup_enabled = true where id = $1")
    .bind(tenant_id)
    .execute(&mut *tx)
    .await?;

// Enqueue/Schedule as part of the same transaction
workflow.enqueue_using(&mut *tx, &input).await?;
workflow.schedule_using(&mut *tx, &nightly, &input).await?;

tx.commit().await?;
# Ok(())
//}