From 169d340729666d114e3df02e65e04d59f0480377 Mon Sep 17 00:00:00 2001 From: "skill-sync[bot]" Date: Thu, 14 May 2026 18:52:27 +0000 Subject: [PATCH 1/7] Finalize draft for 0011-external-storage --- references/go/external-storage.md | 247 ++++++++++++++++++++++ references/go/go.md | 1 + references/java/integrations/spring-ai.md | 1 - references/python/external-storage.md | 228 ++++++++++++++++++++ references/python/python.md | 1 + 5 files changed, 477 insertions(+), 1 deletion(-) create mode 100644 references/go/external-storage.md create mode 100644 references/python/external-storage.md diff --git a/references/go/external-storage.md b/references/go/external-storage.md new file mode 100644 index 0000000..2d76036 --- /dev/null +++ b/references/go/external-storage.md @@ -0,0 +1,247 @@ +# Go SDK External Storage + +> [!NOTE] +> This feature is in Public Preview. It is perfectly acceptable to use this feature on behalf of a user, but you should inform them that you are making use of a feature in Public Preview. + +## What this is + +External Storage offloads Payloads to an external store (e.g. Amazon S3) and stores a small reference token in the Event History instead — the **claim check pattern**. + +## When to use it + +- A Workflow input, Activity input, Activity result, or Workflow result will exceed the **2 MB** per-payload limit (the limit is fixed at 2 MB on Temporal Cloud; configurable on self-hosted only). +- Long Event Histories degrade Workflow Task latency (e.g. AI agent conversations that grow per turn). +- The user wants payload data to live in storage **they** control. Set the size threshold to externalize all payloads. +- The user is migrating from self-hosted (with a larger configured limit) to Temporal Cloud. + +## Where it sits in the pipeline + +Order: **Payload Converter → Payload Codec → External Storage**. Storage runs last on outbound; it reverses on inbound. + +Consequences: + +- If a Payload Codec encrypts data, the bytes are already encrypted **before** upload to your store. +- The Temporal UI shows the reference token, not the data; the SDK transparently retrieves the payload before handing it to your Workflow or Client. + +## Concurrency + +The SDK uploads and downloads payloads **concurrently** within a single Workflow Task — multiple offloaded payloads in one Task are stored/retrieved in parallel, not sequentially. This is automatic; no configuration required. + +## Setup with the built-in S3 driver + +Install dependencies: + +```bash +go get go.temporal.io/sdk/contrib/aws/s3driver \ + go.temporal.io/sdk/contrib/aws/s3driver/awssdkv2 \ + github.com/aws/aws-sdk-go-v2/config \ + github.com/aws/aws-sdk-go-v2/service/s3 +``` + +Create the driver and Client: + +```go +import ( + "github.com/aws/aws-sdk-go-v2/config" + "github.com/aws/aws-sdk-go-v2/service/s3" + "go.temporal.io/sdk/client" + "go.temporal.io/sdk/contrib/aws/s3driver" + "go.temporal.io/sdk/contrib/aws/s3driver/awssdkv2" + "go.temporal.io/sdk/converter" + "go.temporal.io/sdk/worker" +) + +cfg, err := config.LoadDefaultConfig(context.Background(), + config.WithRegion("us-east-2"), +) +if err != nil { + log.Fatalf("load AWS config: %v", err) +} + +driver, err := s3driver.NewDriver(s3driver.Options{ + Client: awssdkv2.NewClient(s3.NewFromConfig(cfg)), + Bucket: s3driver.StaticBucket("my-temporal-payloads"), +}) +if err != nil { + log.Fatalf("create S3 driver: %v", err) +} + +c, err := client.Dial(client.Options{ + HostPort: "localhost:7233", + ExternalStorage: converter.ExternalStorage{ + Drivers: []converter.StorageDriver{driver}, + }, +}) +if err != nil { + log.Fatalf("connect to Temporal: %v", err) +} +defer c.Close() + +w := worker.New(c, "my-task-queue", worker.Options{}) +``` + +The S3 driver uses standard AWS credentials from the environment (env vars, IAM role, or AWS config file). + +Workflows and Activities running on the Worker use the driver automatically — no changes to business logic. + +## Payload size threshold + +- Default: **256 KiB**. +- Set `PayloadSizeThreshold: 1` to externalize **all** payloads regardless of size. +- `PayloadSizeThreshold: 0` is **interpreted as the default (256 KiB)** — it does **not** mean "externalize everything". + +```go +c, err := client.Dial(client.Options{ + ExternalStorage: converter.ExternalStorage{ + Drivers: []converter.StorageDriver{driver}, + PayloadSizeThreshold: 1, + }, +}) +``` + +## Multiple drivers and migration + +When you register more than one driver, you **must** supply a `DriverSelector` implementing `StorageDriverSelector`. The selector chooses which driver stores each payload. Unselected drivers remain available for **retrieval** — this is how you migrate between storage backends without losing access to existing claims. + +- Return `nil` from the selector to keep a specific payload inline in Event History. + +```go +type PreferredSelector struct { + preferred converter.StorageDriver +} + +func (s *PreferredSelector) SelectDriver( + ctx converter.StorageDriverStoreContext, + payload *commonpb.Payload, +) (converter.StorageDriver, error) { + return s.preferred, nil +} + +func MultipleDriversSetup(preferredDriver, legacyDriver converter.StorageDriver) converter.ExternalStorage { + return converter.ExternalStorage{ + Drivers: []converter.StorageDriver{preferredDriver, legacyDriver}, + DriverSelector: &PreferredSelector{preferred: preferredDriver}, + } +} +``` + +## Custom storage driver + +Implement `converter.StorageDriver` with **four** methods: + +- `Name() string` — unique identifier for **this driver instance**, stored in the claim reference so the SDK can route retrieval. Renaming after payloads are stored **breaks retrieval**. +- `Type() string` — identifier for the driver **implementation**, same across all instances regardless of configuration (e.g. `"aws.s3driver"`, `"local-disk"`). +- `Store(ctx, payloads) ([]StorageDriverClaim, error)` — upload each Payload protobuf and return one claim per payload. A claim is a `map[string]string` the driver uses to locate the payload later. +- `Retrieve(ctx, claims) ([]*commonpb.Payload, error)` — download bytes using claim data and reconstruct each Payload. + +Inside `Store()`, marshal each payload with `proto.Marshal(payload)`; in `Retrieve()`, reconstruct with `proto.Unmarshal(data, payload)`. The application data has already been serialized by the Payload Converter and Payload Codec before it reaches the driver. + +`ctx.Target` provides identity information. Type-switch over `StorageDriverWorkflowInfo` and `StorageDriverActivityInfo` to access the namespace / Workflow ID / Activity ID. `StorageDriverActivityInfo` is only used for standalone (non-workflow-bound) Activities; Activities started by a Workflow get `StorageDriverWorkflowInfo`. + +Worked example — local-disk driver (development/testing only): + +```go +type LocalDiskStorageDriver struct { + storeDir string +} + +func NewLocalDiskStorageDriver(storeDir string) converter.StorageDriver { + return &LocalDiskStorageDriver{storeDir: storeDir} +} + +func (d *LocalDiskStorageDriver) Name() string { return "my-local-disk" } +func (d *LocalDiskStorageDriver) Type() string { return "local-disk" } + +func (d *LocalDiskStorageDriver) Store( + ctx converter.StorageDriverStoreContext, + payloads []*commonpb.Payload, +) ([]converter.StorageDriverClaim, error) { + dir := d.storeDir + switch info := ctx.Target.(type) { + case converter.StorageDriverWorkflowInfo: + if info.WorkflowID != "" { + dir = filepath.Join(d.storeDir, info.Namespace, info.WorkflowID) + } + case converter.StorageDriverActivityInfo: + if info.ActivityID != "" { + dir = filepath.Join(d.storeDir, info.Namespace, info.ActivityID) + } + } + if err := os.MkdirAll(dir, 0o755); err != nil { + return nil, fmt.Errorf("create store directory: %w", err) + } + + claims := make([]converter.StorageDriverClaim, len(payloads)) + for i, payload := range payloads { + key := uuid.NewString() + ".bin" + filePath := filepath.Join(dir, key) + data, err := proto.Marshal(payload) + if err != nil { + return nil, fmt.Errorf("marshal payload: %w", err) + } + if err := os.WriteFile(filePath, data, 0o644); err != nil { + return nil, fmt.Errorf("write payload: %w", err) + } + claims[i] = converter.StorageDriverClaim{ + ClaimData: map[string]string{"path": filePath}, + } + } + return claims, nil +} + +func (d *LocalDiskStorageDriver) Retrieve( + ctx converter.StorageDriverRetrieveContext, + claims []converter.StorageDriverClaim, +) ([]*commonpb.Payload, error) { + payloads := make([]*commonpb.Payload, len(claims)) + for i, claim := range claims { + filePath := claim.ClaimData["path"] + data, err := os.ReadFile(filePath) + if err != nil { + return nil, fmt.Errorf("read payload: %w", err) + } + payload := &commonpb.Payload{} + if err := proto.Unmarshal(data, payload); err != nil { + return nil, fmt.Errorf("unmarshal payload: %w", err) + } + payloads[i] = payload + } + return payloads, nil +} +``` + +You can package a custom driver as a [plugin](/develop/plugins-guide) for reuse across services. + +## Codec Server with External Storage + +When Workers and Clients use External Storage, Event History contains reference tokens — not payload data. For the Web UI and CLI to display decoded payloads, the Codec Server must download from external storage **and** decode through the Payload Codec in the correct order. + +Build the Codec Server with `NewPayloadHTTPHandler` and `PayloadHTTPHandlerOptions`. Pass it your storage drivers, your pre-storage codecs (the Payload Codecs your Workers use), and any post-storage codecs (applied by a proxy after external storage). + +When configured with storage drivers, the handler exposes: + +- **`/download`** — retrieves payload data from external storage and decodes it through the Payload Codec. The Web UI calls this when a user clicks to view the full payload behind a reference. +- **`/decode`** — decodes encoded payloads and, by default, retrieves storage references inline. Pass `?preserveStorageRefs=true` to return storage references as-is without retrieval. +- **`/encode`** — applies the Payload Codec, then uploads payloads exceeding the threshold and replaces them with reference tokens. + +**Don't use `NewPayloadHTTPHandler` as a remote Data Converter or remote codec target for your Workers** — it runs the full encode-store-encode and decode-retrieve-decode pipeline. For remote codecs use `NewPayloadCodecHTTPHandler` separately. If you need both, run both handlers, configured with the same codecs. + +## Lifecycle management + +Temporal does **not** auto-delete payloads from your store. Configure a TTL on your bucket: + +``` +TTL > Maximum Workflow Run Timeout + Namespace Retention Period +``` + +Example: Run Timeout 14 days + Namespace retention 30 days → set TTL to at least 44 days. + +For Workflows with no finite Run Timeout, there is no safe finite TTL. Use Continue-as-New so the new run uploads fresh payloads and the old run's payloads only need to survive its retention period. + +## Anti-patterns + +- **Don't change `Name()` after payloads have been stored.** The name is embedded in the claim reference; renaming breaks retrieval of existing claims. +- **Don't use `PayloadSizeThreshold: 0` to mean "externalize all".** `0` is interpreted as the default (256 KiB). Use `PayloadSizeThreshold: 1`. +- **Don't register multiple drivers without a `DriverSelector`.** The selector is required when there are multiple drivers. +- **Don't point a Worker's remote codec at `NewPayloadHTTPHandler`.** Use `NewPayloadCodecHTTPHandler` for remote codec endpoints. +- **Don't omit a TTL on the bucket.** Payloads are orphaned otherwise; orphaned objects can also remain if a request fails after upload. diff --git a/references/go/go.md b/references/go/go.md index 6c42bed..903e992 100644 --- a/references/go/go.md +++ b/references/go/go.md @@ -250,5 +250,6 @@ See `references/go/testing.md` for info on writing tests. - **`references/go/testing.md`** - TestWorkflowEnvironment, time-skipping, activity mocking - **`references/go/advanced-features.md`** - Schedules, worker tuning, and more - **`references/go/data-handling.md`** - Data converters, payload codecs, encryption +- **`references/go/external-storage.md`** - Claim-check pattern for large payloads (S3 driver, custom drivers, codec-server handling) - **`references/go/versioning.md`** - Patching API (`workflow.GetVersion`), Worker Versioning - **`references/go/determinism-protection.md`** - Information on **`workflowcheck`** tool to help statically check for determinism issues. diff --git a/references/java/integrations/spring-ai.md b/references/java/integrations/spring-ai.md index 5ee0704..ae5154f 100644 --- a/references/java/integrations/spring-ai.md +++ b/references/java/integrations/spring-ai.md @@ -217,7 +217,6 @@ Media image = new Media(MimeTypeUtils.IMAGE_PNG, URI.create("https://cdn.example For anything larger than a small thumbnail, route the bytes to a binary store from an Activity and pass only the URL across the conversation. - ## Vector stores, embeddings, and MCP When the corresponding Spring AI modules (`spring-ai-rag`, `spring-ai-mcp`) are on the classpath, the integration registers Activities for vector stores, embeddings, and MCP tool calls automatically. Inject the matching Spring AI types into your Activities or Workflows and use them as you would in any Spring AI application — each operation executes through a Temporal Activity. diff --git a/references/python/external-storage.md b/references/python/external-storage.md new file mode 100644 index 0000000..0a3e1b5 --- /dev/null +++ b/references/python/external-storage.md @@ -0,0 +1,228 @@ +# Python SDK External Storage + +> [!NOTE] +> This feature is in Public Preview. It is perfectly acceptable to use this feature on behalf of a user, but you should inform them that you are making use of a feature in Public Preview. + +## What this is + +External Storage offloads Payloads to an external store (e.g. Amazon S3) and stores a small reference token in the Event History instead — the **claim check pattern**. + +## When to use it + +- A Workflow input, Activity input, Activity result, or Workflow result will exceed the **2 MB** per-payload limit (fixed at 2 MB on Temporal Cloud; configurable on self-hosted only). +- Long Event Histories degrade Workflow Task latency (e.g. AI agent conversations growing per turn). +- The user wants payload data to live in storage **they** control. Set the size threshold to externalize all payloads. +- The user is migrating from self-hosted (with a larger configured limit) to Temporal Cloud. + +## Where it sits in the pipeline + +Order: **Payload Converter → Payload Codec → External Storage**. Storage runs last on outbound; it reverses on inbound. + +Consequences: + +- If a Payload Codec encrypts data, the bytes are already encrypted **before** upload. +- The Temporal UI displays the reference token, not the data; the SDK retrieves the payload transparently before handing it to your Workflow or Client. + +## Concurrency + +The SDK uploads and downloads payloads **concurrently** within a single Workflow Task — multiple offloaded payloads in one Task are stored or retrieved in parallel, not sequentially. This is automatic. + +## Setup with the built-in S3 driver + +Install the `aioboto3` extra: + +```bash +python -m pip install "temporalio[aioboto3]" +``` + +Create the driver, attach it to a `DataConverter`, and pass the converter to both Client and Worker: + +```python +import aioboto3 +import dataclasses +from temporalio.client import Client, ClientConfig +from temporalio.contrib.aioboto3 import new_aioboto3_client +from temporalio.converter import DataConverter +from temporalio.external_storage import ExternalStorage, S3StorageDriver +from temporalio.worker import Worker + +session = aioboto3.Session(profile_name=AWS_PROFILE, region_name=AWS_REGION) +async with session.client("s3") as s3_client: + driver = S3StorageDriver( + client=new_aioboto3_client(s3_client), + bucket="my-temporal-payloads", + ) + + data_converter = dataclasses.replace( + DataConverter.default, + external_storage=ExternalStorage(drivers=[driver]), + ) + + client_config = ClientConfig.load_client_connect_config() + client = await Client.connect(**client_config, data_converter=data_converter) + + worker = Worker( + client, + task_queue="my-task-queue", + workflows=[], + activities=[], + ) +``` + +The S3 driver uses standard AWS credentials from the environment (env vars, IAM role, or AWS config file). + +Workflows and Activities on the Worker use the driver automatically — no business-logic changes. + +## Payload size threshold + +- Default: **256 KiB**. +- Set `payload_size_threshold=0` to externalize **all** payloads regardless of size. + +```python +data_converter = dataclasses.replace( + DataConverter.default, + external_storage=ExternalStorage( + drivers=[driver], + payload_size_threshold=0, + ), +) +``` + +## Multiple drivers and migration + +When you register more than one driver, you **must** supply a `driver_selector` function. The selector chooses which driver stores each payload. Unselected drivers remain available for **retrieval** — this is how you migrate between storage backends without losing access to existing claims. + +- Return `None` from the selector to keep a specific payload inline in Event History. + +```python +preferred_driver = S3StorageDriver(client=s3_client, bucket="my-bucket") +legacy_driver = LegacyStorageDriver() + +ExternalStorage( + drivers=[preferred_driver, legacy_driver], + driver_selector=lambda context, payload: preferred_driver, +) +``` + +## Custom storage driver + +Extend `StorageDriver` and implement **three** methods: + +- `name() -> str` — unique identifier for the driver, stored in the claim reference so the SDK can route retrieval. Renaming after payloads are stored **breaks retrieval**. +- `async store(context, payloads) -> list[StorageDriverClaim]` — upload each Payload and return one claim per payload. A claim is a `dict[str, str]` the driver uses to locate the payload later. +- `async retrieve(context, claims) -> list[Payload]` — download bytes using claim data and reconstruct each Payload. + +Inside `store()`, serialize each payload with `payload.SerializeToString()`; in `retrieve()`, reconstruct with `payload.ParseFromString(data)`. The application data has already been serialized by the Payload Converter and Payload Codec before reaching the driver. + +`context.target` provides identity information (namespace, Workflow ID, or Activity ID). Check the target type with `isinstance(target, StorageDriverWorkflowInfo)`; the Workflow info exposes `target.namespace` and `target.id`. Use this to scope storage keys per Workflow. + +Worked example — local-disk driver (development/testing only): + +```python +import os +import uuid +from typing import Sequence + +from temporalio.api.common.v1 import Payload +from temporalio.external_storage import ( + StorageDriver, + StorageDriverClaim, + StorageDriverRetrieveContext, + StorageDriverStoreContext, + StorageDriverWorkflowInfo, +) + + +class LocalDiskStorageDriver(StorageDriver): + def __init__(self, store_dir: str = "/tmp/temporal-payload-store") -> None: + self._store_dir = store_dir + + def name(self) -> str: + return "local-disk" + + async def store( + self, + context: StorageDriverStoreContext, + payloads: Sequence[Payload], + ) -> list[StorageDriverClaim]: + os.makedirs(self._store_dir, exist_ok=True) + + prefix = self._store_dir + target = context.target + if isinstance(target, StorageDriverWorkflowInfo) and target.id: + prefix = os.path.join(self._store_dir, target.namespace, target.id) + os.makedirs(prefix, exist_ok=True) + + claims = [] + for payload in payloads: + key = f"{uuid.uuid4()}.bin" + file_path = os.path.join(prefix, key) + with open(file_path, "wb") as f: + f.write(payload.SerializeToString()) + claims.append(StorageDriverClaim(claim_data={"path": file_path})) + return claims + + async def retrieve( + self, + context: StorageDriverRetrieveContext, + claims: Sequence[StorageDriverClaim], + ) -> list[Payload]: + payloads = [] + for claim in claims: + file_path = claim.claim_data["path"] + with open(file_path, "rb") as f: + raw = f.read() + payload = Payload() + payload.ParseFromString(raw) + payloads.append(payload) + return payloads +``` + +Wire the custom driver into the Data Converter the same way as the S3 driver: + +```python +data_converter = dataclasses.replace( + DataConverter.default, + external_storage=ExternalStorage( + drivers=[LocalDiskStorageDriver()], + ), +) +``` + +You can package a custom driver as a [plugin](/develop/plugins-guide) for reuse across services. + +## Codec Server with External Storage + +When Workers and Clients use External Storage, Event History contains reference tokens — not payload data. For the Web UI and CLI to show decoded payloads, the Codec Server must download from external storage **and** decode through the Payload Codec in the correct order. + +Build the Codec Server with a payload HTTP handler that accepts your storage drivers, your pre-storage codecs (the Payload Codecs your Workers use), and any post-storage codecs (applied by a proxy after external storage). The handler applies them in the correct order across all endpoints. + +Endpoints exposed when storage drivers are configured: + +- **`/download`** — retrieves payload data from external storage and decodes it through the Payload Codec. The Web UI calls this when a user clicks to view the full payload behind a reference. +- **`/decode`** — decodes encoded payloads and, by default, retrieves storage references inline. Pass `?preserveStorageRefs=true` to return storage references as-is without retrieval. +- **`/encode`** — applies the Payload Codec, then uploads payloads exceeding the threshold and replaces them with reference tokens. + +**Don't point a Worker's remote codec at the storage-aware handler** — it runs the full encode-store-encode and decode-retrieve-decode pipeline. Run a separate non-storage codec HTTP handler for remote codecs, configured with the same codecs. + +The [Python Codec Server sample](https://github.com/temporalio/samples-python/blob/main/encryption/codec_server.py) demonstrates a Codec Server implementation. + +## Lifecycle management + +Temporal does **not** auto-delete payloads from your store. Configure a TTL on your bucket: + +``` +TTL > Maximum Workflow Run Timeout + Namespace Retention Period +``` + +Example: Run Timeout 14 days + Namespace retention 30 days → set TTL to at least 44 days. + +For Workflows with no finite Run Timeout, there is no safe finite TTL. Use Continue-as-New so the new run uploads fresh payloads and the old run's payloads only need to survive its retention period. + +## Anti-patterns + +- **Don't change the value returned by `name()` after payloads have been stored.** The name is embedded in the claim reference; renaming breaks retrieval of existing claims. +- **Don't use `payload_size_threshold=1` to mean "externalize all"** — use `payload_size_threshold=0`. (This sentinel differs from Go, where `0` is the default and `1` externalizes all.) +- **Don't register multiple drivers without a `driver_selector`.** The selector is required when there is more than one driver. +- **Don't pass the storage-aware payload HTTP handler as a Worker's remote codec target.** Use a separate non-storage codec HTTP handler for that role. +- **Don't omit a TTL on the bucket.** Payloads can be orphaned if a request fails after upload. diff --git a/references/python/python.md b/references/python/python.md index 5493387..0561bd2 100644 --- a/references/python/python.md +++ b/references/python/python.md @@ -179,6 +179,7 @@ See `references/python/testing.md` for info on writing tests. - **`references/python/sync-vs-async.md`** - Sync vs async activities, event loop blocking, executor configuration - **`references/python/advanced-features.md`** - Schedules, worker tuning, and more - **`references/python/data-handling.md`** - Data converters, Pydantic, payload encryption +- **`references/python/external-storage.md`** - Claim-check pattern for large payloads (S3 driver, custom drivers, codec-server handling) - **`references/python/versioning.md`** - Patching API, workflow type versioning, Worker Versioning - **`references/python/determinism-protection.md`** - Python sandbox specifics, forbidden operations, pass-through imports - **`references/python/ai-patterns.md`** - LLM integration, Pydantic data converter, AI workflow patterns From dc0061fe7e1d1725fd517c5ba51f5f418fc5b48d Mon Sep 17 00:00:00 2001 From: Brian Strauch Date: Fri, 21 Aug 2026 14:45:42 -0700 Subject: [PATCH 2/7] Fix Python external storage examples --- references/python/external-storage.md | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/references/python/external-storage.md b/references/python/external-storage.md index 0a3e1b5..569728a 100644 --- a/references/python/external-storage.md +++ b/references/python/external-storage.md @@ -41,9 +41,9 @@ Create the driver, attach it to a `DataConverter`, and pass the converter to bot import aioboto3 import dataclasses from temporalio.client import Client, ClientConfig -from temporalio.contrib.aioboto3 import new_aioboto3_client -from temporalio.converter import DataConverter -from temporalio.external_storage import ExternalStorage, S3StorageDriver +from temporalio.contrib.aws.s3driver import S3StorageDriver +from temporalio.contrib.aws.s3driver.aioboto3 import new_aioboto3_client +from temporalio.converter import DataConverter, ExternalStorage from temporalio.worker import Worker session = aioboto3.Session(profile_name=AWS_PROFILE, region_name=AWS_REGION) @@ -95,7 +95,10 @@ When you register more than one driver, you **must** supply a `driver_selector` - Return `None` from the selector to keep a specific payload inline in Event History. ```python -preferred_driver = S3StorageDriver(client=s3_client, bucket="my-bucket") +preferred_driver = S3StorageDriver( + client=new_aioboto3_client(s3_client), + bucket="my-bucket", +) legacy_driver = LegacyStorageDriver() ExternalStorage( @@ -124,7 +127,7 @@ import uuid from typing import Sequence from temporalio.api.common.v1 import Payload -from temporalio.external_storage import ( +from temporalio.converter import ( StorageDriver, StorageDriverClaim, StorageDriverRetrieveContext, @@ -205,7 +208,7 @@ Endpoints exposed when storage drivers are configured: **Don't point a Worker's remote codec at the storage-aware handler** — it runs the full encode-store-encode and decode-retrieve-decode pipeline. Run a separate non-storage codec HTTP handler for remote codecs, configured with the same codecs. -The [Python Codec Server sample](https://github.com/temporalio/samples-python/blob/main/encryption/codec_server.py) demonstrates a Codec Server implementation. +The [Python External Storage sample](https://github.com/temporalio/samples-python/tree/main/external_storage) demonstrates a storage-aware Codec Server implementation. ## Lifecycle management From d3e0f69db28eca47bdac2af35c0d69a3b89ecbff Mon Sep 17 00:00:00 2001 From: Brian Strauch Date: Mon, 24 Aug 2026 10:50:47 -0700 Subject: [PATCH 3/7] Add TypeScript external storage guidance --- references/go/external-storage.md | 4 +- references/python/external-storage.md | 4 +- references/typescript/external-storage.md | 226 ++++++++++++++++++++++ references/typescript/typescript.md | 1 + 4 files changed, 231 insertions(+), 4 deletions(-) create mode 100644 references/typescript/external-storage.md diff --git a/references/go/external-storage.md b/references/go/external-storage.md index 2d76036..24b6cf2 100644 --- a/references/go/external-storage.md +++ b/references/go/external-storage.md @@ -5,13 +5,13 @@ ## What this is -External Storage offloads Payloads to an external store (e.g. Amazon S3) and stores a small reference token in the Event History instead — the **claim check pattern**. +External Storage uses the **claim check pattern**: it offloads each Payload to an external store (e.g. Amazon S3), records a small reference token (the "claim check") in Event History, and uses that token to retrieve the Payload when needed. The SDK handles storage and retrieval transparently. ## When to use it - A Workflow input, Activity input, Activity result, or Workflow result will exceed the **2 MB** per-payload limit (the limit is fixed at 2 MB on Temporal Cloud; configurable on self-hosted only). - Long Event Histories degrade Workflow Task latency (e.g. AI agent conversations that grow per turn). -- The user wants payload data to live in storage **they** control. Set the size threshold to externalize all payloads. +- The user wants payload data to live in storage **they** control. Set `PayloadSizeThreshold: 1` to externalize all payloads (`0` selects the default 256 KiB threshold in Go). - The user is migrating from self-hosted (with a larger configured limit) to Temporal Cloud. ## Where it sits in the pipeline diff --git a/references/python/external-storage.md b/references/python/external-storage.md index 569728a..5b1de91 100644 --- a/references/python/external-storage.md +++ b/references/python/external-storage.md @@ -5,13 +5,13 @@ ## What this is -External Storage offloads Payloads to an external store (e.g. Amazon S3) and stores a small reference token in the Event History instead — the **claim check pattern**. +External Storage uses the **claim check pattern**: it offloads each Payload to an external store (e.g. Amazon S3), records a small reference token (the "claim check") in Event History, and uses that token to retrieve the Payload when needed. The SDK handles storage and retrieval transparently. ## When to use it - A Workflow input, Activity input, Activity result, or Workflow result will exceed the **2 MB** per-payload limit (fixed at 2 MB on Temporal Cloud; configurable on self-hosted only). - Long Event Histories degrade Workflow Task latency (e.g. AI agent conversations growing per turn). -- The user wants payload data to live in storage **they** control. Set the size threshold to externalize all payloads. +- The user wants payload data to live in storage **they** control. Set `payload_size_threshold=0` to externalize all payloads. - The user is migrating from self-hosted (with a larger configured limit) to Temporal Cloud. ## Where it sits in the pipeline diff --git a/references/typescript/external-storage.md b/references/typescript/external-storage.md new file mode 100644 index 0000000..0b08673 --- /dev/null +++ b/references/typescript/external-storage.md @@ -0,0 +1,226 @@ +# TypeScript SDK External Storage + +> [!NOTE] +> This TypeScript SDK feature is Pre-release. It is acceptable to use on behalf of a user, but inform them that its APIs and configuration may change before General Availability. + +## What this is + +External Storage uses the **claim check pattern**: it offloads each Payload to an external store (e.g. Amazon S3 or Google Cloud Storage), records a small reference token (the "claim check") in Event History, and uses that token to retrieve the Payload when needed. The SDK handles storage and retrieval transparently. + +## When to use it + +- A Workflow input, Activity input, Activity result, or Workflow result will exceed the **2 MB** per-payload limit (fixed at 2 MB on Temporal Cloud; configurable on self-hosted only). +- Long Event Histories degrade Workflow Task latency (e.g. AI agent conversations growing per turn). +- The user wants payload data to live in storage **they** control. Set `payloadSizeThreshold: 0` to externalize all payloads. +- The user is migrating from self-hosted (with a larger configured limit) to Temporal Cloud. + +## Where it sits in the pipeline + +Order: **Payload Converter → Payload Codec → External Storage**. Storage runs last on outbound; it reverses on inbound. + +Consequences: + +- If a Payload Codec encrypts data, the bytes are already encrypted **before** upload. +- The Temporal UI displays the reference token, not the data; the SDK retrieves the payload transparently before handing it to your Workflow or Client. +- Every Client and Worker that might read an offloaded payload needs the same External Storage configuration. + +## Setup with a built-in driver + +The TypeScript SDK provides first-party drivers for Amazon S3 and Google Cloud Storage. Install one driver, its SDK adapter, and the cloud provider's SDK. Keep all `@temporalio/*` packages on the same version. + +Amazon S3: + +```bash +npm install @temporalio/external-storage-s3 \ + @temporalio/external-storage-s3-aws-sdk \ + @aws-sdk/client-s3 +``` + +Google Cloud Storage: + +```bash +npm install @temporalio/external-storage-gcs \ + @temporalio/external-storage-gcs-google-sdk \ + @google-cloud/storage +``` + +### Amazon S3 driver + +```typescript +import { S3Client } from '@aws-sdk/client-s3'; +import { S3StorageDriver } from '@temporalio/external-storage-s3'; +import { AwsSdkS3StorageDriverClient } from '@temporalio/external-storage-s3-aws-sdk'; + +const s3Client = new S3Client({ region: 'us-east-2' }); + +const driver = new S3StorageDriver({ + client: new AwsSdkS3StorageDriverClient(s3Client), + bucket: 'my-temporal-payloads', +}); +``` + +The AWS SDK reads standard credentials from environment variables, an IAM role, or the AWS config file. + +### Google Cloud Storage driver + +```typescript +import { Storage } from '@google-cloud/storage'; +import { GcsStorageDriver } from '@temporalio/external-storage-gcs'; +import { GoogleCloudGcsStorageDriverClient } from '@temporalio/external-storage-gcs-google-sdk'; + +const storage = new Storage(); + +const driver = new GcsStorageDriver({ + client: new GoogleCloudGcsStorageDriverClient(storage), + bucket: 'my-temporal-payloads', +}); +``` + +The Google Cloud SDK reads Application Default Credentials. + +For either driver, `bucket` can be a function instead of a string. The function receives the store context and Payload and returns a bucket name, allowing runtime routing. + +### Configure the Client and Worker + +Create one Data Converter configuration and pass it to both the Client and Worker: + +```typescript +import { Client, Connection } from '@temporalio/client'; +import { ExternalStorage } from '@temporalio/common'; +import { Worker } from '@temporalio/worker'; + +const dataConverter = { + externalStorage: new ExternalStorage({ drivers: [driver] }), +}; + +const connection = await Connection.connect(); +const client = new Client({ connection, dataConverter }); + +const worker = await Worker.create({ + workflowsPath: require.resolve('./workflows'), + taskQueue: 'my-task-queue', + dataConverter, +}); +``` + +External Storage runs outside the Workflow sandbox, so pass the driver object directly. Workflows and Activities use it automatically; business logic does not change. + +## Built-in driver behavior + +The S3 and GCS drivers: + +- Upload and download Payloads concurrently. +- Address objects by a SHA-256 hash of their contents, deduplicating identical Payloads. +- Verify the content hash during retrieval. +- Reject any single Payload larger than `maxPayloadSize`, which defaults to **50 MiB**. +- Include diagnostic metadata in storage errors. + +The External Storage threshold does not override `maxPayloadSize`. Configure the backing store and driver for the largest Payload the application needs to support. + +## Payload size threshold + +- Default: **256 KiB**. +- Set `payloadSizeThreshold: 0` to externalize **all** Payloads regardless of size. +- Payloads whose serialized size is **greater than or equal to** the threshold are eligible for external storage. +- The measured size includes Payload metadata after Payload Converter and Payload Codec processing, not only the raw application value. + +```typescript +const dataConverter = { + externalStorage: new ExternalStorage({ + drivers: [driver], + payloadSizeThreshold: 0, + }), +}; +``` + +## Multiple drivers and migration + +When registering more than one driver, supply a `driverSelector`. The selector chooses which driver stores each Payload. Unselected registered drivers remain available for **retrieval**, which supports migrations without losing access to existing claims. + +- Return `null` from the selector to keep a specific Payload inline in Event History. +- Every registered driver must have a distinct `name`. +- `S3StorageDriver` defaults its name to `"aws.s3driver"`; when registering two S3 drivers, set `driverName` on at least one. + +```typescript +const preferredDriver = new S3StorageDriver({ + client: new AwsSdkS3StorageDriverClient(s3Client), + bucket: 'my-bucket', +}); +const legacyDriver = new LegacyStorageDriver(); + +const externalStorage = new ExternalStorage({ + drivers: [preferredDriver, legacyDriver], + driverSelector: () => preferredDriver, +}); +``` + +Useful routing patterns include driver migration, hot/cold storage tiers, per-tenant storage, and selecting S3 or GCS based on the runtime environment. + +## Custom storage driver + +Implement the `StorageDriver` interface with two readonly properties and two methods: + +- `name: string` — unique identifier for **this driver instance**, stored in the reference so the SDK can route retrieval. Changing it after Payloads are stored **breaks retrieval**. +- `type: string` — stable identifier for the driver implementation, shared by all instances of that implementation and reported in Worker heartbeats (e.g. `"aws.s3driver"`). +- `store(context, payloads): Promise` — serialize and upload each Payload, then return one claim per Payload. Each claim contains string key-value data sufficient to find the object later. +- `retrieve(context, claims): Promise` — download and reconstruct one Payload per claim, preserving input order. + +The `store()` context includes an optional `abortSignal` and `target`. The target is a discriminated union: + +- Check `target.kind` for `"workflow"` or `"activity"`. +- Read `namespace`, `id`, `runId`, and `type` to scope storage keys. + +Honor `abortSignal` in storage calls so sibling operations can be cancelled after the first failure. Content-addressable keys can make retries idempotent and deduplicate identical Payloads. + +Return exactly one claim for each Payload passed to `store()` and exactly one Payload for each claim passed to `retrieve()`. Store the complete serialized Payload protobuf: application data has already passed through the Payload Converter and Payload Codec before reaching the driver. + +## Multi-region durability with Amazon S3 + +For regional-failure tolerance, configure S3 Cross-Region Replication and an S3 Multi-Region Access Point (MRAP), then use the MRAP ARN as `bucket`. + +MRAP requests require a SigV4A signer. The AWS SDK for JavaScript does not bundle one, so install and register it at application startup: + +```bash +npm install @aws-sdk/signature-v4a +``` + +```typescript +import '@aws-sdk/signature-v4a'; +``` + +Then configure the driver with the MRAP ARN: + +```typescript +const driver = new S3StorageDriver({ + client: new AwsSdkS3StorageDriverClient(s3Client), + bucket: 'arn:aws:s3::123456789012:accesspoint/mfzwi23gnjvgw.mrap', +}); +``` + +`@aws-sdk/signature-v4-crt` is an alternative backed by the AWS Common Runtime. The AWS SDK prefers it when both signer implementations are installed. + +Cross-region replication is eventually consistent. Activities reading newly written Payloads from another region need an appropriate Retry Policy. Replication, versioning, and Replication Time Control can add significant cost. + +## Lifecycle and failure handling + +Temporal does **not** automatically delete Payloads from the external store. Configure a bucket lifecycle policy with: + +``` +TTL > Maximum Workflow Run Timeout + Namespace Retention Period +``` + +Example: Run Timeout 14 days + Namespace retention 30 days → set TTL to at least 44 days. + +For Workflows with no finite Run Timeout, there is no safe finite TTL. Use Continue-as-New so the new run uploads fresh Payloads and the old run's Payloads only need to survive its retention period. + +The SDK does not retry a failed `store()` or `retrieve()` call within the same Task attempt. The failure fails the current Workflow Task or Activity Task attempt; Temporal then retries the Task as a whole. Storage operations should therefore be idempotent. + +## Anti-patterns + +- **Don't change a driver's `name` after Payloads have been stored.** The name is embedded in references; changing it breaks retrieval. +- **Don't use `payloadSizeThreshold: 1` to mean "externalize all".** TypeScript uses `0` for that purpose. +- **Don't register duplicate driver names.** Give each instance a unique `name` or `driverName`. +- **Don't register multiple drivers without a `driverSelector`.** Construction fails when more than one driver is registered without one. +- **Don't omit External Storage configuration from a Client or Worker that may retrieve offloaded data.** It cannot resolve the reference without the matching driver. +- **Don't assume the 2 MB Temporal limit is the built-in driver's maximum.** The S3 and GCS drivers default `maxPayloadSize` to 50 MiB. +- **Don't omit a lifecycle policy.** Payloads are otherwise retained indefinitely, and failed requests can leave orphaned objects. diff --git a/references/typescript/typescript.md b/references/typescript/typescript.md index 96fc089..9e11cc9 100644 --- a/references/typescript/typescript.md +++ b/references/typescript/typescript.md @@ -180,5 +180,6 @@ See `references/typescript/testing.md` for info on writing tests. - **`references/typescript/testing.md`** - TestWorkflowEnvironment, time-skipping, activity mocking - **`references/typescript/advanced-features.md`** - Schedules, worker tuning, and more - **`references/typescript/data-handling.md`** - Data converters, payload encryption, etc. +- **`references/typescript/external-storage.md`** - Claim-check pattern for large Payloads (S3/GCS drivers, custom drivers, multi-region durability) - **`references/typescript/versioning.md`** - Patching API, workflow type versioning, Worker Versioning - **`references/typescript/determinism-protection.md`** - V8 sandbox and bundling From 72cd3e630dfac3f27bde9abaef094814c912307a Mon Sep 17 00:00:00 2001 From: Brian Strauch Date: Mon, 24 Aug 2026 11:10:59 -0700 Subject: [PATCH 4/7] Address review findings on external storage references Python: - Import ClientConfig from temporalio.envconfig, not temporalio.client. load_client_connect_config() is a staticmethod on the envconfig class; the temporalio.client.ClientConfig TypedDict has no such member, so the snippet raised AttributeError. Follow main's env-config convention (setdefault target_host) from #261. - Register real Workflow/Activity placeholders. Worker() with empty workflows and activities raises "At least one activity, Nexus service, or workflow must be specified", and wrap the setup in async main(). Go: - Cover the GCS driver (contrib/gcp/gcsdriver + gcssdk), which the SDK ships and the docs install alongside S3. - Load client options with envconfig.MustLoadDefaultClientOptions() and note that Workers inherit External Storage from their Client. Align coverage across all three languages, each of which was missing something the others had: - 50 MiB MaxPayloadSize/max_payload_size ceiling and the matching anti-pattern (Go, Python). - Store/Retrieve are not retried within a Task attempt; the Task retries as a whole, so storage must be idempotent (Go, Python). - Multi-region durability with CRR + an MRAP ARN (Go, Python). - Distinct driver names when registering two drivers of the same kind (Go, Python). - Codec Server guidance (TypeScript), including that neither the TypeScript nor Python SDK ships a storage-aware handler. - Built-in driver behavior sections (concurrency, content-addressed keys, integrity checks, diagnostics) in Go and Python. - ctx.Context on the Go driver contexts, mirroring TypeScript's abortSignal guidance; optional type() override in Python. Also: standardize the TypeScript Public Preview admonition on the repo's wording, drop the transplanted `payloadSizeThreshold: 1` anti-pattern (TypeScript compares >=, so 1 behaves like 0), replace site-relative plugins-guide links with absolute URLs, refresh the index pointers, and revert an unrelated whitespace change in the Spring AI reference. Co-Authored-By: Claude Opus 5 (1M context) --- references/go/external-storage.md | 136 +++++++++++++++++----- references/go/go.md | 2 +- references/java/integrations/spring-ai.md | 1 + references/python/external-storage.md | 127 +++++++++++++------- references/python/python.md | 2 +- references/typescript/external-storage.md | 32 ++++- references/typescript/typescript.md | 2 +- 7 files changed, 226 insertions(+), 76 deletions(-) diff --git a/references/go/external-storage.md b/references/go/external-storage.md index 24b6cf2..956e4ac 100644 --- a/references/go/external-storage.md +++ b/references/go/external-storage.md @@ -5,7 +5,7 @@ ## What this is -External Storage uses the **claim check pattern**: it offloads each Payload to an external store (e.g. Amazon S3), records a small reference token (the "claim check") in Event History, and uses that token to retrieve the Payload when needed. The SDK handles storage and retrieval transparently. +External Storage uses the **claim check pattern**: it offloads each Payload to an external store (e.g. Amazon S3 or Google Cloud Storage), records a small reference token (the "claim check") in Event History, and uses that token to retrieve the Payload when needed. The SDK handles storage and retrieval transparently. ## When to use it @@ -22,14 +22,13 @@ Consequences: - If a Payload Codec encrypts data, the bytes are already encrypted **before** upload to your store. - The Temporal UI shows the reference token, not the data; the SDK transparently retrieves the payload before handing it to your Workflow or Client. +- Every Client and Worker that might read an offloaded payload needs the same External Storage configuration. -## Concurrency +## Setup with a built-in driver -The SDK uploads and downloads payloads **concurrently** within a single Workflow Task — multiple offloaded payloads in one Task are stored/retrieved in parallel, not sequentially. This is automatic; no configuration required. +The Go SDK ships drivers for Amazon S3 and Google Cloud Storage. Only the driver setup differs between the two; everything after that is identical. -## Setup with the built-in S3 driver - -Install dependencies: +Amazon S3: ```bash go get go.temporal.io/sdk/contrib/aws/s3driver \ @@ -38,17 +37,22 @@ go get go.temporal.io/sdk/contrib/aws/s3driver \ github.com/aws/aws-sdk-go-v2/service/s3 ``` -Create the driver and Client: +Google Cloud Storage: + +```bash +go get go.temporal.io/sdk/contrib/gcp/gcsdriver \ + go.temporal.io/sdk/contrib/gcp/gcsdriver/gcssdk \ + cloud.google.com/go/storage +``` + +### Amazon S3 driver ```go import ( "github.com/aws/aws-sdk-go-v2/config" "github.com/aws/aws-sdk-go-v2/service/s3" - "go.temporal.io/sdk/client" "go.temporal.io/sdk/contrib/aws/s3driver" "go.temporal.io/sdk/contrib/aws/s3driver/awssdkv2" - "go.temporal.io/sdk/converter" - "go.temporal.io/sdk/worker" ) cfg, err := config.LoadDefaultConfig(context.Background(), @@ -65,13 +69,53 @@ driver, err := s3driver.NewDriver(s3driver.Options{ if err != nil { log.Fatalf("create S3 driver: %v", err) } +``` + +The AWS SDK reads standard credentials from the environment (env vars, IAM role, or AWS config file). + +### Google Cloud Storage driver -c, err := client.Dial(client.Options{ - HostPort: "localhost:7233", - ExternalStorage: converter.ExternalStorage{ - Drivers: []converter.StorageDriver{driver}, - }, +```go +import ( + "cloud.google.com/go/storage" + "go.temporal.io/sdk/contrib/gcp/gcsdriver" + "go.temporal.io/sdk/contrib/gcp/gcsdriver/gcssdk" +) + +gcsClient, err := storage.NewClient(context.Background()) +if err != nil { + log.Fatalf("create GCS client: %v", err) +} + +driver, err := gcsdriver.NewDriver(gcsdriver.Options{ + Client: gcssdk.NewClient(gcsClient), + Bucket: gcsdriver.StaticBucket("my-temporal-payloads"), }) +if err != nil { + log.Fatalf("create GCS driver: %v", err) +} +``` + +The Google Cloud SDK reads Application Default Credentials. + +For either driver, pass a `BucketFunc` as `Bucket` instead of `StaticBucket` to route payloads at runtime. The function receives the store context and the payload and returns a bucket name. + +### Configure the Client and Worker + +```go +import ( + "go.temporal.io/sdk/client" + "go.temporal.io/sdk/contrib/envconfig" + "go.temporal.io/sdk/converter" + "go.temporal.io/sdk/worker" +) + +opts := envconfig.MustLoadDefaultClientOptions() +opts.ExternalStorage = converter.ExternalStorage{ + Drivers: []converter.StorageDriver{driver}, +} + +c, err := client.Dial(opts) if err != nil { log.Fatalf("connect to Temporal: %v", err) } @@ -80,23 +124,34 @@ defer c.Close() w := worker.New(c, "my-task-queue", worker.Options{}) ``` -The S3 driver uses standard AWS credentials from the environment (env vars, IAM role, or AWS config file). +A Worker inherits External Storage from the Client it is created with. When your Workers run in their own process, repeat this setup there — a Client or Worker without the matching driver cannot resolve a reference. Workflows and Activities running on the Worker use the driver automatically — no changes to business logic. +## Built-in driver behavior + +Both the S3 and GCS drivers: + +- Upload and download payloads **concurrently**. Multiple offloaded payloads in a single Workflow Task are stored or retrieved in parallel, not sequentially. +- Address objects by a SHA-256 hash of the contents, scoped by Namespace, Workflow ID, and Run ID, and verify that hash on retrieval. One Run passing the same payload to several Activities uploads it once; a different Run, Workflow, or Namespace stores its own copy, so storage scales with the number of Runs rather than with how often a Run passes a payload around. +- Reject any single payload larger than `MaxPayloadSize`, which defaults to **50 MiB**. `PayloadSizeThreshold` does not raise this ceiling — set `MaxPayloadSize` for the largest payload the application must support, and size the backing store to match. +- Include diagnostic metadata, such as the AWS region, in storage errors. + ## Payload size threshold - Default: **256 KiB**. - Set `PayloadSizeThreshold: 1` to externalize **all** payloads regardless of size. - `PayloadSizeThreshold: 0` is **interpreted as the default (256 KiB)** — it does **not** mean "externalize everything". +- The size compared against the threshold is that of the serialized Payload, including its metadata, not just your data. ```go -c, err := client.Dial(client.Options{ - ExternalStorage: converter.ExternalStorage{ - Drivers: []converter.StorageDriver{driver}, - PayloadSizeThreshold: 1, - }, -}) +opts := envconfig.MustLoadDefaultClientOptions() +opts.ExternalStorage = converter.ExternalStorage{ + Drivers: []converter.StorageDriver{driver}, + PayloadSizeThreshold: 1, +} + +c, err := client.Dial(opts) ``` ## Multiple drivers and migration @@ -104,6 +159,7 @@ c, err := client.Dial(client.Options{ When you register more than one driver, you **must** supply a `DriverSelector` implementing `StorageDriverSelector`. The selector chooses which driver stores each payload. Unselected drivers remain available for **retrieval** — this is how you migrate between storage backends without losing access to existing claims. - Return `nil` from the selector to keep a specific payload inline in Event History. +- Every registered driver must have a distinct `Name()`; duplicates are rejected when the Client or Worker is constructed. `s3driver` defaults its name to `"aws.s3driver"` and `gcsdriver` to `"gcp.gcsdriver"`, so registering two drivers of the same kind requires setting `DriverName` on at least one. ```go type PreferredSelector struct { @@ -125,18 +181,22 @@ func MultipleDriversSetup(preferredDriver, legacyDriver converter.StorageDriver) } ``` +Useful routing patterns include driver migration, hot/cold storage tiers, per-tenant storage, and selecting S3 or GCS based on the runtime environment. + ## Custom storage driver Implement `converter.StorageDriver` with **four** methods: - `Name() string` — unique identifier for **this driver instance**, stored in the claim reference so the SDK can route retrieval. Renaming after payloads are stored **breaks retrieval**. -- `Type() string` — identifier for the driver **implementation**, same across all instances regardless of configuration (e.g. `"aws.s3driver"`, `"local-disk"`). -- `Store(ctx, payloads) ([]StorageDriverClaim, error)` — upload each Payload protobuf and return one claim per payload. A claim is a `map[string]string` the driver uses to locate the payload later. -- `Retrieve(ctx, claims) ([]*commonpb.Payload, error)` — download bytes using claim data and reconstruct each Payload. +- `Type() string` — identifier for the driver **implementation**, same across all instances regardless of configuration (e.g. `"aws.s3driver"`, `"local-disk"`). It is reported in Worker heartbeats. +- `Store(ctx, payloads) ([]StorageDriverClaim, error)` — upload each Payload protobuf and return one claim per payload, in the same order. A claim is a `map[string]string` the driver uses to locate the payload later. +- `Retrieve(ctx, claims) ([]*commonpb.Payload, error)` — download bytes using claim data and reconstruct each Payload, one per claim, in the same order. Inside `Store()`, marshal each payload with `proto.Marshal(payload)`; in `Retrieve()`, reconstruct with `proto.Unmarshal(data, payload)`. The application data has already been serialized by the Payload Converter and Payload Codec before it reaches the driver. -`ctx.Target` provides identity information. Type-switch over `StorageDriverWorkflowInfo` and `StorageDriverActivityInfo` to access the namespace / Workflow ID / Activity ID. `StorageDriverActivityInfo` is only used for standalone (non-workflow-bound) Activities; Activities started by a Workflow get `StorageDriverWorkflowInfo`. +`ctx.Context` carries the context of the operation that triggered the driver call — pass it to your storage calls so cancellation and deadlines propagate, and so sibling operations stop after the first failure. + +`ctx.Target` provides identity information. Type-switch over `StorageDriverWorkflowInfo` and `StorageDriverActivityInfo` to access the namespace / Workflow ID / Activity ID, and use it to scope storage keys. `StorageDriverActivityInfo` is only used for standalone (non-workflow-bound) Activities; Activities started by a Workflow get `StorageDriverWorkflowInfo`. Worked example — local-disk driver (development/testing only): @@ -210,7 +270,22 @@ func (d *LocalDiskStorageDriver) Retrieve( } ``` -You can package a custom driver as a [plugin](/develop/plugins-guide) for reuse across services. +You can package a custom driver as a [plugin](https://docs.temporal.io/develop/plugins-guide) for reuse across services. + +## Multi-region durability with Amazon S3 + +For regional-failure tolerance, configure S3 Cross-Region Replication (CRR) and an S3 Multi-Region Access Point (MRAP), then pass the MRAP ARN as the bucket: + +```go +driver, err := s3driver.NewDriver(s3driver.Options{ + Client: awssdkv2.NewClient(s3.NewFromConfig(cfg)), + Bucket: s3driver.StaticBucket("arn:aws:s3::123456789012:accesspoint/mfzwi23gnjvgw.mrap"), +}) +``` + +The AWS SDK for Go v2 uses SigV4A signing automatically when the bucket value is an MRAP ARN, so no additional client configuration is required. + +Cross-region replication is eventually consistent. Activities reading newly written Payloads from another region need an appropriate Retry Policy. Replication, versioning, and Replication Time Control can add significant cost. ## Codec Server with External Storage @@ -226,7 +301,7 @@ When configured with storage drivers, the handler exposes: **Don't use `NewPayloadHTTPHandler` as a remote Data Converter or remote codec target for your Workers** — it runs the full encode-store-encode and decode-retrieve-decode pipeline. For remote codecs use `NewPayloadCodecHTTPHandler` separately. If you need both, run both handlers, configured with the same codecs. -## Lifecycle management +## Lifecycle and failure handling Temporal does **not** auto-delete payloads from your store. Configure a TTL on your bucket: @@ -238,10 +313,15 @@ Example: Run Timeout 14 days + Namespace retention 30 days → set TTL to at lea For Workflows with no finite Run Timeout, there is no safe finite TTL. Use Continue-as-New so the new run uploads fresh payloads and the old run's payloads only need to survive its retention period. +The SDK does not retry a failed `Store` or `Retrieve` call within the same Task attempt. The failure fails the current Workflow Task or Activity Task attempt; Temporal then retries the Task as a whole, and the new attempt retries the storage operation along with it. For Activities, the Retry Policy controls the timing. Storage operations should therefore be idempotent — content-addressable keys are one way to get that. + ## Anti-patterns - **Don't change `Name()` after payloads have been stored.** The name is embedded in the claim reference; renaming breaks retrieval of existing claims. - **Don't use `PayloadSizeThreshold: 0` to mean "externalize all".** `0` is interpreted as the default (256 KiB). Use `PayloadSizeThreshold: 1`. - **Don't register multiple drivers without a `DriverSelector`.** The selector is required when there are multiple drivers. +- **Don't register duplicate driver names.** Two same-kind drivers share a default name; set `DriverName` on at least one. +- **Don't omit External Storage configuration from a Client or Worker that may retrieve offloaded data.** It cannot resolve the reference without the matching driver. +- **Don't assume the 2 MB Temporal limit is the driver's maximum.** The S3 and GCS drivers reject payloads above `MaxPayloadSize`, which defaults to 50 MiB. - **Don't point a Worker's remote codec at `NewPayloadHTTPHandler`.** Use `NewPayloadCodecHTTPHandler` for remote codec endpoints. - **Don't omit a TTL on the bucket.** Payloads are orphaned otherwise; orphaned objects can also remain if a request fails after upload. diff --git a/references/go/go.md b/references/go/go.md index 903e992..8882868 100644 --- a/references/go/go.md +++ b/references/go/go.md @@ -250,6 +250,6 @@ See `references/go/testing.md` for info on writing tests. - **`references/go/testing.md`** - TestWorkflowEnvironment, time-skipping, activity mocking - **`references/go/advanced-features.md`** - Schedules, worker tuning, and more - **`references/go/data-handling.md`** - Data converters, payload codecs, encryption -- **`references/go/external-storage.md`** - Claim-check pattern for large payloads (S3 driver, custom drivers, codec-server handling) +- **`references/go/external-storage.md`** - Claim-check pattern for large payloads (S3 and GCS drivers, custom drivers, codec-server handling, multi-region durability) - **`references/go/versioning.md`** - Patching API (`workflow.GetVersion`), Worker Versioning - **`references/go/determinism-protection.md`** - Information on **`workflowcheck`** tool to help statically check for determinism issues. diff --git a/references/java/integrations/spring-ai.md b/references/java/integrations/spring-ai.md index ae5154f..5ee0704 100644 --- a/references/java/integrations/spring-ai.md +++ b/references/java/integrations/spring-ai.md @@ -217,6 +217,7 @@ Media image = new Media(MimeTypeUtils.IMAGE_PNG, URI.create("https://cdn.example For anything larger than a small thumbnail, route the bytes to a binary store from an Activity and pass only the URL across the conversation. + ## Vector stores, embeddings, and MCP When the corresponding Spring AI modules (`spring-ai-rag`, `spring-ai-mcp`) are on the classpath, the integration registers Activities for vector stores, embeddings, and MCP tool calls automatically. Inject the matching Spring AI types into your Activities or Workflows and use them as you would in any Spring AI application — each operation executes through a Temporal Activity. diff --git a/references/python/external-storage.md b/references/python/external-storage.md index 5b1de91..2eb12a2 100644 --- a/references/python/external-storage.md +++ b/references/python/external-storage.md @@ -22,14 +22,11 @@ Consequences: - If a Payload Codec encrypts data, the bytes are already encrypted **before** upload. - The Temporal UI displays the reference token, not the data; the SDK retrieves the payload transparently before handing it to your Workflow or Client. - -## Concurrency - -The SDK uploads and downloads payloads **concurrently** within a single Workflow Task — multiple offloaded payloads in one Task are stored or retrieved in parallel, not sequentially. This is automatic. +- Every Client and Worker that might read an offloaded payload needs the same External Storage configuration. ## Setup with the built-in S3 driver -Install the `aioboto3` extra: +The Python SDK ships an Amazon S3 driver (there is no built-in GCS driver — use a custom driver for other backends). Install the `aioboto3` extra: ```bash python -m pip install "temporalio[aioboto3]" @@ -38,45 +35,69 @@ python -m pip install "temporalio[aioboto3]" Create the driver, attach it to a `DataConverter`, and pass the converter to both Client and Worker: ```python -import aioboto3 +import asyncio import dataclasses -from temporalio.client import Client, ClientConfig + +import aioboto3 +from temporalio.client import Client from temporalio.contrib.aws.s3driver import S3StorageDriver from temporalio.contrib.aws.s3driver.aioboto3 import new_aioboto3_client from temporalio.converter import DataConverter, ExternalStorage +from temporalio.envconfig import ClientConfig from temporalio.worker import Worker -session = aioboto3.Session(profile_name=AWS_PROFILE, region_name=AWS_REGION) -async with session.client("s3") as s3_client: - driver = S3StorageDriver( - client=new_aioboto3_client(s3_client), - bucket="my-temporal-payloads", - ) - - data_converter = dataclasses.replace( - DataConverter.default, - external_storage=ExternalStorage(drivers=[driver]), - ) - - client_config = ClientConfig.load_client_connect_config() - client = await Client.connect(**client_config, data_converter=data_converter) - - worker = Worker( - client, - task_queue="my-task-queue", - workflows=[], - activities=[], - ) +from activities.greet import greet +from workflows.greeting import GreetingWorkflow + + +async def main() -> None: + session = aioboto3.Session(region_name="us-east-2") + async with session.client("s3") as s3_client: + driver = S3StorageDriver( + client=new_aioboto3_client(s3_client), + bucket="my-temporal-payloads", + ) + + data_converter = dataclasses.replace( + DataConverter.default, + external_storage=ExternalStorage(drivers=[driver]), + ) + + connect_config = ClientConfig.load_client_connect_config() + connect_config.setdefault("target_host", "localhost:7233") + client = await Client.connect(**connect_config, data_converter=data_converter) + + worker = Worker( + client, + task_queue="my-task-queue", + workflows=[GreetingWorkflow], + activities=[greet], + ) + await worker.run() + + +if __name__ == "__main__": + asyncio.run(main()) ``` -The S3 driver uses standard AWS credentials from the environment (env vars, IAM role, or AWS config file). +`ClientConfig` for connection settings comes from `temporalio.envconfig`, not `temporalio.client`. The S3 driver uses standard AWS credentials from the environment (env vars, IAM role, or AWS config file); pass `profile_name=` to `aioboto3.Session` to select a named profile. Keep the `async with session.client("s3")` block open for as long as the Worker runs — the driver uses that client for every upload and download. Workflows and Activities on the Worker use the driver automatically — no business-logic changes. +## Built-in driver behavior + +The S3 driver: + +- Uploads and downloads payloads **concurrently**. Multiple offloaded payloads in a single Workflow Task are stored or retrieved in parallel, not sequentially. +- Addresses objects by a SHA-256 hash of their contents, segmented by Namespace and Workflow/Activity identifiers, and validates payload integrity on retrieval. +- Rejects any single payload larger than `max_payload_size`, which defaults to **50 MiB**. `payload_size_threshold` does not raise this ceiling — set `max_payload_size` for the largest payload the application must support, and size the backing store to match. +- Includes diagnostic metadata, such as the AWS region, in error messages. + ## Payload size threshold - Default: **256 KiB**. - Set `payload_size_threshold=0` to externalize **all** payloads regardless of size. +- Payloads whose serialized size is **greater than or equal to** the threshold are eligible; smaller ones stay inline. The measured size includes Payload metadata, not just your data. ```python data_converter = dataclasses.replace( @@ -93,11 +114,13 @@ data_converter = dataclasses.replace( When you register more than one driver, you **must** supply a `driver_selector` function. The selector chooses which driver stores each payload. Unselected drivers remain available for **retrieval** — this is how you migrate between storage backends without losing access to existing claims. - Return `None` from the selector to keep a specific payload inline in Event History. +- Every registered driver must have a distinct name; duplicates raise `ValueError` at construction. `S3StorageDriver` defaults its name to `"aws.s3driver"`, so registering two S3 drivers requires passing `driver_name=` to at least one. ```python preferred_driver = S3StorageDriver( client=new_aioboto3_client(s3_client), bucket="my-bucket", + driver_name="s3-primary", ) legacy_driver = LegacyStorageDriver() @@ -107,17 +130,21 @@ ExternalStorage( ) ``` +Useful routing patterns include driver migration, hot/cold storage tiers, and per-tenant storage. + ## Custom storage driver Extend `StorageDriver` and implement **three** methods: - `name() -> str` — unique identifier for the driver, stored in the claim reference so the SDK can route retrieval. Renaming after payloads are stored **breaks retrieval**. -- `async store(context, payloads) -> list[StorageDriverClaim]` — upload each Payload and return one claim per payload. A claim is a `dict[str, str]` the driver uses to locate the payload later. -- `async retrieve(context, claims) -> list[Payload]` — download bytes using claim data and reconstruct each Payload. +- `async store(context, payloads) -> list[StorageDriverClaim]` — upload each Payload and return one claim per payload, in the same order. A claim is a `dict[str, str]` the driver uses to locate the payload later. +- `async retrieve(context, claims) -> list[Payload]` — download bytes using claim data and reconstruct each Payload, one per claim, in the same order. + +`type() -> str` is optional and defaults to the class name. Override it with a stable identifier shared by every instance of the implementation (e.g. `"aws.s3driver"`) so the driver reports the same type as its equivalents in other languages. Inside `store()`, serialize each payload with `payload.SerializeToString()`; in `retrieve()`, reconstruct with `payload.ParseFromString(data)`. The application data has already been serialized by the Payload Converter and Payload Codec before reaching the driver. -`context.target` provides identity information (namespace, Workflow ID, or Activity ID). Check the target type with `isinstance(target, StorageDriverWorkflowInfo)`; the Workflow info exposes `target.namespace` and `target.id`. Use this to scope storage keys per Workflow. +`context.target` provides identity information (namespace, Workflow ID, or Activity ID). Check the target type with `isinstance(target, StorageDriverWorkflowInfo)`; the Workflow info exposes `target.namespace` and `target.id`. Use this to scope storage keys per Workflow. Within that scope, content-addressable keys (such as a SHA-256 hash of the payload bytes) deduplicate identical payloads and make retries idempotent. Worked example — local-disk driver (development/testing only): @@ -143,6 +170,9 @@ class LocalDiskStorageDriver(StorageDriver): def name(self) -> str: return "local-disk" + def type(self) -> str: + return "local-disk" + async def store( self, context: StorageDriverStoreContext, @@ -192,25 +222,38 @@ data_converter = dataclasses.replace( ) ``` -You can package a custom driver as a [plugin](/develop/plugins-guide) for reuse across services. +You can package a custom driver as a [plugin](https://docs.temporal.io/develop/plugins-guide) for reuse across services. + +## Multi-region durability with Amazon S3 + +For regional-failure tolerance, configure S3 Cross-Region Replication (CRR) and an S3 Multi-Region Access Point (MRAP), then pass the MRAP ARN as `bucket`: + +```python +driver = S3StorageDriver( + client=new_aioboto3_client(s3_client), + bucket="arn:aws:s3::123456789012:accesspoint/mfzwi23gnjvgw.mrap", +) +``` + +`aioboto3` (via `botocore`) uses SigV4A signing automatically when the bucket value is an MRAP ARN. Make sure `botocore` is recent enough to support SigV4A. + +Cross-region replication is eventually consistent. Activities reading newly written payloads from another region need an appropriate Retry Policy. Replication, versioning, and Replication Time Control can add significant cost. ## Codec Server with External Storage When Workers and Clients use External Storage, Event History contains reference tokens — not payload data. For the Web UI and CLI to show decoded payloads, the Codec Server must download from external storage **and** decode through the Payload Codec in the correct order. -Build the Codec Server with a payload HTTP handler that accepts your storage drivers, your pre-storage codecs (the Payload Codecs your Workers use), and any post-storage codecs (applied by a proxy after external storage). The handler applies them in the correct order across all endpoints. +The Python SDK does not ship a storage-aware Codec Server handler — implement the routes yourself (e.g. with `aiohttp`), giving them your storage drivers, your pre-storage codecs (the Payload Codecs your Workers use), and any post-storage codecs (applied by a proxy after external storage). The [Python External Storage sample](https://github.com/temporalio/samples-python/tree/main/external_storage) has a working implementation (`payload_routes` in `handler.py`) to copy from. -Endpoints exposed when storage drivers are configured: +Endpoints to expose when storage drivers are configured: - **`/download`** — retrieves payload data from external storage and decodes it through the Payload Codec. The Web UI calls this when a user clicks to view the full payload behind a reference. -- **`/decode`** — decodes encoded payloads and, by default, retrieves storage references inline. Pass `?preserveStorageRefs=true` to return storage references as-is without retrieval. +- **`/decode`** — decodes encoded payloads and, by default, retrieves storage references inline. Support `?preserveStorageRefs=true` to return storage references as-is without retrieval; the Web UI uses it to render history without downloading every blob. - **`/encode`** — applies the Payload Codec, then uploads payloads exceeding the threshold and replaces them with reference tokens. **Don't point a Worker's remote codec at the storage-aware handler** — it runs the full encode-store-encode and decode-retrieve-decode pipeline. Run a separate non-storage codec HTTP handler for remote codecs, configured with the same codecs. -The [Python External Storage sample](https://github.com/temporalio/samples-python/tree/main/external_storage) demonstrates a storage-aware Codec Server implementation. - -## Lifecycle management +## Lifecycle and failure handling Temporal does **not** auto-delete payloads from your store. Configure a TTL on your bucket: @@ -222,10 +265,16 @@ Example: Run Timeout 14 days + Namespace retention 30 days → set TTL to at lea For Workflows with no finite Run Timeout, there is no safe finite TTL. Use Continue-as-New so the new run uploads fresh payloads and the old run's payloads only need to survive its retention period. +The SDK does not retry a failed `store()` or `retrieve()` call within the same Task attempt. The failure fails the current Workflow Task or Activity Task attempt; Temporal then retries the Task as a whole, and the new attempt retries the storage operation along with it. For Activities, the Retry Policy controls the timing. Storage operations should therefore be idempotent — content-addressable keys are one way to get that. + ## Anti-patterns - **Don't change the value returned by `name()` after payloads have been stored.** The name is embedded in the claim reference; renaming breaks retrieval of existing claims. - **Don't use `payload_size_threshold=1` to mean "externalize all"** — use `payload_size_threshold=0`. (This sentinel differs from Go, where `0` is the default and `1` externalizes all.) - **Don't register multiple drivers without a `driver_selector`.** The selector is required when there is more than one driver. +- **Don't register duplicate driver names.** Two `S3StorageDriver` instances share a default name; pass `driver_name=` to at least one. +- **Don't omit External Storage configuration from a Client or Worker that may retrieve offloaded data.** It cannot resolve the reference without the matching driver. +- **Don't assume the 2 MB Temporal limit is the driver's maximum.** The S3 driver rejects payloads above `max_payload_size`, which defaults to 50 MiB. +- **Don't import `ClientConfig` from `temporalio.client` for connection settings.** `load_client_connect_config()` lives on `temporalio.envconfig.ClientConfig`. - **Don't pass the storage-aware payload HTTP handler as a Worker's remote codec target.** Use a separate non-storage codec HTTP handler for that role. - **Don't omit a TTL on the bucket.** Payloads can be orphaned if a request fails after upload. diff --git a/references/python/python.md b/references/python/python.md index 0561bd2..4b8afe8 100644 --- a/references/python/python.md +++ b/references/python/python.md @@ -179,7 +179,7 @@ See `references/python/testing.md` for info on writing tests. - **`references/python/sync-vs-async.md`** - Sync vs async activities, event loop blocking, executor configuration - **`references/python/advanced-features.md`** - Schedules, worker tuning, and more - **`references/python/data-handling.md`** - Data converters, Pydantic, payload encryption -- **`references/python/external-storage.md`** - Claim-check pattern for large payloads (S3 driver, custom drivers, codec-server handling) +- **`references/python/external-storage.md`** - Claim-check pattern for large payloads (S3 driver, custom drivers, codec-server handling, multi-region durability) - **`references/python/versioning.md`** - Patching API, workflow type versioning, Worker Versioning - **`references/python/determinism-protection.md`** - Python sandbox specifics, forbidden operations, pass-through imports - **`references/python/ai-patterns.md`** - LLM integration, Pydantic data converter, AI workflow patterns diff --git a/references/typescript/external-storage.md b/references/typescript/external-storage.md index 0b08673..52b8fe4 100644 --- a/references/typescript/external-storage.md +++ b/references/typescript/external-storage.md @@ -1,7 +1,7 @@ # TypeScript SDK External Storage > [!NOTE] -> This TypeScript SDK feature is Pre-release. It is acceptable to use on behalf of a user, but inform them that its APIs and configuration may change before General Availability. +> This feature is in Public Preview. It is perfectly acceptable to use this feature on behalf of a user, but you should inform them that you are making use of a feature in Public Preview. ## What this is @@ -33,6 +33,7 @@ Amazon S3: ```bash npm install @temporalio/external-storage-s3 \ @temporalio/external-storage-s3-aws-sdk \ + @temporalio/envconfig \ @aws-sdk/client-s3 ``` @@ -41,6 +42,7 @@ Google Cloud Storage: ```bash npm install @temporalio/external-storage-gcs \ @temporalio/external-storage-gcs-google-sdk \ + @temporalio/envconfig \ @google-cloud/storage ``` @@ -82,21 +84,27 @@ For either driver, `bucket` can be a function instead of a string. The function ### Configure the Client and Worker -Create one Data Converter configuration and pass it to both the Client and Worker: +Create one Data Converter configuration and pass it to both the Client and Worker. Load connection settings with `loadClientConnectConfig()`, and remember that `NativeConnection` carries no namespace, so the Worker needs `namespace` passed explicitly: ```typescript import { Client, Connection } from '@temporalio/client'; import { ExternalStorage } from '@temporalio/common'; -import { Worker } from '@temporalio/worker'; +import { loadClientConnectConfig } from '@temporalio/envconfig'; +import { NativeConnection, Worker } from '@temporalio/worker'; const dataConverter = { externalStorage: new ExternalStorage({ drivers: [driver] }), }; -const connection = await Connection.connect(); -const client = new Client({ connection, dataConverter }); +const config = loadClientConnectConfig(); +const connection = await Connection.connect(config.connectionOptions); +const client = new Client({ connection, namespace: config.namespace, dataConverter }); + +const workerConnection = await NativeConnection.connect(config.connectionOptions); const worker = await Worker.create({ + connection: workerConnection, + namespace: config.namespace, workflowsPath: require.resolve('./workflows'), taskQueue: 'my-task-queue', dataConverter, @@ -201,6 +209,18 @@ const driver = new S3StorageDriver({ Cross-region replication is eventually consistent. Activities reading newly written Payloads from another region need an appropriate Retry Policy. Replication, versioning, and Replication Time Control can add significant cost. +## Codec Server with External Storage + +When Workers and Clients use External Storage, Event History contains reference tokens — not payload data. A plain codec server that only implements `/encode` and `/decode` leaves the Web UI and CLI showing raw reference tokens. + +The TypeScript SDK does not ship a codec-server handler, so implement the routes yourself (e.g. with Express), wiring in your storage drivers, your pre-storage codecs (the Payload Codecs your Workers use), and any post-storage codecs (applied by a proxy after external storage): + +- **`/download`** — retrieves payload data from external storage and decodes it through the Payload Codec. The Web UI calls this when a user clicks to view the full payload behind a reference. +- **`/decode`** — decodes encoded payloads and, by default, retrieves storage references inline. Support `?preserveStorageRefs=true` to return storage references as-is without retrieval; the Web UI uses it to render history without downloading every blob. +- **`/encode`** — applies the Payload Codec, then uploads payloads exceeding the threshold and replaces them with reference tokens. + +**Don't point a Worker's remote codec at the storage-aware handler** — it runs the full encode-store-encode and decode-retrieve-decode pipeline. Serve remote codecs from a separate non-storage endpoint, configured with the same codecs. + ## Lifecycle and failure handling Temporal does **not** automatically delete Payloads from the external store. Configure a bucket lifecycle policy with: @@ -218,9 +238,9 @@ The SDK does not retry a failed `store()` or `retrieve()` call within the same T ## Anti-patterns - **Don't change a driver's `name` after Payloads have been stored.** The name is embedded in references; changing it breaks retrieval. -- **Don't use `payloadSizeThreshold: 1` to mean "externalize all".** TypeScript uses `0` for that purpose. - **Don't register duplicate driver names.** Give each instance a unique `name` or `driverName`. - **Don't register multiple drivers without a `driverSelector`.** Construction fails when more than one driver is registered without one. - **Don't omit External Storage configuration from a Client or Worker that may retrieve offloaded data.** It cannot resolve the reference without the matching driver. - **Don't assume the 2 MB Temporal limit is the built-in driver's maximum.** The S3 and GCS drivers default `maxPayloadSize` to 50 MiB. +- **Don't point a Worker's remote codec at a storage-aware codec-server handler.** Serve remote codecs from a separate non-storage endpoint. - **Don't omit a lifecycle policy.** Payloads are otherwise retained indefinitely, and failed requests can leave orphaned objects. diff --git a/references/typescript/typescript.md b/references/typescript/typescript.md index 9e11cc9..318068c 100644 --- a/references/typescript/typescript.md +++ b/references/typescript/typescript.md @@ -180,6 +180,6 @@ See `references/typescript/testing.md` for info on writing tests. - **`references/typescript/testing.md`** - TestWorkflowEnvironment, time-skipping, activity mocking - **`references/typescript/advanced-features.md`** - Schedules, worker tuning, and more - **`references/typescript/data-handling.md`** - Data converters, payload encryption, etc. -- **`references/typescript/external-storage.md`** - Claim-check pattern for large Payloads (S3/GCS drivers, custom drivers, multi-region durability) +- **`references/typescript/external-storage.md`** - Claim-check pattern for large Payloads (S3 and GCS drivers, custom drivers, codec-server handling, multi-region durability) - **`references/typescript/versioning.md`** - Patching API, workflow type versioning, Worker Versioning - **`references/typescript/determinism-protection.md`** - V8 sandbox and bundling From e4793f9f6ecf0e8695d38ae874c232febf42d0e2 Mon Sep 17 00:00:00 2001 From: Brian Strauch Date: Mon, 24 Aug 2026 11:23:39 -0700 Subject: [PATCH 5/7] Harden external storage driver examples --- references/go/external-storage.md | 24 +++++++++++++++++++----- references/python/external-storage.md | 19 ++++++++++++++----- 2 files changed, 33 insertions(+), 10 deletions(-) diff --git a/references/go/external-storage.md b/references/go/external-storage.md index 956e4ac..ec6de97 100644 --- a/references/go/external-storage.md +++ b/references/go/external-storage.md @@ -196,7 +196,7 @@ Inside `Store()`, marshal each payload with `proto.Marshal(payload)`; in `Retrie `ctx.Context` carries the context of the operation that triggered the driver call — pass it to your storage calls so cancellation and deadlines propagate, and so sibling operations stop after the first failure. -`ctx.Target` provides identity information. Type-switch over `StorageDriverWorkflowInfo` and `StorageDriverActivityInfo` to access the namespace / Workflow ID / Activity ID, and use it to scope storage keys. `StorageDriverActivityInfo` is only used for standalone (non-workflow-bound) Activities; Activities started by a Workflow get `StorageDriverWorkflowInfo`. +`ctx.Target` provides identity information. Type-switch over `StorageDriverWorkflowInfo` and `StorageDriverActivityInfo` to access the namespace / Workflow ID / Activity ID, and use it to scope storage keys. Hash or encode identifiers before using them as path segments because identifiers can contain path separators or traversal sequences. `StorageDriverActivityInfo` is only used for standalone (non-workflow-bound) Activities; Activities started by a Workflow get `StorageDriverWorkflowInfo`. Worked example — local-disk driver (development/testing only): @@ -205,6 +205,11 @@ type LocalDiskStorageDriver struct { storeDir string } +func safePathSegment(value string) string { + sum := sha256.Sum256([]byte(value)) + return hex.EncodeToString(sum[:]) +} + func NewLocalDiskStorageDriver(storeDir string) converter.StorageDriver { return &LocalDiskStorageDriver{storeDir: storeDir} } @@ -220,11 +225,19 @@ func (d *LocalDiskStorageDriver) Store( switch info := ctx.Target.(type) { case converter.StorageDriverWorkflowInfo: if info.WorkflowID != "" { - dir = filepath.Join(d.storeDir, info.Namespace, info.WorkflowID) + dir = filepath.Join( + d.storeDir, + safePathSegment(info.Namespace), + safePathSegment(info.WorkflowID), + ) } case converter.StorageDriverActivityInfo: if info.ActivityID != "" { - dir = filepath.Join(d.storeDir, info.Namespace, info.ActivityID) + dir = filepath.Join( + d.storeDir, + safePathSegment(info.Namespace), + safePathSegment(info.ActivityID), + ) } } if err := os.MkdirAll(dir, 0o755); err != nil { @@ -233,12 +246,13 @@ func (d *LocalDiskStorageDriver) Store( claims := make([]converter.StorageDriverClaim, len(payloads)) for i, payload := range payloads { - key := uuid.NewString() + ".bin" - filePath := filepath.Join(dir, key) data, err := proto.Marshal(payload) if err != nil { return nil, fmt.Errorf("marshal payload: %w", err) } + sum := sha256.Sum256(data) + key := hex.EncodeToString(sum[:]) + ".bin" + filePath := filepath.Join(dir, key) if err := os.WriteFile(filePath, data, 0o644); err != nil { return nil, fmt.Errorf("write payload: %w", err) } diff --git a/references/python/external-storage.md b/references/python/external-storage.md index 2eb12a2..5fd2f10 100644 --- a/references/python/external-storage.md +++ b/references/python/external-storage.md @@ -144,13 +144,13 @@ Extend `StorageDriver` and implement **three** methods: Inside `store()`, serialize each payload with `payload.SerializeToString()`; in `retrieve()`, reconstruct with `payload.ParseFromString(data)`. The application data has already been serialized by the Payload Converter and Payload Codec before reaching the driver. -`context.target` provides identity information (namespace, Workflow ID, or Activity ID). Check the target type with `isinstance(target, StorageDriverWorkflowInfo)`; the Workflow info exposes `target.namespace` and `target.id`. Use this to scope storage keys per Workflow. Within that scope, content-addressable keys (such as a SHA-256 hash of the payload bytes) deduplicate identical payloads and make retries idempotent. +`context.target` provides identity information (namespace, Workflow ID, or Activity ID). Check the target type with `isinstance(target, StorageDriverWorkflowInfo)`; the Workflow info exposes `target.namespace` and `target.id`. Use this to scope storage keys per Workflow, but hash or encode identifiers before using them as path segments because identifiers can contain path separators or traversal sequences. Within that scope, content-addressable keys (such as a SHA-256 hash of the payload bytes) deduplicate identical payloads and make retries idempotent. Worked example — local-disk driver (development/testing only): ```python +import hashlib import os -import uuid from typing import Sequence from temporalio.api.common.v1 import Payload @@ -163,6 +163,10 @@ from temporalio.converter import ( ) +def safe_path_segment(value: str) -> str: + return hashlib.sha256(value.encode("utf-8")).hexdigest() + + class LocalDiskStorageDriver(StorageDriver): def __init__(self, store_dir: str = "/tmp/temporal-payload-store") -> None: self._store_dir = store_dir @@ -183,15 +187,20 @@ class LocalDiskStorageDriver(StorageDriver): prefix = self._store_dir target = context.target if isinstance(target, StorageDriverWorkflowInfo) and target.id: - prefix = os.path.join(self._store_dir, target.namespace, target.id) + prefix = os.path.join( + self._store_dir, + safe_path_segment(target.namespace), + safe_path_segment(target.id), + ) os.makedirs(prefix, exist_ok=True) claims = [] for payload in payloads: - key = f"{uuid.uuid4()}.bin" + data = payload.SerializeToString() + key = f"{hashlib.sha256(data).hexdigest()}.bin" file_path = os.path.join(prefix, key) with open(file_path, "wb") as f: - f.write(payload.SerializeToString()) + f.write(data) claims.append(StorageDriverClaim(claim_data={"path": file_path})) return claims From 9f47be3131c636326c9b9a9a8b20d8d47a93e738 Mon Sep 17 00:00:00 2001 From: Brian Strauch Date: Mon, 24 Aug 2026 11:54:58 -0700 Subject: [PATCH 6/7] Fix correctness bugs in external storage references Address code-review findings on the new external storage docs: - Go: add missing "context" and "log" imports to the S3 driver, GCS driver, and client/worker setup snippets, which presented complete import lists but failed to compile. - Go: add go.temporal.io/sdk/contrib/envconfig to both go get lines; it is a separate module and is imported by the setup snippet. - Go: give the local-disk worked example an import block, and introduce the commonpb alias at its first use in the selector example. - Go and Python: validate claim data in Retrieve/retrieve so a hand-crafted reference payload cannot read files outside the store directory, matching the hardening already applied to Store/store. - Python: the Worker inherits the Data Converter from its Client and takes no data_converter argument; the prose said to pass it to both. Verified by compiling every Go snippet against sdk-go and exercising both path guards. Co-Authored-By: Claude Opus 5 (1M context) --- references/go/external-storage.md | 53 ++++++++++++++++++++++++++- references/python/external-storage.md | 14 ++++++- 2 files changed, 64 insertions(+), 3 deletions(-) diff --git a/references/go/external-storage.md b/references/go/external-storage.md index ec6de97..b58e629 100644 --- a/references/go/external-storage.md +++ b/references/go/external-storage.md @@ -33,6 +33,7 @@ Amazon S3: ```bash go get go.temporal.io/sdk/contrib/aws/s3driver \ go.temporal.io/sdk/contrib/aws/s3driver/awssdkv2 \ + go.temporal.io/sdk/contrib/envconfig \ github.com/aws/aws-sdk-go-v2/config \ github.com/aws/aws-sdk-go-v2/service/s3 ``` @@ -42,6 +43,7 @@ Google Cloud Storage: ```bash go get go.temporal.io/sdk/contrib/gcp/gcsdriver \ go.temporal.io/sdk/contrib/gcp/gcsdriver/gcssdk \ + go.temporal.io/sdk/contrib/envconfig \ cloud.google.com/go/storage ``` @@ -49,6 +51,9 @@ go get go.temporal.io/sdk/contrib/gcp/gcsdriver \ ```go import ( + "context" + "log" + "github.com/aws/aws-sdk-go-v2/config" "github.com/aws/aws-sdk-go-v2/service/s3" "go.temporal.io/sdk/contrib/aws/s3driver" @@ -77,6 +82,9 @@ The AWS SDK reads standard credentials from the environment (env vars, IAM role, ```go import ( + "context" + "log" + "cloud.google.com/go/storage" "go.temporal.io/sdk/contrib/gcp/gcsdriver" "go.temporal.io/sdk/contrib/gcp/gcsdriver/gcssdk" @@ -104,6 +112,8 @@ For either driver, pass a `BucketFunc` as `Bucket` instead of `StaticBucket` to ```go import ( + "log" + "go.temporal.io/sdk/client" "go.temporal.io/sdk/contrib/envconfig" "go.temporal.io/sdk/converter" @@ -162,6 +172,12 @@ When you register more than one driver, you **must** supply a `DriverSelector` i - Every registered driver must have a distinct `Name()`; duplicates are rejected when the Client or Worker is constructed. `s3driver` defaults its name to `"aws.s3driver"` and `gcsdriver` to `"gcp.gcsdriver"`, so registering two drivers of the same kind requires setting `DriverName` on at least one. ```go +import ( + commonpb "go.temporal.io/api/common/v1" + + "go.temporal.io/sdk/converter" +) + type PreferredSelector struct { preferred converter.StorageDriver } @@ -198,9 +214,25 @@ Inside `Store()`, marshal each payload with `proto.Marshal(payload)`; in `Retrie `ctx.Target` provides identity information. Type-switch over `StorageDriverWorkflowInfo` and `StorageDriverActivityInfo` to access the namespace / Workflow ID / Activity ID, and use it to scope storage keys. Hash or encode identifiers before using them as path segments because identifiers can contain path separators or traversal sequences. `StorageDriverActivityInfo` is only used for standalone (non-workflow-bound) Activities; Activities started by a Workflow get `StorageDriverWorkflowInfo`. +Validate claim data in `Retrieve()` as untrusted input. A driver that resolves a filesystem path, object key, or URL straight out of the claim will follow whatever a hand-crafted reference payload puts there, so re-check that the resolved location stays inside the store the driver owns. + Worked example — local-disk driver (development/testing only): ```go +import ( + "crypto/sha256" + "encoding/hex" + "fmt" + "os" + "path/filepath" + "strings" + + commonpb "go.temporal.io/api/common/v1" + "google.golang.org/protobuf/proto" + + "go.temporal.io/sdk/converter" +) + type LocalDiskStorageDriver struct { storeDir string } @@ -214,6 +246,22 @@ func NewLocalDiskStorageDriver(storeDir string) converter.StorageDriver { return &LocalDiskStorageDriver{storeDir: storeDir} } +// resolvePath rejects claim data that points outside the store directory. +func (d *LocalDiskStorageDriver) resolvePath(claimPath string) (string, error) { + root, err := filepath.Abs(d.storeDir) + if err != nil { + return "", fmt.Errorf("resolve store directory: %w", err) + } + resolved, err := filepath.Abs(claimPath) + if err != nil { + return "", fmt.Errorf("resolve claim path: %w", err) + } + if resolved != root && !strings.HasPrefix(resolved, root+string(os.PathSeparator)) { + return "", fmt.Errorf("claim path %q escapes the store directory", claimPath) + } + return resolved, nil +} + func (d *LocalDiskStorageDriver) Name() string { return "my-local-disk" } func (d *LocalDiskStorageDriver) Type() string { return "local-disk" } @@ -269,7 +317,10 @@ func (d *LocalDiskStorageDriver) Retrieve( ) ([]*commonpb.Payload, error) { payloads := make([]*commonpb.Payload, len(claims)) for i, claim := range claims { - filePath := claim.ClaimData["path"] + filePath, err := d.resolvePath(claim.ClaimData["path"]) + if err != nil { + return nil, err + } data, err := os.ReadFile(filePath) if err != nil { return nil, fmt.Errorf("read payload: %w", err) diff --git a/references/python/external-storage.md b/references/python/external-storage.md index 5fd2f10..20866fb 100644 --- a/references/python/external-storage.md +++ b/references/python/external-storage.md @@ -32,7 +32,7 @@ The Python SDK ships an Amazon S3 driver (there is no built-in GCS driver — us python -m pip install "temporalio[aioboto3]" ``` -Create the driver, attach it to a `DataConverter`, and pass the converter to both Client and Worker: +Create the driver, attach it to a `DataConverter`, and pass the converter to `Client.connect`. A Worker inherits the Data Converter from the Client it is created with — `Worker` takes no `data_converter` argument of its own: ```python import asyncio @@ -146,6 +146,8 @@ Inside `store()`, serialize each payload with `payload.SerializeToString()`; in `context.target` provides identity information (namespace, Workflow ID, or Activity ID). Check the target type with `isinstance(target, StorageDriverWorkflowInfo)`; the Workflow info exposes `target.namespace` and `target.id`. Use this to scope storage keys per Workflow, but hash or encode identifiers before using them as path segments because identifiers can contain path separators or traversal sequences. Within that scope, content-addressable keys (such as a SHA-256 hash of the payload bytes) deduplicate identical payloads and make retries idempotent. +Treat claim data in `retrieve()` as untrusted input. A driver that resolves a filesystem path, object key, or URL straight out of the claim will follow whatever a hand-crafted reference payload puts there, so re-check that the resolved location stays inside the store the driver owns. + Worked example — local-disk driver (development/testing only): ```python @@ -171,6 +173,14 @@ class LocalDiskStorageDriver(StorageDriver): def __init__(self, store_dir: str = "/tmp/temporal-payload-store") -> None: self._store_dir = store_dir + def _resolve_path(self, claim_path: str) -> str: + """Reject claim data that points outside the store directory.""" + root = os.path.realpath(self._store_dir) + resolved = os.path.realpath(claim_path) + if resolved != root and not resolved.startswith(root + os.sep): + raise ValueError(f"claim path {claim_path!r} escapes the store directory") + return resolved + def name(self) -> str: return "local-disk" @@ -211,7 +221,7 @@ class LocalDiskStorageDriver(StorageDriver): ) -> list[Payload]: payloads = [] for claim in claims: - file_path = claim.claim_data["path"] + file_path = self._resolve_path(claim.claim_data["path"]) with open(file_path, "rb") as f: raw = f.read() payload = Payload() From a635c0c6a0b2776a1a9f62b687b87339d700a49e Mon Sep 17 00:00:00 2001 From: Brian Strauch Date: Mon, 24 Aug 2026 11:59:21 -0700 Subject: [PATCH 7/7] Route large-payload triage to the external storage references The new external storage docs were only reachable from the language index files, so the paths an agent actually takes when a user hits a payload limit still sent it to hand-roll the claim-check pattern. - core/error-reference.md: TMPRL1103 recovery now points at built-in External Storage before manual reference passing. - core/gotchas.md: the payload-limit fix notes the SDK does this for you in Go, Python, and TypeScript. - core/patterns.md: Large Data Handling leads with the SDK-native option and scopes the manual pattern to the cases that need it. Also link the Go external storage sample from the Codec Server section, matching what the Python reference already does. Co-Authored-By: Claude Opus 5 (1M context) --- references/core/error-reference.md | 2 +- references/core/gotchas.md | 2 ++ references/core/patterns.md | 2 ++ references/go/external-storage.md | 2 ++ 4 files changed, 7 insertions(+), 1 deletion(-) diff --git a/references/core/error-reference.md b/references/core/error-reference.md index 29a40b7..5c108c9 100644 --- a/references/core/error-reference.md +++ b/references/core/error-reference.md @@ -5,7 +5,7 @@ | **Non-determinism** | TMPRL1100 | `WorkflowTaskFailed` in history | Replay doesn't match history | Analyze error first. **If accidental**: fix code to match history → restart worker. **If intentional v2 change**: terminate → start fresh workflow. | https://github.com/temporalio/rules/blob/main/rules/TMPRL1100.md | | **Deadlock** | TMPRL1101 | `WorkflowTaskFailed` in history, worker logs | Workflow blocked too long (deadlock detected) | Remove blocking operations from workflow code (no I/O, no sleep, no threading locks). Use Temporal primitives instead. | https://github.com/temporalio/rules/blob/main/rules/TMPRL1101.md | | **Unfinished handlers** | TMPRL1102 | `WorkflowTaskFailed` in history | Workflow completed while update/signal handlers still running | Ensure all handlers complete before workflow finishes. Use `workflow.wait_condition()` to wait for handler completion. | https://github.com/temporalio/rules/blob/main/rules/TMPRL1102.md | -| **Payload overflow** | TMPRL1103 | `WorkflowTaskFailed` or `ActivityTaskFailed` in history | Payload size limit exceeded (default 2MB) | Reduce payload size. Use external storage (S3, database) for large data and pass references instead. | https://github.com/temporalio/rules/blob/main/rules/TMPRL1103.md | +| **Payload overflow** | TMPRL1103 | `WorkflowTaskFailed` or `ActivityTaskFailed` in history | Payload size limit exceeded (default 2MB) | Reduce payload size. Use the SDK's built-in External Storage where available (see `references/{your_language}/external-storage.md`; Go, Python, and TypeScript), or pass references to external storage yourself (see the Large Data Handling pattern in `references/core/patterns.md`). | https://github.com/temporalio/rules/blob/main/rules/TMPRL1103.md | | **Workflow code bug** | | `WorkflowTaskFailed` in history | Bug in workflow logic | Fix code → Restart worker → Workflow auto-resumes | | | **Missing workflow** | | Worker logs | Workflow not registered | Add to worker.py → Restart worker | | | **Missing activity** | | Worker logs | Activity not registered | Add to worker.py → Restart worker | | diff --git a/references/core/gotchas.md b/references/core/gotchas.md index 677362f..4a6a454 100644 --- a/references/core/gotchas.md +++ b/references/core/gotchas.md @@ -212,3 +212,5 @@ See language-specific gotchas for details. - Workflow history growing unboundedly **The Fix**: Store large data externally (S3/GCS) and pass references, use compression codecs, or chunk data across multiple activities. See the Large Data Handling pattern in `references/core/patterns.md`. + +Before hand-rolling reference passing, check whether the SDK does it for you: the Go, Python, and TypeScript SDKs have built-in External Storage that applies the claim-check pattern automatically. See `references/{your_language}/external-storage.md`, if available. diff --git a/references/core/patterns.md b/references/core/patterns.md index 7e7c7a3..1922106 100644 --- a/references/core/patterns.md +++ b/references/core/patterns.md @@ -368,6 +368,8 @@ This ensures that on replay, already-completed steps are skipped. - Max 4MB per gRPC message - Max 50MB for workflow history (aim for < 10MB) +**Check for SDK support first**: the Go, Python, and TypeScript SDKs have built-in External Storage that applies the claim-check pattern for you — Payloads over a size threshold are offloaded to S3 or GCS and replaced in Event History with a small reference, with no changes to Workflow or Activity code. Prefer it where it exists; see `references/{your_language}/external-storage.md`, if available. The rest of this section applies when you need explicit control over which data is offloaded, or when your SDK has no built-in support. + **Key Principle**: Large data should never flow through workflow history. Activities read and write large data directly, passing only small references through the workflow. **Wrong Approach**: diff --git a/references/go/external-storage.md b/references/go/external-storage.md index b58e629..cc0dd46 100644 --- a/references/go/external-storage.md +++ b/references/go/external-storage.md @@ -366,6 +366,8 @@ When configured with storage drivers, the handler exposes: **Don't use `NewPayloadHTTPHandler` as a remote Data Converter or remote codec target for your Workers** — it runs the full encode-store-encode and decode-retrieve-decode pipeline. For remote codecs use `NewPayloadCodecHTTPHandler` separately. If you need both, run both handlers, configured with the same codecs. +The [Go External Storage sample](https://github.com/temporalio/samples-go/tree/main/external-storage) is a working end-to-end setup to copy from: a Worker with an S3 driver behind a zlib Payload Codec, a Codec Server built on `NewPayloadHTTPHandler` (`codec-server/main.go`), and a mock S3 service so it runs locally without an AWS account. + ## Lifecycle and failure handling Temporal does **not** auto-delete payloads from your store. Configure a TTL on your bucket: