diff --git a/sdk/src/main/java/software/amazon/lambda/durable/offload/PayloadOffloadRetryExecutor.java b/sdk/src/main/java/software/amazon/lambda/durable/offload/PayloadOffloadRetryExecutor.java new file mode 100644 index 000000000..6b0df53cf --- /dev/null +++ b/sdk/src/main/java/software/amazon/lambda/durable/offload/PayloadOffloadRetryExecutor.java @@ -0,0 +1,88 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.offload; + +import java.time.Duration; +import java.util.Objects; +import java.util.concurrent.TimeUnit; +import java.util.function.Supplier; +import software.amazon.lambda.durable.exception.PayloadOffloadException; +import software.amazon.lambda.durable.exception.RetryablePayloadOffloadException; +import software.amazon.lambda.durable.retry.RetryDecision; +import software.amazon.lambda.durable.retry.RetryStrategy; + +final class PayloadOffloadRetryExecutor { + static final Sleeper DEFAULT_SLEEPER = delay -> { + if (delay.getSeconds() > 0) { + TimeUnit.SECONDS.sleep(delay.getSeconds()); + } + if (delay.getNano() > 0) { + TimeUnit.NANOSECONDS.sleep(delay.getNano()); + } + }; + + private final RetryStrategy retryStrategy; + private final Sleeper sleeper; + + PayloadOffloadRetryExecutor(RetryStrategy retryStrategy, Sleeper sleeper) { + this.retryStrategy = Objects.requireNonNull(retryStrategy, "retryStrategy cannot be null"); + this.sleeper = Objects.requireNonNull(sleeper, "sleeper cannot be null"); + } + + T execute(String action, Supplier operation) { + int attempt = 1; + while (true) { + try { + return operation.get(); + } catch (RetryablePayloadOffloadException failure) { + var decision = makeRetryDecision(action, failure, attempt); + if (!decision.shouldRetry()) { + throw failure; + } + waitForRetry(action, failure, attempt, decision.delay()); + attempt++; + } + } + } + + private RetryDecision makeRetryDecision(String action, RetryablePayloadOffloadException failure, int attempt) { + try { + var decision = retryStrategy.makeRetryDecision(failure, attempt); + if (decision == null) { + throw new PayloadOffloadException( + String.format("Retry strategy returned null for payload %s attempt %d", action, attempt)); + } + return decision; + } catch (PayloadOffloadException e) { + throw e; + } catch (RuntimeException e) { + throw new PayloadOffloadException( + String.format("Retry strategy failed for payload %s attempt %d", action, attempt), e); + } + } + + private void waitForRetry(String action, RetryablePayloadOffloadException failure, int attempt, Duration delay) { + if (delay == null || delay.isNegative()) { + throw new PayloadOffloadException(String.format( + "Retry strategy returned an invalid delay for payload %s attempt %d", action, attempt)); + } + if (delay.isZero()) { + return; + } + try { + sleeper.sleep(delay); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + var interrupted = new RetryablePayloadOffloadException( + String.format("Interrupted while waiting to retry payload %s after attempt %d", action, attempt), + e); + interrupted.addSuppressed(failure); + throw interrupted; + } + } + + @FunctionalInterface + interface Sleeper { + void sleep(Duration delay) throws InterruptedException; + } +} diff --git a/sdk/src/main/java/software/amazon/lambda/durable/offload/RetryPayloadOffloader.java b/sdk/src/main/java/software/amazon/lambda/durable/offload/RetryPayloadOffloader.java new file mode 100644 index 000000000..e138aea2b --- /dev/null +++ b/sdk/src/main/java/software/amazon/lambda/durable/offload/RetryPayloadOffloader.java @@ -0,0 +1,32 @@ +// 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.lambda.durable.retry.RetryStrategy; + +/** A payload-offloader decorator that retries explicitly retryable storage failures. */ +public final class RetryPayloadOffloader implements PayloadOffloader { + private final PayloadOffloader delegate; + private final PayloadOffloadRetryExecutor retryExecutor; + + public RetryPayloadOffloader(PayloadOffloader delegate, RetryStrategy retryStrategy) { + this(delegate, retryStrategy, PayloadOffloadRetryExecutor.DEFAULT_SLEEPER); + } + + RetryPayloadOffloader( + PayloadOffloader delegate, RetryStrategy retryStrategy, PayloadOffloadRetryExecutor.Sleeper sleeper) { + this.delegate = Objects.requireNonNull(delegate, "delegate cannot be null"); + retryExecutor = new PayloadOffloadRetryExecutor(retryStrategy, sleeper); + } + + @Override + public OffloadedPayload offload(String serializedPayload, PayloadOffloadContext context) { + return retryExecutor.execute("store", () -> delegate.offload(serializedPayload, context)); + } + + @Override + public String load(OffloadedPayload payload, PayloadOffloadContext context) { + return retryExecutor.execute("load", () -> delegate.load(payload, context)); + } +} diff --git a/sdk/src/main/java/software/amazon/lambda/durable/offload/filesystem/FieldMatchMode.java b/sdk/src/main/java/software/amazon/lambda/durable/offload/filesystem/FieldMatchMode.java new file mode 100644 index 000000000..7c51b68db --- /dev/null +++ b/sdk/src/main/java/software/amazon/lambda/durable/offload/filesystem/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.offload.filesystem; + +/** 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/offload/filesystem/FileSystemPathEncoding.java b/sdk/src/main/java/software/amazon/lambda/durable/offload/filesystem/FileSystemPathEncoding.java new file mode 100644 index 000000000..ac7f6ab71 --- /dev/null +++ b/sdk/src/main/java/software/amazon/lambda/durable/offload/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.offload.filesystem; + +/** Controls how durable execution ownership is represented in payload file names. */ +public enum FileSystemPathEncoding { + /** Include a bounded, escaped entity prefix followed by a SHA-256 owner digest. */ + URI, + + /** Use only the fixed-length SHA-256 owner digest. */ + HASH +} diff --git a/sdk/src/main/java/software/amazon/lambda/durable/offload/filesystem/FileSystemPayloadOffloader.java b/sdk/src/main/java/software/amazon/lambda/durable/offload/filesystem/FileSystemPayloadOffloader.java new file mode 100644 index 000000000..05982f38e --- /dev/null +++ b/sdk/src/main/java/software/amazon/lambda/durable/offload/filesystem/FileSystemPayloadOffloader.java @@ -0,0 +1,503 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.offload.filesystem; + +import java.io.IOException; +import java.io.InterruptedIOException; +import java.nio.ByteBuffer; +import java.nio.channels.Channels; +import java.nio.channels.ClosedByInterruptException; +import java.nio.charset.StandardCharsets; +import java.nio.file.DirectoryStream; +import java.nio.file.FileAlreadyExistsException; +import java.nio.file.FileSystemException; +import java.nio.file.Files; +import java.nio.file.LinkOption; +import java.nio.file.NoSuchFileException; +import java.nio.file.Path; +import java.nio.file.ReadOnlyFileSystemException; +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.Locale; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import java.util.UUID; +import java.util.regex.Pattern; +import software.amazon.awssdk.services.lambda.model.OperationType; +import software.amazon.lambda.durable.exception.PayloadOffloadException; +import software.amazon.lambda.durable.exception.RetryablePayloadOffloadException; +import software.amazon.lambda.durable.execution.PayloadCodec; +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; +import software.amazon.lambda.durable.offload.SerDesPayloadKind; + +/** + * Stores serialized payloads on a durable, shared filesystem. + * + *

Do not use Lambda's ephemeral {@code /tmp} storage. Use a durable shared mount such as EFS, or S3 Files only when + * its synchronization and crash-durability tradeoffs are acceptable for the workload. + * + *

Payload files are immutable and published with a single {@code CREATE_NEW} write. Each SDK envelope records the + * producing execution and entity plus a SHA-256 content digest. Loads validate ownership, path, file name, and content + * before returning data. + * + *

The configured base path and all ancestors must already exist. Payload files are direct children of that path. The + * filesystem provider must support {@link SecureDirectoryStream}; directory handles remain open through file I/O and + * symbolic-link following is disabled. + */ +public final class FileSystemPayloadOffloader implements PayloadOffloader { + private static final int DEFAULT_CHECKPOINT_ENVELOPE_LIMIT_BYTES = 256 * 1024 - 1024; + private static final int MAX_URI_OWNER_PREFIX_LENGTH = 32; + private static final Pattern SHA_256_DIGEST_PATTERN = Pattern.compile("[0-9a-f]{64}"); + + private final Path basePath; + private final PayloadOffloadMode storageMode; + private final FileSystemPathEncoding pathEncoding; + private final int checkpointEnvelopeLimitBytes; + private final PayloadPreviewGenerator previewGenerator; + + private FileSystemPayloadOffloader(Builder builder) { + basePath = builder.basePath.toAbsolutePath().normalize(); + storageMode = builder.storageMode; + pathEncoding = builder.pathEncoding; + checkpointEnvelopeLimitBytes = builder.checkpointEnvelopeLimitBytes; + 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"); + context = requireContext(context); + var payloadDigest = sha256(serializedPayload); + if (storageMode == PayloadOffloadMode.OVERFLOW) { + var inlinePayload = OffloadedPayload.inline( + serializedPayload, context.durableExecutionArn(), context.entityId(), payloadDigest) + .bindProducer(context, payloadDigest); + if (fitsCheckpoint(inlinePayload)) { + return inlinePayload; + } + } + + var path = resolvePayloadPath(payloadDigest, context); + var preview = generatePreview(serializedPayload, context); + var referencePayload = OffloadedPayload.reference( + path.toString(), preview, context.durableExecutionArn(), context.entityId(), payloadDigest) + .bindProducer(context, payloadDigest); + if (!fitsCheckpoint(referencePayload)) { + throw new PayloadOffloadException( + "Filesystem payload envelope exceeds the checkpoint limit for entity '" + context.entityId() + "'"); + } + try { + writePayload(serializedPayload, path); + return referencePayload; + } catch (IOException e) { + throw classifyIoFailure("store", context, e); + } + } + + @Override + public String load(OffloadedPayload payload, PayloadOffloadContext context) { + Objects.requireNonNull(payload, "payload cannot be null"); + context = requireContext(context); + requireIntegrityMetadata(payload, context); + validateOwner(payload, context); + if (payload.mode() == PayloadStorageMode.INLINE) { + verifyDigest(payload.data(), payload.payloadDigest(), context); + return payload.data(); + } + + var path = basePath.getFileSystem() + .getPath(payload.reference()) + .toAbsolutePath() + .normalize(); + validatePayloadPath(path, payload); + try { + byte[] storedData; + try (var directory = openSecureDirectory(path.getParent()); + var channel = openPayloadForRead(directory.directory(), path.getFileName(), context); + var input = Channels.newInputStream(channel)) { + storedData = input.readAllBytes(); + } + var serialized = new String(storedData, StandardCharsets.UTF_8); + verifyDigest(serialized, payload.payloadDigest(), context); + return serialized; + } catch (IOException e) { + throw classifyIoFailure("load", context, e); + } + } + + private boolean fitsCheckpoint(OffloadedPayload payload) { + return PayloadCodec.envelopeSizeBytes(payload) <= checkpointEnvelopeLimitBytes; + } + + private Map generatePreview(String serializedPayload, PayloadOffloadContext context) { + if (previewGenerator == null) { + return null; + } + try { + return previewGenerator.generate(serializedPayload, context); + } catch (RuntimeException e) { + if (e instanceof RetryablePayloadOffloadException retryablePayloadOffloadException) { + throw retryablePayloadOffloadException; + } + if (e instanceof PayloadOffloadException payloadOffloadException) { + throw payloadOffloadException; + } + throw new PayloadOffloadException( + "Failed to generate filesystem payload preview for entity '" + context.entityId() + "'", e); + } + } + + private static PayloadOffloadContext requireContext(PayloadOffloadContext context) { + if (context == null + || context.durableExecutionArn() == null + || context.durableExecutionArn().isBlank() + || context.entityId() == null + || context.entityId().isBlank()) { + throw new PayloadOffloadException("FileSystemPayloadOffloader requires durableExecutionArn and entityId"); + } + return context; + } + + private static void requireIntegrityMetadata(OffloadedPayload payload, PayloadOffloadContext context) { + if (!payload.hasIntegrityMetadata()) { + throw new PayloadOffloadException("Filesystem payload is missing ownership or digest metadata for entity '" + + context.entityId() + "'"); + } + } + + private static void validateOwner(OffloadedPayload payload, PayloadOffloadContext context) { + var sameOwner = payload.ownerDurableExecutionArn().equals(context.durableExecutionArn()) + && payload.ownerEntityId().equals(context.entityId()); + if (!sameOwner && !acceptsCrossExecutionReference(context)) { + throw new PayloadOffloadException("Filesystem payload belongs to a different durable entity"); + } + } + + private static boolean acceptsCrossExecutionReference(PayloadOffloadContext context) { + return context.payloadKind() == SerDesPayloadKind.INPUT + || context.operationType() == OperationType.CHAINED_INVOKE; + } + + private void validatePayloadPath(Path path, OffloadedPayload payload) { + var expectedPrefix = payloadOwnerPrefix(payload.ownerDurableExecutionArn(), payload.ownerEntityId()) + + "-" + + payload.payloadDigest() + + "-"; + var fileName = path.getFileName(); + if (fileName == null + || path.getParent() == null + || !path.getParent().equals(basePath) + || !fileName.toString().startsWith(expectedPrefix) + || !fileName.toString().endsWith(".payload")) { + throw new PayloadOffloadException("Filesystem payload path is not valid for its declared durable entity"); + } + } + + private Path resolvePayloadPath(String payloadDigest, PayloadOffloadContext context) { + var fileName = payloadOwnerPrefix(context.durableExecutionArn(), context.entityId()) + + "-" + + payloadDigest + + "-" + + UUID.randomUUID() + + ".payload"; + var file = basePath.resolve(fileName).normalize(); + if (!basePath.equals(file.getParent())) { + throw new PayloadOffloadException("Resolved filesystem payload path is outside the configured base path"); + } + return file; + } + + private String payloadOwnerPrefix(String durableExecutionArn, String entityId) { + var ownerDigest = sha256(durableExecutionArn + "\0" + entityId); + if (pathEncoding == FileSystemPathEncoding.HASH) { + return ownerDigest; + } + var readableEntity = encode(entityId); + var readablePrefix = + readableEntity.substring(0, Math.min(readableEntity.length(), MAX_URI_OWNER_PREFIX_LENGTH)); + return readablePrefix + "-" + ownerDigest; + } + + private String encode(String value) { + if (pathEncoding == FileSystemPathEncoding.HASH) { + return sha256(value); + } + var encoded = new StringBuilder(); + for (byte valueByte : value.getBytes(StandardCharsets.UTF_8)) { + int current = valueByte & 0xff; + if (current >= 'a' && current <= 'z' + || current >= 'A' && current <= 'Z' + || current >= '0' && current <= '9' + || current == '-' + || current == '_' + || current == '.' + || current == '~') { + encoded.append((char) current); + } else { + encoded.append('%'); + encoded.append(Character.toUpperCase(Character.forDigit(current >>> 4, 16))); + encoded.append(Character.toUpperCase(Character.forDigit(current & 0xf, 16))); + } + } + if (".".contentEquals(encoded) || "..".contentEquals(encoded)) { + return encoded.toString().replace(".", "%2E"); + } + return encoded.toString(); + } + + private void writePayload(String serializedPayload, Path file) throws IOException { + try (var directory = openSecureDirectory(file.getParent())) { + if (Files.getFileStore(basePath).isReadOnly()) { + throw new PayloadOffloadException("Filesystem payload base path is read-only"); + } + var created = false; + try (var channel = directory + .directory() + .newByteChannel( + file.getFileName(), + Set.of( + StandardOpenOption.CREATE_NEW, + StandardOpenOption.WRITE, + LinkOption.NOFOLLOW_LINKS))) { + created = true; + var buffer = ByteBuffer.wrap(serializedPayload.getBytes(StandardCharsets.UTF_8)); + while (buffer.hasRemaining()) { + channel.write(buffer); + } + } catch (FileAlreadyExistsException failure) { + throw failure; + } catch (IOException failure) { + if (created) { + try { + directory.directory().deleteFile(file.getFileName()); + } catch (IOException cleanupFailure) { + failure.addSuppressed(cleanupFailure); + } + } + throw failure; + } + } catch (ReadOnlyFileSystemException failure) { + throw new PayloadOffloadException("Filesystem payload base path is read-only", failure); + } + } + + private SecureDirectoryHandle openSecureDirectory(Path directory) throws IOException { + if (directory == null || !directory.startsWith(basePath)) { + throw new PayloadOffloadException("Filesystem payload directory is outside the configured base path"); + } + var root = basePath.getRoot(); + if (root == null) { + throw new PayloadOffloadException("Filesystem payload base path must be absolute"); + } + + var openedStreams = new ArrayList>(); + try { + var current = requireSecureDirectoryStream(Files.newDirectoryStream(root), openedStreams); + for (var component : root.relativize(directory)) { + try { + var attributes = readAttributes(current, component); + if (attributes.isSymbolicLink()) { + throw new PayloadOffloadException("Filesystem payload base path cannot contain symbolic links"); + } + if (!attributes.isDirectory()) { + throw new PayloadOffloadException( + "Filesystem payload base path and all ancestors must be directories"); + } + current = requireSecureDirectoryStream( + current.newDirectoryStream(component, LinkOption.NOFOLLOW_LINKS), openedStreams); + } catch (NoSuchFileException missing) { + throw new PayloadOffloadException( + "Filesystem payload base path and all ancestors must already exist", missing); + } + } + return new SecureDirectoryHandle(current, openedStreams); + } catch (IOException | RuntimeException failure) { + closeDirectoryStreams(openedStreams, failure); + throw failure; + } + } + + private java.nio.channels.SeekableByteChannel openPayloadForRead( + SecureDirectoryStream directory, Path fileName, PayloadOffloadContext context) throws IOException { + var attributes = readAttributes(directory, fileName); + if (attributes.isSymbolicLink() || !attributes.isRegularFile()) { + throw new PayloadOffloadException( + "Filesystem payload must be a regular file for entity '" + context.entityId() + "'"); + } + return directory.newByteChannel(fileName, Set.of(StandardOpenOption.READ, LinkOption.NOFOLLOW_LINKS)); + } + + private static java.nio.file.attribute.BasicFileAttributes readAttributes( + SecureDirectoryStream directory, Path path) throws IOException { + var view = directory.getFileAttributeView(path, BasicFileAttributeView.class, LinkOption.NOFOLLOW_LINKS); + if (view == null) { + throw new PayloadOffloadException("FileSystemPayloadOffloader requires basic file attribute support"); + } + return view.readAttributes(); + } + + static PayloadOffloadException classifyIoFailure( + String action, PayloadOffloadContext context, IOException failure) { + var message = "Failed to " + action + " filesystem payload for entity '" + context.entityId() + "'"; + if (isKnownTransientFailure(failure)) { + return new RetryablePayloadOffloadException(message, failure); + } + return new PayloadOffloadException(message, failure); + } + + private static boolean isKnownTransientFailure(IOException failure) { + if (failure instanceof FileAlreadyExistsException + || failure instanceof InterruptedIOException + || failure instanceof ClosedByInterruptException) { + return true; + } + if (!(failure instanceof FileSystemException fileSystemFailure) || fileSystemFailure.getReason() == null) { + return false; + } + var reason = fileSystemFailure.getReason().toLowerCase(Locale.ROOT); + return reason.contains("stale file handle") + || reason.contains("resource temporarily unavailable") + || reason.contains("connection reset") + || reason.contains("connection timed out") + || reason.contains("network is unreachable") + || reason.contains("no route to host") + || reason.contains("transport endpoint is not connected"); + } + + @SuppressWarnings("unchecked") + private static SecureDirectoryStream requireSecureDirectoryStream( + DirectoryStream stream, List> openedStreams) { + openedStreams.add(stream); + if (stream instanceof SecureDirectoryStream secureStream) { + return (SecureDirectoryStream) secureStream; + } + throw new PayloadOffloadException( + "FileSystemPayloadOffloader 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 static void verifyDigest(String serializedPayload, String expectedDigest, PayloadOffloadContext context) { + if (!SHA_256_DIGEST_PATTERN.matcher(expectedDigest).matches() + || !sha256(serializedPayload).equals(expectedDigest)) { + throw new PayloadOffloadException( + "Filesystem payload digest does not match stored content for entity '" + context.entityId() + "'"); + } + } + + 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 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 FileSystemPayloadOffloader}. */ + public static final class Builder { + private final Path basePath; + private PayloadOffloadMode storageMode = PayloadOffloadMode.ALWAYS; + private FileSystemPathEncoding pathEncoding = FileSystemPathEncoding.URI; + private int checkpointEnvelopeLimitBytes = DEFAULT_CHECKPOINT_ENVELOPE_LIMIT_BYTES; + 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; + } + + /** Sets the maximum UTF-8 size allowed for inline and reference checkpoint envelopes. */ + public Builder checkpointEnvelopeLimitBytes(int checkpointEnvelopeLimitBytes) { + if (checkpointEnvelopeLimitBytes < 1 + || checkpointEnvelopeLimitBytes > DEFAULT_CHECKPOINT_ENVELOPE_LIMIT_BYTES) { + throw new IllegalArgumentException("checkpointEnvelopeLimitBytes must be between 1 and " + + DEFAULT_CHECKPOINT_ENVELOPE_LIMIT_BYTES); + } + this.checkpointEnvelopeLimitBytes = checkpointEnvelopeLimitBytes; + return this; + } + + /** Configures built-in structured JSON preview generation. */ + public Builder previewConfig(PreviewConfig previewConfig) { + Objects.requireNonNull(previewConfig, "previewConfig cannot be null"); + previewGenerator = (serialized, context) -> PayloadPreview.buildPreviewFromJson(serialized, previewConfig); + return this; + } + + public Builder previewGenerator(PayloadPreviewGenerator previewGenerator) { + this.previewGenerator = previewGenerator; + return this; + } + + public FileSystemPayloadOffloader build() { + return new FileSystemPayloadOffloader(this); + } + } +} diff --git a/sdk/src/main/java/software/amazon/lambda/durable/offload/filesystem/PayloadOffloadMode.java b/sdk/src/main/java/software/amazon/lambda/durable/offload/filesystem/PayloadOffloadMode.java new file mode 100644 index 000000000..d49bf54ec --- /dev/null +++ b/sdk/src/main/java/software/amazon/lambda/durable/offload/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.offload.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/sdk/src/main/java/software/amazon/lambda/durable/offload/filesystem/PayloadPreview.java b/sdk/src/main/java/software/amazon/lambda/durable/offload/filesystem/PayloadPreview.java new file mode 100644 index 000000000..027e33e2b --- /dev/null +++ b/sdk/src/main/java/software/amazon/lambda/durable/offload/filesystem/PayloadPreview.java @@ -0,0 +1,182 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.offload.filesystem; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.DeserializationFeature; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.SerializationFeature; +import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import software.amazon.lambda.durable.exception.PayloadOffloadException; + +/** Utilities for building compact structured previews for externally stored payloads. */ +public final class PayloadPreview { + private static final ObjectMapper MAPPER = new ObjectMapper() + .registerModule(new JavaTimeModule()) + .disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS) + .disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES); + + private PayloadPreview() {} + + /** Builds a preview from an object using include, exclude, mask, path-matching, and byte-budget rules. */ + public static Map buildPreview(Object value, PreviewConfig config) { + Objects.requireNonNull(config, "config cannot be null"); + final JsonNode root; + try { + root = MAPPER.valueToTree(value); + } catch (IllegalArgumentException e) { + throw new PayloadOffloadException("Failed to convert value for preview generation", e); + } + return buildPreview(root, config); + } + + /** Builds a preview from a JSON string. */ + public static Map buildPreviewFromJson(String value, PreviewConfig config) { + Objects.requireNonNull(value, "value cannot be null"); + Objects.requireNonNull(config, "config cannot be null"); + try { + return buildPreview(MAPPER.readTree(value), config); + } catch (JsonProcessingException e) { + return null; + } + } + + private static Map buildPreview(JsonNode root, PreviewConfig config) { + if (root == null || !root.isObject()) { + return null; + } + + var pairs = new ArrayList(); + collect(root, List.of(), false, config, pairs); + if (pairs.isEmpty()) { + return null; + } + + Map result = new LinkedHashMap<>(); + for (var pair : pairs) { + var candidate = copy(result); + insert(candidate, pair.path(), pair.value()); + if (serializedSize(candidate) > config.maxPreviewBytes()) { + break; + } + result = candidate; + } + return result.isEmpty() ? null : result; + } + + private static void collect( + JsonNode node, + List pathPrefix, + boolean inheritedInclude, + PreviewConfig config, + List pairs) { + if (node == null || node.isNull()) { + return; + } + if (node.isArray()) { + // Container arrays cannot be represented faithfully by the map-based preview shape without preserving + // indices. Omit them instead of collapsing elements onto the same path and silently overwriting values. + return; + } + if (!node.isObject()) { + return; + } + + for (var field : node.properties()) { + var name = field.getKey(); + var path = append(pathPrefix, name); + var masked = isMatched(path, config.mask()); + var excluded = isMatched(path, config.exclude()); + var explicitlyIncluded = isMatched(path, config.include()); + var visible = !excluded + && (masked || inheritedInclude || config.mode() == PreviewMode.INCLUDE_ALL || explicitlyIncluded); + + if (!visible) { + if (!excluded) { + collect(field.getValue(), path, false, config, pairs); + } + continue; + } + if (masked) { + pairs.add(new PreviewEntry(path, config.maskString())); + } else if (isScalarArray(field.getValue())) { + pairs.add(new PreviewEntry(path, MAPPER.convertValue(field.getValue(), Object.class))); + } else if (field.getValue().isContainerNode()) { + collect(field.getValue(), path, inheritedInclude || explicitlyIncluded, config, pairs); + } else { + pairs.add(new PreviewEntry(path, MAPPER.convertValue(field.getValue(), Object.class))); + } + } + } + + private static boolean isScalarArray(JsonNode node) { + if (!node.isArray()) { + return false; + } + for (var item : node) { + if (item.isContainerNode()) { + return false; + } + } + return true; + } + + private static boolean isMatched(List path, List fields) { + for (var field : fields) { + if (field.match() == FieldMatchMode.PATH) { + if (path.equals(field.pathSegments())) { + return true; + } + } else if (path.contains(field.name())) { + return true; + } + } + return false; + } + + private static List append(List path, String field) { + var result = new ArrayList<>(path); + result.add(field); + return List.copyOf(result); + } + + private static int serializedSize(Map preview) { + try { + return MAPPER.writeValueAsBytes(preview).length; + } catch (JsonProcessingException e) { + throw new PayloadOffloadException("Failed to measure preview size", e); + } + } + + @SuppressWarnings("unchecked") + private static Map copy(Map source) { + var copy = new LinkedHashMap(); + for (var entry : source.entrySet()) { + var value = entry.getValue(); + copy.put(entry.getKey(), value instanceof Map nested ? copy((Map) nested) : value); + } + return copy; + } + + @SuppressWarnings("unchecked") + private static void insert(Map result, List path, Object value) { + Map current = result; + for (int index = 0; index < path.size() - 1; index++) { + var existing = current.get(path.get(index)); + if (!(existing instanceof Map)) { + existing = new LinkedHashMap(); + current.put(path.get(index), existing); + } + current = (Map) existing; + } + current.put(path.get(path.size() - 1), value); + } + + private record PreviewEntry(List path, Object value) {} +} diff --git a/sdk/src/main/java/software/amazon/lambda/durable/offload/filesystem/PayloadPreviewGenerator.java b/sdk/src/main/java/software/amazon/lambda/durable/offload/filesystem/PayloadPreviewGenerator.java new file mode 100644 index 000000000..80e904dea --- /dev/null +++ b/sdk/src/main/java/software/amazon/lambda/durable/offload/filesystem/PayloadPreviewGenerator.java @@ -0,0 +1,17 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.offload.filesystem; + +import java.util.Map; +import software.amazon.lambda.durable.offload.PayloadOffloadContext; + +/** + * Generates optional inline preview metadata for an externally stored serialized payload. + * + *

Preview values must be JSON-compatible maps, collections, arrays, or scalar values. The SDK snapshots containers + * and normalizes arrays to immutable lists before encoding the payload envelope. + */ +@FunctionalInterface +public interface PayloadPreviewGenerator { + Map generate(String serializedPayload, PayloadOffloadContext context); +} diff --git a/sdk/src/main/java/software/amazon/lambda/durable/offload/filesystem/PreviewConfig.java b/sdk/src/main/java/software/amazon/lambda/durable/offload/filesystem/PreviewConfig.java new file mode 100644 index 000000000..0fec80e27 --- /dev/null +++ b/sdk/src/main/java/software/amazon/lambda/durable/offload/filesystem/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.offload.filesystem; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Objects; + +/** + * Configuration for {@link PayloadPreview#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 serialized UTF-8 preview size + */ +public record PreviewConfig( + PreviewMode mode, + List include, + List exclude, + List mask, + String maskString, + int maxPreviewBytes) { + public static final String DEFAULT_MASK_STRING = "***"; + public static final int DEFAULT_MAX_PREVIEW_BYTES = 4096; + + public PreviewConfig { + Objects.requireNonNull(mode, "mode cannot be null"); + include = immutableFields(include, "include"); + exclude = immutableFields(exclude, "exclude"); + mask = immutableFields(mask, "mask"); + Objects.requireNonNull(maskString, "maskString cannot be null"); + if (maxPreviewBytes < 0) { + throw new IllegalArgumentException("maxPreviewBytes cannot be negative"); + } + } + + /** Creates a preview configuration builder. */ + public static Builder builder(PreviewMode mode) { + return new Builder(mode); + } + + private static List immutableFields(List fields, String name) { + Objects.requireNonNull(fields, name + " cannot be null"); + if (fields.stream().anyMatch(Objects::isNull)) { + throw new NullPointerException(name + " cannot contain null"); + } + return List.copyOf(fields); + } + + /** Builder for {@link PreviewConfig}. */ + public static final class Builder { + private final PreviewMode mode; + private final List include = new ArrayList<>(); + private final List exclude = new ArrayList<>(); + private final List mask = new ArrayList<>(); + private String maskString = DEFAULT_MASK_STRING; + private int maxPreviewBytes = DEFAULT_MAX_PREVIEW_BYTES; + + private Builder(PreviewMode mode) { + this.mode = Objects.requireNonNull(mode, "mode cannot be null"); + } + + /** Adds fields that should be visible. */ + public Builder include(PreviewField... fields) { + include.addAll(validFields(fields, "include")); + return this; + } + + /** Adds fields that should be hidden. */ + public Builder exclude(PreviewField... fields) { + exclude.addAll(validFields(fields, "exclude")); + return this; + } + + /** Adds fields whose values should be masked. */ + public Builder mask(PreviewField... fields) { + mask.addAll(validFields(fields, "mask")); + return this; + } + + /** Sets the value used for masked fields. */ + public Builder maskString(String maskString) { + this.maskString = Objects.requireNonNull(maskString, "maskString cannot be null"); + return this; + } + + /** Sets the maximum serialized UTF-8 preview size. */ + public Builder maxPreviewBytes(int maxPreviewBytes) { + if (maxPreviewBytes < 0) { + throw new IllegalArgumentException("maxPreviewBytes cannot be negative"); + } + this.maxPreviewBytes = maxPreviewBytes; + return this; + } + + /** Returns the immutable preview configuration. */ + public PreviewConfig build() { + return new PreviewConfig(mode, include, exclude, mask, maskString, maxPreviewBytes); + } + + private static List validFields(PreviewField[] fields, String name) { + Objects.requireNonNull(fields, name + " cannot be null"); + if (Arrays.stream(fields).anyMatch(Objects::isNull)) { + throw new NullPointerException(name + " cannot contain null"); + } + return List.of(fields); + } + } +} diff --git a/sdk/src/main/java/software/amazon/lambda/durable/offload/filesystem/PreviewField.java b/sdk/src/main/java/software/amazon/lambda/durable/offload/filesystem/PreviewField.java new file mode 100644 index 000000000..20869ba13 --- /dev/null +++ b/sdk/src/main/java/software/amazon/lambda/durable/offload/filesystem/PreviewField.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.filesystem; + +import java.util.ArrayList; +import java.util.List; +import java.util.Objects; + +/** + * A field selector used by {@link PreviewConfig}. + * + * @param name a literal field name for {@link FieldMatchMode#ANYWHERE}, or a dot-separated path for + * {@link FieldMatchMode#PATH}; in paths, backslash escapes the following character + * @param match how the selector is matched + */ +public record PreviewField(String name, FieldMatchMode match) { + public PreviewField { + Objects.requireNonNull(name, "name cannot be null"); + Objects.requireNonNull(match, "match cannot be null"); + if (name.isBlank()) { + throw new IllegalArgumentException("name cannot be blank"); + } + if (match == FieldMatchMode.PATH) { + parsePath(name); + } + } + + /** Creates a selector that matches this field name at any depth. */ + public PreviewField(String name) { + this(name, FieldMatchMode.ANYWHERE); + } + + /** Creates a selector that matches this field name at any depth. */ + public static PreviewField anywhere(String name) { + return new PreviewField(name, FieldMatchMode.ANYWHERE); + } + + /** + * Creates a selector that matches an exact dot-separated path. + * + *

Use {@code \.} to match a literal dot in a field name and {@code \\} to match a literal backslash. + */ + public static PreviewField path(String name) { + return new PreviewField(name, FieldMatchMode.PATH); + } + + List pathSegments() { + return parsePath(name); + } + + private static List parsePath(String path) { + var segments = new ArrayList(); + var segment = new StringBuilder(); + var escaped = false; + for (int index = 0; index < path.length(); index++) { + var character = path.charAt(index); + if (escaped) { + segment.append(character); + escaped = false; + } else if (character == '\\') { + escaped = true; + } else if (character == '.') { + segments.add(segment.toString()); + segment.setLength(0); + } else { + segment.append(character); + } + } + if (escaped) { + throw new IllegalArgumentException("path cannot end with an unescaped backslash"); + } + segments.add(segment.toString()); + return List.copyOf(segments); + } +} diff --git a/sdk/src/main/java/software/amazon/lambda/durable/offload/filesystem/PreviewMode.java b/sdk/src/main/java/software/amazon/lambda/durable/offload/filesystem/PreviewMode.java new file mode 100644 index 000000000..9f5cb42a2 --- /dev/null +++ b/sdk/src/main/java/software/amazon/lambda/durable/offload/filesystem/PreviewMode.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.offload.filesystem; + +/** Controls which fields are visible by default in a structured payload preview. */ +public enum PreviewMode { + /** Includes every field unless an exclude rule removes it. */ + INCLUDE_ALL, + + /** Excludes every field unless an include or mask rule selects it. */ + EXCLUDE_ALL +} diff --git a/sdk/src/test/java/software/amazon/lambda/durable/offload/RetryPayloadOffloaderTest.java b/sdk/src/test/java/software/amazon/lambda/durable/offload/RetryPayloadOffloaderTest.java new file mode 100644 index 000000000..3e872f35e --- /dev/null +++ b/sdk/src/test/java/software/amazon/lambda/durable/offload/RetryPayloadOffloaderTest.java @@ -0,0 +1,135 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.offload; + +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.time.Duration; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.jupiter.api.Test; +import software.amazon.lambda.durable.exception.PayloadOffloadException; +import software.amazon.lambda.durable.exception.RetryablePayloadOffloadException; +import software.amazon.lambda.durable.model.OperationIdentifier; +import software.amazon.lambda.durable.model.OperationSubType; +import software.amazon.lambda.durable.retry.RetryDecision; + +class RetryPayloadOffloaderTest { + @Test + void retriesRetryableStoreFailure() { + var attempts = new AtomicInteger(); + var offloader = new PayloadOffloader() { + @Override + public OffloadedPayload offload(String serializedPayload, PayloadOffloadContext context) { + if (attempts.incrementAndGet() == 1) { + throw new RetryablePayloadOffloadException("transient"); + } + return OffloadedPayload.inline(serializedPayload); + } + + @Override + public String load(OffloadedPayload payload, PayloadOffloadContext context) { + return payload.data(); + } + }; + var retrying = new RetryPayloadOffloader( + offloader, + (error, attempt) -> attempt < 2 ? RetryDecision.retry(Duration.ZERO) : RetryDecision.fail(), + delay -> {}); + + assertEquals("stored", retrying.offload("stored", context()).data()); + assertEquals(2, attempts.get()); + } + + @Test + void doesNotRetryPermanentFailure() { + var attempts = new AtomicInteger(); + var offloader = new PayloadOffloader() { + @Override + public OffloadedPayload offload(String serializedPayload, PayloadOffloadContext context) { + attempts.incrementAndGet(); + throw new PayloadOffloadException("permanent"); + } + + @Override + public String load(OffloadedPayload payload, PayloadOffloadContext context) { + return payload.data(); + } + }; + var retrying = new RetryPayloadOffloader( + offloader, (error, attempt) -> RetryDecision.retry(Duration.ZERO), delay -> {}); + + assertThrows(PayloadOffloadException.class, () -> retrying.offload("stored", context())); + assertEquals(1, attempts.get()); + } + + @Test + void retriesRetryableLoadFailure() { + var attempts = new AtomicInteger(); + var offloader = new PayloadOffloader() { + @Override + public OffloadedPayload offload(String serializedPayload, PayloadOffloadContext context) { + return OffloadedPayload.inline(serializedPayload); + } + + @Override + public String load(OffloadedPayload payload, PayloadOffloadContext context) { + if (attempts.incrementAndGet() == 1) { + throw new RetryablePayloadOffloadException("transient"); + } + return payload.data(); + } + }; + var retrying = new RetryPayloadOffloader( + offloader, + (error, attempt) -> attempt < 2 ? RetryDecision.retry(Duration.ZERO) : RetryDecision.fail(), + delay -> {}); + + assertEquals("stored", retrying.load(OffloadedPayload.inline("stored"), context())); + assertEquals(2, attempts.get()); + } + + @Test + void interruptedBackoffRemainsRetryableAndRestoresInterruptFlag() { + Thread.interrupted(); + var initialFailure = new RetryablePayloadOffloadException("transient"); + var offloader = new PayloadOffloader() { + @Override + public OffloadedPayload offload(String serializedPayload, PayloadOffloadContext context) { + throw initialFailure; + } + + @Override + public String load(OffloadedPayload payload, PayloadOffloadContext context) { + return payload.data(); + } + }; + var retrying = new RetryPayloadOffloader( + offloader, (error, attempt) -> RetryDecision.retry(Duration.ofSeconds(1)), delay -> { + throw new InterruptedException("cancelled"); + }); + + try { + var interrupted = + assertThrows(RetryablePayloadOffloadException.class, () -> retrying.offload("stored", context())); + + assertTrue(Thread.currentThread().isInterrupted()); + assertTrue(interrupted.getCause() instanceof InterruptedException); + assertEquals(1, interrupted.getSuppressed().length); + assertSame(initialFailure, interrupted.getSuppressed()[0]); + } finally { + Thread.interrupted(); + } + } + + private static PayloadOffloadContext context() { + 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, + 1); + } +} diff --git a/sdk/src/test/java/software/amazon/lambda/durable/offload/filesystem/FileSystemPayloadOffloaderTest.java b/sdk/src/test/java/software/amazon/lambda/durable/offload/filesystem/FileSystemPayloadOffloaderTest.java new file mode 100644 index 000000000..1d09c3bbc --- /dev/null +++ b/sdk/src/test/java/software/amazon/lambda/durable/offload/filesystem/FileSystemPayloadOffloaderTest.java @@ -0,0 +1,405 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.offload.filesystem; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +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.charset.StandardCharsets; +import java.nio.file.FileSystemException; +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.exception.RetryablePayloadOffloadException; +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(Path.of(payload.reference()).getFileName().toString().startsWith("operation%2Fop%2F1%2Fresult")); + 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 overflowModeAccountsForEnvelopeEscaping() { + var offloader = FileSystemPayloadOffloader.builder(temporaryDirectory) + .storageMode(PayloadOffloadMode.OVERFLOW) + .build(); + var highlyEscapablePayload = "\"".repeat(128 * 1024); + + var payload = offloader.offload(highlyEscapablePayload, context()); + + assertTrue(highlyEscapablePayload.getBytes(StandardCharsets.UTF_8).length < 255 * 1024); + assertEquals(PayloadStorageMode.REFERENCE, payload.mode()); + } + + @Test + void checkpointEnvelopeLimitRejectsValuesAboveServiceSafeMaximum() { + assertThrows(IllegalArgumentException.class, () -> FileSystemPayloadOffloader.builder(temporaryDirectory) + .checkpointEnvelopeLimitBytes(256 * 1024 - 1024 + 1)); + } + + @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(174, path.getFileName().toString().length()); + assertEquals(temporaryDirectory, path.getParent()); + } + + @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 structuredPreviewSupportsSelectionAndMasking() { + var offloader = FileSystemPayloadOffloader.builder(temporaryDirectory) + .previewConfig(PreviewConfig.builder(PreviewMode.EXCLUDE_ALL) + .include(PreviewField.anywhere("id")) + .mask(PreviewField.anywhere("email")) + .build()) + .build(); + var previewContext = context().withOriginalValue(Map.of("id", "123", "email", "secret@example.com")); + + var payload = offloader.offload("{\"id\":\"123\",\"email\":\"secret@example.com\"}", previewContext); + + assertEquals("123", payload.preview().get("id")); + assertEquals("***", payload.preview().get("email")); + } + + @Test + void structuredPreviewUsesSerializedValueInsteadOfOriginalObject() { + var offloader = FileSystemPayloadOffloader.builder(temporaryDirectory) + .previewConfig(PreviewConfig.builder(PreviewMode.INCLUDE_ALL).build()) + .build(); + var previewContext = context().withOriginalValue(Map.of("visible", "value", "secret", "plaintext")); + + var payload = offloader.offload("{\"visible\":\"value\"}", previewContext); + + assertEquals(Map.of("visible", "value"), payload.preview()); + } + + @Test + void customPreviewGeneratorCanExplicitlyUseOriginalObject() { + var offloader = FileSystemPayloadOffloader.builder(temporaryDirectory) + .previewGenerator((serialized, context) -> { + @SuppressWarnings("unchecked") + var original = (Map) context.originalValue(); + return Map.of("secret", original.get("secret")); + }) + .build(); + var previewContext = context().withOriginalValue(Map.of("secret", "plaintext")); + + var payload = offloader.offload("{\"visible\":\"value\"}", previewContext); + + assertEquals("plaintext", payload.preview().get("secret")); + } + + @Test + void oversizedReferencePreviewFailsBeforePublishingFile() throws Exception { + var offloader = FileSystemPayloadOffloader.builder(temporaryDirectory) + .checkpointEnvelopeLimitBytes(512) + .previewGenerator((serialized, context) -> Map.of("large", "x".repeat(1024))) + .build(); + + assertThrows(PayloadOffloadException.class, () -> offloader.offload("stored", context())); + try (var files = Files.walk(temporaryDirectory)) { + assertEquals(0, files.filter(Files::isRegularFile).count()); + } + } + + @Test + void repeatedWritesPublishImmutableFiles() { + var offloader = FileSystemPayloadOffloader.builder(temporaryDirectory).build(); + + var first = offloader.offload("first", context()); + var second = offloader.offload("second", context()); + + assertNotEquals(first.reference(), second.reference()); + assertEquals("first", offloader.load(first, context())); + assertEquals("second", offloader.load(second, context())); + } + + @Test + void loadRejectsReferencesOutsideConfiguredBasePath() { + var offloader = FileSystemPayloadOffloader.builder(temporaryDirectory).build(); + var outside = temporaryDirectory.getParent().resolve("outside.json"); + var validPayload = offloader.offload("stored", context()); + var payload = OffloadedPayload.reference( + outside.toString(), + null, + validPayload.ownerDurableExecutionArn(), + validPayload.ownerEntityId(), + validPayload.payloadDigest()); + + var failure = assertThrows(PayloadOffloadException.class, () -> offloader.load(payload, context())); + + assertFalse(failure instanceof RetryablePayloadOffloadException); + } + + @Test + void loadRejectsSymbolicLinksOutsideConfiguredBasePath() throws Exception { + var basePath = temporaryDirectory.resolve("base"); + var outsideFile = temporaryDirectory.resolve("outside.json"); + Files.createDirectories(basePath); + Files.writeString(outsideFile, "outside"); + var offloader = FileSystemPayloadOffloader.builder(basePath).build(); + var payload = offloader.offload("stored", context()); + var link = Path.of(payload.reference()); + Files.delete(link); + Files.createSymbolicLink(link, outsideFile); + + var failure = assertThrows(PayloadOffloadException.class, () -> offloader.load(payload, context())); + + assertFalse(failure instanceof RetryablePayloadOffloadException); + } + + @Test + void loadRejectsMissingPayloadAsPermanentFailure() throws Exception { + var offloader = FileSystemPayloadOffloader.builder(temporaryDirectory).build(); + var payload = offloader.offload("stored", context()); + Files.delete(Path.of(payload.reference())); + + var failure = assertThrows(PayloadOffloadException.class, () -> offloader.load(payload, context())); + + assertFalse(failure instanceof RetryablePayloadOffloadException); + } + + @Test + void writeEncodesTraversalSegmentsInsideConfiguredBasePath() throws Exception { + var basePath = temporaryDirectory.resolve("base"); + var outsidePath = temporaryDirectory.resolve("outside"); + Files.createDirectory(basePath); + var offloader = FileSystemPayloadOffloader.builder(basePath).build(); + var traversalContext = PayloadOffloadContext.forOperation( + "arn:aws:lambda:us-east-1:123456789012:function:..:$LATEST/durable-execution/outside/invocation", + OperationIdentifier.of("op-1", "step", OperationSubType.STEP), + null, + SerDesPayloadKind.RESULT, + 1); + + var payload = offloader.offload("stored", traversalContext); + + assertEquals(basePath, Path.of(payload.reference()).getParent()); + assertFalse(Files.exists(outsidePath)); + } + + @Test + void writeEncodesFallbackTraversalInsideConfiguredBasePath() throws Exception { + var basePath = temporaryDirectory.resolve("base"); + Files.createDirectory(basePath); + var offloader = FileSystemPayloadOffloader.builder(basePath).build(); + var traversalContext = PayloadOffloadContext.forOperation( + "..", OperationIdentifier.of("op-1", "step", OperationSubType.STEP), null, SerDesPayloadKind.RESULT, 1); + + var payload = offloader.offload("stored", traversalContext); + + assertEquals(basePath, Path.of(payload.reference()).getParent()); + } + + @Test + void writeRejectsConfiguredBasePathThatIsASymbolicLink() throws Exception { + var outsidePath = temporaryDirectory.resolve("outside"); + Files.createDirectories(outsidePath); + var basePath = temporaryDirectory.resolve("base"); + Files.createSymbolicLink(basePath, outsidePath); + var offloader = FileSystemPayloadOffloader.builder(basePath).build(); + + var failure = assertThrows(PayloadOffloadException.class, () -> offloader.offload("stored", context())); + + assertFalse(failure instanceof RetryablePayloadOffloadException); + try (var files = Files.list(outsidePath)) { + assertEquals(0, files.count()); + } + } + + @Test + void writeDoesNotCreateMissingBasePathComponents() { + var missingRoot = temporaryDirectory.resolve("missing"); + var offloader = FileSystemPayloadOffloader.builder(missingRoot.resolve("payloads")) + .build(); + + var failure = assertThrows(PayloadOffloadException.class, () -> offloader.offload("stored", context())); + + assertTrue(failure.getMessage().contains("base path and all ancestors must already exist")); + assertFalse(failure instanceof RetryablePayloadOffloadException); + assertFalse(Files.exists(missingRoot)); + } + + @Test + void writeRejectsConfiguredBasePathThatIsNotDirectoryAsPermanentFailure() throws Exception { + var basePath = temporaryDirectory.resolve("payloads"); + Files.writeString(basePath, "not a directory"); + var offloader = FileSystemPayloadOffloader.builder(basePath).build(); + + var failure = assertThrows(PayloadOffloadException.class, () -> offloader.offload("stored", context())); + + assertFalse(failure instanceof RetryablePayloadOffloadException); + assertTrue(failure.getMessage().contains("must be directories")); + } + + @Test + void readOnlyProviderFailureIsPermanent() { + var providerFailure = new FileSystemException(temporaryDirectory.toString(), null, "Read-only file system"); + + var failure = FileSystemPayloadOffloader.classifyIoFailure("store", context(), providerFailure); + + assertEquals(providerFailure, failure.getCause()); + assertFalse(failure instanceof RetryablePayloadOffloadException); + } + + @Test + void quotaCapacityAndUnknownFailuresArePermanent() { + var quotaFailure = new FileSystemException(temporaryDirectory.toString(), null, "Disk quota exceeded"); + var capacityFailure = new FileSystemException(temporaryDirectory.toString(), null, "No space left on device"); + var unknownFailure = new java.io.IOException("provider configuration failure"); + + assertFalse( + FileSystemPayloadOffloader.classifyIoFailure("store", context(), quotaFailure) + instanceof RetryablePayloadOffloadException); + assertFalse( + FileSystemPayloadOffloader.classifyIoFailure("store", context(), capacityFailure) + instanceof RetryablePayloadOffloadException); + assertFalse( + FileSystemPayloadOffloader.classifyIoFailure("load", context(), unknownFailure) + instanceof RetryablePayloadOffloadException); + } + + @Test + void knownTransientProviderFailureRemainsRetryable() { + var transientFailure = new FileSystemException(temporaryDirectory.toString(), null, "Stale file handle"); + + assertTrue( + FileSystemPayloadOffloader.classifyIoFailure("load", context(), transientFailure) + instanceof RetryablePayloadOffloadException); + } + + @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)); + } + + @Test + void tamperedPayloadFailsDigestValidation() throws Exception { + var offloader = FileSystemPayloadOffloader.builder(temporaryDirectory).build(); + var payload = offloader.offload("stored", context()); + Files.writeString(Path.of(payload.reference()), "tampered"); + + assertThrows(PayloadOffloadException.class, () -> offloader.load(payload, context())); + } + + @Test + void payloadCannotBeLoadedByDifferentEntity() { + var offloader = FileSystemPayloadOffloader.builder(temporaryDirectory).build(); + var payload = offloader.offload("stored", context()); + var otherContext = PayloadOffloadContext.forOperation( + context().durableExecutionArn(), + OperationIdentifier.of("op-2", "other", OperationSubType.STEP), + null, + SerDesPayloadKind.RESULT, + 1); + + assertThrows(PayloadOffloadException.class, () -> offloader.load(payload, otherContext)); + } + + @Test + void chainedInvokeResultCanLoadPayloadOwnedByTargetExecution() { + var offloader = FileSystemPayloadOffloader.builder(temporaryDirectory).build(); + var targetContext = PayloadOffloadContext.forExecution( + "arn:aws:lambda:us-east-1:123456789012:function:target:$LATEST/durable-execution/name/target-id", + "target-id", + "target", + SerDesPayloadKind.OUTPUT); + var payload = offloader.offload("stored", targetContext); + var callerContext = PayloadOffloadContext.forOperation( + context().durableExecutionArn(), + OperationIdentifier.of("invoke", "target", OperationSubType.CHAINED_INVOKE), + null, + SerDesPayloadKind.RESULT, + null); + + assertEquals("stored", offloader.load(payload, callerContext)); + } + + 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/sdk/src/test/java/software/amazon/lambda/durable/offload/filesystem/PayloadPreviewTest.java b/sdk/src/test/java/software/amazon/lambda/durable/offload/filesystem/PayloadPreviewTest.java new file mode 100644 index 000000000..401835799 --- /dev/null +++ b/sdk/src/test/java/software/amazon/lambda/durable/offload/filesystem/PayloadPreviewTest.java @@ -0,0 +1,171 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +package software.amazon.lambda.durable.offload.filesystem; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.time.Instant; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import org.junit.jupiter.api.Test; + +class PayloadPreviewTest { + @Test + void includeAllExcludesAndMasksSelectedFields() { + var config = PreviewConfig.builder(PreviewMode.INCLUDE_ALL) + .exclude(PreviewField.anywhere("internal")) + .mask(PreviewField.anywhere("email")) + .build(); + + var preview = PayloadPreview.buildPreview( + Map.of( + "id", "123", + "email", "customer@example.com", + "nested", Map.of("internal", "secret", "status", "ready")), + config); + + assertEquals("123", preview.get("id")); + assertEquals("***", preview.get("email")); + var nested = nested(preview, "nested"); + assertFalse(nested.containsKey("internal")); + assertEquals("ready", nested.get("status")); + } + + @Test + void excludeAllIncludesExactPathAndScalarArray() { + var config = PreviewConfig.builder(PreviewMode.EXCLUDE_ALL) + .include(PreviewField.path("customer.status"), PreviewField.anywhere("tags")) + .build(); + + var preview = PayloadPreview.buildPreview( + Map.of( + "customer", Map.of("status", "ready", "name", "Ada"), + "tags", List.of("a", "b"), + "ignored", true), + config); + + assertEquals("ready", nested(preview, "customer").get("status")); + assertEquals(List.of("a", "b"), preview.get("tags")); + assertFalse(preview.containsKey("ignored")); + } + + @Test + void excludeAllExactContainerPathIncludesSubtreeWithNestedPolicies() { + var config = PreviewConfig.builder(PreviewMode.EXCLUDE_ALL) + .include(PreviewField.path("customer")) + .exclude(PreviewField.path("customer.internal")) + .mask(PreviewField.path("customer.email")) + .build(); + + var preview = PayloadPreview.buildPreview( + Map.of( + "customer", + Map.of( + "name", "Ada", + "email", "customer@example.com", + "internal", "secret", + "address", Map.of("city", "Seattle")), + "ignored", + true), + config); + + var customer = nested(preview, "customer"); + assertEquals("Ada", customer.get("name")); + assertEquals("***", customer.get("email")); + assertFalse(customer.containsKey("internal")); + assertEquals("Seattle", nested(customer, "address").get("city")); + assertFalse(preview.containsKey("ignored")); + } + + @Test + void literalDotsArePreservedAndEscapedPathsRemainUnambiguous() { + var value = new LinkedHashMap(); + value.put("customer.email", "literal@example.com"); + value.put("customer", Map.of("email", "nested@example.com")); + value.put("profile", Map.of("contact.email", "private@example.com")); + var config = PreviewConfig.builder(PreviewMode.EXCLUDE_ALL) + .include(PreviewField.path("customer\\.email")) + .mask(PreviewField.path("profile.contact\\.email")) + .build(); + + var preview = PayloadPreview.buildPreview(value, config); + + assertEquals("literal@example.com", preview.get("customer.email")); + assertFalse(preview.containsKey("customer")); + assertEquals("***", nested(preview, "profile").get("contact.email")); + } + + @Test + void includeAllPreservesLiteralDotAndBackslashKeys() { + var preview = PayloadPreview.buildPreview( + Map.of("customer.email", "literal", "path\\name", "backslash"), + PreviewConfig.builder(PreviewMode.INCLUDE_ALL).build()); + + assertEquals("literal", preview.get("customer.email")); + assertEquals("backslash", preview.get("path\\name")); + } + + @Test + void temporalValuesUseJacksonSerDesCompatibleEncoding() { + var preview = PayloadPreview.buildPreview( + Map.of("createdAt", Instant.parse("2026-09-01T00:00:00Z")), + PreviewConfig.builder(PreviewMode.INCLUDE_ALL).build()); + + assertEquals("2026-09-01T00:00:00Z", preview.get("createdAt")); + } + + @Test + void containerArraysAreOmittedInsteadOfCollapsingPositions() { + var value = Map.of( + "items", + List.of(Map.of("id", "first", "secret", "one"), Map.of("id", "second", "secret", "two")), + "status", + "ready"); + var config = PreviewConfig.builder(PreviewMode.INCLUDE_ALL) + .include(PreviewField.anywhere("id")) + .mask(PreviewField.anywhere("secret")) + .build(); + + var preview = PayloadPreview.buildPreview(value, config); + + assertFalse(preview.containsKey("items")); + assertEquals("ready", preview.get("status")); + } + + @Test + void previewBudgetUsesFinalNestedJsonSize() { + var config = PreviewConfig.builder(PreviewMode.INCLUDE_ALL) + .maxPreviewBytes(20) + .build(); + var value = new LinkedHashMap(); + value.put("a", Map.of("b", "value")); + value.put("c", "later"); + + var preview = PayloadPreview.buildPreview(value, config); + + assertTrue(preview.containsKey("a")); + assertFalse(preview.containsKey("c")); + } + + @Test + void nonObjectPayloadHasNoStructuredPreview() { + assertNull(PayloadPreview.buildPreview( + List.of("a", "b"), + PreviewConfig.builder(PreviewMode.INCLUDE_ALL).build())); + } + + @Test + void nonJsonSerializedPayloadHasNoStructuredPreview() { + assertNull(PayloadPreview.buildPreviewFromJson( + "raw-error-data", PreviewConfig.builder(PreviewMode.INCLUDE_ALL).build())); + } + + @SuppressWarnings("unchecked") + private static Map nested(Map preview, String field) { + return (Map) preview.get(field); + } +}