diff --git a/README.md b/README.md index 766a71b02..72641689b 100644 --- a/README.md +++ b/README.md @@ -23,6 +23,7 @@ Build resilient, long-running AWS Lambda functions that automatically checkpoint - **Replay Safety** – Functions deterministically resume from checkpoints after interruptions - **Type Safety** – Full generic type support for step results - **Data-Driven Concurrency** – Apply a function across a collection with `map()`, with per-item error isolation and configurable completion criteria +- **Payload Offloading** – Keep large serialized payloads in durable external storage while checkpoints retain compact references ## How It Works @@ -50,6 +51,16 @@ Your durable function extends `DurableHandler` and implements `handleReque ``` +Filesystem payload offloading is available as an optional artifact: + +```xml + + software.amazon.lambda.durable + aws-durable-execution-sdk-java-extra-filesystem-offloader + VERSION + +``` + ### Your First Durable Function ```java @@ -111,6 +122,7 @@ See [Deploy Lambda durable functions with Infrastructure as Code](https://docs.a **Advanced Topics** - [Configuration](docs/advanced/configuration.md) - Customize SDK behaviour +- [Payload Offloading](docs/advanced/configuration.md#payload-offloading) - Store serialized payloads outside checkpoints - [Error Handling](docs/advanced/error-handling.md) - SDK exceptions for handling failures - [Logging](docs/advanced/logging.md) - How to use DurableLogger - [Migrating from 1.x to 2.x](docs/migration-1.x-to-2.x.md) - Upgrade guide for breaking changes since `v1.2.1` diff --git a/coverage-report/pom.xml b/coverage-report/pom.xml index e820594e1..d91211228 100644 --- a/coverage-report/pom.xml +++ b/coverage-report/pom.xml @@ -22,6 +22,11 @@ aws-durable-execution-sdk-java ${project.version} + + software.amazon.lambda.durable + aws-durable-execution-sdk-java-extra-filesystem-offloader + ${project.version} + software.amazon.lambda.durable aws-durable-execution-sdk-java-testing diff --git a/docs/adr/005-filesystem-serdes.md b/docs/adr/005-filesystem-serdes.md index ef0a0a17e..b8e000ca9 100644 --- a/docs/adr/005-filesystem-serdes.md +++ b/docs/adr/005-filesystem-serdes.md @@ -1,8 +1,11 @@ # ADR-005: Payload Offloading for Filesystem Storage -**Status:** Proposed +**Status:** Accepted + **Date:** 2026-07-02 +**Decision:** Approach B, a dedicated `PayloadOffloader` interface with an optional filesystem implementation. + ## Context Issue [#463](https://github.com/aws/aws-durable-execution-sdk-java/issues/463) asks for Java parity with the JavaScript SDK's filesystem-backed SerDes. The JavaScript implementation receives a `SerdesContext` containing a stable durable execution ARN and an entity ID, then stores either inline JSON or a file pointer in the checkpoint payload. That context lets the implementation choose a collision-free path for each operation payload. @@ -407,9 +410,9 @@ This approach gives the SDK one consistent policy for root payloads, operation r | Immediate delivery risk | Lower. Builds on existing customization point. | Higher. Requires new API and more runtime integration. | | Long-term design risk | Higher. Blurs SerDes semantics and may accumulate storage behavior in serializers. | Lower if offloading grows into a first-class feature, but higher if this remains a one-off filesystem parity feature. | -## AI Recommendation +## Decision Rationale -**AI recommendation:** Prefer **Approach B: Create a `PayloadOffloader` interface** if the team is willing to treat payload offloading as a first-class Java SDK capability rather than only a JavaScript parity item. +Approach B was selected because payload offloading is treated as a first-class Java SDK capability rather than only a JavaScript parity item. Reasoning: @@ -419,7 +422,7 @@ Reasoning: - Both approaches use one optional extra package for filesystem-specific code; that is not a differentiator. The package would be either filesystem SerDes or filesystem offloader depending on the chosen approach. The differentiator is that Approach B gives future storage extras such as S3 or DynamoDB offload the same focused core offloader contract instead of encoding storage behavior as more SerDes implementations. - SDK-owned envelopes and two-layer caching make replay behavior easier to test and reason about. -The main reason to choose Approach A is schedule and parity: it is smaller and maps directly to the JavaScript feature request. If the team needs to satisfy #463 quickly with minimal public API design, Approach A is a reasonable incremental step, but it should be documented as payload offloading implemented through SerDes rather than as the long-term ideal boundary. +Approach A remains smaller and maps directly to the JavaScript feature shape, but it would make serialization responsible for external storage and require thread-local context propagation. ## Other Alternatives Considered @@ -459,7 +462,7 @@ Rejected. The backend request/response envelope is protocol data. User payload c Positive: -- Both approaches enable filesystem-backed payload storage without changing the existing `SerDes` interface. +- Filesystem-backed payload storage is available without changing the existing `SerDes` interface. - Filesystem-specific functionality stays out of the core SDK artifact. - The repository gets a repeatable `aws-durable-execution-sdk-java-extra-xxx` artifact pattern for optional packages. - Custom payload implementations get enough context to use external storage safely. @@ -471,15 +474,12 @@ Negative: - Adds executor, context, and caching machinery that must stay deterministic. - Adds at least one Maven module and published artifact to release and document. -- Approach A requires thread-local SerDes context because the existing `SerDes` methods do not accept context. - Repeated `get()` calls may return the same object instance in one invocation. - Filesystem-backed storage introduces operational durability requirements outside the SDK's control. -- Approach A risks overloading the meaning of SerDes. - Approach B requires a larger core SDK design before delivering filesystem parity. Deferred: -- Choosing whether payload offloading is a first-class SDK concept or a parity feature implemented through SerDes. - A fully async Java SerDes or payload pipeline contract. - A separate, explicitly dangerous protocol-envelope customization API. - File cleanup, retention policies, and lifecycle management for offloaded payloads. diff --git a/docs/advanced/configuration.md b/docs/advanced/configuration.md index bc8bf89d0..05a1d0d1e 100644 --- a/docs/advanced/configuration.md +++ b/docs/advanced/configuration.md @@ -17,7 +17,9 @@ public class OrderProcessor extends DurableHandler { return DurableConfig.builder() .withLambdaClientBuilder(lambdaClientBuilder) .withSerDes(new MyCustomSerDes()) // Custom serialization + .withPayloadOffloader(myPayloadOffloader) // Optional external payload storage .withExecutorService(Executors.newFixedThreadPool(10)) // Custom thread pool + .withPayloadOffloadExecutorService(payloadIoExecutor) // Blocking payload I/O .withLoggerConfig(LoggerConfig.withReplayLogging()) // Enable replay logs .build(); } @@ -33,13 +35,86 @@ public class OrderProcessor extends DurableHandler { |-----------------------------|-----------------------------------------|-------------------------------| | `withLambdaClientBuilder()` | Custom AWS Lambda client | Auto-configured Lambda client | | `withSerDes()` | Serializer for step results | Jackson with default settings | +| `withPayloadOffloader()` | External storage for serialized user payloads | Disabled | | `withExecutorService()` | Thread pool for user-defined operations | Cached daemon thread pool | +| `withPayloadOffloadExecutorService()` | Thread pool for blocking payload storage I/O | Cached daemon thread pool | | `withLoggerConfig()` | Logger behavior configuration | Suppress logs during replay | | `withPollingStrategy()` | Backend polling strategy | Exponential backoff: 1s base, 2x rate, FULL jitter, 10s max | | `withCheckpointDelay()` | How often the SDK checkpoints updates | `Duration.ofSeconds(0)` (as soon as possible) | The `withExecutorService()` option configures the thread pool used for running user-defined operations. Internal SDK coordination (checkpoint batching, polling) runs on an SDK-managed thread pool. +### Payload offloading + +`SerDes` remains responsible for converting objects to serialized text. A `PayloadOffloader` runs after serialization +and decides whether that text remains inline or is stored externally. On replay, the SDK resolves the stored reference +before passing the serialized text back to `SerDes`. + +Add the optional filesystem artifact: + +```xml + + software.amazon.lambda.durable + aws-durable-execution-sdk-java-extra-filesystem-offloader + VERSION + +``` + +Configure a durable shared mount: + +```java +import java.nio.file.Path; +import software.amazon.lambda.durable.extra.filesystem.FileSystemPathEncoding; +import software.amazon.lambda.durable.extra.filesystem.FileSystemPayloadOffloader; +import software.amazon.lambda.durable.extra.filesystem.PayloadOffloadMode; + +var offloader = FileSystemPayloadOffloader.builder(Path.of("/mnt/efs/durable-payloads")) + .storageMode(PayloadOffloadMode.OVERFLOW) + .pathEncoding(FileSystemPathEncoding.HASH) + .previewGenerator((serialized, context) -> Map.of( + "entityId", context.entityId(), + "payloadKind", context.payloadKind().name())) + .build(); + +return DurableConfig.builder() + .withSerDes(new JacksonSerDes()) + .withPayloadOffloader(offloader) + .build(); +``` + +`ALWAYS` writes every serialized payload to the filesystem. `OVERFLOW` keeps payloads inline until they approach the +256 KB checkpoint limit, then stores them externally. `URI` path encoding produces readable directories and file names; +`HASH` uses fixed-length SHA-256 names and is safer for long or unusual entity identifiers. + +The global offloader applies to root input/output, operation results, invoke payloads, callback results, +wait-for-condition state, child/map/parallel results, and serialized exception data. Operation configuration can +override it: + +```java +var stepConfig = StepConfig.builder() + .payloadOffloader(otherOffloader) + .build(); + +var inlineStepConfig = StepConfig.builder() + .payloadOffloader(PayloadOffloader.disabled()) + .build(); +``` + +The same `payloadOffloader(...)` option is available on `InvokeConfig`, `CallbackConfig`, +`RunInChildContextConfig`, `MapConfig`, `ParallelConfig`, `ParallelBranchConfig`, and +`WaitForConditionConfig`. + +The SDK uses a versioned checkpoint envelope and continues to read payloads written by older SDK versions as raw +serialized text. Within one Lambda invocation, resolved storage data and deserialized objects are cached, so repeated +`DurableFuture.get()` calls do not repeatedly read the same file. + +> **Do not use Lambda `/tmp` for durable payloads.** It is local to one execution environment and might not exist on +> replay. Use a shared durable filesystem such as EFS. S3 Files can have delayed synchronization and recent writes can +> be lost if the runtime crashes before the mount flushes; use it only when that durability tradeoff is acceptable. + +The SDK does not delete offloaded files. Configure storage lifecycle and retention separately, and keep the mounted +path accessible to every function environment that may replay or consume the payload. + ### Dynamic plugin loading Dynamic plugin loading is an opt-in alternative to registering plugins in application code. Put provider JARs on the application class path, then set `DURABLE_EXECUTION_PLUGINS` to an ordered, comma-separated list of provider names: diff --git a/docs/design.md b/docs/design.md index eaba54af0..68234ac42 100644 --- a/docs/design.md +++ b/docs/design.md @@ -11,6 +11,7 @@ This document explains the internal architecture, threading model, and extension ``` aws-durable-execution-sdk-java/ ├── sdk/ # Core SDK - DurableHandler, DurableContext, operations +├── extra-filesystem-offloader/ # Optional durable filesystem payload storage ├── sdk-testing/ # Test utilities for local and cloud testing ├── sdk-integration-tests/ # Integration tests using LocalDurableTestRunner └── examples/ # Real-world usage patterns as customers would implement them @@ -19,6 +20,7 @@ aws-durable-execution-sdk-java/ | Module | Purpose | Key Classes | |--------|---------|-------------| | `sdk` | Core runtime - extend `DurableHandler`, use `DurableContext` for durable operations | `DurableHandler`, `DurableContext`, `DurableExecutor`, `ExecutionManager` | +| `extra-filesystem-offloader` | Optional payload offloader for durable shared filesystems | `FileSystemPayloadOffloader` | | `sdk-testing` | Test utilities: `LocalDurableTestRunner` (in-memory, simulates re-invocations and time-skipping) and `CloudDurableTestRunner` (executes against deployed Lambda) | `LocalDurableTestRunner`, `CloudDurableTestRunner`, `LocalMemoryExecutionClient`, `TestResult` | | `sdk-integration-tests` | Dogfooding tests - validates the SDK using its own test utilities. Separate module keeps dependencies acyclic: `sdk` → `sdk-testing` → `sdk-integration-tests`. | Test classes only | | `examples` | Real-world usage patterns as customers would implement them, with local and cloud tests | Example handlers, `CloudBasedIntegrationTest` | diff --git a/extra-filesystem-offloader/README.md b/extra-filesystem-offloader/README.md new file mode 100644 index 000000000..fa8d56ad8 --- /dev/null +++ b/extra-filesystem-offloader/README.md @@ -0,0 +1,34 @@ +# Filesystem Payload Offloader + +This optional module stores serialized durable execution payloads on a shared filesystem while the durable checkpoint +contains a compact SDK-owned reference. + +```xml + + software.amazon.lambda.durable + aws-durable-execution-sdk-java-extra-filesystem-offloader + VERSION + +``` + +```java +var offloader = FileSystemPayloadOffloader.builder(Path.of("/mnt/efs/durable-payloads")) + .storageMode(PayloadOffloadMode.OVERFLOW) + .pathEncoding(FileSystemPathEncoding.HASH) + .build(); + +return DurableConfig.builder() + .withPayloadOffloader(offloader) + .build(); +``` + +- `ALWAYS` stores every payload in a file. +- `OVERFLOW` keeps small payloads inline and offloads values near the 256 KB checkpoint limit. +- `URI` creates readable paths. +- `HASH` creates fixed-length SHA-256 path segments. + +Do not use Lambda `/tmp`: it is not shared across environments or guaranteed to survive replay. EFS provides a shared +durable mount. S3 Files can delay synchronization and can lose recent writes if the runtime crashes before a flush, so +use it only when the application accepts that tradeoff. + +The SDK does not delete files. Configure retention and lifecycle management on the backing storage. diff --git a/extra-filesystem-offloader/pom.xml b/extra-filesystem-offloader/pom.xml new file mode 100644 index 000000000..a3afb7ae9 --- /dev/null +++ b/extra-filesystem-offloader/pom.xml @@ -0,0 +1,75 @@ + + + 4.0.0 + + + software.amazon.lambda.durable + aws-durable-execution-sdk-java-parent + 2.1.1-SNAPSHOT + + + aws-durable-execution-sdk-java-extra-filesystem-offloader + jar + + AWS Lambda Durable Execution SDK Filesystem Payload Offloader + Filesystem-backed payload offloader for the AWS Lambda Durable Execution SDK for Java + https://github.com/aws/aws-durable-execution-sdk-java + + + scm:git:https://github.com/aws/aws-durable-execution-sdk-java.git + scm:git:https://github.com/aws/aws-durable-execution-sdk-java.git + https://github.com/aws/aws-durable-execution-sdk-java + + + + + software.amazon.lambda.durable + aws-durable-execution-sdk-java + ${project.version} + + + org.junit.jupiter + junit-jupiter + test + + + + + + + org.apache.maven.plugins + maven-compiler-plugin + + + org.apache.maven.plugins + maven-surefire-plugin + + + org.apache.maven.plugins + maven-source-plugin + + + attach-sources + + jar-no-fork + + + + + + org.apache.maven.plugins + maven-javadoc-plugin + + + attach-javadocs + + jar + + + + + + + diff --git a/extra-filesystem-offloader/src/main/java/software/amazon/lambda/durable/extra/filesystem/FileSystemPathEncoding.java b/extra-filesystem-offloader/src/main/java/software/amazon/lambda/durable/extra/filesystem/FileSystemPathEncoding.java new file mode 100644 index 000000000..e3904928f --- /dev/null +++ b/extra-filesystem-offloader/src/main/java/software/amazon/lambda/durable/extra/filesystem/FileSystemPathEncoding.java @@ -0,0 +1,12 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.extra.filesystem; + +/** Controls how durable execution and entity identifiers are encoded into filesystem paths. */ +public enum FileSystemPathEncoding { + /** Use percent-encoded, human-readable path segments. */ + URI, + + /** Use fixed-length SHA-256 hashes for filesystem-safe path segments. */ + HASH +} diff --git a/extra-filesystem-offloader/src/main/java/software/amazon/lambda/durable/extra/filesystem/FileSystemPayloadOffloader.java b/extra-filesystem-offloader/src/main/java/software/amazon/lambda/durable/extra/filesystem/FileSystemPayloadOffloader.java new file mode 100644 index 000000000..67c715d33 --- /dev/null +++ b/extra-filesystem-offloader/src/main/java/software/amazon/lambda/durable/extra/filesystem/FileSystemPayloadOffloader.java @@ -0,0 +1,180 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.extra.filesystem; + +import java.io.IOException; +import java.net.URLEncoder; +import java.nio.charset.StandardCharsets; +import java.nio.file.AtomicMoveNotSupportedException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.StandardCopyOption; +import java.security.MessageDigest; +import java.security.NoSuchAlgorithmException; +import java.util.HexFormat; +import java.util.Map; +import java.util.Objects; +import java.util.regex.Pattern; +import software.amazon.lambda.durable.exception.PayloadOffloadException; +import software.amazon.lambda.durable.offload.OffloadedPayload; +import software.amazon.lambda.durable.offload.PayloadOffloadContext; +import software.amazon.lambda.durable.offload.PayloadOffloader; +import software.amazon.lambda.durable.offload.PayloadStorageMode; + +/** + * Stores serialized payloads on a durable, shared filesystem. + * + *

Do not use Lambda's ephemeral {@code /tmp} storage. Replays can run in another execution environment where those + * files do not exist. Use a durable shared mount such as EFS, or S3 Files only when its synchronization and crash-loss + * tradeoffs are acceptable for the workload. + */ +public final class FileSystemPayloadOffloader implements PayloadOffloader { + private static final int CHECKPOINT_SIZE_LIMIT_BYTES = 256 * 1024; + private static final int OVERFLOW_THRESHOLD_BYTES = CHECKPOINT_SIZE_LIMIT_BYTES - 1024; + private static final Pattern DURABLE_EXECUTION_ARN_PATTERN = Pattern.compile( + "^arn:[^:]*:lambda:[^:]*:[^:]*:function:([^:/]+):[^:/]+/durable-execution/([^/]+)/([^/]+)$"); + + private final Path basePath; + private final PayloadOffloadMode storageMode; + private final FileSystemPathEncoding pathEncoding; + private final PayloadPreviewGenerator previewGenerator; + + private FileSystemPayloadOffloader(Builder builder) { + this.basePath = builder.basePath.toAbsolutePath().normalize(); + this.storageMode = Objects.requireNonNullElse(builder.storageMode, PayloadOffloadMode.ALWAYS); + this.pathEncoding = Objects.requireNonNullElse(builder.pathEncoding, FileSystemPathEncoding.URI); + this.previewGenerator = builder.previewGenerator; + } + + /** Creates a filesystem offloader builder. */ + public static Builder builder(Path basePath) { + return new Builder(basePath); + } + + @Override + public OffloadedPayload offload(String serializedPayload, PayloadOffloadContext context) { + Objects.requireNonNull(serializedPayload, "serializedPayload cannot be null"); + Objects.requireNonNull(context, "context cannot be null"); + if (storageMode == PayloadOffloadMode.OVERFLOW + && serializedPayload.getBytes(StandardCharsets.UTF_8).length <= OVERFLOW_THRESHOLD_BYTES) { + return OffloadedPayload.inline(serializedPayload); + } + + var path = resolvePayloadPath(context); + writeAtomically(path, serializedPayload, context); + Map preview = + previewGenerator == null ? null : previewGenerator.generate(serializedPayload, context); + return OffloadedPayload.reference(path.toString(), preview); + } + + @Override + public String load(OffloadedPayload payload, PayloadOffloadContext context) { + Objects.requireNonNull(payload, "payload cannot be null"); + Objects.requireNonNull(context, "context cannot be null"); + if (payload.mode() == PayloadStorageMode.INLINE) { + return payload.data(); + } + + var path = Path.of(payload.reference()).toAbsolutePath().normalize(); + if (!path.startsWith(basePath)) { + throw new PayloadOffloadException( + "Payload reference is outside configured base path for entity '" + context.entityId() + "'"); + } + try { + return Files.readString(path, StandardCharsets.UTF_8); + } catch (IOException e) { + throw new PayloadOffloadException( + "Failed to read filesystem payload for entity '" + context.entityId() + "'", e); + } + } + + private Path resolvePayloadPath(PayloadOffloadContext context) { + var directory = resolveExecutionDirectory(context.durableExecutionArn()); + var storageKey = + context.attempt() == null ? context.entityId() : context.entityId() + "/attempt-" + context.attempt(); + return directory + .resolve(encodeSegment(storageKey) + ".json") + .toAbsolutePath() + .normalize(); + } + + private Path resolveExecutionDirectory(String durableExecutionArn) { + if (pathEncoding == FileSystemPathEncoding.URI) { + var matcher = DURABLE_EXECUTION_ARN_PATTERN.matcher(durableExecutionArn); + if (matcher.matches()) { + return basePath.resolve(matcher.group(1)) + .resolve(matcher.group(2)) + .resolve(matcher.group(3)); + } + } + return basePath.resolve(encodeSegment(durableExecutionArn)); + } + + private String encodeSegment(String value) { + if (pathEncoding == FileSystemPathEncoding.HASH) { + try { + var digest = MessageDigest.getInstance("SHA-256").digest(value.getBytes(StandardCharsets.UTF_8)); + return HexFormat.of().formatHex(digest); + } catch (NoSuchAlgorithmException e) { + throw new IllegalStateException("SHA-256 is unavailable", e); + } + } + return URLEncoder.encode(value, StandardCharsets.UTF_8).replace("+", "%20"); + } + + private void writeAtomically(Path path, String serializedPayload, PayloadOffloadContext context) { + Path temporary = null; + try { + Files.createDirectories(path.getParent()); + temporary = Files.createTempFile(path.getParent(), ".durable-payload-", ".tmp"); + Files.writeString(temporary, serializedPayload, StandardCharsets.UTF_8); + try { + Files.move(temporary, path, StandardCopyOption.ATOMIC_MOVE, StandardCopyOption.REPLACE_EXISTING); + } catch (AtomicMoveNotSupportedException e) { + Files.move(temporary, path, StandardCopyOption.REPLACE_EXISTING); + } + } catch (IOException e) { + throw new PayloadOffloadException( + "Failed to write filesystem payload for entity '" + context.entityId() + "'", e); + } finally { + if (temporary != null) { + try { + Files.deleteIfExists(temporary); + } catch (IOException ignored) { + // The target move already succeeded or the original failure is more useful. + } + } + } + } + + /** Builder for {@link FileSystemPayloadOffloader}. */ + public static final class Builder { + private final Path basePath; + private PayloadOffloadMode storageMode; + private FileSystemPathEncoding pathEncoding; + private PayloadPreviewGenerator previewGenerator; + + private Builder(Path basePath) { + this.basePath = Objects.requireNonNull(basePath, "basePath cannot be null"); + } + + public Builder storageMode(PayloadOffloadMode storageMode) { + this.storageMode = Objects.requireNonNull(storageMode, "storageMode cannot be null"); + return this; + } + + public Builder pathEncoding(FileSystemPathEncoding pathEncoding) { + this.pathEncoding = Objects.requireNonNull(pathEncoding, "pathEncoding cannot be null"); + return this; + } + + public Builder previewGenerator(PayloadPreviewGenerator previewGenerator) { + this.previewGenerator = previewGenerator; + return this; + } + + public FileSystemPayloadOffloader build() { + return new FileSystemPayloadOffloader(this); + } + } +} diff --git a/extra-filesystem-offloader/src/main/java/software/amazon/lambda/durable/extra/filesystem/PayloadOffloadMode.java b/extra-filesystem-offloader/src/main/java/software/amazon/lambda/durable/extra/filesystem/PayloadOffloadMode.java new file mode 100644 index 000000000..8321c0387 --- /dev/null +++ b/extra-filesystem-offloader/src/main/java/software/amazon/lambda/durable/extra/filesystem/PayloadOffloadMode.java @@ -0,0 +1,12 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.extra.filesystem; + +/** Controls when serialized payloads are written to the filesystem. */ +public enum PayloadOffloadMode { + /** Always write payloads to the configured filesystem. */ + ALWAYS, + + /** Keep small payloads inline and write only payloads near the checkpoint size limit. */ + OVERFLOW +} diff --git a/extra-filesystem-offloader/src/main/java/software/amazon/lambda/durable/extra/filesystem/PayloadPreviewGenerator.java b/extra-filesystem-offloader/src/main/java/software/amazon/lambda/durable/extra/filesystem/PayloadPreviewGenerator.java new file mode 100644 index 000000000..8887243d3 --- /dev/null +++ b/extra-filesystem-offloader/src/main/java/software/amazon/lambda/durable/extra/filesystem/PayloadPreviewGenerator.java @@ -0,0 +1,12 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.extra.filesystem; + +import java.util.Map; +import software.amazon.lambda.durable.offload.PayloadOffloadContext; + +/** Generates optional inline preview metadata for an externally stored serialized payload. */ +@FunctionalInterface +public interface PayloadPreviewGenerator { + Map generate(String serializedPayload, PayloadOffloadContext context); +} diff --git a/extra-filesystem-offloader/src/test/java/software/amazon/lambda/durable/extra/filesystem/FileSystemPayloadOffloaderTest.java b/extra-filesystem-offloader/src/test/java/software/amazon/lambda/durable/extra/filesystem/FileSystemPayloadOffloaderTest.java new file mode 100644 index 000000000..2a642e782 --- /dev/null +++ b/extra-filesystem-offloader/src/test/java/software/amazon/lambda/durable/extra/filesystem/FileSystemPayloadOffloaderTest.java @@ -0,0 +1,138 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.extra.filesystem; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.Map; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import software.amazon.lambda.durable.exception.PayloadOffloadException; +import software.amazon.lambda.durable.model.OperationIdentifier; +import software.amazon.lambda.durable.model.OperationSubType; +import software.amazon.lambda.durable.offload.OffloadedPayload; +import software.amazon.lambda.durable.offload.PayloadOffloadContext; +import software.amazon.lambda.durable.offload.PayloadStorageMode; +import software.amazon.lambda.durable.offload.SerDesPayloadKind; + +class FileSystemPayloadOffloaderTest { + @TempDir + Path temporaryDirectory; + + @Test + void alwaysModeWritesAndLoadsPayloadUsingReadablePath() { + var offloader = FileSystemPayloadOffloader.builder(temporaryDirectory).build(); + + var payload = offloader.offload("{\"value\":\"stored\"}", context()); + + assertEquals(PayloadStorageMode.REFERENCE, payload.mode()); + assertTrue(payload.reference().contains("test-function")); + assertTrue(payload.reference().contains("execution-name")); + assertTrue(Files.exists(Path.of(payload.reference()))); + assertEquals("{\"value\":\"stored\"}", offloader.load(payload, context())); + } + + @Test + void overflowModeKeepsSmallPayloadInlineAndOffloadsLargePayload() { + var offloader = FileSystemPayloadOffloader.builder(temporaryDirectory) + .storageMode(PayloadOffloadMode.OVERFLOW) + .build(); + + var inline = offloader.offload("small", context()); + var reference = offloader.offload("x".repeat(256 * 1024), context()); + + assertEquals(PayloadStorageMode.INLINE, inline.mode()); + assertEquals("small", inline.data()); + assertEquals(PayloadStorageMode.REFERENCE, reference.mode()); + } + + @Test + void hashModeUsesFixedLengthFilesystemSafeNames() { + var offloader = FileSystemPayloadOffloader.builder(temporaryDirectory) + .pathEncoding(FileSystemPathEncoding.HASH) + .build(); + + var payload = offloader.offload("stored", context()); + var path = Path.of(payload.reference()); + + assertEquals(69, path.getFileName().toString().length()); + assertEquals(64, path.getParent().getFileName().toString().length()); + } + + @Test + void previewMetadataIsCopiedIntoEnvelope() { + var offloader = FileSystemPayloadOffloader.builder(temporaryDirectory) + .previewGenerator((serialized, context) -> Map.of("entity", context.entityId())) + .build(); + + var payload = offloader.offload("stored", context()); + + assertEquals(context().entityId(), payload.preview().get("entity")); + } + + @Test + void repeatedWritesAtomicallyReplaceTheSameEntityFile() { + var offloader = FileSystemPayloadOffloader.builder(temporaryDirectory).build(); + + var first = offloader.offload("first", context()); + var second = offloader.offload("second", context()); + + assertEquals(first.reference(), second.reference()); + assertEquals("second", offloader.load(second, context())); + } + + @Test + void loadRejectsReferencesOutsideConfiguredBasePath() { + var offloader = FileSystemPayloadOffloader.builder(temporaryDirectory).build(); + var outside = temporaryDirectory.getParent().resolve("outside.json"); + var payload = OffloadedPayload.reference(outside.toString(), null); + + assertThrows(PayloadOffloadException.class, () -> offloader.load(payload, context())); + } + + @Test + void differentEntitiesUseDifferentFiles() { + var offloader = FileSystemPayloadOffloader.builder(temporaryDirectory).build(); + var first = offloader.offload("first", context()); + var secondContext = PayloadOffloadContext.forOperation( + context().durableExecutionArn(), + OperationIdentifier.of("op-2", "other", OperationSubType.STEP), + null, + SerDesPayloadKind.RESULT, + 1); + var second = offloader.offload("second", secondContext); + + assertNotEquals(first.reference(), second.reference()); + } + + @Test + void differentAttemptsUseDifferentFiles() { + var offloader = FileSystemPayloadOffloader.builder(temporaryDirectory).build(); + var first = offloader.offload("first", context()); + var secondContext = PayloadOffloadContext.forOperation( + context().durableExecutionArn(), + OperationIdentifier.of("op/1", "step", OperationSubType.STEP), + null, + SerDesPayloadKind.RESULT, + 2); + var second = offloader.offload("second", secondContext); + + assertNotEquals(first.reference(), second.reference()); + assertEquals("first", offloader.load(first, context())); + assertEquals("second", offloader.load(second, secondContext)); + } + + private static PayloadOffloadContext context() { + return PayloadOffloadContext.forOperation( + "arn:aws:lambda:us-east-1:123456789012:function:test-function:$LATEST/durable-execution/execution-name/invocation-id", + OperationIdentifier.of("op/1", "step", OperationSubType.STEP), + null, + SerDesPayloadKind.RESULT, + 1); + } +} diff --git a/pom.xml b/pom.xml index b224efbbd..68d970041 100644 --- a/pom.xml +++ b/pom.xml @@ -40,6 +40,7 @@ sdk + extra-filesystem-offloader sdk-testing sdk-integration-tests otel-plugin diff --git a/sdk-integration-tests/pom.xml b/sdk-integration-tests/pom.xml index 9d0459a07..e4a782bde 100644 --- a/sdk-integration-tests/pom.xml +++ b/sdk-integration-tests/pom.xml @@ -36,6 +36,12 @@ ${project.version} test + + software.amazon.lambda.durable + aws-durable-execution-sdk-java-extra-filesystem-offloader + ${project.version} + test + org.junit.jupiter junit-jupiter diff --git a/sdk-integration-tests/src/test/java/software/amazon/lambda/durable/PayloadOffloaderIntegrationTest.java b/sdk-integration-tests/src/test/java/software/amazon/lambda/durable/PayloadOffloaderIntegrationTest.java new file mode 100644 index 000000000..a8d9294ff --- /dev/null +++ b/sdk-integration-tests/src/test/java/software/amazon/lambda/durable/PayloadOffloaderIntegrationTest.java @@ -0,0 +1,115 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.time.Duration; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import software.amazon.lambda.durable.config.StepConfig; +import software.amazon.lambda.durable.extra.filesystem.FileSystemPayloadOffloader; +import software.amazon.lambda.durable.model.ExecutionStatus; +import software.amazon.lambda.durable.offload.PayloadOffloader; +import software.amazon.lambda.durable.retry.RetryStrategies; +import software.amazon.lambda.durable.testing.LocalDurableTestRunner; + +class PayloadOffloaderIntegrationTest { + @TempDir + Path payloadDirectory; + + @Test + void filesystemOffloaderReplaysStepAndRootOutput() throws IOException { + var stepExecutions = new AtomicInteger(); + var config = DurableConfig.builder() + .withPayloadOffloader( + FileSystemPayloadOffloader.builder(payloadDirectory).build()) + .build(); + var runner = LocalDurableTestRunner.create( + String.class, + (input, context) -> { + var future = context.stepAsync("offloaded-step", String.class, stepContext -> { + stepExecutions.incrementAndGet(); + return "stored-" + input; + }); + var first = future.get(); + var second = future.get(); + context.wait("replay-boundary", Duration.ofSeconds(1)); + return first + ":" + second; + }, + config); + + var result = runner.runUntilComplete("value"); + + assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus()); + assertEquals("stored-value:stored-value", result.getResult(String.class)); + assertEquals("stored-value", result.getOperation("offloaded-step").getStepResult(String.class)); + assertEquals(1, stepExecutions.get()); + try (var files = Files.walk(payloadDirectory)) { + assertTrue(files.filter(Files::isRegularFile).count() >= 2); + } + } + + @Test + void offloadedExceptionIsReconstructedAfterReplay() { + var config = DurableConfig.builder() + .withPayloadOffloader( + FileSystemPayloadOffloader.builder(payloadDirectory).build()) + .build(); + var runner = LocalDurableTestRunner.create( + String.class, + (input, context) -> { + try { + context.step( + "failing-step", + String.class, + stepContext -> { + throw new IllegalStateException("offloaded failure"); + }, + StepConfig.builder() + .retryStrategy(RetryStrategies.Presets.NO_RETRY) + .build()); + return "unreachable"; + } catch (IllegalStateException expected) { + context.wait("replay-after-failure", Duration.ofSeconds(1)); + return expected.getMessage(); + } + }, + config); + + var result = runner.runUntilComplete("value"); + + assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus()); + assertEquals("offloaded failure", result.getResult(String.class)); + } + + @Test + void operationCanDisableGlobalOffloader() { + var config = DurableConfig.builder() + .withPayloadOffloader( + FileSystemPayloadOffloader.builder(payloadDirectory).build()) + .build(); + var runner = LocalDurableTestRunner.create( + String.class, + (input, context) -> context.step( + "inline-step", + String.class, + stepContext -> "inline", + StepConfig.builder() + .payloadOffloader(PayloadOffloader.disabled()) + .build()), + config); + + var result = runner.run("value"); + + assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus()); + assertEquals( + "\"inline\"", + result.getOperation("inline-step").getStepDetails().result()); + } +} diff --git a/sdk-testing/src/main/java/software/amazon/lambda/durable/testing/LocalDurableTestRunner.java b/sdk-testing/src/main/java/software/amazon/lambda/durable/testing/LocalDurableTestRunner.java index f04e1c27b..a6f83add5 100644 --- a/sdk-testing/src/main/java/software/amazon/lambda/durable/testing/LocalDurableTestRunner.java +++ b/sdk-testing/src/main/java/software/amazon/lambda/durable/testing/LocalDurableTestRunner.java @@ -5,6 +5,7 @@ import com.amazonaws.services.lambda.runtime.Context; import java.time.Instant; import java.util.ArrayList; +import java.util.Arrays; import java.util.List; import java.util.UUID; import java.util.function.BiFunction; @@ -19,8 +20,13 @@ import software.amazon.lambda.durable.DurableHandler; import software.amazon.lambda.durable.TypeToken; import software.amazon.lambda.durable.execution.DurableExecutor; +import software.amazon.lambda.durable.execution.PayloadCodec; import software.amazon.lambda.durable.model.DurableExecutionInput; +import software.amazon.lambda.durable.model.DurableExecutionOutput; import software.amazon.lambda.durable.model.ExecutionStatus; +import software.amazon.lambda.durable.model.OperationSubType; +import software.amazon.lambda.durable.offload.PayloadOffloadContext; +import software.amazon.lambda.durable.offload.SerDesPayloadKind; import software.amazon.lambda.durable.plugin.DurableExecutionPlugin; import software.amazon.lambda.durable.serde.SerDes; import software.amazon.lambda.durable.testing.local.LocalMemoryExecutionClient; @@ -43,6 +49,11 @@ public class LocalDurableTestRunner { private final SerDes serDes; private final DurableConfig customerConfig; private final Instant executionStartTime = Instant.now(); + private final String executionName = UUID.randomUUID().toString(); + private final String invocationId = UUID.randomUUID().toString(); + private final String executionArn = String.format( + "arn:aws:lambda:us-east-1:123456789012:function:test:$LATEST/durable-execution/%s/%s", + executionName, invocationId); private LocalDurableTestRunner( TypeToken inputType, @@ -57,17 +68,22 @@ private LocalDurableTestRunner( // Create config that uses customer's configuration but overrides the client with in-memory storage if (customerConfig != null) { // Use customer's config but override the client with our in-memory implementation - this.customerConfig = DurableConfig.builder() + var builder = DurableConfig.builder() .withDurableExecutionClient(storage) .withSerDes(customerConfig.getSerDes()) + .withPayloadOffloadExecutorService(customerConfig.getPayloadOffloadExecutorService()) .withExecutorService(customerConfig.getExecutorService()) .withPollingStrategy(customerConfig.getPollingStrategy()) .withCheckpointDelay(customerConfig.getCheckpointDelay()) .withLoggerConfig(customerConfig.getLoggerConfig()) + .withDeserializeAfterSerialization(customerConfig.shouldDeserializeAfterSerialization()) // Temporary: remove along with the checkpointEmptyMap flag in a future major version. .withCheckpointEmptyMap(customerConfig.shouldCheckpointEmptyMap()) - .withPlugins(customerConfig.getPluginRunner().getPlugins().toArray(new DurableExecutionPlugin[0])) - .build(); + .withPlugins(customerConfig.getPluginRunner().getPlugins().toArray(new DurableExecutionPlugin[0])); + if (customerConfig.getPayloadOffloader() != null) { + builder.withPayloadOffloader(customerConfig.getPayloadOffloader()); + } + this.customerConfig = builder.build(); } else { // Fallback to default config with in-memory client this.customerConfig = @@ -241,8 +257,13 @@ public TestResult run(I input) { var durableInput = createDurableInput(input); var output = DurableExecutor.execute(durableInput, mockLambdaContext(), inputType, handler, customerConfig); + var codec = new PayloadCodec(customerConfig.getPayloadOffloadExecutorService()); - return storage.toTestResult(output, outputType, serDes); + return storage.toTestResult( + resolveOutput(output, durableInput, codec), + outputType, + serDes, + (operation, payload) -> resolveOperationPayload(operation, payload, codec)); } /** @@ -281,7 +302,11 @@ public void simulateFireAndForgetCheckpointLoss(String stepName) { /** Returns the {@link TestOperation} for the given operation name, or null if not found. */ public TestOperation getOperation(String name) { var op = storage.getOperationByName(name); - return op != null ? new TestOperation(op, serDes) : null; + if (op == null) { + return null; + } + var codec = new PayloadCodec(customerConfig.getPayloadOffloadExecutorService()); + return new TestOperation(op, List.of(), serDes, payload -> resolveOperationPayload(op, payload, codec)); } /** Get callback ID for a named callback operation. */ @@ -330,11 +355,6 @@ public void stopChainedInvoke(String name, ErrorObject error) { } private DurableExecutionInput createDurableInput(I input) { - var executionName = UUID.randomUUID().toString(); - var invocationId = UUID.randomUUID().toString(); - var executionArn = String.format( - "arn:aws:lambda:us-east-1:123456789012:function:test:$LATEST/durable-execution/%s/%s", - executionName, invocationId); var inputJson = serDes.serialize(input); var executionOp = Operation.builder() .id(invocationId) @@ -362,6 +382,49 @@ private DurableExecutionInput createDurableInput(I input) { updatedOperationIds); } + private DurableExecutionOutput resolveOutput( + DurableExecutionOutput output, DurableExecutionInput input, PayloadCodec codec) { + if (output.result() == null) { + return output; + } + var executionOperation = input.initialExecutionState().operations().stream() + .filter(operation -> operation.type() == OperationType.EXECUTION) + .findFirst() + .orElseThrow(); + var context = PayloadOffloadContext.forExecution( + input.durableExecutionArn(), + executionOperation.id(), + executionOperation.name(), + SerDesPayloadKind.OUTPUT); + return DurableExecutionOutput.success( + codec.resolveSerializedPayload(output.result(), customerConfig.getPayloadOffloader(), context)); + } + + private String resolveOperationPayload(Operation operation, String payload, PayloadCodec codec) { + var payloadKind = OperationSubType.WAIT_FOR_CONDITION.getValue().equals(operation.subType()) + ? SerDesPayloadKind.STATE + : SerDesPayloadKind.RESULT; + var attempt = operation.stepDetails() != null ? operation.stepDetails().attempt() : null; + var operationSubType = operation.subType() == null + ? null + : Arrays.stream(OperationSubType.values()) + .filter(value -> value.getValue().equals(operation.subType())) + .findFirst() + .orElse(null); + var context = new PayloadOffloadContext( + executionArn, + "operation/" + operation.id() + "/" + + payloadKind.name().toLowerCase().replace('_', '-'), + payloadKind, + operation.id(), + operation.name(), + operation.parentId(), + operation.type(), + operationSubType, + attempt); + return codec.resolveSerializedPayload(payload, customerConfig.getPayloadOffloader(), context); + } + private Context mockLambdaContext() { return null; // Minimal - tests don't need real Lambda context } diff --git a/sdk-testing/src/main/java/software/amazon/lambda/durable/testing/TestOperation.java b/sdk-testing/src/main/java/software/amazon/lambda/durable/testing/TestOperation.java index 31a28b988..171c8f350 100644 --- a/sdk-testing/src/main/java/software/amazon/lambda/durable/testing/TestOperation.java +++ b/sdk-testing/src/main/java/software/amazon/lambda/durable/testing/TestOperation.java @@ -5,6 +5,7 @@ import java.time.Duration; import java.time.Instant; import java.util.List; +import java.util.function.Function; import software.amazon.awssdk.services.lambda.model.CallbackDetails; import software.amazon.awssdk.services.lambda.model.ChainedInvokeDetails; import software.amazon.awssdk.services.lambda.model.ContextDetails; @@ -25,15 +26,23 @@ public class TestOperation { private final Operation operation; private final List events; private final SerDes serDes; + private final Function payloadResolver; public TestOperation(Operation operation, SerDes serDes) { this(operation, List.of(), serDes); } public TestOperation(Operation operation, List events, SerDes serDes) { + this(operation, events, serDes, Function.identity()); + } + + /** Creates an operation wrapper that resolves stored payloads before passing them to SerDes. */ + public TestOperation( + Operation operation, List events, SerDes serDes, Function payloadResolver) { this.operation = operation; this.events = events; this.serDes = serDes; + this.payloadResolver = payloadResolver; } /** Returns the raw history events associated with this operation. */ @@ -119,7 +128,7 @@ public T getStepResult(TypeToken type) { if (details == null || details.result() == null) { return null; } - return serDes.deserialize(details.result(), type); + return serDes.deserialize(payloadResolver.apply(details.result()), type); } /** Returns the step error, or null if the step succeeded or this is not a step operation. */ diff --git a/sdk-testing/src/main/java/software/amazon/lambda/durable/testing/local/LocalMemoryExecutionClient.java b/sdk-testing/src/main/java/software/amazon/lambda/durable/testing/local/LocalMemoryExecutionClient.java index 25f016cd9..1f7405893 100644 --- a/sdk-testing/src/main/java/software/amazon/lambda/durable/testing/local/LocalMemoryExecutionClient.java +++ b/sdk-testing/src/main/java/software/amazon/lambda/durable/testing/local/LocalMemoryExecutionClient.java @@ -12,6 +12,7 @@ import java.util.UUID; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.function.BiFunction; import software.amazon.awssdk.services.lambda.model.CheckpointDurableExecutionResponse; import software.amazon.awssdk.services.lambda.model.CheckpointUpdatedExecutionState; import software.amazon.awssdk.services.lambda.model.GetDurableExecutionStateResponse; @@ -131,9 +132,22 @@ public List getUpdatedOperationIdsSinceLastInvocation() { /** Build TestResult from current state. */ public TestResult toTestResult(DurableExecutionOutput output, TypeToken resultType, SerDes serDes) { + return toTestResult(output, resultType, serDes, (operation, payload) -> payload); + } + + /** Build TestResult from current state, resolving operation payloads before deserialization. */ + public TestResult toTestResult( + DurableExecutionOutput output, + TypeToken resultType, + SerDes serDes, + BiFunction payloadResolver) { var testOperations = existingOperations.values().stream() .filter(op -> op.type() != OperationType.EXECUTION) - .map(op -> new TestOperation(op, eventProcessor.getEventsForOperation(op.id()), serDes)) + .map(op -> new TestOperation( + op, + eventProcessor.getEventsForOperation(op.id()), + serDes, + payload -> payloadResolver.apply(op, payload))) .toList(); return new TestResult<>( output.status(), diff --git a/sdk/src/main/java/software/amazon/lambda/durable/DurableConfig.java b/sdk/src/main/java/software/amazon/lambda/durable/DurableConfig.java index 5101b9fda..77858b24a 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/DurableConfig.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/DurableConfig.java @@ -23,6 +23,7 @@ import software.amazon.lambda.durable.client.DurableExecutionClient; import software.amazon.lambda.durable.client.LambdaDurableFunctionsClient; import software.amazon.lambda.durable.logging.LoggerConfig; +import software.amazon.lambda.durable.offload.PayloadOffloader; import software.amazon.lambda.durable.plugin.DurableExecutionPlugin; import software.amazon.lambda.durable.plugin.PluginRunner; import software.amazon.lambda.durable.retry.PollingStrategies; @@ -92,9 +93,18 @@ public final class DurableConfig { return t; }); + private static final ExecutorService DEFAULT_PAYLOAD_THREAD_POOL = Executors.newCachedThreadPool(r -> { + Thread t = new Thread(r); + t.setName("durable-sdk-payload-" + t.getId()); + t.setDaemon(true); + return t; + }); + private final DurableExecutionClient durableExecutionClient; private final SerDes serDes; + private final PayloadOffloader payloadOffloader; private final ExecutorService executorService; + private final ExecutorService payloadOffloadExecutorService; private final LoggerConfig loggerConfig; private final PollingStrategy pollingStrategy; private final Duration checkpointDelay; @@ -107,8 +117,11 @@ private DurableConfig(Builder builder) { this.durableExecutionClient = Objects.requireNonNullElseGet( builder.durableExecutionClient, DurableConfig::createDefaultDurableExecutionClient); this.serDes = Objects.requireNonNullElseGet(builder.serDes, JacksonSerDes::new); + this.payloadOffloader = builder.payloadOffloader; this.executorService = Objects.requireNonNullElseGet(builder.executorService, DurableConfig::createDefaultExecutor); + this.payloadOffloadExecutorService = + Objects.requireNonNullElse(builder.payloadOffloadExecutorService, DEFAULT_PAYLOAD_THREAD_POOL); this.loggerConfig = Objects.requireNonNullElseGet(builder.loggerConfig, LoggerConfig::defaults); this.pollingStrategy = Objects.requireNonNullElse(builder.pollingStrategy, PollingStrategies.Presets.DEFAULT); this.checkpointDelay = Objects.requireNonNullElseGet(builder.checkpointDelay, () -> Duration.ofSeconds(0)); @@ -155,6 +168,11 @@ public SerDes getSerDes() { return serDes; } + /** Gets the globally configured payload offloader, or null when payload offloading is disabled. */ + public PayloadOffloader getPayloadOffloader() { + return payloadOffloader; + } + /** * Gets the configured ExecutorService. * @@ -164,6 +182,11 @@ public ExecutorService getExecutorService() { return executorService; } + /** Gets the executor used for blocking payload offload and load operations. */ + public ExecutorService getPayloadOffloadExecutorService() { + return payloadOffloadExecutorService; + } + /** * Gets the configured LoggerConfig. * @@ -235,6 +258,9 @@ public void validateConfiguration() { if (getExecutorService() == null) { throw new IllegalStateException("ExecutorService configuration failed"); } + if (getPayloadOffloadExecutorService() == null) { + throw new IllegalStateException("Payload offload ExecutorService configuration failed"); + } } /** @@ -315,7 +341,9 @@ private static ExecutorService createDefaultExecutor() { public static final class Builder { private DurableExecutionClient durableExecutionClient; private SerDes serDes; + private PayloadOffloader payloadOffloader; private ExecutorService executorService; + private ExecutorService payloadOffloadExecutorService; private LoggerConfig loggerConfig; private PollingStrategy pollingStrategy; private Duration checkpointDelay; @@ -381,6 +409,17 @@ public Builder withSerDes(SerDes serDes) { return this; } + /** + * Sets the global payload offloader applied after SerDes processing. + * + * @param payloadOffloader payload offloader + * @return this builder + */ + public Builder withPayloadOffloader(PayloadOffloader payloadOffloader) { + this.payloadOffloader = Objects.requireNonNull(payloadOffloader, "PayloadOffloader cannot be null"); + return this; + } + /** * Sets a custom ExecutorService for running user-defined operations. If not set, a default cached thread pool * will be created. @@ -396,6 +435,18 @@ public Builder withExecutorService(ExecutorService executorService) { return this; } + /** + * Sets the executor used for blocking payload storage operations. + * + * @param executorService payload offload executor + * @return this builder + */ + public Builder withPayloadOffloadExecutorService(ExecutorService executorService) { + this.payloadOffloadExecutorService = + Objects.requireNonNull(executorService, "Payload offload ExecutorService cannot be null"); + return this; + } + /** * Sets a custom LoggerConfig. If not set, defaults to suppressing replay logs. * diff --git a/sdk/src/main/java/software/amazon/lambda/durable/config/CallbackConfig.java b/sdk/src/main/java/software/amazon/lambda/durable/config/CallbackConfig.java index a71962aff..e40668703 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/config/CallbackConfig.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/config/CallbackConfig.java @@ -3,6 +3,7 @@ package software.amazon.lambda.durable.config; import java.time.Duration; +import software.amazon.lambda.durable.offload.PayloadOffloader; import software.amazon.lambda.durable.serde.SerDes; import software.amazon.lambda.durable.util.ParameterValidator; @@ -11,11 +12,13 @@ public class CallbackConfig { private final Duration timeout; private final Duration heartbeatTimeout; private final SerDes serDes; + private final PayloadOffloader payloadOffloader; private CallbackConfig(Builder builder) { this.timeout = builder.timeout; this.heartbeatTimeout = builder.heartbeatTimeout; this.serDes = builder.serDes; + this.payloadOffloader = builder.payloadOffloader; } /** @@ -41,6 +44,11 @@ public SerDes serDes() { return serDes; } + /** Returns the callback result offloader, or null to inherit the global offloader. */ + public PayloadOffloader payloadOffloader() { + return payloadOffloader; + } + /** Creates a new builder with default values. */ public static Builder builder() { return new Builder(null, null, null); @@ -48,7 +56,7 @@ public static Builder builder() { /** Creates a new builder pre-populated with this config's values. */ public Builder toBuilder() { - return new Builder(timeout, heartbeatTimeout, serDes); + return new Builder(timeout, heartbeatTimeout, serDes).payloadOffloader(payloadOffloader); } /** Builder for {@link CallbackConfig}. */ @@ -56,6 +64,7 @@ public static class Builder { private Duration timeout; private Duration heartbeatTimeout; private SerDes serDes; + private PayloadOffloader payloadOffloader; public Builder(Duration timeout, Duration heartbeatTimeout, SerDes serDes) { this.timeout = timeout; @@ -102,6 +111,12 @@ public Builder serDes(SerDes serDes) { return this; } + /** Sets the payload offloader for the callback result. */ + public Builder payloadOffloader(PayloadOffloader payloadOffloader) { + this.payloadOffloader = payloadOffloader; + return this; + } + /** Builds the {@link CallbackConfig} instance. */ public CallbackConfig build() { return new CallbackConfig(this); diff --git a/sdk/src/main/java/software/amazon/lambda/durable/config/InvokeConfig.java b/sdk/src/main/java/software/amazon/lambda/durable/config/InvokeConfig.java index e9dc7af24..102102509 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/config/InvokeConfig.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/config/InvokeConfig.java @@ -2,6 +2,7 @@ // SPDX-License-Identifier: Apache-2.0 package software.amazon.lambda.durable.config; +import software.amazon.lambda.durable.offload.PayloadOffloader; import software.amazon.lambda.durable.serde.SerDes; /** @@ -12,11 +13,13 @@ public class InvokeConfig { private final SerDes payloadSerDes; private final SerDes resultSerDes; + private final PayloadOffloader payloadOffloader; private final String tenantId; public InvokeConfig(Builder builder) { this.payloadSerDes = builder.payloadSerDes; this.resultSerDes = builder.resultSerDes; + this.payloadOffloader = builder.payloadOffloader; this.tenantId = builder.tenantId; } @@ -28,6 +31,11 @@ public SerDes serDes() { return this.resultSerDes; } + /** Returns the offloader used for both the invoke payload and result, or null to inherit the global offloader. */ + public PayloadOffloader payloadOffloader() { + return payloadOffloader; + } + public String tenantId() { return tenantId; } @@ -37,13 +45,14 @@ public static Builder builder() { } public Builder toBuilder() { - return new Builder(payloadSerDes, resultSerDes, tenantId); + return new Builder(payloadSerDes, resultSerDes, tenantId).payloadOffloader(payloadOffloader); } /** Builder for creating InvokeConfig instances. */ public static class Builder { private SerDes payloadSerDes; private SerDes resultSerDes; + private PayloadOffloader payloadOffloader; private String tenantId; private Builder(SerDes payloadSerDes, SerDes resultSerDes, String tenantId) { @@ -96,6 +105,15 @@ public Builder serDes(SerDes resultSerDes) { return this; } + /** + * Sets the offloader for both the invoke payload and result. Use {@link PayloadOffloader#disabled()} to force + * inline storage when a global offloader is configured. + */ + public Builder payloadOffloader(PayloadOffloader payloadOffloader) { + this.payloadOffloader = payloadOffloader; + return this; + } + /** * Builds the InvokeConfig instance. * diff --git a/sdk/src/main/java/software/amazon/lambda/durable/config/MapConfig.java b/sdk/src/main/java/software/amazon/lambda/durable/config/MapConfig.java index 78bdcc293..5894b1a9d 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/config/MapConfig.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/config/MapConfig.java @@ -4,6 +4,7 @@ import java.util.Objects; import java.util.function.BiFunction; +import software.amazon.lambda.durable.offload.PayloadOffloader; import software.amazon.lambda.durable.serde.SerDes; /** @@ -15,6 +16,7 @@ public class MapConfig { private final Integer maxConcurrency; private final CompletionConfig completionConfig; private final SerDes serDes; + private final PayloadOffloader payloadOffloader; private final NestingType nestingType; private final BiFunction itemNamer; @@ -23,6 +25,7 @@ private MapConfig(Builder builder) { this.completionConfig = Objects.requireNonNullElse(builder.completionConfig, CompletionConfig.allCompleted()); this.nestingType = Objects.requireNonNullElse(builder.nestingType, NestingType.NESTED); this.serDes = builder.serDes; + this.payloadOffloader = builder.payloadOffloader; this.itemNamer = builder.itemNamer; if (itemNamer != null && nestingType == NestingType.FLAT) { throw new IllegalArgumentException("itemNamer is not supported with FLAT map nesting"); @@ -44,6 +47,11 @@ public SerDes serDes() { return serDes; } + /** @return the map and iteration result offloader, or null to inherit the global offloader */ + public PayloadOffloader payloadOffloader() { + return payloadOffloader; + } + /** @return nesting type, defaults to {@link NestingType#NESTED} */ public NestingType nestingType() { return nestingType; @@ -70,6 +78,7 @@ public Builder toBuilder() { .maxConcurrency(maxConcurrency) .completionConfig(completionConfig) .serDes(serDes) + .payloadOffloader(payloadOffloader) .nestingType(nestingType) .itemNamer(itemNamer); } @@ -80,6 +89,7 @@ public static class Builder { private Integer maxConcurrency; private CompletionConfig completionConfig; private SerDes serDes; + private PayloadOffloader payloadOffloader; private BiFunction itemNamer; private Builder() {} @@ -114,6 +124,12 @@ public Builder serDes(SerDes serDes) { return this; } + /** Sets the payload offloader for map iteration and aggregate results. */ + public Builder payloadOffloader(PayloadOffloader payloadOffloader) { + this.payloadOffloader = payloadOffloader; + return this; + } + /** * Sets the nesting type for the map operation. * diff --git a/sdk/src/main/java/software/amazon/lambda/durable/config/ParallelBranchConfig.java b/sdk/src/main/java/software/amazon/lambda/durable/config/ParallelBranchConfig.java index 689f9aa54..8e4316118 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/config/ParallelBranchConfig.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/config/ParallelBranchConfig.java @@ -2,6 +2,7 @@ // SPDX-License-Identifier: Apache-2.0 package software.amazon.lambda.durable.config; +import software.amazon.lambda.durable.offload.PayloadOffloader; import software.amazon.lambda.durable.serde.SerDes; /** @@ -11,9 +12,11 @@ */ public class ParallelBranchConfig { private final SerDes serDes; + private final PayloadOffloader payloadOffloader; private ParallelBranchConfig(Builder builder) { this.serDes = builder.serDes; + this.payloadOffloader = builder.payloadOffloader; } /** Returns the custom serializer for this step, or null if not specified (uses default SerDes). */ @@ -21,8 +24,13 @@ public SerDes serDes() { return serDes; } + /** Returns the branch result offloader, or null to inherit the global offloader. */ + public PayloadOffloader payloadOffloader() { + return payloadOffloader; + } + public Builder toBuilder() { - return new Builder(serDes); + return new Builder(serDes).payloadOffloader(payloadOffloader); } /** @@ -37,6 +45,7 @@ public static Builder builder() { /** Builder for creating StepConfig instances. */ public static class Builder { private SerDes serDes; + private PayloadOffloader payloadOffloader; public Builder(SerDes serDes) { this.serDes = serDes; @@ -57,6 +66,12 @@ public Builder serDes(SerDes serDes) { return this; } + /** Sets the payload offloader for the parallel branch result. */ + public Builder payloadOffloader(PayloadOffloader payloadOffloader) { + this.payloadOffloader = payloadOffloader; + return this; + } + /** * Builds the ParallelBranchConfig instance. * diff --git a/sdk/src/main/java/software/amazon/lambda/durable/config/ParallelConfig.java b/sdk/src/main/java/software/amazon/lambda/durable/config/ParallelConfig.java index 863d36972..d504037bd 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/config/ParallelConfig.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/config/ParallelConfig.java @@ -3,6 +3,7 @@ package software.amazon.lambda.durable.config; import java.util.Objects; +import software.amazon.lambda.durable.offload.PayloadOffloader; /** * Configuration options for parallel operations in durable executions. @@ -14,11 +15,13 @@ public class ParallelConfig { private final int maxConcurrency; private final CompletionConfig completionConfig; private final NestingType nestingType; + private final PayloadOffloader payloadOffloader; private ParallelConfig(Builder builder) { this.maxConcurrency = Objects.requireNonNullElse(builder.maxConcurrency, Integer.MAX_VALUE); this.completionConfig = Objects.requireNonNullElseGet(builder.completionConfig, CompletionConfig::allCompleted); this.nestingType = Objects.requireNonNullElse(builder.nestingType, NestingType.NESTED); + this.payloadOffloader = builder.payloadOffloader; } /** @return the maximum number of branches running simultaneously, or -1 for unlimited */ @@ -36,6 +39,11 @@ public NestingType nestingType() { return nestingType; } + /** @return the aggregate result offloader, or null to inherit the global offloader */ + public PayloadOffloader payloadOffloader() { + return payloadOffloader; + } + /** * Creates a new builder for ParallelConfig. * @@ -49,7 +57,8 @@ public Builder toBuilder() { return new Builder() .maxConcurrency(maxConcurrency) .completionConfig(completionConfig) - .nestingType(nestingType); + .nestingType(nestingType) + .payloadOffloader(payloadOffloader); } /** Builder for creating ParallelConfig instances. */ @@ -57,6 +66,7 @@ public static class Builder { private Integer maxConcurrency; private CompletionConfig completionConfig; private NestingType nestingType; + private PayloadOffloader payloadOffloader; private Builder() {} @@ -101,6 +111,12 @@ public Builder nestingType(NestingType nestingType) { return this; } + /** Sets the payload offloader for the aggregate parallel result. */ + public Builder payloadOffloader(PayloadOffloader payloadOffloader) { + this.payloadOffloader = payloadOffloader; + return this; + } + /** * Builds the ParallelConfig instance. * diff --git a/sdk/src/main/java/software/amazon/lambda/durable/config/RunInChildContextConfig.java b/sdk/src/main/java/software/amazon/lambda/durable/config/RunInChildContextConfig.java index 93fde31e8..092f21fa7 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/config/RunInChildContextConfig.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/config/RunInChildContextConfig.java @@ -3,6 +3,7 @@ package software.amazon.lambda.durable.config; import java.util.Objects; +import software.amazon.lambda.durable.offload.PayloadOffloader; import software.amazon.lambda.durable.serde.SerDes; /** @@ -12,10 +13,12 @@ */ public class RunInChildContextConfig { private final SerDes serDes; + private final PayloadOffloader payloadOffloader; private final Boolean isVirtual; private RunInChildContextConfig(Builder builder) { this.serDes = builder.serDes; + this.payloadOffloader = builder.payloadOffloader; this.isVirtual = Objects.requireNonNullElse(builder.isVirtual, false); } @@ -27,13 +30,18 @@ public SerDes serDes() { return serDes; } + /** Returns the child context result offloader, or null to inherit the global offloader. */ + public PayloadOffloader payloadOffloader() { + return payloadOffloader; + } + /** Returns true if the context operation will not be checkpointed, false otherwise. */ public Boolean isVirtual() { return isVirtual; } public Builder toBuilder() { - return new Builder().serDes(serDes).isVirtual(isVirtual); + return new Builder().serDes(serDes).payloadOffloader(payloadOffloader).isVirtual(isVirtual); } /** @@ -48,6 +56,7 @@ public static Builder builder() { /** Builder for creating StepConfig instances. */ public static class Builder { private SerDes serDes; + private PayloadOffloader payloadOffloader; private Boolean isVirtual; private Builder() {} @@ -67,6 +76,12 @@ public Builder serDes(SerDes serDes) { return this; } + /** Sets the payload offloader for the child context result and exception. */ + public Builder payloadOffloader(PayloadOffloader payloadOffloader) { + this.payloadOffloader = payloadOffloader; + return this; + } + /** * Sets whether the context is virtual (not checkpointed) or not. * diff --git a/sdk/src/main/java/software/amazon/lambda/durable/config/StepConfig.java b/sdk/src/main/java/software/amazon/lambda/durable/config/StepConfig.java index 92a90a643..49b53f771 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/config/StepConfig.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/config/StepConfig.java @@ -2,6 +2,7 @@ // SPDX-License-Identifier: Apache-2.0 package software.amazon.lambda.durable.config; +import software.amazon.lambda.durable.offload.PayloadOffloader; import software.amazon.lambda.durable.retry.RetryStrategies; import software.amazon.lambda.durable.retry.RetryStrategy; import software.amazon.lambda.durable.serde.SerDes; @@ -16,11 +17,13 @@ public class StepConfig { private final RetryStrategy retryStrategy; private final StepSemantics semanticsPerRetry; private final SerDes serDes; + private final PayloadOffloader payloadOffloader; private StepConfig(Builder builder) { this.retryStrategy = builder.retryStrategy; this.semanticsPerRetry = builder.semanticsPerRetry; this.serDes = builder.serDes; + this.payloadOffloader = builder.payloadOffloader; } /** Returns the retry strategy for this step, or the default strategy if not specified. */ @@ -38,8 +41,13 @@ public SerDes serDes() { return serDes; } + /** Returns the operation payload offloader, or null to inherit the handler configuration. */ + public PayloadOffloader payloadOffloader() { + return payloadOffloader; + } + public Builder toBuilder() { - return new Builder(retryStrategy, semanticsPerRetry, serDes); + return new Builder(retryStrategy, semanticsPerRetry, serDes).payloadOffloader(payloadOffloader); } /** @@ -56,6 +64,7 @@ public static class Builder { private RetryStrategy retryStrategy; private StepSemantics semanticsPerRetry; private SerDes serDes; + private PayloadOffloader payloadOffloader; public Builder(RetryStrategy retryStrategy, StepSemantics semanticsPerRetry, SerDes serDes) { this.retryStrategy = retryStrategy; @@ -100,6 +109,15 @@ public Builder serDes(SerDes serDes) { return this; } + /** + * Sets the payload offloader for this step. Use {@link PayloadOffloader#disabled()} to force inline storage + * when a global offloader is configured. + */ + public Builder payloadOffloader(PayloadOffloader payloadOffloader) { + this.payloadOffloader = payloadOffloader; + return this; + } + /** * Builds the StepConfig instance. * diff --git a/sdk/src/main/java/software/amazon/lambda/durable/config/WaitForConditionConfig.java b/sdk/src/main/java/software/amazon/lambda/durable/config/WaitForConditionConfig.java index 1561199f9..18096fc14 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/config/WaitForConditionConfig.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/config/WaitForConditionConfig.java @@ -2,6 +2,7 @@ // SPDX-License-Identifier: Apache-2.0 package software.amazon.lambda.durable.config; +import software.amazon.lambda.durable.offload.PayloadOffloader; import software.amazon.lambda.durable.retry.WaitForConditionWaitStrategy; import software.amazon.lambda.durable.retry.WaitStrategies; import software.amazon.lambda.durable.serde.SerDes; @@ -16,11 +17,13 @@ public class WaitForConditionConfig { private final WaitForConditionWaitStrategy waitStrategy; private final SerDes serDes; + private final PayloadOffloader payloadOffloader; private final T initialState; private WaitForConditionConfig(Builder builder) { this.waitStrategy = builder.waitStrategy; this.serDes = builder.serDes; + this.payloadOffloader = builder.payloadOffloader; this.initialState = builder.initialState; } @@ -37,6 +40,11 @@ public SerDes serDes() { return serDes; } + /** Returns the state offloader, or null to inherit the global offloader. */ + public PayloadOffloader payloadOffloader() { + return payloadOffloader; + } + /** Returns the initial state object, or null if not specified. */ public T initialState() { return initialState; @@ -52,6 +60,7 @@ public Builder toBuilder() { var b = new Builder(); b.waitStrategy = this.waitStrategy; b.serDes = this.serDes; + b.payloadOffloader = this.payloadOffloader; b.initialState = this.initialState; return b; } @@ -69,6 +78,7 @@ public static Builder builder() { public static class Builder { private WaitForConditionWaitStrategy waitStrategy; private SerDes serDes; + private PayloadOffloader payloadOffloader; private T initialState; private Builder() {} @@ -100,6 +110,12 @@ public Builder serDes(SerDes serDes) { return this; } + /** Sets the payload offloader for checkpointed condition state. */ + public Builder payloadOffloader(PayloadOffloader payloadOffloader) { + this.payloadOffloader = payloadOffloader; + return this; + } + /** * Sets the initial state for the waitForCondition operation. The initial state will be null if it's not set. * diff --git a/sdk/src/main/java/software/amazon/lambda/durable/context/DurableContextImpl.java b/sdk/src/main/java/software/amazon/lambda/durable/context/DurableContextImpl.java index 0c79165ec..2c28b9e28 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/context/DurableContextImpl.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/context/DurableContextImpl.java @@ -140,6 +140,11 @@ public DurableFuture stepAsync( if (config.serDes() == null) { config = config.toBuilder().serDes(getDurableConfig().getSerDes()).build(); } + if (config.payloadOffloader() == null) { + config = config.toBuilder() + .payloadOffloader(getDurableConfig().getPayloadOffloader()) + .build(); + } var operationId = nextOperationId(); // Create and start step operation with TypeToken @@ -181,6 +186,11 @@ public DurableFuture invokeAsync( .payloadSerDes(getDurableConfig().getSerDes()) .build(); } + if (config.payloadOffloader() == null) { + config = config.toBuilder() + .payloadOffloader(getDurableConfig().getPayloadOffloader()) + .build(); + } var operationId = nextOperationId(); // Create and start invoke operation @@ -202,6 +212,11 @@ public DurableCallbackFuture createCallback(String name, TypeToken res if (config.serDes() == null) { config = config.toBuilder().serDes(getDurableConfig().getSerDes()).build(); } + if (config.payloadOffloader() == null) { + config = config.toBuilder() + .payloadOffloader(getDurableConfig().getPayloadOffloader()) + .build(); + } var operationId = nextOperationId(); var operation = new CallbackOperation<>( @@ -242,6 +257,11 @@ private DurableFuture runInChildContextAsync( if (config.serDes() == null) { config = config.toBuilder().serDes(getDurableConfig().getSerDes()).build(); } + if (config.payloadOffloader() == null) { + config = config.toBuilder() + .payloadOffloader(getDurableConfig().getPayloadOffloader()) + .build(); + } var operationId = nextOperationId(); @@ -265,6 +285,11 @@ public DurableFuture> mapAsync( if (config.serDes() == null) { config = config.toBuilder().serDes(getDurableConfig().getSerDes()).build(); } + if (config.payloadOffloader() == null) { + config = config.toBuilder() + .payloadOffloader(getDurableConfig().getPayloadOffloader()) + .build(); + } // Convert to List for deterministic index-based access var itemList = List.copyOf(items); @@ -286,6 +311,11 @@ public DurableFuture> mapAsync( @Override public ParallelDurableFuture parallel(String name, ParallelConfig config) { Objects.requireNonNull(config, "config cannot be null"); + if (config.payloadOffloader() == null) { + config = config.toBuilder() + .payloadOffloader(getDurableConfig().getPayloadOffloader()) + .build(); + } var operationId = nextOperationId(); var parallelOp = new ParallelOperation( @@ -357,6 +387,11 @@ public DurableFuture waitForConditionAsync( if (config.serDes() == null) { config = config.toBuilder().serDes(getDurableConfig().getSerDes()).build(); } + if (config.payloadOffloader() == null) { + config = config.toBuilder() + .payloadOffloader(getDurableConfig().getPayloadOffloader()) + .build(); + } var operationId = nextOperationId(); var operation = new WaitForConditionOperation<>( diff --git a/sdk/src/main/java/software/amazon/lambda/durable/exception/PayloadOffloadException.java b/sdk/src/main/java/software/amazon/lambda/durable/exception/PayloadOffloadException.java new file mode 100644 index 000000000..1a94ddfd6 --- /dev/null +++ b/sdk/src/main/java/software/amazon/lambda/durable/exception/PayloadOffloadException.java @@ -0,0 +1,14 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.exception; + +/** Thrown when a serialized payload cannot be stored in or loaded from external storage. */ +public class PayloadOffloadException extends DurableExecutionException { + public PayloadOffloadException(String message, Throwable cause) { + super(message, cause); + } + + public PayloadOffloadException(String message) { + super(message); + } +} diff --git a/sdk/src/main/java/software/amazon/lambda/durable/execution/DurableExecutor.java b/sdk/src/main/java/software/amazon/lambda/durable/execution/DurableExecutor.java index 649c7a600..9aae7345d 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/execution/DurableExecutor.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/execution/DurableExecutor.java @@ -12,7 +12,6 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import software.amazon.awssdk.services.lambda.model.ErrorObject; -import software.amazon.awssdk.services.lambda.model.Operation; import software.amazon.awssdk.services.lambda.model.OperationAction; import software.amazon.awssdk.services.lambda.model.OperationType; import software.amazon.awssdk.services.lambda.model.OperationUpdate; @@ -26,11 +25,12 @@ import software.amazon.lambda.durable.logging.DurableLogger; import software.amazon.lambda.durable.model.DurableExecutionInput; import software.amazon.lambda.durable.model.DurableExecutionOutput; +import software.amazon.lambda.durable.offload.PayloadOffloadContext; +import software.amazon.lambda.durable.offload.SerDesPayloadKind; import software.amazon.lambda.durable.plugin.InvocationEndInfo; import software.amazon.lambda.durable.plugin.InvocationInfo; import software.amazon.lambda.durable.plugin.InvocationStatus; import software.amazon.lambda.durable.plugin.PluginRunner; -import software.amazon.lambda.durable.serde.SerDes; import software.amazon.lambda.durable.util.ExceptionHelper; /** @@ -76,8 +76,7 @@ public static DurableExecutionOutput execute( I userInput = null; Throwable inputFailure = null; try { - userInput = extractUserInput( - executionManager.getExecutionOperation(), config.getSerDes(), inputType); + userInput = extractUserInput(executionManager, config, inputType); } catch (Throwable t) { inputFailure = t; } @@ -159,11 +158,18 @@ public static DurableExecutionOutput execute( cause, pluginExecutionInput.get(), null); - return DurableExecutionOutput.failure(buildErrorObject(cause, config.getSerDes())); + return DurableExecutionOutput.failure( + buildErrorObject(cause, executionManager, config)); } // user handler complete successfully logger.debug("Execution completed"); - var outputPayload = config.getSerDes().serialize(result); + var outputPayload = executionManager + .getPayloadCodec() + .serialize( + result, + config.getSerDes(), + config.getPayloadOffloader(), + executionContext(executionManager, SerDesPayloadKind.OUTPUT)); var output = DurableExecutionOutput.success(handleLargePayload(executionManager, outputPayload)); fireOnInvocationEnd( @@ -228,7 +234,7 @@ private static String handleLargePayload(ExecutionManager executionManager, Stri return outputPayload; } - private static ErrorObject buildErrorObject(Throwable e, SerDes serDes) { + private static ErrorObject buildErrorObject(Throwable e, ExecutionManager executionManager, DurableConfig config) { // exceptions thrown from operations, e.g. Step if (e instanceof DurableOperationException durableOperationException) { return durableOperationException.getErrorObject(); @@ -237,16 +243,43 @@ private static ErrorObject buildErrorObject(Throwable e, SerDes serDes) { return unrecoverableDurableExecutionException.getErrorObject(); } // exceptions thrown from non-operation code - return ExceptionHelper.buildErrorObject(e, serDes); + return ErrorObject.builder() + .errorType(e.getClass().getName()) + .errorMessage(e.getMessage()) + .errorData(executionManager + .getPayloadCodec() + .serialize( + e, + config.getSerDes(), + config.getPayloadOffloader(), + executionContext(executionManager, SerDesPayloadKind.EXCEPTION))) + .stackTrace(ExceptionHelper.serializeStackTrace(e.getStackTrace())) + .build(); } - private static I extractUserInput(Operation executionOp, SerDes serDes, TypeToken inputType) { + private static I extractUserInput( + ExecutionManager executionManager, DurableConfig config, TypeToken inputType) { + var executionOp = executionManager.getExecutionOperation(); if (executionOp.executionDetails() == null) { throw new IllegalDurableOperationException("EXECUTION operation missing executionDetails"); } var inputPayload = executionOp.executionDetails().inputPayload(); - return serDes.deserialize(inputPayload, inputType); + return executionManager + .getPayloadCodec() + .deserialize( + inputPayload, + inputType, + config.getSerDes(), + config.getPayloadOffloader(), + executionContext(executionManager, SerDesPayloadKind.INPUT)); + } + + private static PayloadOffloadContext executionContext( + ExecutionManager executionManager, SerDesPayloadKind payloadKind) { + var operation = executionManager.getExecutionOperation(); + return PayloadOffloadContext.forExecution( + executionManager.getDurableExecutionArn(), operation.id(), operation.name(), payloadKind); } /** diff --git a/sdk/src/main/java/software/amazon/lambda/durable/execution/ExecutionManager.java b/sdk/src/main/java/software/amazon/lambda/durable/execution/ExecutionManager.java index 1c45cb0d6..7d7458d73 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/execution/ExecutionManager.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/execution/ExecutionManager.java @@ -62,6 +62,7 @@ public class ExecutionManager implements SafeCloseable { private final Context lambdaContext; private final AtomicReference executionMode; private final DurableConfig durableConfig; + private final PayloadCodec payloadCodec; private final Set updatedOperationIdsSinceLastInvocation; // ===== Thread Coordination ===== @@ -75,6 +76,7 @@ public class ExecutionManager implements SafeCloseable { public ExecutionManager(DurableExecutionInput input, DurableConfig config, Context lambdaContext) { durableConfig = config; + payloadCodec = new PayloadCodec(config.getPayloadOffloadExecutorService()); this.durableExecutionArn = input.durableExecutionArn(); this.lambdaContext = lambdaContext; @@ -205,6 +207,11 @@ public Operation getExecutionOperation() { return executionOp; } + /** Returns the invocation-scoped payload serialization and offload pipeline. */ + public PayloadCodec getPayloadCodec() { + return payloadCodec; + } + /** * Checks whether there are any cached operations for the given parent context ID. Used to initialize per-context * replay state — a context starts in replay mode if the ExecutionManager has cached operations belonging to it. @@ -322,6 +329,7 @@ public void close() { validateRunningThreads(); checkpointManager.shutdown(); + payloadCodec.clear(); } private void validateRunningThreads() { diff --git a/sdk/src/main/java/software/amazon/lambda/durable/execution/PayloadCodec.java b/sdk/src/main/java/software/amazon/lambda/durable/execution/PayloadCodec.java new file mode 100644 index 000000000..de2bcf913 --- /dev/null +++ b/sdk/src/main/java/software/amazon/lambda/durable/execution/PayloadCodec.java @@ -0,0 +1,204 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.execution; + +import java.nio.charset.StandardCharsets; +import java.security.MessageDigest; +import java.security.NoSuchAlgorithmException; +import java.util.HexFormat; +import java.util.Map; +import java.util.Objects; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutorService; +import java.util.function.Supplier; +import software.amazon.lambda.durable.TypeToken; +import software.amazon.lambda.durable.exception.PayloadOffloadException; +import software.amazon.lambda.durable.exception.SerDesException; +import software.amazon.lambda.durable.offload.OffloadedPayload; +import software.amazon.lambda.durable.offload.PayloadOffloadContext; +import software.amazon.lambda.durable.offload.PayloadOffloader; +import software.amazon.lambda.durable.offload.PayloadOffloaders; +import software.amazon.lambda.durable.offload.PayloadStorageMode; +import software.amazon.lambda.durable.offload.SerDesPayloadKind; +import software.amazon.lambda.durable.serde.JacksonSerDes; +import software.amazon.lambda.durable.serde.SerDes; +import software.amazon.lambda.durable.util.ExceptionHelper; + +/** + * Invocation-scoped pipeline that composes object serialization, payload offloading, loading, and caching. + * + *

Legacy raw serialized strings remain readable. New offloaded values use a reserved, versioned prefix so arbitrary + * JSON payloads cannot be mistaken for SDK envelopes. + */ +public final class PayloadCodec { + private static final String ENVELOPE_PREFIX = "@aws-durable-payload:v1:"; + private static final Object NULL_VALUE = new Object(); + private static final TypeToken OFFLOADED_PAYLOAD_TYPE = TypeToken.get(OffloadedPayload.class); + + private final ExecutorService executorService; + private final SerDes envelopeSerDes = new JacksonSerDes(); + private final Map loadedPayloadCache = new ConcurrentHashMap<>(); + private final Map deserializedPayloadCache = new ConcurrentHashMap<>(); + + public PayloadCodec(ExecutorService executorService) { + this.executorService = Objects.requireNonNull(executorService, "executorService cannot be null"); + } + + /** Returns whether a checkpoint value uses the SDK payload offload envelope. */ + public static boolean isOffloadEnvelope(String checkpointPayload) { + return checkpointPayload != null && checkpointPayload.startsWith(ENVELOPE_PREFIX); + } + + /** Serializes and optionally offloads a value. */ + public String serialize(Object value, SerDes serDes, PayloadOffloader offloader, PayloadOffloadContext context) { + var serialized = serDes.serialize(value); + var effectiveOffloader = effectiveOffloader(offloader); + if (effectiveOffloader == null) { + return serialized; + } + + var payload = runOffloadTask(() -> effectiveOffloader.offload(serialized, context), "store", context); + if (payload == null) { + throw new PayloadOffloadException("Payload offloader returned null for " + describe(context)); + } + var checkpointPayload = ENVELOPE_PREFIX + envelopeSerDes.serialize(payload); + loadedPayloadCache.put(payloadCacheKey(context, checkpointPayload), serialized); + return checkpointPayload; + } + + /** Deserializes a raw legacy payload or an SDK offload envelope. */ + @SuppressWarnings("unchecked") + public T deserialize( + String checkpointPayload, + TypeToken typeToken, + SerDes serDes, + PayloadOffloader offloader, + PayloadOffloadContext context) { + var serialized = resolve(checkpointPayload, offloader, context); + var key = new DeserializedCacheKey( + context.durableExecutionArn(), + context.entityId(), + context.payloadKind(), + context.attempt(), + typeToken.getType().getTypeName(), + hash(serialized)); + var cached = deserializedPayloadCache.get(key); + if (cached != null) { + return cached == NULL_VALUE ? null : (T) cached; + } + + var deserialized = serDes.deserialize(serialized, typeToken); + deserializedPayloadCache.putIfAbsent(key, deserialized == null ? NULL_VALUE : deserialized); + return deserialized; + } + + /** Resolves an SDK envelope to the serialized text produced by SerDes without deserializing the object. */ + public String resolveSerializedPayload( + String checkpointPayload, PayloadOffloader offloader, PayloadOffloadContext context) { + return resolve(checkpointPayload, offloader, context); + } + + /** Clears invocation-scoped payload caches. */ + public void clear() { + loadedPayloadCache.clear(); + deserializedPayloadCache.clear(); + } + + private String resolve(String checkpointPayload, PayloadOffloader offloader, PayloadOffloadContext context) { + if (checkpointPayload == null || !checkpointPayload.startsWith(ENVELOPE_PREFIX)) { + return checkpointPayload; + } + + var key = payloadCacheKey(context, checkpointPayload); + var cached = loadedPayloadCache.get(key); + if (cached != null) { + return cached; + } + + OffloadedPayload payload; + try { + payload = envelopeSerDes.deserialize( + checkpointPayload.substring(ENVELOPE_PREFIX.length()), OFFLOADED_PAYLOAD_TYPE); + } catch (SerDesException e) { + throw new PayloadOffloadException("Invalid payload offload envelope for " + describe(context), e); + } + + final String serialized; + if (payload.mode() == PayloadStorageMode.INLINE) { + serialized = payload.data(); + } else { + var effectiveOffloader = effectiveOffloader(offloader); + if (effectiveOffloader == null) { + throw new PayloadOffloadException( + "Payload uses external storage but no payload offloader is configured for " + + describe(context)); + } + serialized = runOffloadTask(() -> effectiveOffloader.load(payload, context), "load", context); + if (serialized == null) { + throw new PayloadOffloadException("Payload offloader returned null while loading " + describe(context)); + } + } + loadedPayloadCache.putIfAbsent(key, serialized); + return serialized; + } + + private T runOffloadTask(Supplier task, String action, PayloadOffloadContext context) { + try { + return CompletableFuture.supplyAsync(task, executorService).join(); + } catch (RuntimeException throwable) { + var cause = throwable instanceof CompletionException + ? ExceptionHelper.unwrapCompletableFuture(throwable) + : throwable; + if (cause instanceof PayloadOffloadException payloadOffloadException) { + throw payloadOffloadException; + } + throw new PayloadOffloadException("Failed to " + action + " payload for " + describe(context), cause); + } + } + + private static PayloadOffloader effectiveOffloader(PayloadOffloader offloader) { + return offloader == null || PayloadOffloaders.isDisabled(offloader) ? null : offloader; + } + + private static PayloadCacheKey payloadCacheKey(PayloadOffloadContext context, String checkpointPayload) { + return new PayloadCacheKey( + context.durableExecutionArn(), + context.entityId(), + context.payloadKind(), + context.attempt(), + hash(checkpointPayload)); + } + + private static String describe(PayloadOffloadContext context) { + return context.payloadKind() + " payload '" + context.entityId() + "'"; + } + + private static String hash(String value) { + if (value == null) { + return "null"; + } + try { + var digest = MessageDigest.getInstance("SHA-256"); + return HexFormat.of().formatHex(digest.digest(value.getBytes(StandardCharsets.UTF_8))); + } catch (NoSuchAlgorithmException e) { + throw new IllegalStateException("SHA-256 is unavailable", e); + } + } + + private record PayloadCacheKey( + String durableExecutionArn, + String entityId, + SerDesPayloadKind payloadKind, + Integer attempt, + String checkpointPayloadHash) {} + + private record DeserializedCacheKey( + String durableExecutionArn, + String entityId, + SerDesPayloadKind payloadKind, + Integer attempt, + String targetType, + String serializedPayloadHash) {} +} diff --git a/sdk/src/main/java/software/amazon/lambda/durable/offload/OffloadedPayload.java b/sdk/src/main/java/software/amazon/lambda/durable/offload/OffloadedPayload.java new file mode 100644 index 000000000..16ee05e37 --- /dev/null +++ b/sdk/src/main/java/software/amazon/lambda/durable/offload/OffloadedPayload.java @@ -0,0 +1,45 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.offload; + +import java.util.Map; +import java.util.Objects; + +/** + * SDK-owned representation of inline or externally stored serialized payload data. + * + * @param mode payload storage mode + * @param data inline serialized data, when {@code mode} is {@link PayloadStorageMode#INLINE} + * @param reference external storage reference, when {@code mode} is {@link PayloadStorageMode#REFERENCE} + * @param preview optional inline preview metadata + */ +public record OffloadedPayload(PayloadStorageMode mode, String data, String reference, Map preview) { + + public OffloadedPayload { + Objects.requireNonNull(mode, "mode cannot be null"); + preview = preview == null ? null : Map.copyOf(preview); + if (mode == PayloadStorageMode.INLINE) { + Objects.requireNonNull(data, "data cannot be null for an inline payload"); + if (reference != null) { + throw new IllegalArgumentException("reference must be null for an inline payload"); + } + } else { + if (reference == null || reference.isBlank()) { + throw new IllegalArgumentException("reference cannot be blank for an externally stored payload"); + } + if (data != null) { + throw new IllegalArgumentException("data must be null for an externally stored payload"); + } + } + } + + /** Creates an inline payload. */ + public static OffloadedPayload inline(String data) { + return new OffloadedPayload(PayloadStorageMode.INLINE, data, null, null); + } + + /** Creates an externally stored payload. */ + public static OffloadedPayload reference(String reference, Map preview) { + return new OffloadedPayload(PayloadStorageMode.REFERENCE, null, reference, preview); + } +} diff --git a/sdk/src/main/java/software/amazon/lambda/durable/offload/PayloadOffloadContext.java b/sdk/src/main/java/software/amazon/lambda/durable/offload/PayloadOffloadContext.java new file mode 100644 index 000000000..1bfb07cbf --- /dev/null +++ b/sdk/src/main/java/software/amazon/lambda/durable/offload/PayloadOffloadContext.java @@ -0,0 +1,75 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.offload; + +import java.util.Objects; +import software.amazon.awssdk.services.lambda.model.OperationType; +import software.amazon.lambda.durable.model.OperationIdentifier; +import software.amazon.lambda.durable.model.OperationSubType; + +/** + * Stable identity and operation metadata for a payload being stored or loaded. + * + * @param durableExecutionArn durable execution ARN + * @param entityId stable identifier for this payload within the durable execution + * @param payloadKind role of the payload + * @param operationId operation identifier, or the execution operation identifier for root payloads + * @param operationName operation name, when present + * @param parentId parent context identifier, when present + * @param operationType durable operation type + * @param operationSubType durable operation subtype, or null for root execution payloads + * @param attempt current attempt for retryable operations, when available + */ +public record PayloadOffloadContext( + String durableExecutionArn, + String entityId, + SerDesPayloadKind payloadKind, + String operationId, + String operationName, + String parentId, + OperationType operationType, + OperationSubType operationSubType, + Integer attempt) { + + public PayloadOffloadContext { + Objects.requireNonNull(durableExecutionArn, "durableExecutionArn cannot be null"); + Objects.requireNonNull(entityId, "entityId cannot be null"); + Objects.requireNonNull(payloadKind, "payloadKind cannot be null"); + Objects.requireNonNull(operationId, "operationId cannot be null"); + Objects.requireNonNull(operationType, "operationType cannot be null"); + } + + /** Creates context for a root execution input, output, or exception payload. */ + public static PayloadOffloadContext forExecution( + String durableExecutionArn, String executionOperationId, String executionName, SerDesPayloadKind kind) { + return new PayloadOffloadContext( + durableExecutionArn, + "execution/" + executionOperationId + "/" + kind.entitySuffix(), + kind, + executionOperationId, + executionName, + null, + OperationType.EXECUTION, + null, + null); + } + + /** Creates context for a durable operation payload. */ + public static PayloadOffloadContext forOperation( + String durableExecutionArn, + OperationIdentifier operation, + String parentId, + SerDesPayloadKind kind, + Integer attempt) { + return new PayloadOffloadContext( + durableExecutionArn, + "operation/" + operation.operationId() + "/" + kind.entitySuffix(), + kind, + operation.operationId(), + operation.name(), + parentId, + operation.operationType(), + operation.subType(), + attempt); + } +} diff --git a/sdk/src/main/java/software/amazon/lambda/durable/offload/PayloadOffloader.java b/sdk/src/main/java/software/amazon/lambda/durable/offload/PayloadOffloader.java new file mode 100644 index 000000000..61fa57142 --- /dev/null +++ b/sdk/src/main/java/software/amazon/lambda/durable/offload/PayloadOffloader.java @@ -0,0 +1,26 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.offload; + +/** + * Stores and loads serialized durable execution payloads. + * + *

Implementations receive serialized text after {@code SerDes} processing. They may keep it inline or replace it + * with a reference to external storage. A returned reference must keep the same meaning for the lifetime of every + * checkpoint that contains it. Implementations that overwrite storage should include + * {@link PayloadOffloadContext#attempt()} or another immutable version in the storage key when a payload can be + * updated. + */ +public interface PayloadOffloader { + + /** Stores serialized payload data or returns it inline. */ + OffloadedPayload offload(String serializedPayload, PayloadOffloadContext context); + + /** Loads serialized payload data from an externally stored payload. */ + String load(OffloadedPayload payload, PayloadOffloadContext context); + + /** Returns a sentinel that disables a globally configured offloader for a specific operation. */ + static PayloadOffloader disabled() { + return PayloadOffloaders.disabled(); + } +} diff --git a/sdk/src/main/java/software/amazon/lambda/durable/offload/PayloadOffloaders.java b/sdk/src/main/java/software/amazon/lambda/durable/offload/PayloadOffloaders.java new file mode 100644 index 000000000..b51db0596 --- /dev/null +++ b/sdk/src/main/java/software/amazon/lambda/durable/offload/PayloadOffloaders.java @@ -0,0 +1,37 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.offload; + +/** Factory methods for built-in payload offloader policies. */ +public final class PayloadOffloaders { + private static final PayloadOffloader DISABLED = new DisabledPayloadOffloader(); + + private PayloadOffloaders() {} + + /** Returns a sentinel that forces payloads to remain in the normal inline checkpoint format. */ + public static PayloadOffloader disabled() { + return DISABLED; + } + + /** Returns whether the supplied offloader is the disabled sentinel. */ + public static boolean isDisabled(PayloadOffloader offloader) { + return offloader == DISABLED; + } + + private static final class DisabledPayloadOffloader implements PayloadOffloader { + @Override + public OffloadedPayload offload(String serializedPayload, PayloadOffloadContext context) { + return OffloadedPayload.inline(serializedPayload); + } + + @Override + public String load(OffloadedPayload payload, PayloadOffloadContext context) { + return payload.data(); + } + + @Override + public String toString() { + return "PayloadOffloader.disabled()"; + } + } +} diff --git a/sdk/src/main/java/software/amazon/lambda/durable/offload/PayloadStorageMode.java b/sdk/src/main/java/software/amazon/lambda/durable/offload/PayloadStorageMode.java new file mode 100644 index 000000000..82f798a94 --- /dev/null +++ b/sdk/src/main/java/software/amazon/lambda/durable/offload/PayloadStorageMode.java @@ -0,0 +1,9 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.offload; + +/** Identifies whether a serialized payload is stored inline or in external storage. */ +public enum PayloadStorageMode { + INLINE, + REFERENCE +} diff --git a/sdk/src/main/java/software/amazon/lambda/durable/offload/SerDesPayloadKind.java b/sdk/src/main/java/software/amazon/lambda/durable/offload/SerDesPayloadKind.java new file mode 100644 index 000000000..aa9b56e5c --- /dev/null +++ b/sdk/src/main/java/software/amazon/lambda/durable/offload/SerDesPayloadKind.java @@ -0,0 +1,23 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.offload; + +/** Identifies the role of a serialized user payload in a durable execution. */ +public enum SerDesPayloadKind { + INPUT("input"), + OUTPUT("output"), + RESULT("result"), + INVOKE_PAYLOAD("invoke-payload"), + STATE("state"), + EXCEPTION("exception"); + + private final String entitySuffix; + + SerDesPayloadKind(String entitySuffix) { + this.entitySuffix = entitySuffix; + } + + String entitySuffix() { + return entitySuffix; + } +} diff --git a/sdk/src/main/java/software/amazon/lambda/durable/operation/BaseDurableOperation.java b/sdk/src/main/java/software/amazon/lambda/durable/operation/BaseDurableOperation.java index 35a71f0da..33e77393b 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/operation/BaseDurableOperation.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/operation/BaseDurableOperation.java @@ -113,6 +113,11 @@ public String getName() { return operationIdentifier.name(); } + /** Gets the complete operation identifier. */ + protected OperationIdentifier getOperationIdentifier() { + return operationIdentifier; + } + /** Gets the parent context. */ protected DurableContextImpl getContext() { return durableContext; diff --git a/sdk/src/main/java/software/amazon/lambda/durable/operation/CallbackOperation.java b/sdk/src/main/java/software/amazon/lambda/durable/operation/CallbackOperation.java index 9d9481fb9..2c821f4bb 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/operation/CallbackOperation.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/operation/CallbackOperation.java @@ -26,7 +26,7 @@ public CallbackOperation( TypeToken resultTypeToken, CallbackConfig config, DurableContextImpl durableContext) { - super(operationIdentifier, resultTypeToken, config.serDes(), durableContext); + super(operationIdentifier, resultTypeToken, config.serDes(), config.payloadOffloader(), durableContext); this.config = config; } diff --git a/sdk/src/main/java/software/amazon/lambda/durable/operation/ChildContextOperation.java b/sdk/src/main/java/software/amazon/lambda/durable/operation/ChildContextOperation.java index 8c299cfa4..a2d5fe4ab 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/operation/ChildContextOperation.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/operation/ChildContextOperation.java @@ -77,6 +77,7 @@ public ChildContextOperation( operationIdentifier, resultTypeToken, config.serDes(), + config.payloadOffloader(), durableContext, parentOperation, config.isVirtual()); diff --git a/sdk/src/main/java/software/amazon/lambda/durable/operation/ConcurrencyOperation.java b/sdk/src/main/java/software/amazon/lambda/durable/operation/ConcurrencyOperation.java index 646b0685a..ce1ab8d85 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/operation/ConcurrencyOperation.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/operation/ConcurrencyOperation.java @@ -28,6 +28,7 @@ import software.amazon.lambda.durable.execution.ThreadType; import software.amazon.lambda.durable.model.OperationIdentifier; import software.amazon.lambda.durable.model.OperationSubType; +import software.amazon.lambda.durable.offload.PayloadOffloader; import software.amazon.lambda.durable.serde.SerDes; import software.amazon.lambda.durable.util.ExceptionHelper; @@ -105,7 +106,27 @@ protected ConcurrencyOperation( int maxConcurrency, Function shouldComplete, NestingType nestingType) { - super(operationIdentifier, resultTypeToken, resultSerDes, durableContext); + this( + operationIdentifier, + resultTypeToken, + resultSerDes, + null, + durableContext, + maxConcurrency, + shouldComplete, + nestingType); + } + + protected ConcurrencyOperation( + OperationIdentifier operationIdentifier, + TypeToken resultTypeToken, + SerDes resultSerDes, + PayloadOffloader payloadOffloader, + DurableContextImpl durableContext, + int maxConcurrency, + Function shouldComplete, + NestingType nestingType) { + super(operationIdentifier, resultTypeToken, resultSerDes, payloadOffloader, durableContext); this.maxConcurrency = maxConcurrency; this.shouldComplete = Objects.requireNonNull(shouldComplete, "shouldComplete cannot be null"); this.operationIdGenerator = new OperationIdGenerator(getOperationId()); @@ -151,6 +172,30 @@ protected ChildContextOperation createItem( this); } + protected ChildContextOperation createItem( + String operationId, + String name, + Function function, + TypeToken resultType, + SerDes serDes, + PayloadOffloader payloadOffloader, + OperationSubType branchSubType) { + if (payloadOffloader == null) { + return createItem(operationId, name, function, resultType, serDes, branchSubType); + } + return new ChildContextOperation<>( + OperationIdentifier.of(operationId, name, branchSubType), + function, + resultType, + RunInChildContextConfig.builder() + .serDes(serDes) + .payloadOffloader(payloadOffloader) + .isVirtual(nestingType == NestingType.FLAT) + .build(), + rootContext, + this); + } + /** Called when the concurrency operation completes. Subclasses define checkpointing behavior. */ protected abstract void handleCompletion(CompletionConfig.CompletionDecision completionDecision); @@ -168,8 +213,19 @@ protected ChildContextOperation enqueueItem( SerDes serDes, OperationSubType branchSubType, boolean skipped) { + return enqueueItem(name, function, resultType, serDes, null, branchSubType, skipped); + } + + protected ChildContextOperation enqueueItem( + String name, + Function function, + TypeToken resultType, + SerDes serDes, + PayloadOffloader payloadOffloader, + OperationSubType branchSubType, + boolean skipped) { var operationId = this.operationIdGenerator.nextOperationId(); - var childOp = createItem(operationId, name, function, resultType, serDes, branchSubType); + var childOp = createItem(operationId, name, function, resultType, serDes, payloadOffloader, branchSubType); branches.add(childOp); if (!skipped) { logger.debug("Item enqueued {}", name); diff --git a/sdk/src/main/java/software/amazon/lambda/durable/operation/InvokeOperation.java b/sdk/src/main/java/software/amazon/lambda/durable/operation/InvokeOperation.java index 9e2c54ace..23abd189d 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/operation/InvokeOperation.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/operation/InvokeOperation.java @@ -14,6 +14,7 @@ import software.amazon.lambda.durable.exception.InvokeStoppedException; import software.amazon.lambda.durable.exception.InvokeTimedOutException; import software.amazon.lambda.durable.model.OperationIdentifier; +import software.amazon.lambda.durable.offload.SerDesPayloadKind; import software.amazon.lambda.durable.serde.SerDes; /** @@ -35,7 +36,7 @@ public InvokeOperation( TypeToken resultTypeToken, InvokeConfig config, DurableContextImpl durableContext) { - super(operationIdentifier, resultTypeToken, config.serDes(), durableContext); + super(operationIdentifier, resultTypeToken, config.serDes(), config.payloadOffloader(), durableContext); this.functionName = functionName; this.payload = payload; @@ -70,7 +71,7 @@ private void startInvocation() { .functionName(functionName) .tenantId(invokeConfig.tenantId()) .build()) - .payload(payloadSerDes.serialize(this.payload)); + .payload(serializePayload(this.payload, payloadSerDes, SerDesPayloadKind.INVOKE_PAYLOAD, null)); sendOperationUpdate(update); } diff --git a/sdk/src/main/java/software/amazon/lambda/durable/operation/MapOperation.java b/sdk/src/main/java/software/amazon/lambda/durable/operation/MapOperation.java index 2f665547a..8dbdc2982 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/operation/MapOperation.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/operation/MapOperation.java @@ -26,6 +26,7 @@ import software.amazon.lambda.durable.model.MapResult; import software.amazon.lambda.durable.model.OperationIdentifier; import software.amazon.lambda.durable.model.OperationSubType; +import software.amazon.lambda.durable.offload.PayloadOffloader; import software.amazon.lambda.durable.serde.SerDes; import software.amazon.lambda.durable.util.ExceptionHelper; import software.amazon.lambda.durable.util.ParameterValidator; @@ -50,6 +51,7 @@ public class MapOperation extends ConcurrencyOperation> { private final DurableContext.MapFunction function; private final TypeToken itemResultType; private final SerDes serDes; + private final PayloadOffloader payloadOffloader; private final List iterationNames; private volatile MapResult cachedResult; @@ -82,6 +84,7 @@ public MapOperation( operationIdentifier, new TypeToken<>() {}, config.serDes(), + config.payloadOffloader(), durableContext, config.maxConcurrency(), config.completionConfig().completionDecisionFunction(), @@ -96,6 +99,7 @@ public MapOperation( this.function = function; this.itemResultType = itemResultType; this.serDes = config.serDes(); + this.payloadOffloader = config.payloadOffloader(); this.iterationNames = Collections.unmodifiableList(new ArrayList<>(iterationNames)); if (this.iterationNames.size() != this.items.size()) { throw new IllegalArgumentException("iterationNames must have one entry per item"); @@ -152,6 +156,7 @@ private void addUnskippedItems(List resultItems) childCtx -> function.apply(item, index, childCtx), itemResultType, serDes, + payloadOffloader, OperationSubType.MAP_ITERATION, skip); } diff --git a/sdk/src/main/java/software/amazon/lambda/durable/operation/ParallelOperation.java b/sdk/src/main/java/software/amazon/lambda/durable/operation/ParallelOperation.java index 41a687e6b..d2467ca7d 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/operation/ParallelOperation.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/operation/ParallelOperation.java @@ -59,6 +59,7 @@ public ParallelOperation( operationIdentifier, TypeToken.get(ParallelResult.class), resultSerDes, + config.payloadOffloader(), durableContext, config.maxConcurrency(), config.completionConfig().completionDecisionFunction(), @@ -187,6 +188,9 @@ public DurableFuture branch( && (partialResult.statuses().size() <= nextBranchIndex || partialResult.statuses().get(nextBranchIndex) == ParallelResult.Status.SKIPPED); var serDes = config.serDes() == null ? getContext().getDurableConfig().getSerDes() : config.serDes(); - return enqueueItem(name, func, resultType, serDes, OperationSubType.PARALLEL_BRANCH, skip); + var offloader = config.payloadOffloader() == null + ? getContext().getDurableConfig().getPayloadOffloader() + : config.payloadOffloader(); + return enqueueItem(name, func, resultType, serDes, offloader, OperationSubType.PARALLEL_BRANCH, skip); } } diff --git a/sdk/src/main/java/software/amazon/lambda/durable/operation/SerializableDurableOperation.java b/sdk/src/main/java/software/amazon/lambda/durable/operation/SerializableDurableOperation.java index 6457c996d..88f230aaa 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/operation/SerializableDurableOperation.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/operation/SerializableDurableOperation.java @@ -9,7 +9,12 @@ import software.amazon.lambda.durable.TypeToken; import software.amazon.lambda.durable.context.DurableContextImpl; import software.amazon.lambda.durable.exception.SerDesException; +import software.amazon.lambda.durable.execution.PayloadCodec; import software.amazon.lambda.durable.model.OperationIdentifier; +import software.amazon.lambda.durable.offload.PayloadOffloadContext; +import software.amazon.lambda.durable.offload.PayloadOffloader; +import software.amazon.lambda.durable.offload.PayloadOffloaders; +import software.amazon.lambda.durable.offload.SerDesPayloadKind; import software.amazon.lambda.durable.serde.SerDes; import software.amazon.lambda.durable.util.ExceptionHelper; @@ -38,6 +43,7 @@ protected record SerializedResult(String serialized, T deserialized) {} private final TypeToken resultTypeToken; private final SerDes resultSerDes; + private final PayloadOffloader payloadOffloader; /** * Constructs a new durable operation. @@ -52,7 +58,16 @@ protected SerializableDurableOperation( TypeToken resultTypeToken, SerDes resultSerDes, DurableContextImpl durableContext) { - this(operationIdentifier, resultTypeToken, resultSerDes, durableContext, null, false); + this(operationIdentifier, resultTypeToken, resultSerDes, null, durableContext, null, false); + } + + protected SerializableDurableOperation( + OperationIdentifier operationIdentifier, + TypeToken resultTypeToken, + SerDes resultSerDes, + PayloadOffloader payloadOffloader, + DurableContextImpl durableContext) { + this(operationIdentifier, resultTypeToken, resultSerDes, payloadOffloader, durableContext, null, false); } /** @@ -72,9 +87,21 @@ protected SerializableDurableOperation( DurableContextImpl durableContext, BaseDurableOperation parentOperation, boolean isVirtual) { + this(operationIdentifier, resultTypeToken, resultSerDes, null, durableContext, parentOperation, isVirtual); + } + + protected SerializableDurableOperation( + OperationIdentifier operationIdentifier, + TypeToken resultTypeToken, + SerDes resultSerDes, + PayloadOffloader payloadOffloader, + DurableContextImpl durableContext, + BaseDurableOperation parentOperation, + boolean isVirtual) { super(operationIdentifier, durableContext, parentOperation, isVirtual); this.resultTypeToken = resultTypeToken; this.resultSerDes = resultSerDes; + this.payloadOffloader = payloadOffloader; } /** @@ -85,8 +112,23 @@ protected SerializableDurableOperation( * @throws SerDesException if deserialization fails */ protected T deserializeResult(String result) { + return deserializeResult(result, SerDesPayloadKind.RESULT, null); + } + + /** Deserializes a result with explicit payload kind and attempt metadata. */ + protected T deserializeResult(String result, SerDesPayloadKind payloadKind, Integer attempt) { try { - return resultSerDes.deserialize(result, resultTypeToken); + if (!usesPayloadCodec(result)) { + return resultSerDes.deserialize(result, resultTypeToken); + } + return executionManager + .getPayloadCodec() + .deserialize( + result, + resultTypeToken, + resultSerDes, + payloadOffloader, + payloadContext(payloadKind, attempt)); } catch (SerDesException e) { logger.warn( "Failed to deserialize {} result for operation name '{}'. Ensure the result is properly encoded.", @@ -106,11 +148,28 @@ protected T deserializeResult(String result) { * @return the serialized string and the deserialized result */ protected SerializedResult serializeAndDeserializeResult(T result) { - var serialized = resultSerDes.serialize(result); - var deserialized = shouldDeserializeAfterSerialization() ? deserializeResult(serialized) : result; + return serializeAndDeserializeResult(result, SerDesPayloadKind.RESULT, null); + } + + /** Serializes a result with explicit payload kind and attempt metadata. */ + protected SerializedResult serializeAndDeserializeResult( + T result, SerDesPayloadKind payloadKind, Integer attempt) { + var serialized = serializePayload(result, resultSerDes, payloadKind, attempt); + var deserialized = + shouldDeserializeAfterSerialization() ? deserializeResult(serialized, payloadKind, attempt) : result; return new SerializedResult<>(serialized, deserialized); } + /** Serializes an operation-owned payload with the operation's offloader policy. */ + protected String serializePayload(Object value, SerDes serDes, SerDesPayloadKind payloadKind, Integer attempt) { + if (!hasActivePayloadOffloader()) { + return serDes.serialize(value); + } + return executionManager + .getPayloadCodec() + .serialize(value, serDes, payloadOffloader, payloadContext(payloadKind, attempt)); + } + /** * Serializes a throwable into an {@link ErrorObject} for checkpointing. * @@ -119,9 +178,19 @@ protected SerializedResult serializeAndDeserializeResult(T result) { */ @SuppressWarnings("ThrowableNotThrown") protected ErrorObject serializeException(Throwable throwable) { - var error = ExceptionHelper.buildErrorObject(throwable, resultSerDes); + return serializeException(throwable, null); + } + + /** Serializes a throwable with attempt metadata. */ + protected ErrorObject serializeException(Throwable throwable, Integer attempt) { + var error = ErrorObject.builder() + .errorType(throwable.getClass().getName()) + .errorMessage(throwable.getMessage()) + .errorData(serializePayload(throwable, resultSerDes, SerDesPayloadKind.EXCEPTION, attempt)) + .stackTrace(ExceptionHelper.serializeStackTrace(throwable.getStackTrace())) + .build(); if (shouldDeserializeAfterSerialization()) { - deserializeException(error); + deserializeException(error, attempt); } return error; } @@ -139,6 +208,11 @@ private boolean shouldDeserializeAfterSerialization() { * @return the reconstructed throwable, or null if reconstruction is not possible */ protected Throwable deserializeException(ErrorObject errorObject) { + return deserializeException(errorObject, null); + } + + /** Deserializes a throwable with attempt metadata. */ + protected Throwable deserializeException(ErrorObject errorObject, Integer attempt) { Throwable original = null; if (errorObject == null) { return original; @@ -153,8 +227,19 @@ protected Throwable deserializeException(ErrorObject errorObject) { Class exceptionClass = Class.forName(errorType); if (Throwable.class.isAssignableFrom(exceptionClass)) { - original = - resultSerDes.deserialize(errorData, TypeToken.get(exceptionClass.asSubclass(Throwable.class))); + var exceptionType = TypeToken.get(exceptionClass.asSubclass(Throwable.class)); + if (usesPayloadCodec(errorData)) { + original = executionManager + .getPayloadCodec() + .deserialize( + errorData, + exceptionType, + resultSerDes, + payloadOffloader, + payloadContext(SerDesPayloadKind.EXCEPTION, attempt)); + } else { + original = resultSerDes.deserialize(errorData, exceptionType); + } if (original != null) { original.setStackTrace(ExceptionHelper.deserializeStackTrace(errorObject.stackTrace())); @@ -168,5 +253,22 @@ protected Throwable deserializeException(ErrorObject errorObject) { return original; } + private boolean usesPayloadCodec(String checkpointPayload) { + return hasActivePayloadOffloader() || PayloadCodec.isOffloadEnvelope(checkpointPayload); + } + + private boolean hasActivePayloadOffloader() { + return payloadOffloader != null && !PayloadOffloaders.isDisabled(payloadOffloader); + } + + private PayloadOffloadContext payloadContext(SerDesPayloadKind kind, Integer attempt) { + return PayloadOffloadContext.forOperation( + executionManager.getDurableExecutionArn(), + getOperationIdentifier(), + getContext().getParentId(), + kind, + attempt); + } + public abstract T get(); } diff --git a/sdk/src/main/java/software/amazon/lambda/durable/operation/StepOperation.java b/sdk/src/main/java/software/amazon/lambda/durable/operation/StepOperation.java index 467a87b94..c25603f13 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/operation/StepOperation.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/operation/StepOperation.java @@ -25,6 +25,7 @@ import software.amazon.lambda.durable.execution.ThreadType; import software.amazon.lambda.durable.logging.DurableLogger; import software.amazon.lambda.durable.model.OperationIdentifier; +import software.amazon.lambda.durable.offload.SerDesPayloadKind; import software.amazon.lambda.durable.util.ExceptionHelper; /** @@ -47,7 +48,7 @@ public StepOperation( TypeToken resultTypeToken, StepConfig config, DurableContextImpl durableContext) { - super(operationIdentifier, resultTypeToken, config.serDes(), durableContext); + super(operationIdentifier, resultTypeToken, config.serDes(), config.payloadOffloader(), durableContext); this.function = function; this.config = config; @@ -117,7 +118,7 @@ private void executeStepLogic(int attempt) { // through onUserFunctionEnd; retry/checkpoint handling stays outside the boundary. T result = runUserFunction(attempt, () -> function.apply(stepContext)); - handleStepSucceeded(result); + handleStepSucceeded(result, attempt); } catch (Throwable e) { handleStepFailure(e, attempt); } @@ -144,8 +145,8 @@ private void checkpointStarted() { } } - private void handleStepSucceeded(T result) { - var serializedResult = serializeAndDeserializeResult(result); + private void handleStepSucceeded(T result, int attempt) { + var serializedResult = serializeAndDeserializeResult(result, SerDesPayloadKind.RESULT, attempt); // Send SUCCEED var successUpdate = @@ -170,7 +171,7 @@ private void handleStepFailure(Throwable exception, int attempt) { if (exception instanceof DurableOperationException durableOperationException) { errorObject = durableOperationException.getErrorObject(); } else { - errorObject = serializeException(exception); + errorObject = serializeException(exception, attempt); } var retryDecision = config.retryStrategy().makeRetryDecision(exception, attempt); @@ -205,8 +206,9 @@ public T get() { if (op.status() == OperationStatus.SUCCEEDED) { var stepDetails = op.stepDetails(); var result = (stepDetails != null) ? stepDetails.result() : null; + var attempt = stepDetails != null ? stepDetails.attempt() : null; - return deserializeResult(result); + return deserializeResult(result, SerDesPayloadKind.RESULT, attempt); } else { var errorObject = op.stepDetails().error(); @@ -216,7 +218,8 @@ public T get() { } // Attempt to reconstruct and throw the original exception - Throwable original = deserializeException(errorObject); + var attempt = op.stepDetails() != null ? op.stepDetails().attempt() : null; + Throwable original = deserializeException(errorObject, attempt); if (original != null) { ExceptionHelper.sneakyThrow(original); } diff --git a/sdk/src/main/java/software/amazon/lambda/durable/operation/WaitForConditionOperation.java b/sdk/src/main/java/software/amazon/lambda/durable/operation/WaitForConditionOperation.java index 6a653c37b..17e2b424d 100644 --- a/sdk/src/main/java/software/amazon/lambda/durable/operation/WaitForConditionOperation.java +++ b/sdk/src/main/java/software/amazon/lambda/durable/operation/WaitForConditionOperation.java @@ -23,6 +23,7 @@ import software.amazon.lambda.durable.logging.DurableLogger; import software.amazon.lambda.durable.model.OperationIdentifier; import software.amazon.lambda.durable.model.WaitForConditionResult; +import software.amazon.lambda.durable.offload.SerDesPayloadKind; import software.amazon.lambda.durable.util.ExceptionHelper; /** @@ -46,7 +47,7 @@ public WaitForConditionOperation( TypeToken resultTypeToken, WaitForConditionConfig config, DurableContextImpl durableContext) { - super(operationIdentifier, resultTypeToken, config.serDes(), durableContext); + super(operationIdentifier, resultTypeToken, config.serDes(), config.payloadOffloader(), durableContext); this.checkFunc = checkFunc; this.config = config; @@ -76,12 +77,14 @@ public T get() { if (op.status() == OperationStatus.SUCCEEDED) { var stepDetails = op.stepDetails(); var result = (stepDetails != null) ? stepDetails.result() : null; - return deserializeResult(result); + var attempt = stepDetails != null ? stepDetails.attempt() : null; + return deserializeResult(result, SerDesPayloadKind.STATE, attempt); } else { var errorObject = op.stepDetails().error(); // Attempt to reconstruct and throw the original exception - Throwable original = deserializeException(errorObject); + var attempt = op.stepDetails() != null ? op.stepDetails().attempt() : null; + Throwable original = deserializeException(errorObject, attempt); if (original != null) { ExceptionHelper.sneakyThrow(original); } @@ -97,7 +100,7 @@ private void resumeCheckLoop(Operation existing) { var checkpointData = stepDetails != null ? stepDetails.result() : null; T currentState; // Get current state if (checkpointData != null) { - currentState = deserializeResult(checkpointData); + currentState = deserializeResult(checkpointData, SerDesPayloadKind.STATE, attempt - 1); } else { currentState = config.initialState(); } @@ -131,7 +134,8 @@ private void executeCheckLogic(T currentState, int attempt) { runUserFunction(attempt, () -> checkFunc.apply(currentState, stepContext)); // Normalize the value through SerDes so first execution matches replay. - var serializedState = serializeAndDeserializeResult(result.value()); + var serializedState = + serializeAndDeserializeResult(result.value(), SerDesPayloadKind.STATE, attempt); T deserializedValue = serializedState.deserialized(); if (result.isDone()) { @@ -161,7 +165,7 @@ private void executeCheckLogic(T currentState, int attempt) { .thenRun(() -> executeCheckLogic(deserializedValue, attempt + 1)); } } catch (Throwable e) { - handleCheckFailure(e); + handleCheckFailure(e, attempt); } } }; @@ -169,7 +173,7 @@ private void executeCheckLogic(T currentState, int attempt) { runUserHandler(userHandler, ThreadType.STEP); } - private void handleCheckFailure(Throwable exception) { + private void handleCheckFailure(Throwable exception, int attempt) { exception = ExceptionHelper.unwrapCompletableFuture(exception); if (exception instanceof SuspendExecutionException suspendExecutionException) { throw suspendExecutionException; @@ -180,7 +184,7 @@ private void handleCheckFailure(Throwable exception) { final var errorObject = (exception instanceof DurableOperationException durableOpEx) ? durableOpEx.getErrorObject() - : serializeException(exception); + : serializeException(exception, attempt); // Checkpoint FAIL var failUpdate = OperationUpdate.builder().action(OperationAction.FAIL).error(errorObject); diff --git a/sdk/src/test/java/software/amazon/lambda/durable/execution/PayloadCodecTest.java b/sdk/src/test/java/software/amazon/lambda/durable/execution/PayloadCodecTest.java new file mode 100644 index 000000000..faca8a7fc --- /dev/null +++ b/sdk/src/test/java/software/amazon/lambda/durable/execution/PayloadCodecTest.java @@ -0,0 +1,201 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.execution; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; +import software.amazon.lambda.durable.TypeToken; +import software.amazon.lambda.durable.exception.PayloadOffloadException; +import software.amazon.lambda.durable.model.OperationIdentifier; +import software.amazon.lambda.durable.model.OperationSubType; +import software.amazon.lambda.durable.offload.OffloadedPayload; +import software.amazon.lambda.durable.offload.PayloadOffloadContext; +import software.amazon.lambda.durable.offload.PayloadOffloader; +import software.amazon.lambda.durable.offload.SerDesPayloadKind; +import software.amazon.lambda.durable.serde.JacksonSerDes; + +class PayloadCodecTest { + private ExecutorService executor; + + @AfterEach + void shutdownExecutor() { + if (executor != null) { + executor.shutdownNow(); + } + } + + @Test + void legacyPayloadRemainsReadable() { + var codec = codec(); + var value = codec.deserialize( + "{\"value\":\"legacy\"}", TypeToken.get(TestValue.class), new JacksonSerDes(), null, context()); + + assertEquals("legacy", value.value()); + } + + @Test + void referencePayloadIsLoadedOnceAndDeserializedObjectIsCached() { + var offloader = new InMemoryOffloader(); + var writer = codec(); + var payload = writer.serialize(new TestValue("stored"), new JacksonSerDes(), offloader, context()); + writer.clear(); + + var reader = codec(); + var first = + reader.deserialize(payload, TypeToken.get(TestValue.class), new JacksonSerDes(), offloader, context()); + var second = + reader.deserialize(payload, TypeToken.get(TestValue.class), new JacksonSerDes(), offloader, context()); + + assertEquals("stored", first.value()); + assertSame(first, second); + assertEquals(1, offloader.loadCount.get()); + } + + @Test + void changedSerializedDataInvalidatesObjectCache() { + var codec = codec(); + var serDes = new JacksonSerDes(); + + var first = codec.deserialize("{\"value\":\"one\"}", TypeToken.get(TestValue.class), serDes, null, context()); + var second = codec.deserialize("{\"value\":\"two\"}", TypeToken.get(TestValue.class), serDes, null, context()); + + assertEquals("one", first.value()); + assertEquals("two", second.value()); + } + + @Test + void attemptsHaveIndependentCachesWhenReferenceIsReused() { + var codec = codec(); + var serDes = new JacksonSerDes(); + var offloader = new PayloadOffloader() { + @Override + public OffloadedPayload offload(String serializedPayload, PayloadOffloadContext context) { + return OffloadedPayload.reference("memory://shared", null); + } + + @Override + public String load(OffloadedPayload payload, PayloadOffloadContext context) { + return context.attempt() == 1 ? "{\"value\":\"one\"}" : "{\"value\":\"two\"}"; + } + }; + + var firstPayload = codec.serialize(new TestValue("one"), serDes, offloader, context(1)); + var secondPayload = codec.serialize(new TestValue("two"), serDes, offloader, context(2)); + + assertEquals(firstPayload, secondPayload); + assertEquals( + "one", + codec.deserialize(firstPayload, TypeToken.get(TestValue.class), serDes, offloader, context(1)) + .value()); + assertEquals( + "two", + codec.deserialize(secondPayload, TypeToken.get(TestValue.class), serDes, offloader, context(2)) + .value()); + } + + @Test + void disabledOffloaderKeepsLegacyInlineFormat() { + var codec = codec(); + var payload = + codec.serialize(new TestValue("inline"), new JacksonSerDes(), PayloadOffloader.disabled(), context()); + + assertEquals("{\"value\":\"inline\"}", payload); + } + + @Test + void externalEnvelopeRequiresConfiguredOffloader() { + var offloader = new InMemoryOffloader(); + var writer = codec(); + var payload = writer.serialize(new TestValue("stored"), new JacksonSerDes(), offloader, context()); + writer.clear(); + + var reader = codec(); + assertThrows( + PayloadOffloadException.class, + () -> reader.deserialize( + payload, TypeToken.get(TestValue.class), new JacksonSerDes(), null, context())); + } + + @Test + void offloadRunsOnConfiguredExecutor() { + var threadName = new AtomicReference(); + executor = Executors.newSingleThreadExecutor(r -> new Thread(r, "payload-io-test")); + var codec = new PayloadCodec(executor); + var offloader = new InMemoryOffloader() { + @Override + public OffloadedPayload offload(String serializedPayload, PayloadOffloadContext context) { + threadName.set(Thread.currentThread().getName()); + return super.offload(serializedPayload, context); + } + }; + + codec.serialize(new TestValue("stored"), new JacksonSerDes(), offloader, context()); + + assertTrue(threadName.get().startsWith("payload-io-test")); + } + + @Test + void malformedEnvelopeReportsPayloadIdentity() { + var codec = codec(); + var error = assertThrows( + PayloadOffloadException.class, + () -> codec.deserialize( + "@aws-durable-payload:v1:not-json", + TypeToken.get(TestValue.class), + new JacksonSerDes(), + null, + context())); + + assertTrue(error.getMessage().contains("operation/op-1/result")); + } + + private PayloadCodec codec() { + executor = Executors.newCachedThreadPool(); + return new PayloadCodec(executor); + } + + private static PayloadOffloadContext context() { + return context(1); + } + + private static PayloadOffloadContext context(int attempt) { + return PayloadOffloadContext.forOperation( + "arn:aws:lambda:us-east-1:123456789012:function:test:$LATEST/durable-execution/name/invocation", + OperationIdentifier.of("op-1", "step", OperationSubType.STEP), + null, + SerDesPayloadKind.RESULT, + attempt); + } + + record TestValue(String value) {} + + private static class InMemoryOffloader implements PayloadOffloader { + private final Map values = new ConcurrentHashMap<>(); + private final AtomicInteger sequence = new AtomicInteger(); + private final AtomicInteger loadCount = new AtomicInteger(); + + @Override + public OffloadedPayload offload(String serializedPayload, PayloadOffloadContext context) { + var reference = "memory://" + sequence.incrementAndGet(); + values.put(reference, serializedPayload); + return OffloadedPayload.reference(reference, null); + } + + @Override + public String load(OffloadedPayload payload, PayloadOffloadContext context) { + loadCount.incrementAndGet(); + return values.get(payload.reference()); + } + } +}