typeToken, SerDesContext context) {
+ observedContexts.add(context);
+ return (T) data;
+ }
+ };
+ }
+}
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..7ed5a58c3 100644
--- a/sdk/src/main/java/software/amazon/lambda/durable/DurableConfig.java
+++ b/sdk/src/main/java/software/amazon/lambda/durable/DurableConfig.java
@@ -95,6 +95,7 @@ public final class DurableConfig {
private final DurableExecutionClient durableExecutionClient;
private final SerDes serDes;
private final ExecutorService executorService;
+ private final ExecutorService serDesExecutorService;
private final LoggerConfig loggerConfig;
private final PollingStrategy pollingStrategy;
private final Duration checkpointDelay;
@@ -109,6 +110,7 @@ private DurableConfig(Builder builder) {
this.serDes = Objects.requireNonNullElseGet(builder.serDes, JacksonSerDes::new);
this.executorService =
Objects.requireNonNullElseGet(builder.executorService, DurableConfig::createDefaultExecutor);
+ this.serDesExecutorService = builder.serDesExecutorService;
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));
@@ -164,6 +166,15 @@ public ExecutorService getExecutorService() {
return executorService;
}
+ /**
+ * Gets the executor used for customer SerDes calls and blocking payload storage I/O.
+ *
+ * @return the configured executor, or {@code null} when SerDes calls execute inline
+ */
+ public ExecutorService getSerDesExecutorService() {
+ return serDesExecutorService;
+ }
+
/**
* Gets the configured LoggerConfig.
*
@@ -235,6 +246,10 @@ public void validateConfiguration() {
if (getExecutorService() == null) {
throw new IllegalStateException("ExecutorService configuration failed");
}
+ if (getSerDesExecutorService() != null && getSerDesExecutorService() == getExecutorService()) {
+ throw new IllegalStateException(
+ "SerDes ExecutorService must be different from the user operation ExecutorService");
+ }
}
/**
@@ -316,6 +331,7 @@ public static final class Builder {
private DurableExecutionClient durableExecutionClient;
private SerDes serDes;
private ExecutorService executorService;
+ private ExecutorService serDesExecutorService;
private LoggerConfig loggerConfig;
private PollingStrategy pollingStrategy;
private Duration checkpointDelay;
@@ -396,6 +412,21 @@ public Builder withExecutorService(ExecutorService executorService) {
return this;
}
+ /**
+ * Sets a dedicated executor for SerDes calls and blocking payload storage I/O.
+ *
+ * The SerDes executor must be different from the user operation executor to avoid deadlock when operation
+ * threads synchronously wait for serialization.
+ *
+ * @param executorService dedicated SerDes ExecutorService
+ * @return This builder
+ */
+ public Builder withSerDesExecutorService(ExecutorService executorService) {
+ this.serDesExecutorService =
+ Objects.requireNonNull(executorService, "SerDes 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/exception/RetryableSerDesException.java b/sdk/src/main/java/software/amazon/lambda/durable/exception/RetryableSerDesException.java
new file mode 100644
index 000000000..be06a762f
--- /dev/null
+++ b/sdk/src/main/java/software/amazon/lambda/durable/exception/RetryableSerDesException.java
@@ -0,0 +1,19 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+package software.amazon.lambda.durable.exception;
+
+/**
+ * Indicates a transient serialization or deserialization failure that may succeed when retried.
+ *
+ *
{@link software.amazon.lambda.durable.serde.RetrySerDes} retries only this exception type. Other
+ * {@link SerDesException} instances are treated as permanent failures.
+ */
+public class RetryableSerDesException extends SerDesException {
+ public RetryableSerDesException(String message, Throwable cause) {
+ super(message, cause);
+ }
+
+ public RetryableSerDesException(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 d8db91326..5144aa1e1 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
@@ -32,6 +32,8 @@
import software.amazon.lambda.durable.plugin.PluginInfoConverter;
import software.amazon.lambda.durable.plugin.PluginRunner;
import software.amazon.lambda.durable.serde.SerDes;
+import software.amazon.lambda.durable.serde.SerDesContext;
+import software.amazon.lambda.durable.serde.SerDesRunner;
import software.amazon.lambda.durable.util.ExceptionHelper;
/**
@@ -60,6 +62,11 @@ public static DurableExecutionOutput execute(
var isFirstInvocation = !executionManager.isReplaying();
var requestId = lambdaContext != null ? lambdaContext.getAwsRequestId() : null;
var executionArn = input.durableExecutionArn();
+ var executionOperationId = executionManager.getExecutionOperation().id();
+ var inputSerDesContext = new SerDesContext(executionArn, executionOperationId + "/input");
+ var outputSerDesContext = new SerDesContext(executionArn, executionOperationId + "/output");
+ var exceptionSerDesContext = new SerDesContext(executionArn, executionOperationId + "/exception");
+ var serDesRunner = executionManager.getSerDesRunner();
executionManager.registerActiveThread(null);
// Captured for onInvocationEnd, which runs outside the handler thread below.
@@ -78,7 +85,11 @@ public static DurableExecutionOutput execute(
Throwable inputFailure = null;
try {
userInput = extractUserInput(
- executionManager.getExecutionOperation(), config.getSerDes(), inputType);
+ executionManager.getExecutionOperation(),
+ config.getSerDes(),
+ inputType,
+ serDesRunner,
+ inputSerDesContext);
} catch (Throwable t) {
inputFailure = t;
}
@@ -171,11 +182,12 @@ public static DurableExecutionOutput execute(
cause,
pluginExecutionInput.get(),
null);
- return DurableExecutionOutput.failure(buildErrorObject(cause, config.getSerDes()));
+ return DurableExecutionOutput.failure(buildErrorObject(
+ cause, config.getSerDes(), serDesRunner, exceptionSerDesContext));
}
// user handler complete successfully
logger.debug("Execution completed");
- var outputPayload = config.getSerDes().serialize(result);
+ var outputPayload = serDesRunner.serialize(config.getSerDes(), result, outputSerDesContext);
var output =
DurableExecutionOutput.success(handleLargePayload(executionManager, outputPayload));
fireOnInvocationEnd(
@@ -254,7 +266,8 @@ private static String handleLargePayload(ExecutionManager executionManager, Stri
return outputPayload;
}
- private static ErrorObject buildErrorObject(Throwable e, SerDes serDes) {
+ private static ErrorObject buildErrorObject(
+ Throwable e, SerDes serDes, SerDesRunner serDesRunner, SerDesContext serDesContext) {
// exceptions thrown from operations, e.g. Step
if (e instanceof DurableOperationException durableOperationException) {
return durableOperationException.getErrorObject();
@@ -263,16 +276,22 @@ private static ErrorObject buildErrorObject(Throwable e, SerDes serDes) {
return unrecoverableDurableExecutionException.getErrorObject();
}
// exceptions thrown from non-operation code
- return ExceptionHelper.buildErrorObject(e, serDes);
+ var errorData = serDesRunner.serialize(serDes, e, serDesContext);
+ return ExceptionHelper.buildErrorObject(e, errorData);
}
- private static I extractUserInput(Operation executionOp, SerDes serDes, TypeToken inputType) {
+ private static I extractUserInput(
+ Operation executionOp,
+ SerDes serDes,
+ TypeToken inputType,
+ SerDesRunner serDesRunner,
+ SerDesContext serDesContext) {
if (executionOp.executionDetails() == null) {
throw new IllegalDurableOperationException("EXECUTION operation missing executionDetails");
}
var inputPayload = executionOp.executionDetails().inputPayload();
- return serDes.deserialize(inputPayload, inputType);
+ return serDesRunner.deserialize(serDes, inputPayload, inputType, serDesContext);
}
/**
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 0e9d8426e..09d6e54c7 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
@@ -30,6 +30,7 @@
import software.amazon.lambda.durable.model.SafeCloseable;
import software.amazon.lambda.durable.operation.BaseDurableOperation;
import software.amazon.lambda.durable.plugin.PluginInfoConverter;
+import software.amazon.lambda.durable.serde.SerDesRunner;
/**
* Central manager for durable execution coordination.
@@ -65,6 +66,7 @@ public class ExecutionManager implements SafeCloseable {
private final DurableConfig durableConfig;
private final Set updatedOperationIdsSinceLastInvocation;
private final Set initialOperationIds;
+ private final SerDesRunner serDesRunner;
// ===== Thread Coordination =====
private final Map registeredOperations = new ConcurrentHashMap<>();
@@ -81,6 +83,7 @@ public ExecutionManager(DurableExecutionInput input, DurableConfig config, Conte
durableConfig = config;
this.durableExecutionArn = input.durableExecutionArn();
this.lambdaContext = lambdaContext;
+ this.serDesRunner = new SerDesRunner(config.getSerDesExecutorService());
// Store the set of operation IDs updated since the last successful invocation
this.updatedOperationIdsSinceLastInvocation =
@@ -129,6 +132,11 @@ public String getDurableExecutionArn() {
return durableExecutionArn;
}
+ /** Returns the invocation-scoped SerDes runner. */
+ public SerDesRunner getSerDesRunner() {
+ return serDesRunner;
+ }
+
/** Returns {@code true} if the execution is currently replaying completed operations. */
public boolean isReplaying() {
return executionMode.get() == ExecutionMode.REPLAY;
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 5cd40820e..92cbe1cb3 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
@@ -30,6 +30,8 @@
import software.amazon.lambda.durable.plugin.PluginInfoConverter;
import software.amazon.lambda.durable.plugin.PluginRunner;
import software.amazon.lambda.durable.plugin.UserFunctionOutcome;
+import software.amazon.lambda.durable.serde.SerDesContext;
+import software.amazon.lambda.durable.serde.SerDesRunner;
import software.amazon.lambda.durable.util.ExceptionHelper;
/**
@@ -113,6 +115,16 @@ public String getName() {
return operationIdentifier.name();
}
+ /** Returns the context used for one durable payload belonging to this operation. */
+ protected SerDesContext getSerDesContext(String payloadKind) {
+ return new SerDesContext(executionManager.getDurableExecutionArn(), getOperationId() + "/" + payloadKind);
+ }
+
+ /** Returns the invocation-scoped SerDes runner. */
+ protected SerDesRunner getSerDesRunner() {
+ return executionManager.getSerDesRunner();
+ }
+
/** Gets the parent context. */
protected DurableContextImpl getContext() {
return durableContext;
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..f99b6f514 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
@@ -70,7 +70,7 @@ private void startInvocation() {
.functionName(functionName)
.tenantId(invokeConfig.tenantId())
.build())
- .payload(payloadSerDes.serialize(this.payload));
+ .payload(getSerDesRunner().serialize(payloadSerDes, this.payload, getSerDesContext("invoke-payload")));
sendOperationUpdate(update);
}
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..eba38b72b 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
@@ -86,7 +86,7 @@ protected SerializableDurableOperation(
*/
protected T deserializeResult(String result) {
try {
- return resultSerDes.deserialize(result, resultTypeToken);
+ return getSerDesRunner().deserialize(resultSerDes, result, resultTypeToken, getSerDesContext("result"));
} catch (SerDesException e) {
logger.warn(
"Failed to deserialize {} result for operation name '{}'. Ensure the result is properly encoded.",
@@ -106,7 +106,7 @@ protected T deserializeResult(String result) {
* @return the serialized string and the deserialized result
*/
protected SerializedResult serializeAndDeserializeResult(T result) {
- var serialized = resultSerDes.serialize(result);
+ var serialized = getSerDesRunner().serialize(resultSerDes, result, getSerDesContext("result"));
var deserialized = shouldDeserializeAfterSerialization() ? deserializeResult(serialized) : result;
return new SerializedResult<>(serialized, deserialized);
}
@@ -119,7 +119,8 @@ protected SerializedResult serializeAndDeserializeResult(T result) {
*/
@SuppressWarnings("ThrowableNotThrown")
protected ErrorObject serializeException(Throwable throwable) {
- var error = ExceptionHelper.buildErrorObject(throwable, resultSerDes);
+ var errorData = getSerDesRunner().serialize(resultSerDes, throwable, getSerDesContext("exception"));
+ var error = ExceptionHelper.buildErrorObject(throwable, errorData);
if (shouldDeserializeAfterSerialization()) {
deserializeException(error);
}
@@ -153,8 +154,12 @@ 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)));
+ original = getSerDesRunner()
+ .deserialize(
+ resultSerDes,
+ errorData,
+ TypeToken.get(exceptionClass.asSubclass(Throwable.class)),
+ getSerDesContext("exception"));
if (original != null) {
original.setStackTrace(ExceptionHelper.deserializeStackTrace(errorObject.stackTrace()));
diff --git a/sdk/src/main/java/software/amazon/lambda/durable/serde/FieldMatchMode.java b/sdk/src/main/java/software/amazon/lambda/durable/serde/FieldMatchMode.java
new file mode 100644
index 000000000..36cf1e726
--- /dev/null
+++ b/sdk/src/main/java/software/amazon/lambda/durable/serde/FieldMatchMode.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.serde;
+
+/** Controls how a {@link PreviewField} matches a field in a structured value. */
+public enum FieldMatchMode {
+ /** Matches the field name at any depth in the object tree. */
+ ANYWHERE,
+
+ /** Matches the exact dot-separated path from the root object. */
+ PATH
+}
diff --git a/sdk/src/main/java/software/amazon/lambda/durable/serde/FileSystemPathEncoding.java b/sdk/src/main/java/software/amazon/lambda/durable/serde/FileSystemPathEncoding.java
new file mode 100644
index 000000000..8e7629a8f
--- /dev/null
+++ b/sdk/src/main/java/software/amazon/lambda/durable/serde/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.serde;
+
+/** Controls how durable execution and entity identifiers are encoded into filesystem paths. */
+public enum FileSystemPathEncoding {
+ /** Percent-encode identifiers to keep paths human-readable. */
+ URI,
+
+ /** Replace identifiers with fixed-length SHA-256 hashes. */
+ HASH
+}
diff --git a/sdk/src/main/java/software/amazon/lambda/durable/serde/FileSystemSerDes.java b/sdk/src/main/java/software/amazon/lambda/durable/serde/FileSystemSerDes.java
new file mode 100644
index 000000000..4f08173b4
--- /dev/null
+++ b/sdk/src/main/java/software/amazon/lambda/durable/serde/FileSystemSerDes.java
@@ -0,0 +1,544 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+package software.amazon.lambda.durable.serde;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.JsonToken;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.ObjectReader;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.Channels;
+import java.nio.charset.CodingErrorAction;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.AccessDeniedException;
+import java.nio.file.DirectoryStream;
+import java.nio.file.FileAlreadyExistsException;
+import java.nio.file.FileSystemLoopException;
+import java.nio.file.Files;
+import java.nio.file.InvalidPathException;
+import java.nio.file.LinkOption;
+import java.nio.file.NoSuchFileException;
+import java.nio.file.NotDirectoryException;
+import java.nio.file.Path;
+import java.nio.file.SecureDirectoryStream;
+import java.nio.file.StandardOpenOption;
+import java.nio.file.attribute.BasicFileAttributeView;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.ArrayList;
+import java.util.HexFormat;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.UUID;
+import java.util.function.Function;
+import java.util.regex.Pattern;
+import software.amazon.lambda.durable.TypeToken;
+import software.amazon.lambda.durable.exception.RetryableSerDesException;
+import software.amazon.lambda.durable.exception.SerDesException;
+
+/**
+ * A SerDes that stores checkpoint payloads on a durable shared filesystem.
+ *
+ * Do not use Lambda's ephemeral {@code /tmp} storage. The base path must be available to every execution environment
+ * that can serialize or deserialize the payload, such as an EFS mount or an S3 Files mount whose synchronization
+ * tradeoffs are acceptable for the workload.
+ *
+ *
The filesystem provider must support {@link SecureDirectoryStream}. Directory components are traversed relative to
+ * held parent handles with symbolic-link following disabled, and file I/O uses {@link LinkOption#NOFOLLOW_LINKS}.
+ *
+ *
Initial invocation input is serialized normally when no {@link SerDesContext} is available, because the durable
+ * execution ARN does not exist until after invocation starts. SDK-managed operation and output payloads are processed
+ * using the configured storage mode.
+ */
+public final class FileSystemSerDes implements SerDes {
+ private static final String ENVELOPE_MARKER = "__durable_execution_filesystem_serdes";
+ private static final int ENVELOPE_VERSION = 1;
+ private static final int DEFAULT_CHECKPOINT_ENVELOPE_LIMIT_BYTES = 256 * 1024 - 1024;
+ private static final ObjectMapper ENVELOPE_MAPPER = new ObjectMapper();
+ private static final ObjectReader ENVELOPE_READER = ENVELOPE_MAPPER
+ .reader()
+ .with(DeserializationFeature.FAIL_ON_TRAILING_TOKENS)
+ .with(DeserializationFeature.FAIL_ON_READING_DUP_TREE_KEY);
+ private static final Pattern SHA_256_DIGEST_PATTERN = Pattern.compile("[0-9a-f]{64}");
+
+ private final Path basePath;
+ private final FileSystemSerDesMode storageMode;
+ private final FileSystemPathEncoding pathEncoding;
+ private final SerDes delegate;
+ private final int checkpointEnvelopeLimitBytes;
+ private final Function> previewGenerator;
+
+ private FileSystemSerDes(Builder builder) {
+ basePath = builder.basePath.toAbsolutePath().normalize();
+ storageMode = builder.storageMode;
+ pathEncoding = builder.pathEncoding;
+ delegate = builder.delegate;
+ checkpointEnvelopeLimitBytes = builder.checkpointEnvelopeLimitBytes;
+ previewGenerator = builder.previewGenerator;
+ }
+
+ /** Creates a builder rooted at the given durable shared filesystem path. */
+ public static Builder builder(Path basePath) {
+ return new Builder(basePath);
+ }
+
+ @Override
+ public String serialize(Object value) {
+ return delegate.serialize(value);
+ }
+
+ @Override
+ public String serialize(Object value, SerDesContext context) {
+ var serialized = context == null ? delegate.serialize(value) : delegate.serialize(value, context);
+ if (serialized == null) {
+ return null;
+ }
+
+ if (context == null) {
+ return serialized;
+ }
+
+ if (storageMode == FileSystemSerDesMode.OVERFLOW) {
+ var inlineEnvelope = inlineEnvelope(serialized);
+ if (fitsCheckpoint(inlineEnvelope)) {
+ return inlineEnvelope;
+ }
+ }
+
+ return fileEnvelope(value, serialized, context);
+ }
+
+ @Override
+ public T deserialize(String data, TypeToken typeToken) {
+ return deserialize(data, typeToken, null);
+ }
+
+ @Override
+ public T deserialize(String data, TypeToken typeToken, SerDesContext context) {
+ if (data == null) {
+ return null;
+ }
+
+ var envelope = parseEnvelope(data);
+ if (envelope == null) {
+ return delegate.deserialize(data, typeToken);
+ }
+ if (envelope.hasNonNull("data")) {
+ var serialized = envelope.get("data").textValue();
+ verifyDigest(serialized, envelope.get("sha256").textValue());
+ return deserializeDelegate(serialized, typeToken, context);
+ }
+
+ var serialized = readPayload(envelope.get("file").textValue());
+ verifyDigest(serialized, envelope.get("sha256").textValue());
+ return deserializeDelegate(serialized, typeToken, context);
+ }
+
+ private T deserializeDelegate(String data, TypeToken typeToken, SerDesContext context) {
+ return context == null ? delegate.deserialize(data, typeToken) : delegate.deserialize(data, typeToken, context);
+ }
+
+ private String inlineEnvelope(String serialized) {
+ var envelope = ENVELOPE_MAPPER.createObjectNode();
+ envelope.put(ENVELOPE_MARKER, ENVELOPE_VERSION);
+ envelope.put("data", serialized);
+ envelope.put("sha256", sha256(serialized));
+ return writeEnvelope(envelope);
+ }
+
+ private String fileEnvelope(Object value, String serialized, SerDesContext context) {
+ var digest = sha256(serialized);
+ var file = payloadPath(context, digest);
+
+ var envelope = ENVELOPE_MAPPER.createObjectNode();
+ envelope.put(ENVELOPE_MARKER, ENVELOPE_VERSION);
+ envelope.put("file", file.toString());
+ envelope.put("sha256", digest);
+ if (previewGenerator != null) {
+ var preview = previewGenerator.apply(value);
+ if (preview != null) {
+ envelope.set("preview", ENVELOPE_MAPPER.valueToTree(preview));
+ }
+ }
+ var encoded = writeEnvelope(envelope);
+ if (!fitsCheckpoint(encoded)) {
+ throw new SerDesException("Filesystem SerDes file envelope exceeds the checkpoint payload limit");
+ }
+
+ writePayload(file, serialized);
+ return encoded;
+ }
+
+ private JsonNode parseEnvelope(String data) {
+ final JsonNode node;
+ try {
+ node = ENVELOPE_READER.readTree(data);
+ } catch (JsonProcessingException e) {
+ if (containsFilesystemMarkerField(data)) {
+ throw new SerDesException("Malformed filesystem SerDes envelope", e);
+ }
+ return null;
+ }
+ if (node == null || !node.isObject() || !node.has(ENVELOPE_MARKER)) {
+ return null;
+ }
+
+ if (!node.get(ENVELOPE_MARKER).isIntegralNumber()
+ || !node.get(ENVELOPE_MARKER).canConvertToInt()
+ || node.get(ENVELOPE_MARKER).intValue() != ENVELOPE_VERSION
+ || !isValidEnvelope(node)) {
+ throw new SerDesException("Malformed filesystem SerDes envelope");
+ }
+ return node;
+ }
+
+ private static boolean isValidEnvelope(JsonNode node) {
+ var hasData = node.has("data") && node.get("data").isTextual();
+ var hasFile = node.has("file") && node.get("file").isTextual();
+ if (hasData == hasFile) {
+ return false;
+ }
+ if (!node.has("sha256")
+ || !node.get("sha256").isTextual()
+ || !SHA_256_DIGEST_PATTERN
+ .matcher(node.get("sha256").textValue())
+ .matches()) {
+ return false;
+ }
+ if (hasData) {
+ return node.size() == 3;
+ }
+ var hasPreview = node.has("preview");
+ return (!hasPreview || node.get("preview").isObject()) && node.size() == (hasPreview ? 4 : 3);
+ }
+
+ private static boolean containsFilesystemMarkerField(String data) {
+ try (var parser = ENVELOPE_MAPPER.createParser(data)) {
+ if (parser.nextToken() != JsonToken.START_OBJECT) {
+ return false;
+ }
+ var depth = 1;
+ while (parser.nextToken() != null) {
+ var token = parser.currentToken();
+ if (token == JsonToken.FIELD_NAME && depth == 1 && ENVELOPE_MARKER.equals(parser.currentName())) {
+ return true;
+ }
+ if (token == JsonToken.START_OBJECT || token == JsonToken.START_ARRAY) {
+ depth++;
+ } else if (token == JsonToken.END_OBJECT || token == JsonToken.END_ARRAY) {
+ depth--;
+ if (depth == 0) {
+ return false;
+ }
+ }
+ }
+ } catch (IOException ignored) {
+ // The caller delegates malformed input unless a top-level marker field was observed before the failure.
+ }
+ return false;
+ }
+
+ private static void verifyDigest(String serialized, String expected) {
+ if (!expected.equals(sha256(serialized))) {
+ throw new SerDesException("Filesystem SerDes payload digest does not match stored content");
+ }
+ }
+
+ private Path payloadPath(SerDesContext context, String digest) {
+ var ownerDigest = sha256(context.durableExecutionArn() + "\0" + context.entityId());
+ var ownerPrefix = ownerDigest;
+ if (pathEncoding == FileSystemPathEncoding.URI) {
+ var encodedEntity = percentEncode(context.entityId());
+ ownerPrefix = encodedEntity.substring(0, Math.min(32, encodedEntity.length())) + "-" + ownerDigest;
+ }
+ var fileName = ownerPrefix + "-" + digest + "-" + UUID.randomUUID() + ".json";
+ var file = basePath.resolve(fileName).toAbsolutePath().normalize();
+ if (!file.startsWith(basePath)) {
+ throw new SerDesException("Filesystem SerDes path escapes the configured base path");
+ }
+ return file;
+ }
+
+ private void writePayload(Path file, String serialized) {
+ try {
+ try (var secureDirectory = openSecureDirectory(file.getParent())) {
+ var created = false;
+ rejectSymbolicLinkIfPresent(secureDirectory.directory(), file.getFileName(), "payload file");
+ try (var channel = secureDirectory
+ .directory()
+ .newByteChannel(
+ file.getFileName(),
+ Set.of(
+ StandardOpenOption.CREATE_NEW,
+ StandardOpenOption.WRITE,
+ LinkOption.NOFOLLOW_LINKS))) {
+ created = true;
+ var buffer = ByteBuffer.wrap(serialized.getBytes(StandardCharsets.UTF_8));
+ while (buffer.hasRemaining()) {
+ channel.write(buffer);
+ }
+ } catch (FileAlreadyExistsException failure) {
+ throw failure;
+ } catch (IOException failure) {
+ if (created) {
+ try {
+ secureDirectory.directory().deleteFile(file.getFileName());
+ } catch (IOException cleanupFailure) {
+ failure.addSuppressed(cleanupFailure);
+ }
+ }
+ throw failure;
+ }
+ }
+ } catch (IOException e) {
+ throw classifyFileSystemFailure("store", e);
+ }
+ }
+
+ private String readPayload(String fileValue) {
+ final Path file;
+ try {
+ var candidate = basePath.getFileSystem().getPath(fileValue);
+ if (!candidate.isAbsolute()) {
+ throw new SerDesException("Filesystem SerDes file path must be absolute");
+ }
+ file = candidate.normalize();
+ } catch (InvalidPathException | SecurityException failure) {
+ throw new SerDesException("Filesystem SerDes file path is invalid", failure);
+ }
+ if (!file.startsWith(basePath)) {
+ throw new SerDesException("Filesystem SerDes file is outside the configured base path");
+ }
+ try {
+ byte[] storedData;
+ try (var secureDirectory = openSecureDirectory(file.getParent())) {
+ rejectSymbolicLinkIfPresent(secureDirectory.directory(), file.getFileName(), "payload file");
+ try (var channel = secureDirectory
+ .directory()
+ .newByteChannel(
+ file.getFileName(),
+ Set.of(StandardOpenOption.READ, LinkOption.NOFOLLOW_LINKS));
+ var input = Channels.newInputStream(channel)) {
+ storedData = input.readAllBytes();
+ }
+ }
+ return StandardCharsets.UTF_8
+ .newDecoder()
+ .onMalformedInput(CodingErrorAction.REPORT)
+ .onUnmappableCharacter(CodingErrorAction.REPORT)
+ .decode(ByteBuffer.wrap(storedData))
+ .toString();
+ } catch (IOException e) {
+ throw classifyFileSystemFailure("load", e);
+ }
+ }
+
+ private static SerDesException classifyFileSystemFailure(String action, IOException failure) {
+ var message = "Failed to " + action + " filesystem SerDes payload";
+ if (failure instanceof AccessDeniedException
+ || failure instanceof NotDirectoryException
+ || failure instanceof FileSystemLoopException) {
+ return new SerDesException(message, failure);
+ }
+ return new RetryableSerDesException(message, failure);
+ }
+
+ private SecureDirectoryHandle openSecureDirectory(Path directory) throws IOException {
+ if (directory == null || !directory.startsWith(basePath)) {
+ throw new SerDesException("Filesystem SerDes directory is outside the configured base path");
+ }
+ var root = basePath.getRoot();
+ if (root == null) {
+ throw new SerDesException("Filesystem SerDes base path must be absolute");
+ }
+
+ var openedStreams = new ArrayList>();
+ try {
+ var current = requireSecureDirectoryStream(Files.newDirectoryStream(root), openedStreams);
+ for (var component : root.relativize(directory)) {
+ rejectSymbolicLinkIfPresent(current, component, "directory");
+ var next = current.newDirectoryStream(component, LinkOption.NOFOLLOW_LINKS);
+ current = requireSecureDirectoryStream(next, openedStreams);
+ }
+ return new SecureDirectoryHandle(current, openedStreams);
+ } catch (IOException | RuntimeException failure) {
+ closeDirectoryStreams(openedStreams, failure);
+ throw failure;
+ }
+ }
+
+ private static void rejectSymbolicLinkIfPresent(
+ SecureDirectoryStream directory, Path entry, String description) throws IOException {
+ var attributes = directory.getFileAttributeView(entry, BasicFileAttributeView.class, LinkOption.NOFOLLOW_LINKS);
+ if (attributes == null) {
+ throw new SerDesException("Filesystem provider cannot inspect " + description + " without following links");
+ }
+ try {
+ if (attributes.readAttributes().isSymbolicLink()) {
+ throw new SerDesException("Filesystem SerDes " + description + " cannot be a symbolic link");
+ }
+ } catch (NoSuchFileException ignored) {
+ // Missing entries are handled by the caller as either creatable directories or retryable read failures.
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private static SecureDirectoryStream requireSecureDirectoryStream(
+ DirectoryStream stream, List> openedStreams) {
+ openedStreams.add(stream);
+ if (stream instanceof SecureDirectoryStream secureStream) {
+ return (SecureDirectoryStream) secureStream;
+ }
+ throw new SerDesException("FileSystemSerDes requires a filesystem provider with SecureDirectoryStream support");
+ }
+
+ private static void closeDirectoryStreams(List> streams, Throwable failure) {
+ for (int index = streams.size() - 1; index >= 0; index--) {
+ try {
+ streams.get(index).close();
+ } catch (IOException closeFailure) {
+ failure.addSuppressed(closeFailure);
+ }
+ }
+ }
+
+ private boolean fitsCheckpoint(String envelope) {
+ return envelope.getBytes(StandardCharsets.UTF_8).length <= checkpointEnvelopeLimitBytes;
+ }
+
+ private static String percentEncode(String value) {
+ var bytes = value.getBytes(StandardCharsets.UTF_8);
+ var encoded = new StringBuilder(bytes.length);
+ for (byte raw : bytes) {
+ int valueByte = raw & 0xff;
+ if ((valueByte >= 'a' && valueByte <= 'z')
+ || (valueByte >= 'A' && valueByte <= 'Z')
+ || (valueByte >= '0' && valueByte <= '9')
+ || valueByte == '-'
+ || valueByte == '_'
+ || valueByte == '.'
+ || valueByte == '~') {
+ encoded.append((char) valueByte);
+ } else {
+ encoded.append('%');
+ encoded.append(Character.toUpperCase(Character.forDigit((valueByte >>> 4) & 0xf, 16)));
+ encoded.append(Character.toUpperCase(Character.forDigit(valueByte & 0xf, 16)));
+ }
+ }
+ return encoded.toString();
+ }
+
+ private static String sha256(String value) {
+ 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);
+ }
+ }
+
+ private static String writeEnvelope(JsonNode envelope) {
+ try {
+ return ENVELOPE_MAPPER.writeValueAsString(envelope);
+ } catch (IOException e) {
+ throw new SerDesException("Failed to create filesystem SerDes envelope", e);
+ }
+ }
+
+ private static final class SecureDirectoryHandle implements AutoCloseable {
+ private final SecureDirectoryStream directory;
+ private final List> openedStreams;
+
+ private SecureDirectoryHandle(
+ SecureDirectoryStream directory, List> openedStreams) {
+ this.directory = directory;
+ this.openedStreams = List.copyOf(openedStreams);
+ }
+
+ private SecureDirectoryStream directory() {
+ return directory;
+ }
+
+ @Override
+ public void close() throws IOException {
+ IOException failure = null;
+ for (int index = openedStreams.size() - 1; index >= 0; index--) {
+ try {
+ openedStreams.get(index).close();
+ } catch (IOException closeFailure) {
+ if (failure == null) {
+ failure = closeFailure;
+ } else {
+ failure.addSuppressed(closeFailure);
+ }
+ }
+ }
+ if (failure != null) {
+ throw failure;
+ }
+ }
+ }
+
+ /** Builder for {@link FileSystemSerDes}. */
+ public static final class Builder {
+ private final Path basePath;
+ private FileSystemSerDesMode storageMode = FileSystemSerDesMode.ALWAYS;
+ private FileSystemPathEncoding pathEncoding = FileSystemPathEncoding.URI;
+ private SerDes delegate = new JacksonSerDes();
+ private int checkpointEnvelopeLimitBytes = DEFAULT_CHECKPOINT_ENVELOPE_LIMIT_BYTES;
+ private Function> previewGenerator;
+
+ private Builder(Path basePath) {
+ this.basePath = Objects.requireNonNull(basePath, "basePath cannot be null");
+ }
+
+ public Builder storageMode(FileSystemSerDesMode 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 delegate(SerDes delegate) {
+ this.delegate = Objects.requireNonNull(delegate, "delegate cannot be null");
+ return this;
+ }
+
+ /** Sets the maximum UTF-8 size of an inline or file checkpoint envelope. */
+ public Builder checkpointEnvelopeLimitBytes(int checkpointEnvelopeLimitBytes) {
+ if (checkpointEnvelopeLimitBytes <= 0) {
+ throw new IllegalArgumentException("checkpointEnvelopeLimitBytes must be positive");
+ }
+ if (checkpointEnvelopeLimitBytes > DEFAULT_CHECKPOINT_ENVELOPE_LIMIT_BYTES) {
+ throw new IllegalArgumentException(
+ "checkpointEnvelopeLimitBytes cannot exceed " + DEFAULT_CHECKPOINT_ENVELOPE_LIMIT_BYTES);
+ }
+ this.checkpointEnvelopeLimitBytes = checkpointEnvelopeLimitBytes;
+ return this;
+ }
+
+ public Builder previewGenerator(Function> previewGenerator) {
+ this.previewGenerator = Objects.requireNonNull(previewGenerator, "previewGenerator cannot be null");
+ return this;
+ }
+
+ /** Configures structured preview generation from the original value. */
+ public Builder previewConfig(PreviewConfig previewConfig) {
+ Objects.requireNonNull(previewConfig, "previewConfig cannot be null");
+ this.previewGenerator = value -> SerDesPreview.buildPreview(value, previewConfig);
+ return this;
+ }
+
+ public FileSystemSerDes build() {
+ return new FileSystemSerDes(this);
+ }
+ }
+}
diff --git a/sdk/src/main/java/software/amazon/lambda/durable/serde/FileSystemSerDesMode.java b/sdk/src/main/java/software/amazon/lambda/durable/serde/FileSystemSerDesMode.java
new file mode 100644
index 000000000..3f4ed7ae6
--- /dev/null
+++ b/sdk/src/main/java/software/amazon/lambda/durable/serde/FileSystemSerDesMode.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.serde;
+
+/** Controls when {@link FileSystemSerDes} stores serialized data on the filesystem. */
+public enum FileSystemSerDesMode {
+ /** Store every SDK-managed payload on the filesystem. */
+ ALWAYS,
+
+ /** Keep small payloads inline and store only payloads that exceed the checkpoint threshold. */
+ OVERFLOW
+}
diff --git a/sdk/src/main/java/software/amazon/lambda/durable/serde/PreviewConfig.java b/sdk/src/main/java/software/amazon/lambda/durable/serde/PreviewConfig.java
new file mode 100644
index 000000000..7d2f8ebf4
--- /dev/null
+++ b/sdk/src/main/java/software/amazon/lambda/durable/serde/PreviewConfig.java
@@ -0,0 +1,113 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+package software.amazon.lambda.durable.serde;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * Configuration for {@link SerDesPreview#buildPreview(Object, PreviewConfig)}.
+ *
+ * @param mode whether fields are included or excluded by default
+ * @param include fields made visible in {@link PreviewMode#EXCLUDE_ALL} mode
+ * @param exclude fields hidden from the preview
+ * @param mask fields made visible with their values replaced by {@code maskString}
+ * @param maskString replacement for masked field values
+ * @param maxPreviewBytes maximum estimated UTF-8 size of accepted preview entries
+ */
+public record PreviewConfig(
+ PreviewMode mode,
+ List include,
+ List exclude,
+ List