activities/execution

235 notes in this subject, read out of this brain and free to use. This is page 4 of 4.

Activity cancellation requirements in Rust

Canceling an Activity from within a Workflow requires that the Activity Execution sends Heartbeats and sets a Heartbeat Timeout. If the Heartbeat is not invoked, the Activity cannot receive a cancellation request. When any non-immediate Activity is executed, the Activity Execution should send Heartbeats and set a Heartbeat Timeout to ensure that the server knows it is still working.

Cancel Activity from Workflow example in Rust

Example of canceling an Activity from a Workflow in Rust: ```rust fn activity_opts() -> ActivityOptions { ActivityOptions::with_start_to_close_timeout(Duration::from_secs(300)) .heartbeat_timeout(Duration::from_secs(5)) .build() } #[workflow_methods] impl CancellationWorkflow { #[run] pub async fn run(ctx: &mut WorkflowContext<Self>, _input: ()) -> WorkflowResult<String> { let mut activity_fut = ctx.start_activity( CancellationActivities::long_running_cancellable_activity, (), activity_opts(), ); temporalio_sdk::workflows::select! { result = &mut activity_fut => { let value = result?; Ok(value) } reason = ctx.cancelled() => { activity_fut.cancel(); let cleanup_result = ctx .start_activity( CancellationActivities::cleanup, (), ActivityOptions::start_to_close_timeout(Duration::from_secs(10)), ) .await?; Ok(format!("Cancelled (reason={reason}), {cleanup_result}")) } } } } ``` This example shows a Workflow that starts a cancellable Activity with a heartbeat timeout of 5 seconds. It uses select! to listen for either the Activity completion or a cancellation signal. When cancelled, it calls activity_fut.cancel() to cancel the Activity and then runs a cleanup Activity.

Activity Task events generated when calling a proxied Activity

When you call a proxied Activity function from a workflow, it schedules an Activity Task rather than executing the Activity code directly. This generates three Activity Task related Events in the Workflow Execution Event History: ActivityTaskScheduled, ActivityTaskStarted, and ActivityTaskCompleted.

Dynamic Activity reference example in TypeScript

export async function DynamicWorkflow(activityName, ...args) { const acts = proxyActivities(/* activityOptions */); // these are equivalent await acts.activity1(); await acts['activity1'](); let result = await acts[activityName](...args); return result; } This example shows how to reference Activities dynamically by string name, with both dot and bracket notation shown as equivalent.

Dynamic Activity reference in TypeScript

Since Activities are referenced by their string name, you can reference them dynamically to get the result of an Activity Execution. You can call an Activity using both dot notation (acts.activity1()) and bracket notation (acts['activity1']()). This allows you to reference Activities by variable name: await acts[activityName](...args).

Getting results from an Activity Execution

The call to spawn an Activity Execution generates a ScheduleActivityTask Command and provides the Workflow with an Awaitable. Workflow Executions can either block progress until the result is available through the Awaitable or continue progressing and make use of the result when it becomes available.

Required Activity Timeouts

Activity Execution semantics rely on several timeout parameters. The only required value that needs to be set is either a Schedule-To-Close Timeout or a Start-To-Close Timeout. These values are set in the Activity Options.

Activity inputs and return values are recorded in Workflow history

Every Activity call you make is recorded in the Workflow's execution history, including the parameters you pass in and the value that comes back. This history allows Temporal to recover a Workflow after a failure. Because the entire history must be stored and replayed, you should avoid passing large objects as Activity inputs or return values to keep payloads small and help your Workflows replay and recover efficiently.

Activities must be idempotent

Activities should be written to be idempotent: calling them multiple times with the same input should have the same effect as calling them once. This is important because the Worker may run many Activity executions at the same time using the same Activity function code, and Temporal can retry an Activity if it fails or times out.

How to start an Activity Execution in TypeScript

Activity Executions are called from within a Workflow Definition. You do not call an Activity function directly. Instead, use the proxyActivities function to pass in the types of your Activities and Activity options. This returns an Activity Handle, a type-safe proxy object with the same function names and signatures as your real activities. You can then call your Activities from the Activity Handle as if they were normal async functions.

proxyActivities usage example in TypeScript

import { proxyActivities } from '@temporalio/workflow'; import type * as activities from './activities'; const activityHandle = proxyActivities<typeof activities>({ startToCloseTimeout: '1 minute', }); const { greet } = activityHandle; export async function example(name: string): Promise<string> { return await greet(name); } This example shows how to import only the activity types (not functions), create an Activity Handle with proxyActivities, destructure individual activity functions, and call them from a workflow.

Count Standalone Activities - TypeScript

Use `client.activity.count(query)` to count Standalone Activity Executions matching a List Filter query. Returns the total count of executions (running, completed, failed, etc.), not the number of queued tasks. The same query syntax works for both listing and counting. Example: ```typescript const { count } = await client.activity.count(query); console.log(`Total activities: ${count}`); ```

Execute Standalone Activity with type checking - TypeScript

Call `client.activity.typed()` with the Activity type (using `typeof` on imported activities) to get a typed Activity Client interface. This provides TypeScript compilation-time checking of Activity names and argument types. Then call the `execute` method to durably enqueue and execute the Activity, waiting for the result. Example: ```typescript import { Connection, Client } from '@temporalio/client'; import { loadClientConnectConfig } from '@temporalio/envconfig'; import * as activities from './activities'; import { nanoid } from 'nanoid'; const config = loadClientConnectConfig(); const connection = await Connection.connect(config.connectionOptions); const client = new Client({ connection }); const activitiesClient = client.activity.typed<typeof activities>(); const taskQueue = 'hello-standalone-activities'; const activityOptions = { taskQueue, startToCloseTimeout: '10s', }; const activityId = nanoid(); const result = await activitiesClient.execute('greet', { ...activityOptions, id: activityId, args: ['World'], }); ```

Execute Standalone Activity without type checking - TypeScript

Call `execute` or `start` directly on `client.activity` without using the typed interface. When called this way, neither the Activity name nor argument types are checked client-side. This allows execution when Activity types are not available.

Start Standalone Activity without waiting for result - TypeScript

Use `client.activity.start()` or `activitiesClient.start()` (typed interface) to durably enqueue a Standalone Activity job without waiting for execution. This returns an activity handle. Example: ```typescript const handle = await activitiesClient.start('greet', { ...activityOptions, id: activityId, args: ['Temporal'], }); ```

Get handle to existing Standalone Activity - TypeScript

Use `client.activity.getHandle(activityId)` to create a handle to a previously started Standalone Activity. This method takes an optional type argument to constrain the Activity result type, but correctness is not verified. The handle can then be used to wait for the result, describe, cancel, or terminate the Activity. Example: ```typescript const newHandle = client.activity.getHandle<string>(activityId); ```

Wait for Standalone Activity result - TypeScript

Call `await handle.result()` to wait for a Standalone Activity to be executed and fetch the result. Calling `execute` is equivalent to calling `start` followed by `await handle.result()`. Example: ```typescript console.log(await handle.result()); // Hello, Temporal! ```

List Standalone Activities - TypeScript

Use `client.activity.list(query)` to list Standalone Activity Executions matching a List Filter query. Returns an AsyncIterable yielding ActivityExecutionInfo entries. Only Standalone Activity Executions are returned; Activities running inside Workflows are not included. Example: ```typescript const query = 'TaskQueue="hello-standalone-activities"'; for await (const a of client.activity.list(query)) { console.log( `${a.activityId} | ${a.activityRunId} | ${a.activityType} | ${a.status} | ${a.closeTime?.toISOString()}`, ); } ```

Standalone Activities not available in Workflow Visibility

When listing or counting Activities, only Standalone Activity Executions are returned. Activities running inside Workflows are not included in the results.

Activity Heartbeat purpose

An Activity Heartbeat is a ping from the Worker Process executing the Activity to the Temporal Service. Each Heartbeat informs the Temporal Service that the Activity Execution is making progress and the Worker has not crashed. If the Temporal Service does not receive a Heartbeat within the Heartbeat Timeout period, the Activity will be considered timed out and another Activity Task Execution may be scheduled according to the Retry Policy. Activity Cancellations are delivered to Activities from the Temporal Service when they Heartbeat, so Activities that don't Heartbeat cannot be notified of Cancellation requests.

Activity Timeout types in TypeScript SDK

There are three types of Activity Timeouts available in ActivityOptions: Schedule-To-Close Timeout (maximum time from when the Activity Task is scheduled to when the server receives completion), Start-To-Close Timeout (maximum time from when a Worker polls the Activity Task to when the server receives completion), and Schedule-To-Start Timeout (maximum time from when an Activity Task is scheduled to when a Worker polls it; non-retryable by design). An Activity Execution must have either the Start-To-Close or the Schedule-To-Close Timeout set.

Set Activity Timeouts with proxyActivities

To set Activity Timeouts in TypeScript, use the proxyActivities() API with ActivityOptions properties: scheduleToCloseTimeout, startToCloseTimeout, and scheduleToStartTimeout. Example: const { myActivity } = proxyActivities<typeof activities>({ scheduleToCloseTimeout: '5m', startToCloseTimeout: '30s', scheduleToStartTimeout: '60s', });

Heartbeat throttling may delay cancellation

Heartbeats may not always be sent to the Temporal Service—they may be throttled by the Worker. Heartbeat throttling may lead to Cancellation getting delivered later than expected.

Set Activity Heartbeat Timeout

A Heartbeat Timeout works in conjunction with Activity Heartbeats. If the Temporal Server doesn't receive a Heartbeat before expiration of the Heartbeat Timeout, the Activity is considered timed out and another Activity Task Execution may be scheduled according to the Retry Policy. To set an Activity's Heartbeat Timeout in TypeScript, set the ActivityOptions.heartbeatTimeout property when creating the corresponding Activity proxy functions using proxyActivities(). Example: const { myLongRunningActivity } = proxyActivities<typeof activities>({ heartbeatTimeout: '30s', });

Activity Heartbeat checkpoint details

An Activity may optionally checkpoint its progression by providing a details argument to the heartbeat() function. Should the Activity Execution timeout and get retried, the Temporal Server will provide the details from the last Heartbeat to the next Activity Execution, allowing the Activity to efficiently resume its work. Example: export async function myActivity(): Promise<void> { const startingPoint = activityInfo().heartbeatDetails?.progress ?? 1; for (let progress = startingPoint; progress <= 1000; ++progress) { await sleep('1s'); heartbeat({ progress }); } }

Send Activity Heartbeat in TypeScript

To Heartbeat an Activity Execution in TypeScript, call the heartbeat() function from the Activity implementation. Example: export async function myActivity(): Promise<void> { for (let progress = 1; progress <= 1000; ++progress) { await sleep('1s'); heartbeat(); } }

Handle Activity Cancellation in TypeScript

In the TypeScript SDK, Activity implementations must opt in to observe cancellation. Use one of the following: await Context.current().cancelled (rejects with CancelledFailure), await Context.current().sleep(ms) (rejects on cancellation), or pass Context.current().cancellationSignal (an AbortSignal) to libraries that support abort. Activities receive cancellation notifications when they heartbeat; if an Activity doesn't heartbeat, delivery of cancellation notification can be delayed.

Cancel Activity in tests

MockActivityEnvironment exposes a .cancel() method that cancels the Activity Context. Use this to test whether an Activity reacts correctly to cancellation.

Example: Testing Activity cancellation

import { CancelledFailure, sleep } from '@temporalio/activity'; import { MockActivityEnvironment } from '@temporalio/testing'; import assert from 'assert'; async function activityFoo(): Promise<void> { heartbeat(6); await sleep(100); } const env = new MockActivityEnvironment(); env.on('heartbeat', (d: unknown) => { assert(d === 6); }); await assert.rejects(env.run(activityFoo), (err) => { assert.ok(err instanceof CancelledFailure); });

MockActivityEnvironment for testing Activities

Activities can be tested in isolation using MockActivityEnvironment, which provides a way to mock the Activity context, listen to Heartbeats, and cancel the Activity without creating a Worker. The constructor accepts an optional partial Activity Info object.

MockActivityEnvironment.run() method

Use MockActivityEnvironment.run() to run a function in an Activity Context. The method takes the function to run and its parameters as arguments.

Example: Testing Activity with MockActivityEnvironment

import { activityInfo } from '@temporalio/activity'; import { MockActivityEnvironment } from '@temporalio/testing'; import assert from 'assert'; async function activityFoo(a: number, b: number): Promise<number> { return a + b + activityInfo().attempt; } const env = new MockActivityEnvironment({ attempt: 2 }); const result = await env.run(activityFoo, 5, 35); assert.equal(result, 42);

Listen to Heartbeats in Activity tests

MockActivityEnvironment is an EventEmitter that emits a 'heartbeat' event. Use the event listener to verify Heartbeats emitted by the Activity during testing. MockActivityEnvironment does not throttle Heartbeats, unlike a Worker.

Example: Listening to Heartbeats in Activity test

import { heartbeat } from '@temporalio/activity'; import assert from 'assert'; async function activityFoo(): Promise<void> { heartbeat(6); } const env = new MockActivityEnvironment(); env.on('heartbeat', (d: unknown) => { assert(d === 6); }); await env.run(activityFoo);

Cancel an Activity from a Workflow - requirements

To cancel an Activity from within a Workflow, the Activity Execution must send Heartbeats and set a Heartbeat Timeout. If the Heartbeat is not invoked, the Activity cannot receive a cancellation request. When any non-immediate Activity is executed, the Activity Execution should send Heartbeats and set a Heartbeat Timeout to ensure that the server knows it is still working.

Eager Activity Start behavior

Eager Activity Start may happen automatically if the Worker processing a Workflow Task has also registered the Activity Definition being called. If it does, it may try to reserve an Activity Slot for the execution of the Activity, and the server may respond to the Workflow Task completion with the Activity Task for the worker to execute immediately.

Excessive Activity Heartbeats anti-pattern

Each Heartbeat counts as one Action. Only use Heartbeats for long-running Activities (10+ minutes) where you need to detect Worker failures and track progress. Short-running Activities that complete in seconds or minutes don't need Heartbeats.

Activity timeout types and definitions

Each Activity timeout controls the maximum duration of a different aspect of Activity Execution. There are three available timeouts: 1. Schedule-To-Close Timeout: The maximum amount of time allowed for the overall Activity Execution. 2. Start-To-Close Timeout: The maximum time allowed for a single Activity Task Execution. 3. Schedule-To-Start Timeout: The maximum amount of time allowed from when an Activity Task is scheduled to when a Worker starts that Activity Task. This timeout is non-retryable by design. An Activity Execution must have either the Start-To-Close or the Schedule-To-Close Timeout set.

Activity Heartbeat enables Cancellation delivery

Activity Cancellations are delivered to Activities from the Temporal Service when they Heartbeat. Activities that do not Heartbeat cannot receive a Cancellation. Heartbeat throttling may lead to Cancellation being delivered later than expected.

Cancel an Activity from a Workflow - error handling and behavior

When an Activity is canceled, an error is raised in the Activity at the next available opportunity. If cleanup logic needs to be performed, it can be done in a finally clause or inside a caught cancel error. However, for the Activity to appear canceled, the exception needs to be re-raised. Local Activities currently do not support cancellation.

ExecuteActivity returns command batched at yield point, not immediate execution

Calling ExecuteActivity by itself returns a command that gets batched with others when there is a yield point in the Workflow definition. Futures will not run until selector.Select(ctx) is called, so no calls to ExecuteActivity result in commands being sent to the server until selector.Select is called.

Standalone Activities in Ruby SDK

Standalone Activities are Activities that run independently, without being orchestrated by a Workflow. Instead of starting an Activity from within a Workflow Definition, you start a Standalone Activity directly from a Temporal Client. The way you write the Activity and register it with a Worker is identical to Workflow Activities.

Rust parallel Activity execution with join

Example of executing multiple activities in parallel in Rust: ```rust use temporalio_sdk::workflows::join; #[run] pub async fn run(ctx: &mut WorkflowContext<Self>) -> WorkflowResult<String> { let name = ctx.state(|s| s.name.clone()); // Execute an activity let greeting = ctx.start_activity( MyActivities::greet, name, ActivityOptions::start_to_close_timeout(Duration::from_secs(30)) ); let language = ctx.start_activity( MyActivities::call_greeting_service, ActivityLanguages::English, ActivityOptions::start_to_close_timeout(Duration::from_secs(30)) ); // Run in parallel let (greeting_res, language_res) = join!(greeting, language); } ``` This example shows how to store Futures without awaiting immediately and await them together for concurrent execution.

Use async primitives for advanced Activity patterns in Rust

Use direct .await in most cases for Activity execution. More advanced patterns, like parallel execution or cancellation, can be built using Rust's async primitives.

Activity execution returns Future in Rust

In Rust, Activities are executed using ctx.start_activity(...), which returns a Future that can be awaited.

Activity Execution generates three Event History events

Spawning an Activity Execution generates the ScheduleActivityTask Command, which results in three Activity Task related Events in the Workflow Execution Event History: ActivityTaskScheduled, ActivityTaskStarted, and ActivityTaskClosed.

Large Activity payloads impact Workflow performance

Values passed to Activities as input parameters or returned as results are recorded in the Workflow Execution history. This history is replayed to Workflow Workers during recovery. Large payloads can negatively impact Workflow performance. Be mindful of the size of data passed to and from Activities.

Required Activity timeouts in Rust

Activity Execution requires at least one of these timeouts to be set: Schedule-To-Close Timeout or Start-To-Close Timeout. These are configured as part of the Activity options when scheduling the Activity.

Activity timeout options in Rust

Available Activity timeout options in Rust are: start_to_close_timeout, schedule_to_close_timeout, with_start_to_close_timeout, with_schedule_to_close_timeout.

Rust Activity execution with schedule_to_close_timeout example

Example of executing an activity in Rust with schedule_to_close_timeout: ```rust #[workflow_methods] impl GreetingWorkflow { #[init] fn new(_ctx: &WorkflowContextView, name: String) -> Self { Self { name } } #[run] pub async fn run(ctx: &mut WorkflowContext<Self>) -> WorkflowResult<String> { let name = ctx.state(|s| s.name.clone()); // Execute an activity let greeting = ctx.start_activity( MyActivities::greet, name, ActivityOptions::schedule_to_close_timeout(Duration::from_secs(30)) ).await?; println!("{}", greeting); Ok(greeting) } } ``` This example shows Activity execution with schedule_to_close_timeout.

Workflows can await Activity results immediately or later in Rust

Spawning an Activity Execution returns a Future to the Workflow. Workflows can either await the result immediately (blocking progress), or store the Future and await it later to allow concurrent execution.

Activity inheritance priority order

Activity priority is resolved in this order (highest precedence first): 1) Fairness weight overrides on the Task Queue (fairness_weight only), 2) Value set explicitly in Activity options, 3) Inherited from calling Workflow, 4) Default value (priority_key=3, fairness_key="", fairness_weight=1.0).

Ruby SDK: Set activity priority

In Ruby, set activity priority using the priority parameter: ```ruby client.start_activity( MyActivity, "input-arg", id: "my-workflow-id", task_queue: "my-task-queue", priority: Temporalio::Priority.new( priority_key: 3, fairness_key: "a-key", fairness_weight: 3.14 ) ) ```

Python SDK: Set activity priority

In Python, set activity priority using the priority parameter in execute_activity(): ```python await workflow.execute_activity( say_hello, "hi", priority=Priority(priority_key=3, fairness_key="a-key", fairness_weight=3.14), start_to_close_timeout=timedelta(seconds=5), ) ```

Go SDK: Set activity priority

In Go, set activity priority using the Priority field in ActivityOptions: ```go ao := workflow.ActivityOptions{ StartToCloseTimeout: time.Minute, Priority: temporal.Priority{ PriorityKey: 1, FairnessKey: "a-key", FairnessWeight: 3.14, }, } ctx := workflow.WithActivityOptions(ctx, ao) err := workflow.ExecuteActivity(ctx, MyActivity).Get(ctx, nil) ```

Give your agent this brain