activities/asynchronous-completion

49 notes, read out of this brain and free to use. Each one was extracted from a source and is re-checked against its exam.

Mark Activity as asynchronously completing in .NET

To mark an Activity as completing asynchronously in .NET, capture the task token from ActivityExecutionContext.Current.Info.TaskToken and then throw CompleteAsyncException(). This signals that the Activity will be completed by an external system.

CompleteAsyncException usage in .NET

CompleteAsyncException is a special exception thrown inside an Activity to indicate that the Activity Execution will be completed by an external system rather than completing when the Activity Function returns.

Get handle for asynchronous Activity completion in .NET

Use the ITemporalClient.GetAsyncActivityHandle() method with the captured task token to obtain a handle to an Activity for external completion. This handle allows you to call CompleteAsync(), FailAsync(), HeartbeatAsync(), or ReportCancellationAsync() methods to update the Activity.

Complete Activity asynchronously in .NET

Call await handle.CompleteAsync(value) on the Activity handle to complete an Activity that was marked for asynchronous completion. The value parameter is the completion result of the Activity.

Task Token vs Namespace/Workflow Id/Activity Id for async Activity

When performing asynchronous Activity completion, you can identify the Activity using either a Task Token or a combination of Namespace, Workflow Id, and Activity Id. The Task Token is obtained from ActivityExecutionContext.Current.Info.TaskToken.

.NET Activities - Asynchronous completion supported

The .NET SDK supports asynchronous Activity completion, which is documented as a distinct feature topic.

Get Task Token in Go activity

To retrieve the Task Token needed for asynchronous activity completion in Go, use the GetInfo() API from the go.temporal.io/sdk/activity package: activityInfo := activity.GetInfo(ctx); taskToken := activityInfo.TaskToken. The Task Token should be sent to the external service that will complete the Activity.

Return activity.ErrResultPending for async completion

To indicate that an Activity is completing asynchronously in Go, return the error activity.ErrResultPending from the Activity Function: return "", activity.ErrResultPending

CompleteActivity function parameters

The CompleteActivity function takes four parameters: context.Context, taskToken (binary TaskToken field from ActivityInfo struct), result (return value matching the Activity function's return type), and err (error code if Activity terminates with error). If err is not null, the result field value is ignored.

Complete async activity in Go using Temporal Client

To complete an asynchronous Activity in Go: instantiate a Temporal service client with client.Dial(client.Options{}), then call temporalClient.CompleteActivity(context.Background(), taskToken, result, nil). The same client can be reused to complete multiple Activities and should be created once per process as it is a heavyweight object.

Fail async activity in Go

To fail an asynchronous Activity in Go, call client.CompleteActivity(context.Background(), taskToken, nil, err) with the Task Token and error. When err is not null, the result field value is ignored.

ActivityCompletionClient interface usage

To complete an Activity asynchronously in Java SDK, set the ActivityCompletionClient interface to the complete() method. The complete() method takes a task token (byte[]) and the result value to complete the Activity with.

doNotCompleteOnReturn() method

Call the doNotCompleteOnReturn() method during an Activity Execution to prevent the Activity Execution from completing when its method returns. When this method is called, the Activity Execution does not complete when its method returns, allowing external systems to complete it later.

Get task token from Activity context

Retrieve the task token from the Activity execution context by calling context.getTaskToken(). This returns a byte[] that can be used as a correlation token to complete the activity asynchronously.

Asynchronous Activity completion example with ForkJoinPool

Example showing asynchronous Activity completion in Java SDK: ```java @Override public String composeGreeting(String greeting, String name) { // Get the activity execution context ActivityExecutionContext context = Activity.getExecutionContext(); // Set a correlation token that can be used to complete the activity asynchronously byte[] taskToken = context.getTaskToken(); // Execute asynchronously using ForkJoinPool ForkJoinPool.commonPool().execute(() -> composeGreetingAsync(taskToken, greeting, name)); context.doNotCompleteOnReturn(); // Return value is ignored when doNotCompleteOnReturn() is set return "ignored"; } // Method that will complete action execution using the defined ActivityCompletionClient private void composeGreetingAsync(byte[] taskToken, String greeting, String name) { String result = greeting + " " + name + "!"; // Complete the activity using ActivityCompletionClient completionClient.complete(taskToken, result); } ``` This example demonstrates how to schedule async work using ForkJoinPool, prevent automatic completion on method return, and then complete the Activity later from async code using the task token.

Async.function and Async.procedure for parallel activity invocation

The Temporal Java SDK provides the Async class which includes static methods used to invoke any Activity asynchronously. The calls return a result of type Promise, which is similar to Java Future and CompletionStage. Use Async.function for Activities that return a result, and Async.procedure for Activities that return void.

Parallel activity execution example

Example of invoking multiple Activities in parallel: ```java public void processFile(Arguments args) { List<Promise<String>> localNamePromises = new ArrayList<>(); List<String> processedNames = null; try { // Download all files in parallel. for (String sourceFilename : args.getSourceFilenames()) { Promise<String> localName = Async.function(activities::download, args.getSourceBucketName(), sourceFilename); localNamePromises.add(localName); } List<String> localNames = new ArrayList<>(); for (Promise<String> localName : localNamePromises) { localNames.add(localName.get()); } processedNames = activities.processFiles(localNames); // Upload all results in parallel. List<Promise<Void>> uploadedList = new ArrayList<>(); for (String processedName : processedNames) { Promise<Void> uploaded = Async.procedure( activities::upload, args.getTargetBucketName(), args.getTargetFilename(), processedName); uploadedList.add(uploaded); } // Wait for all uploads to complete. Promise.allOf(uploadedList).get(); } finally { for (Promise<String> localNamePromise : localNamePromises) { // Skip files that haven't completed downloading. if (localNamePromise.isCompleted()) { activities.deleteLocalFile(localNamePromise.get()); } } if (processedNames != null) { for (String processedName : processedNames) { activities.deleteLocalFile(processedName); } } } } ```

Get activity execution results with Promise.get()

To get the results of an asynchronously invoked Activity method, use the Promise get method to block until the Activity method result is available. This returns the result of the Activity Execution when it is ready.

Asynchronous activity completion with doNotCompleteOnReturn

Sometimes an Activity Execution lifecycle goes beyond a synchronous method invocation. For example, a request can be put in a queue and later a reply comes and is picked up by a different Worker process. To model this, call ActivityExecutionContext.doNotCompleteOnReturn() from the original Activity thread to indicate that an Activity should not be completed upon its method return. Then later, when replies come, complete the Activity using the ActivityCompletionClient. To correlate Activity invocation with completion, use either a TaskToken or Workflow and Activity Ids.

doNotCompleteOnReturn and manual completion example

Example of using doNotCompleteOnReturn for asynchronous activity completion: ```java public class FileProcessingActivitiesImpl implements FileProcessingActivities { public String download(String bucketName, String remoteName, String localName) { ActivityExecutionContext ctx = Activity.getExecutionContext(); byte[] taskToken = ctx.getInfo().getTaskToken(); asyncDownloadFileFromS3(taskToken, bucketName, remoteName, localDirectory + localName); ctx.doNotCompleteOnReturn(); return "ignored"; } } ``` When the download is complete, complete or fail the Activity from a different process: ```java public <R> void completeActivity(byte[] taskToken, R result) { completionClient.complete(taskToken, result); } public void failActivity(byte[] taskToken, Exception failure) { completionClient.completeExceptionally(taskToken, failure); } ```

Activity::doNotCompleteOnReturn() prevents immediate completion

Calling Activity::doNotCompleteOnReturn() in an Activity function prevents the Activity from completing when the function returns. When this method is invoked, the return value is ignored, and the Activity remains waiting for an external system to complete it.

Activity::getInfo()->taskToken retrieves the task token

Activity::getInfo()->taskToken retrieves the task token from the Activity context. This token can be encoded (for example with base64_encode) and sent to an external service that will later use it to complete the Activity.

Complete Activity by token using WorkflowClient

Use the ActivityCompletionClient obtained from WorkflowClient to complete Activities asynchronously. Call completeByToken($taskToken, $result) to complete the Activity successfully with a result.

Fail asynchronous Activity with completeExceptionallyByToken

To fail an asynchronously completed Activity, call completeExceptionallyByToken($taskToken, $exception) on the ActivityCompletionClient, passing the task token and an exception object.

PHP example: Asynchronous Activity completion with task token

Example of implementing an asynchronously completed Activity: ```php class GreetingActivity implements GreetingActivityInterface { private LoggerInterface $logger; public function __construct() { $this->logger = new Logger(); } public function composeGreeting(string $greeting, string $name): string { $this->logger->info(sprintf('GreetingActivity token: %s', base64_encode(Activity::getInfo()->taskToken))); Activity::doNotCompleteOnReturn(); return 'ignored'; } } ``` This Activity logs the encoded task token and calls doNotCompleteOnReturn() to prevent the Activity from completing when the function returns.

PHP example: Complete Activity by token from external system

Example of completing an asynchronous Activity from an external system: ```php $client = $this->workflowClient->newActivityCompletionClient(); $client->completeByToken( base64_decode($input->getArgument('token')), $input->getArgument('message') ); ``` The newActivityCompletionClient() method obtains the ActivityCompletionClient from the WorkflowClient, then completeByToken() is called with the decoded task token and the activity result.

PHP example: Fail asynchronous Activity by token

Example of failing an asynchronous Activity: ```php $activityClient->completeExceptionallyByToken($taskToken, new \Error("activity failed")); ``` Call completeExceptionallyByToken() with the task token and an exception to fail the Activity.

Asynchronous Activity completion overview

Asynchronous Activity Completion enables an Activity Function to return without the Activity Execution completing. This allows external systems to complete the activity at a later time.

Three steps for asynchronous activity completion

To implement asynchronous activity completion: (1) The Activity provides the external system with identifying information, either a Task Token or a combination of Namespace, Workflow Id, and Activity Id. (2) The Activity Function completes in a way that identifies it as waiting to be completed by an external system. (3) The Temporal Client is used to Heartbeat and complete the Activity.

Mark Activity as completing asynchronously with raise_complete_async

To mark an Activity as completing asynchronously in Python, call activity.raise_complete_async() inside the Activity Function. First, capture the task token using captured_token = activity.info().task_token, then call activity.raise_complete_async().

Get async activity handle in Python

To update an Activity outside the Activity Function, use the get_async_activity_handle() method on the Temporal Client with the captured task token: handle = my_client.get_async_activity_handle(task_token=captured_token).

Async activity handle methods

On an async activity handle, you can call the following methods: complete() to complete the Activity with a value, heartbeat() to send a heartbeat, fail() to fail the Activity, or report_cancellation() to report that the Activity was cancelled.

Complete async activity with handle.complete()

To complete an asynchronous Activity using the handle, call await handle.complete(value) where value is the completion result to return from the Activity.

Python async event loop blocks on synchronous calls

The Python async event loop runs in a thread and executes all tasks in its thread. When any task is running, the event loop is blocked and no other tasks can run simultaneously. The event loop can only pass control flow when the await keyword is executed. If a program makes a blocking call such as file I/O, synchronous network requests, or waits for user input, the entire event loop must wait until that execution completes.

Blocking async event loop turns program into synchronous execution

Blocking the async event loop in Python turns your asynchronous program into a synchronous program that executes serially, defeating the purpose of using asyncio. This can lead to potential deadlock and unpredictable behavior that causes tasks to be unable to execute. Debugging these issues is difficult and time consuming because locating the source of the blocking call may not be immediately evident.

Use async-safe HTTP libraries in async Activities

Making an HTTP call with the requests library within an asynchronous Activity blocks the event loop. If you want to make an HTTP call from within an asynchronous Activity, use an async-safe HTTP library such as aiohttp or httpx. Otherwise, use a synchronous Activity.

Asynchronous Activity example with aiohttp

This asynchronous Activity uses the aiohttp library to make async-safe HTTP requests and manages a ClientSession: ```python import aiohttp import urllib.parse from temporalio import activity class TranslateActivities: def __init__(self, session: aiohttp.ClientSession): self.session = session @activity.defn async def greet_in_spanish(self, name: str) -> str: greeting = await self.call_service("get-spanish-greeting", name) return greeting # Utility method for making calls to the microservices async def call_service(self, stem: str, name: str) -> str: base = f"http://localhost:9999/{stem}" url = f"{base}?name={urllib.parse.quote(name)}" async with self.session.get(url) as response: translation = await response.text() if response.status >= 400: raise ApplicationError( f"HTTP Error {response.status}: {translation}", # We want to have Temporal automatically retry 5xx but not 4xx non_retryable=response.status < 500, ) return translation ```

Run blocking code in async Activity using run_in_executor or to_thread

If your Activity is asynchronous and you need to run blocking code inside it without changing it to synchronous, you can use Python utility functions: loop.run_in_executor() or asyncio.to_thread().

When to use async Activities

Asynchronous Activities have many advantages such as potential speed up of execution. However, making unsafe calls within the async event loop can cause sporadic and difficult to diagnose bugs. Use asynchronous Activities only when you are certain that your Activities are async safe and do not make blocking calls. If you experience bugs that may result from an unsafe call in an asynchronous Activity, convert it to a synchronous Activity to see if the issue resolves.

Asynchronous Activity completion in Ruby

Asynchronous Activity Completion enables the Activity Function to return without the Activity Execution completing. This is useful when an Activity needs to be completed by an external system. The process involves three steps: (1) The Activity provides the external system with identifying information (either a Task Token, or a combination of Namespace, Workflow Id, and Activity Id), (2) The Activity Function completes in a way that marks it as waiting for external completion, and (3) The Temporal Client is used to Heartbeat and complete the Activity.

Raise CompleteAsyncError to mark Activity for asynchronous completion

To mark an Activity as completing asynchronously in Ruby, capture the task token from Temporalio::Activity::Context.current.info.task_token, then raise Temporalio::Activity::CompleteAsyncError inside the Activity.

Get async activity handle from client in Ruby

Use the async_activity_handle method on the Temporal Client, passing the captured task token as an argument, to get a handle to an Activity that is completing asynchronously.

Complete asynchronous Activity with handle in Ruby

Once you have an async activity handle, you can call the following methods on it: complete, fail, report_cancellation, or heartbeat. For example, handle.complete('completion value') completes the Activity with a value.

Ruby async activity completion example

Example showing asynchronous Activity completion in Ruby: ```ruby # Inside the Activity captured_token = Temporalio::Activity::Context.current.info.task_token raise Temporalio::Activity::CompleteAsyncError # Outside the Activity, using the client handle = my_client.async_activity_handle(captured_token) handle.complete('completion value') ```

Asynchronous Activity Completion overview

Asynchronous Activity Completion enables the Activity Function to return without the Activity Execution completing. This requires three steps: (1) the Activity provides identifying information to the external system (either a Task Token or a combination of Namespace, Workflow Id, and Activity Id), (2) the Activity Function completes in a way that identifies it as waiting for completion by an external system, and (3) the Temporal Client is used to heartbeat and complete the Activity.

CompleteAsyncError in TypeScript activities

To asynchronously complete an Activity in TypeScript, throw CompleteAsyncError from the Activity Function. This signals that the Activity is waiting to be completed by an external system rather than completing normally.

AsyncCompletionClient.complete method

Call AsyncCompletionClient.complete to asynchronously complete an Activity from an external system. Pass the task token and the result value to complete the Activity with.

Get task token from Activity using activityInfo()

Inside an Activity Function, call activityInfo().taskToken to obtain the task token needed to complete the Activity asynchronously from an external system.

Asynchronous Activity Completion example in TypeScript

import { CompleteAsyncError, activityInfo } from '@temporalio/activity'; import { Client } from '@temporalio/client'; export async function doSomethingAsync(): Promise<string> { const taskToken = activityInfo().taskToken; setTimeout(() => doSomeWork(taskToken), 1000); throw new CompleteAsyncError(); } // this work could be done in a different process or on a different machine async function doSomeWork(taskToken: Uint8Array): Promise<void> { const client = new Client(); // does some work... await client.activity.complete(taskToken, "Job's done!"); } This example shows how to schedule async work, extract the task token from activityInfo(), throw CompleteAsyncError to return from the Activity, and then complete the Activity from external code using the client.

Give your agent this brain