Skip to main content

External Storage

View Markdown

The Temporal Service enforces a 2 MB per-payload limit by default. This limit is configurable on self-hosted deployments. When your Workflows or Activities handle data larger than the limit, you can offload payloads to external storage, such as Amazon S3, and pass a small reference token through the Event History instead. This page shows you how to set up External Storage with Amazon S3 and how to implement a custom storage driver.

For a conceptual overview of External Storage and its use cases, see External Storage.

Rust API availability

The Rust API requires the experimental Cargo feature. The examples on this page target Rust SDK revision 51009f8, which includes APIs absent from the published Rust SDK 1.1.2 crates. Use the pinned dependencies in the example project to try this implementation. See Current limitations before using it.

Store and retrieve large payloads with Amazon S3​

The Rust SDK accepts a custom implementation of StorageDriver. The example project provides an S3 driver built with the AWS SDK for Rust; it is example code, rather than a built-in Temporal driver.

Prerequisites​

  • An S3 bucket accessible to every Client and Worker that handles stored payloads, with s3:PutObject and s3:GetObject permissions for the temporal-payloads/ prefix.
  • AWS credentials and a Region configured for the AWS SDK for Rust, for example through an IAM role and AWS_REGION.
  • The example project's pinned dependencies, with its s3 feature enabled. The project enables aws-config's credentials-login feature so it can use credentials from aws login.

Keep stored payloads available for the entire Workflow lifetime, including retries, replay, and retention. See Lifecycle management and Durable External Storage when planning retention and regional recovery.

Create the storage driver​

Create an AWS S3 client and pass it and your bucket name to the example's S3StorageDriver. The AWS SDK loads credentials and Region settings through its default configuration chain.

rust-sandbox/external-storage/src/s3_driver.rs

let config = aws_config::load_defaults(aws_config::BehaviorVersion::latest()).await;
let driver = S3StorageDriver {
    client: aws_sdk_s3::Client::new(&config),
    bucket,
};

Configure the Client and Worker​

Wrap the driver in Arc<dyn StorageDriver>, create ExternalStorage, and attach it to the Data Converter. Set that converter on the Client options before constructing the Client. The Worker uses the Client's Data Converter.

rust-sandbox/external-storage/src/main.rs

let storage = ExternalStorage::new(driver)?;
let data_converter = DataConverter::default().with_external_storage(storage);
let (connection_options, mut client_options) =
    ClientOptions::load_from_config(Default::default())
        .map_err(|error| anyhow::anyhow!("{error}"))?;
client_options.data_converter = data_converter;
let connection = Connection::connect(connection_options).await?;
let client = Client::new(connection, client_options)?;

let runtime = Runtime::from_current_tokio(Default::default())?;
let worker_options = WorkerOptions::new("external-storage")
    .register_workflow::<LargePayloadWorkflow>()?
    .register_activities(LargePayloadActivities)
    .build();
let mut worker = Worker::new(&runtime, client.clone(), worker_options)?;

Configure compatible drivers on every Client and Worker process that can send or receive an offloaded payload, including Clients that fetch Workflow results. Drivers run outside Workflow Definitions.

The complete example registers a Workflow and an echo Activity, starts a Worker, and sends a 3 MiB string through Workflow input, Activity input and result, and Workflow result. Run it against a local Temporal Service:

cd rust-sandbox/external-storage
S3_BUCKET=my-temporal-payloads AWS_REGION=us-east-1 cargo run --features s3

The example loads Temporal connection settings through ClientOptions::load_from_config. See Connect to a development Temporal Service for connection configuration.

Implement a custom storage driver​

Implement StorageDriver from temporalio_common::data_converters. A driver must be thread safe (Send + Sync) and implement four methods:

  • name() returns a nonempty name unique among the registered drivers. The SDK writes this name into the reference token to route retrieval. Keep it stable after storing payloads.
  • driver_type() identifies the implementation, separately from the configured instance name.
  • store() accepts Vec<Payload> and returns one StorageDriverClaim per payload, in the same order.
  • retrieve() accepts claims and returns the original payloads, in the same order.

Both storage methods return BoxFuture<'static, Result<...>>. Clone the state needed by the future before entering async move, rather than borrowing self or the context in the returned future.

Store payloads and return claims​

Serialize the entire protobuf Payload, including its metadata, with prost::Message::encode_to_vec. The SDK has already applied the Payload Converter and any Payload Codec before handing the payload to your driver.

Return a StorageDriverClaim::new(HashMap<String, String>) with enough information to locate that payload later. Claim data is stored in Event History without encryption in this Rust implementation. Do not include credentials or other secrets in claims.

StorageDriverStoreContext::target provides Workflow or Standalone Activity identity when available. You can use it to organize storage keys, but handle None. Retrieval receives no target: a claim and the driver's configuration must be sufficient to retrieve a payload.

Retrieve payloads and verify their integrity​

Read the stored bytes, validate their integrity, and reconstruct the original message with Payload::decode. Return an error for a missing object, invalid claim, failed download, or corrupt payload. The SDK then applies the Payload Codec and Payload Converter to recover the application value.

Acquire a permit from context.limiter for each storage request and hold it until that request completes. The current implementation does not enforce concurrency limits, but drivers should use the limiter interface. The example drivers issue their requests sequentially within each call.

Use local disk for development​

This driver uses SHA-256 keys and verifies downloaded bytes before decoding. It stores identical payloads under the same key, so repeated storage during replay reuses that key.

Use it only for local development. Every process must share the same directory, and this example does not provide atomic writes or production storage durability.

rust-sandbox/external-storage/src/local_disk_driver.rs

// For local development only. All processes must share this directory.
#[derive(Clone)]
pub struct LocalDiskStorageDriver {
    pub directory: PathBuf,
}

impl StorageDriver for LocalDiskStorageDriver {
    fn name(&self) -> &str {
        "local-disk"
    }
    fn driver_type(&self) -> &str {
        "example.local-disk"
    }

    fn store(
        &self,
        context: &StorageDriverStoreContext,
        payloads: Vec<Payload>,
    ) -> BoxFuture<'static, Result<Vec<StorageDriverClaim>, StorageDriverError>> {
        let directory = self.directory.clone();
        let limiter = context.limiter.clone();
        Box::pin(async move {
            tokio::fs::create_dir_all(&directory).await?;
            let mut claims = Vec::with_capacity(payloads.len());
            for payload in payloads {
                let _permit = limiter.acquire(&payload).await?;
                let bytes = payload.encode_to_vec();
                let key = format!("{:x}", Sha256::digest(&bytes));
                // Identical payloads use the same key, including during replay.
                tokio::fs::write(directory.join(&key), bytes).await?;
                claims.push(StorageDriverClaim::new(HashMap::from([(
                    "key".to_owned(),
                    key,
                )])));
            }
            Ok(claims)
        })
    }

    fn retrieve(
        &self,
        context: &StorageDriverRetrieveContext,
        claims: Vec<StorageDriverClaim>,
    ) -> BoxFuture<'static, Result<Vec<Payload>, StorageDriverError>> {
        let directory = self.directory.clone();
        let limiter = context.limiter.clone();
        Box::pin(async move {
            let mut payloads = Vec::with_capacity(claims.len());
            for claim in claims {
                let _permit = limiter.acquire(&claim).await?;
                let key = claim
                    .claim_data
                    .get("key")
                    .ok_or_else(|| anyhow!("claim is missing its key"))?;
                // Accept only a hash, never a path supplied by a claim.
                if !(key.len() == 64 && key.bytes().all(|b| b.is_ascii_hexdigit())) {
                    return Err(anyhow!("invalid payload key").into());
                }
                let bytes = tokio::fs::read(directory.join(key)).await?;
                if !(format!("{:x}", Sha256::digest(&bytes)) == *key) {
                    return Err(anyhow!("payload checksum mismatch").into());
                }
                payloads.push(Payload::decode(bytes.as_slice())?);
            }
            Ok(payloads)
        })
    }
}

Run the complete example with this driver by leaving S3_BUCKET unset:

cd rust-sandbox/external-storage
cargo run

Use Amazon S3 as the backing store​

The S3 example uses the same protobuf serialization and SHA-256 verification. It stores objects under a fixed prefix and reads from the configured bucket. Keep the bucket configuration stable for existing claims, or retain the old driver when migrating to a new bucket.

rust-sandbox/external-storage/src/s3_driver.rs

// Example driver, not a built-in Temporal driver.
#[derive(Clone)]
pub struct S3StorageDriver {
    pub client: Client,
    pub bucket: String,
}

impl StorageDriver for S3StorageDriver {
    fn name(&self) -> &str {
        "s3-payloads"
    }
    fn driver_type(&self) -> &str {
        "example.s3"
    }

    fn store(
        &self,
        context: &StorageDriverStoreContext,
        payloads: Vec<Payload>,
    ) -> BoxFuture<'static, Result<Vec<StorageDriverClaim>, StorageDriverError>> {
        let driver = self.clone();
        let limiter = context.limiter.clone();
        Box::pin(async move {
            let mut claims = Vec::with_capacity(payloads.len());
            for payload in payloads {
                let _permit = limiter.acquire(&payload).await?;
                let bytes = payload.encode_to_vec();
                let checksum = format!("{:x}", Sha256::digest(&bytes));
                let key = format!("temporal-payloads/{checksum}");
                driver
                    .client
                    .put_object()
                    .bucket(&driver.bucket)
                    .key(&key)
                    .body(ByteStream::from(bytes))
                    .send()
                    .await
                    .map_err(|e| anyhow::Error::new(e).context("S3 upload failed"))?;
                claims.push(StorageDriverClaim::new(HashMap::from([
                    ("key".to_owned(), key),
                    ("sha256".to_owned(), checksum),
                ])));
            }
            Ok(claims)
        })
    }

    fn retrieve(
        &self,
        context: &StorageDriverRetrieveContext,
        claims: Vec<StorageDriverClaim>,
    ) -> BoxFuture<'static, Result<Vec<Payload>, StorageDriverError>> {
        let driver = self.clone();
        let limiter = context.limiter.clone();
        Box::pin(async move {
            let mut payloads = Vec::with_capacity(claims.len());
            for claim in claims {
                let _permit = limiter.acquire(&claim).await?;
                let key = claim
                    .claim_data
                    .get("key")
                    .ok_or_else(|| anyhow!("claim is missing its key"))?;
                let checksum = claim
                    .claim_data
                    .get("sha256")
                    .ok_or_else(|| anyhow!("claim is missing its checksum"))?;
                if !(checksum.len() == 64
                    && checksum.bytes().all(|b| b.is_ascii_hexdigit())
                    && *key == format!("temporal-payloads/{checksum}"))
                {
                    return Err(anyhow!("invalid S3 claim").into());
                }
                let response = driver
                    .client
                    .get_object()
                    .bucket(&driver.bucket)
                    .key(key)
                    .send()
                    .await
                    .map_err(|e| anyhow::Error::new(e).context("S3 download failed"))?;
                let bytes = response
                    .body
                    .collect()
                    .await
                    .map_err(|e| anyhow::Error::new(e).context("S3 body read failed"))?
                    .into_bytes();
                if !(format!("{:x}", Sha256::digest(&bytes)) == *checksum) {
                    return Err(anyhow!("payload checksum mismatch").into());
                }
                payloads.push(Payload::decode(bytes.as_ref())?);
            }
            Ok(payloads)
        })
    }
}

Configure the payload size threshold​

By default, the SDK offloads payloads whose encoded size is at least 256 KiB. The size includes Payload metadata, and is measured after any Payload Codec has run. Smaller payloads stay inline in Event History.

Use with_payload_size_threshold to change the threshold. Set it to 0 to offload every payload.

rust-sandbox/external-storage/src/configuration.rs

let storage = ExternalStorage::new(driver)?.with_payload_size_threshold(256 * 1024);
let data_converter = DataConverter::default().with_external_storage(storage);

Use multiple storage drivers​

Register multiple drivers with ExternalStorage::with_selector. A StorageDriverSelector chooses the driver used for new payloads that meet the size threshold. Return None to keep an eligible payload inline.

During migration, select the preferred driver for new writes and keep the legacy driver registered for retrieval. Each driver must have a distinct, stable name. Retain legacy drivers and their backing stores for as long as existing references can be read, including during replay.

rust-sandbox/external-storage/src/configuration.rs

struct PreferDriver(Arc<dyn StorageDriver>);

impl StorageDriverSelector for PreferDriver {
    fn select_driver(
        &self,
        _context: &StorageDriverSelectContext,
        _payload: &Payload,
    ) -> Option<Arc<dyn StorageDriver>> {
        Some(self.0.clone())
    }
}

pub fn migration_storage(
    preferred: Arc<dyn StorageDriver>,
    legacy: Arc<dyn StorageDriver>,
) -> Result<ExternalStorage, ExternalStorageConfigError> {
    ExternalStorage::with_selector(
        vec![preferred.clone(), legacy],
        Arc::new(PreferDriver(preferred)),
    )
}

Current limitations​

The pinned Rust implementation has these limitations:

  • Storage request concurrency limits are not enforced. Bound requests in your driver if needed.
  • Reference tokens do not pass through Payload Codecs. Claim data is written to Event History unencrypted, and Rust does not recognize a reference token that another SDK has encoded with a Payload Codec.
  • Workflow replay stores payloads again. Plan for repeated writes and their storage costs. Content-addressed keys, as used in the example drivers, reuse object keys for identical payloads.
  • Fetching a Workflow's Event History returns stored reference tokens, rather than resolving them to original payloads.

These limitations are documented on the SDK's ExternalStorage type.