diff --git a/CHANGELOG.md b/CHANGELOG.md index d7f26fc..98ead6c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,4 +1,11 @@ ## Unreleased +* Change the default large-payload externalization threshold from 900,000 bytes to 256 KiB, matching .NET. +* Align the large-payload worker estimator threshold with the 256 KiB storage default, resolve externalized entity-operation failure traces, recover verified ambiguous blob uploads, and recognize `$root` payload tokens. +* Preserve transient verification-download failures during ambiguous blob-upload recovery instead of reporting the original permanent precondition failure. +* Propagate transient large-payload upload failures without completing work items; abandon failed activity and orchestration completions for scheduler redelivery instead of terminating the worker polling thread. +* Ensure the export container exists before every upload, allowing export retries to recover containers deleted between uploads. +* Preserve custom and empty property objects in exported failure details, including nested failures, matching .NET's history conversion. +* Capture the default continuous-export completion-time lower bound when creation options are constructed, matching .NET and preserving the window through worker queue delays. * Recreate a deleted large-payload container and retry the upload once, without allowing stale concurrent failures to invalidate a newly recreated container. * Preserve backend timestamp precision in orchestration history and client metadata, including export blob names, while retaining existing orchestration replay timestamp behavior. * Export entity operations and locks using the .NET-compatible `EventSent`/`EventRaised` message representation instead of Java-native entity event objects, preserving missing stack traces in failure details. diff --git a/azure-blob-payloads/README.md b/azure-blob-payloads/README.md new file mode 100644 index 0000000..f21831e --- /dev/null +++ b/azure-blob-payloads/README.md @@ -0,0 +1,100 @@ +# Durable Task Azure Blob Payloads for Java + +The `durabletask-azure-blob-payloads` module transparently externalizes large Durable Task payloads to Azure Blob Storage. Payloads below the configured threshold continue to travel inline; larger payloads are replaced with opaque blob references and restored before user code receives them. + +## Install + +Add the module alongside the Durable Task client and your backend-specific extension: + +```groovy +implementation 'com.microsoft:durabletask-azure-blob-payloads:1.0.0' +``` + +The module includes Azure Blob Storage support. Applications that use managed identity should also add `com.azure:azure-identity`. + +## Usage + +Configure the same storage location for every client and worker that exchanges externalized payloads: + +```java +LargePayloadStorageOptions payloadOptions = new LargePayloadStorageOptions() + .setConnectionString(System.getenv("PAYLOAD_STORAGE_CONNECTION_STRING")) + .setContainerName("durabletask-payloads") + .setThresholdBytes(256 * 1024); + +PayloadStore payloadStore = new BlobPayloadStore(payloadOptions); + +DurableTaskGrpcClientBuilder clientBuilder = new DurableTaskGrpcClientBuilder(); +DurableTaskSchedulerClientExtensions.useDurableTaskScheduler(clientBuilder, schedulerConnectionString); +LargePayloadClientExtensions.useExternalizedPayloads(clientBuilder, payloadStore, payloadOptions); + +DurableTaskGrpcWorkerBuilder workerBuilder = new DurableTaskGrpcWorkerBuilder(); +DurableTaskSchedulerWorkerExtensions.useDurableTaskScheduler(workerBuilder, schedulerConnectionString); +LargePayloadWorkerExtensions.useExternalizedPayloads(workerBuilder, payloadStore, payloadOptions); +``` + +For identity-based authentication, configure an account URI and `TokenCredential` instead of a connection string: + +```java +LargePayloadStorageOptions payloadOptions = new LargePayloadStorageOptions() + .setAccountUri(URI.create("https://.blob.core.windows.net")) + .setCredential(new DefaultAzureCredentialBuilder().build()); +``` + +## Defaults and limits + +- Payloads of at least 256 KiB (262,144 bytes) are externalized by default. +- The threshold can be configured up to 1 MiB. +- The default maximum externalized payload size is 10 MiB. +- Payloads are gzip-compressed by default. +- The default container name is `durabletask-payloads`. +- The container is created automatically on the first upload. + +The storage identity needs permission to create the container and read and write blobs. Payload blobs are retained after orchestration processing, so configure an Azure Storage lifecycle policy appropriate for the application's retention requirements. + +## Upload failures + +Transient upload failures (including HTTP 408, 429, and 5xx responses) propagate without sending a worker +completion, allowing the work item to be retried rather than recording a task failure. Permanent activity and +orchestration upload failures become non-retriable failure completions. Entity upload failures propagate without +completing the batch, matching .NET. The worker logs failed completions and abandons the affected work item for +scheduler redelivery without terminating its polling thread. + +Uploads use conditional creation to prevent overwriting an existing payload. If a PUT commits but its response is +lost, a retried PUT can return HTTP 412. The store recovers only after verifying that the stored bytes and content +encoding exactly match the attempted upload. A verification download failure propagates with its own status so +transient failures remain retryable. Payload tokens also support Azure's `$root` container. + +## Offline regression tests + +```text +./gradlew :azure-blob-payloads:test :exporthistory:test :client:test -PskipSigning +``` + +These tests do not require Azure or emulators. The blob-upload regressions run the real Azure SDK against an +in-memory HTTP transport that commits a PUT and loses its response, exercising SDK retries with compression on +and off. In-process gRPC tests cover `$root` tokens through client, worker, and streamed history paths and resolved +nested failure traces in decoded JSON and JSONL exports. An orchestration replay regression also verifies that +entity callers receive the original nested failure diagnostics after work-item payload resolution. + +## Integration tests + +With the DTS emulator on port 4001 and Azurite Blob Storage on port 10000, run: + +```text +./gradlew :azure-blob-payloads:integrationTest -PskipSigning +``` + +The suite exercises payload round trips, activity and sub-orchestration outputs, events, queries, custom status, +and payload limits. It also injects a transient worker upload failure and verifies actual scheduler redelivery +and successful completion for both activity and orchestration outputs. + +## Sample + +See [`LargePayloadSample`](../samples/src/main/java/io/durabletask/samples/LargePayloadSample.java): + +```text +./gradlew :samples:runLargePayloadSample +``` + +The sample uses the DTS emulator and Azurite and verifies a payload larger than 1 MiB through a client, orchestration, and activity round trip. diff --git a/azure-blob-payloads/build.gradle b/azure-blob-payloads/build.gradle index 61d033e..61fbfc1 100644 --- a/azure-blob-payloads/build.gradle +++ b/azure-blob-payloads/build.gradle @@ -47,6 +47,7 @@ dependencies { testImplementation 'org.mockito:mockito-core:5.21.0' testImplementation 'org.mockito:mockito-junit-jupiter:5.21.0' + testImplementation "io.grpc:grpc-inprocess:${grpcVersion}" testImplementation project(':azuremanaged') } diff --git a/azure-blob-payloads/src/main/java/com/microsoft/durabletask/azureblobpayloads/BlobPayloadStore.java b/azure-blob-payloads/src/main/java/com/microsoft/durabletask/azureblobpayloads/BlobPayloadStore.java index 2e969e8..902e0b4 100644 --- a/azure-blob-payloads/src/main/java/com/microsoft/durabletask/azureblobpayloads/BlobPayloadStore.java +++ b/azure-blob-payloads/src/main/java/com/microsoft/durabletask/azureblobpayloads/BlobPayloadStore.java @@ -8,6 +8,7 @@ import com.azure.storage.blob.BlobServiceClient; import com.azure.storage.blob.BlobServiceClientBuilder; import com.azure.storage.blob.models.BlobDownloadResponse; +import com.azure.storage.blob.models.BlobDownloadHeaders; import com.azure.storage.blob.models.BlobErrorCode; import com.azure.storage.blob.models.BlobHttpHeaders; import com.azure.storage.blob.models.BlobRequestConditions; @@ -20,6 +21,7 @@ import java.io.IOException; import java.io.InputStream; import java.nio.charset.StandardCharsets; +import java.util.Arrays; import java.util.UUID; import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.locks.ReentrantLock; @@ -41,10 +43,10 @@ public final class BlobPayloadStore extends PayloadStore { // Blob name is UUID.randomUUID().toString().replace("-", ""): exactly 32 lowercase hex chars. // Container name follows Azure rules: 3-63 chars, lowercase alphanumerics and single hyphens, - // must start and end with alphanumeric (see isValidContainerName). + // must start and end with alphanumeric, or the special $root container. // Full token grammar: blob:v1::<32-lowercase-hex> private static final Pattern TOKEN_PATTERN = Pattern.compile( - "^blob:v1:[a-z0-9](?:[a-z0-9]|-(?=[a-z0-9])){1,61}[a-z0-9]:[0-9a-f]{32}$"); + "^blob:v1:(?:\\$root|[a-z0-9](?:[a-z0-9]|-(?=[a-z0-9])){1,61}[a-z0-9]):[0-9a-f]{32}$"); private final BlobContainerClient containerClient; private final LargePayloadStorageOptions options; @@ -160,17 +162,59 @@ private void uploadBlob(BlobClient blob, byte[] payloadBytes) throws IOException byte[] compressedBytes = compressedBuffer.toByteArray(); BlobHttpHeaders headers = new BlobHttpHeaders().setContentEncoding(CONTENT_ENCODING_GZIP); try (InputStream stream = new ByteArrayInputStream(compressedBytes)) { - blob.uploadWithResponse( - stream, compressedBytes.length, null, headers, null, null, conditions, null, Context.NONE); + try { + blob.uploadWithResponse( + stream, compressedBytes.length, null, headers, null, null, conditions, null, Context.NONE); + } catch (BlobStorageException e) { + if (isPreconditionFailure(e) && blobContentMatches(blob, compressedBytes, CONTENT_ENCODING_GZIP)) { + return; + } + throw e; + } } } else { try (InputStream stream = new ByteArrayInputStream(payloadBytes)) { - blob.uploadWithResponse( - stream, payloadBytes.length, null, null, null, null, conditions, null, Context.NONE); + try { + blob.uploadWithResponse( + stream, payloadBytes.length, null, null, null, null, conditions, null, Context.NONE); + } catch (BlobStorageException e) { + if (isPreconditionFailure(e) && blobContentMatches(blob, payloadBytes, null)) { + return; + } + throw e; + } } } } + private static boolean isPreconditionFailure(BlobStorageException e) { + return e.getStatusCode() == 412; + } + + private static boolean blobContentMatches(BlobClient blob, byte[] expectedBytes, String expectedContentEncoding) { + // Verification failures must propagate with their own status/cause. In particular, + // a transient GET failure must not turn the ambiguous PUT into a permanent 412. + ByteArrayOutputStream output = new ByteArrayOutputStream(); + BlobDownloadResponse response = blob.downloadStreamWithResponse( + output, + null, + null, + null, + false, + null, + Context.NONE); + BlobDownloadHeaders headers = response.getDeserializedHeaders(); + String actualEncoding = headers == null ? null : headers.getContentEncoding(); + if (expectedContentEncoding == null) { + if (actualEncoding != null && !actualEncoding.isEmpty()) { + return false; + } + } else if (!expectedContentEncoding.equalsIgnoreCase(actualEncoding)) { + return false; + } + return Arrays.equals(expectedBytes, output.toByteArray()); + } + private Object ensureContainerExists() { Object generation = this.containerGeneration.get(); if (generation != null) { diff --git a/azure-blob-payloads/src/main/java/com/microsoft/durabletask/azureblobpayloads/LargePayloadInterceptor.java b/azure-blob-payloads/src/main/java/com/microsoft/durabletask/azureblobpayloads/LargePayloadInterceptor.java index aa346c1..fed8ea9 100644 --- a/azure-blob-payloads/src/main/java/com/microsoft/durabletask/azureblobpayloads/LargePayloadInterceptor.java +++ b/azure-blob-payloads/src/main/java/com/microsoft/durabletask/azureblobpayloads/LargePayloadInterceptor.java @@ -192,16 +192,17 @@ private Object externalizeActivityResponse(ActivityResponse r) { ? "Permanent payload storage failure" : "Transient payload storage failure"; logger.log(Level.WARNING, prefix + " while externalizing activity response.", ex); - // Convert to a failure response so the orchestration sees a failed activity. - // Permanent failures are non-retriable; transient failures are retriable so - // the sidecar can re-dispatch the work item. + if (!permanent) { + throw ex; + } + // Only permanent failures complete the activity; transient failures leave it uncompleted. return r.toBuilder() .clearResult() .setFailureDetails(TaskFailureDetails.newBuilder() .setErrorType(ex.getClass().getName()) .setErrorMessage(prefix + ": " + ex.getMessage()) .setStackTrace(StringValue.of(getStackTraceString(ex))) - .setIsNonRetriable(permanent) + .setIsNonRetriable(true) .build()) .build(); } @@ -235,9 +236,10 @@ private Object externalizeOrchestratorResponse(OrchestratorResponse r) { ? "Permanent payload storage failure" : "Transient payload storage failure"; logger.log(Level.WARNING, prefix + " while externalizing orchestrator response.", ex); - // Replace with a single Failed completion. - // Permanent failures are non-retriable; transient failures are retriable so - // the sidecar can re-dispatch the orchestration. + if (!permanent) { + throw ex; + } + // Only permanent failures replace the actions with a terminal failed completion. return OrchestratorResponse.newBuilder() .setInstanceId(r.getInstanceId()) .setCompletionToken(r.getCompletionToken()) @@ -249,7 +251,7 @@ private Object externalizeOrchestratorResponse(OrchestratorResponse r) { .setErrorType(ex.getClass().getName()) .setErrorMessage(prefix + ": " + ex.getMessage()) .setStackTrace(StringValue.of(getStackTraceString(ex))) - .setIsNonRetriable(permanent) + .setIsNonRetriable(true) .build()) .build()) .build()) @@ -438,18 +440,8 @@ private Object externalizeEntityBatchResult(EntityBatchResult r) { ? "Permanent payload storage failure" : "Transient payload storage failure"; logger.log(Level.WARNING, prefix + " while externalizing entity batch result.", ex); - // Convert to a failure result so the sidecar records the entity failure. - // Permanent failures are non-retriable; transient failures are retriable so - // the sidecar can re-dispatch the entity work item. - return EntityBatchResult.newBuilder() - .setCompletionToken(r.getCompletionToken()) - .setFailureDetails(TaskFailureDetails.newBuilder() - .setErrorType(ex.getClass().getName()) - .setErrorMessage(prefix + ": " + ex.getMessage()) - .setStackTrace(StringValue.of(getStackTraceString(ex))) - .setIsNonRetriable(permanent) - .build()) - .build(); + // .NET propagates all entity upload failures, leaving the batch uncompleted. + throw ex; } } @@ -855,6 +847,16 @@ private HistoryEvent resolveEventPayloads(HistoryEvent e) { } } break; + case ENTITYOPERATIONFAILED: + if (e.hasEntityOperationFailed() && e.getEntityOperationFailed().hasFailureDetails()) { + EntityOperationFailedEvent efailed = e.getEntityOperationFailed(); + TaskFailureDetails fd = resolveFailureDetails(efailed.getFailureDetails()); + if (fd != efailed.getFailureDetails()) { + return e.toBuilder().setEntityOperationFailed( + efailed.toBuilder().setFailureDetails(fd)).build(); + } + } + break; case HISTORYSTATE: if (e.hasHistoryState()) { HistoryStateEvent hs = e.getHistoryState(); diff --git a/azure-blob-payloads/src/main/java/com/microsoft/durabletask/azureblobpayloads/LargePayloadStorageOptions.java b/azure-blob-payloads/src/main/java/com/microsoft/durabletask/azureblobpayloads/LargePayloadStorageOptions.java index 35dea44..534a886 100644 --- a/azure-blob-payloads/src/main/java/com/microsoft/durabletask/azureblobpayloads/LargePayloadStorageOptions.java +++ b/azure-blob-payloads/src/main/java/com/microsoft/durabletask/azureblobpayloads/LargePayloadStorageOptions.java @@ -31,7 +31,7 @@ public final class LargePayloadStorageOptions { private static final int ONE_MIB = 1024 * 1024; - private int thresholdBytes = 900_000; + private int thresholdBytes = 256 * 1024; private int maxPayloadBytes = 10 * ONE_MIB; private String connectionString; private URI accountUri; @@ -41,7 +41,7 @@ public final class LargePayloadStorageOptions { /** * Gets the threshold in bytes at which payloads are externalized to blob storage. - * Payloads smaller than this threshold are sent inline. Default is 900,000 bytes. + * Payloads smaller than this threshold are sent inline. Default is 256 KiB (262,144 bytes). * * @return the threshold in bytes */ diff --git a/azure-blob-payloads/src/test/java/com/microsoft/durabletask/azureblobpayloads/BlobPayloadStoreTest.java b/azure-blob-payloads/src/test/java/com/microsoft/durabletask/azureblobpayloads/BlobPayloadStoreTest.java index 2154d5d..6b03d70 100644 --- a/azure-blob-payloads/src/test/java/com/microsoft/durabletask/azureblobpayloads/BlobPayloadStoreTest.java +++ b/azure-blob-payloads/src/test/java/com/microsoft/durabletask/azureblobpayloads/BlobPayloadStoreTest.java @@ -121,6 +121,7 @@ void upload_preconditionFailed_throwsPayloadStorageException() { doThrow(preconditionFailed).when(mockBlobClient).uploadWithResponse( any(InputStream.class), anyLong(), isNull(), any(BlobHttpHeaders.class), isNull(), isNull(), any(BlobRequestConditions.class), isNull(), any()); + mockDownloadResponse("different stored content".getBytes(StandardCharsets.UTF_8), "gzip"); BlobPayloadStore store = new BlobPayloadStore(mockContainerClient, options); PayloadStorageException ex = assertThrows(PayloadStorageException.class, @@ -129,6 +130,39 @@ void upload_preconditionFailed_throwsPayloadStorageException() { "Exception message should indicate a blob-name collision: " + ex.getMessage()); } + @ParameterizedTest + @ValueSource(booleans = {true, false}) + void upload_preconditionFailedAfterCommittedPut_recoversWhenStoredContentMatches(boolean compressed) { + options.setCompressionEnabled(compressed); + BlobStorageException preconditionFailed = mock(BlobStorageException.class); + when(preconditionFailed.getStatusCode()).thenReturn(412); + final byte[][] attemptedUpload = new byte[1][]; + doAnswer(invocation -> { + InputStream stream = invocation.getArgument(0); + attemptedUpload[0] = stream.readAllBytes(); + throw preconditionFailed; + }).when(mockBlobClient).uploadWithResponse( + any(InputStream.class), anyLong(), isNull(), + nullable(BlobHttpHeaders.class), isNull(), isNull(), any(BlobRequestConditions.class), isNull(), any()); + doAnswer(invocation -> { + OutputStream out = invocation.getArgument(0); + out.write(attemptedUpload[0]); + BlobDownloadHeaders downloadHeaders = mock(BlobDownloadHeaders.class); + when(downloadHeaders.getContentEncoding()).thenReturn(compressed ? "gzip" : null); + BlobDownloadResponse downloadResponse = mock(BlobDownloadResponse.class); + when(downloadResponse.getDeserializedHeaders()).thenReturn(downloadHeaders); + return downloadResponse; + }).when(mockBlobClient).downloadStreamWithResponse( + any(OutputStream.class), isNull(), isNull(), isNull(), eq(false), isNull(), any()); + + BlobPayloadStore store = new BlobPayloadStore(mockContainerClient, options); + String token = store.upload("payload committed before retry"); + + assertTrue(store.isKnownPayloadToken(token)); + verify(mockBlobClient).downloadStreamWithResponse( + any(OutputStream.class), isNull(), isNull(), isNull(), eq(false), isNull(), any()); + } + @Test void upload_tokenContainsGuidBlobName() { BlobPayloadStore store = new BlobPayloadStore(mockContainerClient, options); @@ -142,6 +176,30 @@ void upload_tokenContainsGuidBlobName() { assertTrue(parts[1].matches("[0-9a-f]{32}"), "Blob name should be a GUID without dashes: " + parts[1]); } + @ParameterizedTest + @ValueSource(booleans = {true, false}) + void upload_download_roundTrip_rootContainerTokenIsRecognized(boolean compressed) throws Exception { + when(mockContainerClient.getBlobContainerName()).thenReturn("$root"); + options.setCompressionEnabled(compressed); + String originalPayload = "root container payload"; + final byte[][] capturedBytes = new byte[1][]; + doAnswer(inv -> { + InputStream stream = inv.getArgument(0); + capturedBytes[0] = stream.readAllBytes(); + return null; + }).when(mockBlobClient).uploadWithResponse( + any(InputStream.class), anyLong(), isNull(), + nullable(BlobHttpHeaders.class), isNull(), isNull(), any(BlobRequestConditions.class), isNull(), any()); + + BlobPayloadStore store = new BlobPayloadStore(mockContainerClient, options); + String token = store.upload(originalPayload); + + assertTrue(token.startsWith("blob:v1:$root:")); + assertTrue(store.isKnownPayloadToken(token)); + mockDownloadResponse(capturedBytes[0], compressed ? "gzip" : null); + assertEquals(originalPayload, store.download(token)); + } + @Test void upload_containerAlreadyExists_succeeds() { BlobStorageException conflict = mock(BlobStorageException.class); @@ -454,6 +512,13 @@ void isKnownPayloadToken_validToken_returnsTrue() { "blob:v1:container:0123456789abcdef0123456789abcdef")); } + @Test + void isKnownPayloadToken_rootContainer_returnsTrue() { + BlobPayloadStore store = new BlobPayloadStore(mockContainerClient, options); + assertTrue(store.isKnownPayloadToken( + "blob:v1:$root:0123456789abcdef0123456789abcdef")); + } + @Test void isKnownPayloadToken_nonToken_returnsFalse() { BlobPayloadStore store = new BlobPayloadStore(mockContainerClient, options); diff --git a/azure-blob-payloads/src/test/java/com/microsoft/durabletask/azureblobpayloads/BlobPayloadStoreTransportTest.java b/azure-blob-payloads/src/test/java/com/microsoft/durabletask/azureblobpayloads/BlobPayloadStoreTransportTest.java new file mode 100644 index 0000000..584f5d4 --- /dev/null +++ b/azure-blob-payloads/src/test/java/com/microsoft/durabletask/azureblobpayloads/BlobPayloadStoreTransportTest.java @@ -0,0 +1,75 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. +package com.microsoft.durabletask.azureblobpayloads; + +import com.azure.storage.blob.models.BlobStorageException; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import static org.junit.jupiter.api.Assertions.*; + +class BlobPayloadStoreTransportTest { + @ParameterizedTest + @ValueSource(booleans = {true, false}) + void committedPutWithLostReply_isRetriedVerifiedAndReadable(boolean compressed) { + OfflineBlobService service = new OfflineBlobService(); + service.loseFirstPutResponse = true; + BlobPayloadStore store = service.createStore("payloads", compressed); + String payload = "original payload \u00e9\n".repeat(200); + + String token = store.upload(payload); + + assertEquals(2, service.putUrls.size(), "The Azure SDK must retry the ambiguous PUT"); + assertEquals(service.putUrls.get(0), service.putUrls.get(1), "Retries must use the same blob URI"); + assertArrayEquals(service.putBodies.get(0), service.putBodies.get(1), "The SDK must replay the same body"); + assertEquals(1, service.downloads, "The 412 must trigger content verification"); + assertTrue(store.isKnownPayloadToken(token)); + assertEquals(payload, store.download(token), "The returned token must actually be usable"); + } + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + void committedPutWithLostReply_rejectsDifferentBytes(boolean compressed) { + OfflineBlobService service = new OfflineBlobService(); + service.loseFirstPutResponse = true; + service.changeStoredBytes = true; + assertCollisionRejected(service, compressed); + } + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + void committedPutWithLostReply_rejectsDifferentEncoding(boolean compressed) { + OfflineBlobService service = new OfflineBlobService(); + service.loseFirstPutResponse = true; + service.changeStoredEncoding = true; + assertCollisionRejected(service, compressed); + } + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + void verificationDownloadFailure_preservesTransientStatus(boolean compressed) { + OfflineBlobService service = new OfflineBlobService(); + service.loseFirstPutResponse = true; + service.downloadStatus = 503; + BlobPayloadStore store = service.createStore("payloads", compressed); + + PayloadStorageException failure = assertThrows(PayloadStorageException.class, + () -> store.upload("committed payload")); + + assertInstanceOf(BlobStorageException.class, failure.getCause()); + assertEquals(503, ((BlobStorageException) failure.getCause()).getStatusCode()); + assertFalse(LargePayloadInterceptor.isPermanentStorageFailure(failure)); + assertEquals(2, service.putUrls.size()); + assertEquals(2, service.downloads, "Verification must exhaust the configured GET retries"); + } + + private static void assertCollisionRejected(OfflineBlobService service, boolean compressed) { + BlobPayloadStore store = service.createStore("payloads", compressed); + PayloadStorageException failure = assertThrows(PayloadStorageException.class, + () -> store.upload("attempted payload")); + assertEquals(412, ((BlobStorageException) failure.getCause()).getStatusCode()); + assertTrue(LargePayloadInterceptor.isPermanentStorageFailure(failure)); + assertEquals(2, service.putUrls.size()); + assertEquals(1, service.downloads); + } +} diff --git a/azure-blob-payloads/src/test/java/com/microsoft/durabletask/azureblobpayloads/LargePayloadIntegrationTest.java b/azure-blob-payloads/src/test/java/com/microsoft/durabletask/azureblobpayloads/LargePayloadIntegrationTest.java index a005482..c01afb2 100644 --- a/azure-blob-payloads/src/test/java/com/microsoft/durabletask/azureblobpayloads/LargePayloadIntegrationTest.java +++ b/azure-blob-payloads/src/test/java/com/microsoft/durabletask/azureblobpayloads/LargePayloadIntegrationTest.java @@ -14,8 +14,11 @@ import org.junit.jupiter.api.Test; import java.time.Duration; +import java.io.IOException; +import java.io.UncheckedIOException; import java.util.List; import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicInteger; import static org.junit.jupiter.api.Assertions.*; @@ -71,12 +74,12 @@ void tearDown() { @Test void roundTrip_largeInput_activityEchoesBack() throws TimeoutException { - // Arrange: create a 1.5 MB payload that exceeds the default 900KB threshold + // Arrange: create a 1.5 MB payload that exceeds the default 256 KiB threshold String largePayload = generatePayload(1_500_000); LargePayloadStorageOptions payloadOptions = new LargePayloadStorageOptions() .setConnectionString(AZURITE_CONNECTION_STRING) - .setThresholdBytes(900_000); + .setThresholdBytes(256 * 1024); PayloadStore store = new BlobPayloadStore(payloadOptions); @@ -129,7 +132,7 @@ void roundTrip_smallInput_passedInline() throws TimeoutException { LargePayloadStorageOptions payloadOptions = new LargePayloadStorageOptions() .setConnectionString(AZURITE_CONNECTION_STRING) - .setThresholdBytes(900_000); + .setThresholdBytes(256 * 1024); PayloadStore store = new BlobPayloadStore(payloadOptions); @@ -171,7 +174,7 @@ void chunking_fanOutOrchestration_manyActivities() throws TimeoutException { LargePayloadStorageOptions payloadOptions = new LargePayloadStorageOptions() .setConnectionString(AZURITE_CONNECTION_STRING) - .setThresholdBytes(900_000); + .setThresholdBytes(256 * 1024); PayloadStore store = new BlobPayloadStore(payloadOptions); @@ -506,7 +509,7 @@ void compressionDisabled_roundTrip() throws TimeoutException { LargePayloadStorageOptions payloadOptions = new LargePayloadStorageOptions() .setConnectionString(AZURITE_CONNECTION_STRING) - .setThresholdBytes(900_000) + .setThresholdBytes(256 * 1024) .setCompressionEnabled(false); PayloadStore store = new BlobPayloadStore(payloadOptions); @@ -775,12 +778,93 @@ void largeActivityOutput_exceedsChunkSize_externalizesWithLP() throws TimeoutExc assertEquals(veryLargePayload.length(), instance.readOutputAs(Integer.class)); } + @Test + void transientActivityUploadFailure_isRedeliveredAndCompletes() throws TimeoutException { + verifyTransientUploadRecovery(true); + } + + @Test + void transientOrchestrationUploadFailure_isRedeliveredAndCompletes() throws TimeoutException { + verifyTransientUploadRecovery(false); + } + + private void verifyTransientUploadRecovery(boolean activityOutput) throws TimeoutException { + String payload = generatePayload(1_500_000); + LargePayloadStorageOptions options = createDefaultPayloadOptions(); + PayloadStore realStore = new BlobPayloadStore(options); + AtomicInteger uploadAttempts = new AtomicInteger(); + AtomicInteger executions = new AtomicInteger(); + PayloadStore failingOnceStore = new PayloadStore() { + @Override + public String upload(String value) { + if (uploadAttempts.incrementAndGet() == 1) { + throw new UncheckedIOException(new IOException("Injected transient upload failure")); + } + return realStore.upload(value); + } + + @Override + public String download(String token) { + return realStore.download(token); + } + + @Override + public boolean isKnownPayloadToken(String value) { + return realStore.isKnownPayloadToken(value); + } + }; + String name = activityOutput ? "RetryActivityUpload" : "RetryOrchestrationUpload"; + DurableTaskGrpcWorkerBuilder workerBuilder = createWorkerBuilder(); + LargePayloadWorkerExtensions.useExternalizedPayloads(workerBuilder, failingOnceStore, options); + workerBuilder.addOrchestration(new TaskOrchestrationFactory() { + @Override + public String getName() { return name; } + + @Override + public TaskOrchestration create() { + return ctx -> { + if (activityOutput) { + ctx.complete(ctx.callActivity(name, null, String.class).await()); + } else { + executions.incrementAndGet(); + ctx.complete(payload); + } + }; + } + }); + workerBuilder.addActivity(new TaskActivityFactory() { + @Override + public String getName() { return name; } + + @Override + public TaskActivity create() { + return ctx -> { + executions.incrementAndGet(); + return payload; + }; + } + }); + worker = workerBuilder.build(); + worker.start(); + DurableTaskGrpcClientBuilder clientBuilder = createClientBuilder(); + LargePayloadClientExtensions.useExternalizedPayloads(clientBuilder, realStore, options); + client = clientBuilder.build(); + + String instanceId = client.scheduleNewOrchestrationInstance(name); + OrchestrationMetadata result = client.waitForInstanceCompletion(instanceId, Duration.ofSeconds(90), true); + + assertEquals(OrchestrationRuntimeStatus.COMPLETED, result.getRuntimeStatus()); + assertEquals(payload, result.readOutputAs(String.class)); + assertTrue(uploadAttempts.get() >= 2, "Failed upload must be retried"); + assertTrue(executions.get() >= 2, "Scheduler must redeliver the uncompleted work item"); + } + // ==================== Helpers ==================== private LargePayloadStorageOptions createDefaultPayloadOptions() { return new LargePayloadStorageOptions() .setConnectionString(AZURITE_CONNECTION_STRING) - .setThresholdBytes(900_000); + .setThresholdBytes(256 * 1024); } private DurableTaskGrpcWorkerBuilder createWorkerBuilder() { diff --git a/azure-blob-payloads/src/test/java/com/microsoft/durabletask/azureblobpayloads/LargePayloadInterceptorTest.java b/azure-blob-payloads/src/test/java/com/microsoft/durabletask/azureblobpayloads/LargePayloadInterceptorTest.java index 595a047..fa3093b 100644 --- a/azure-blob-payloads/src/test/java/com/microsoft/durabletask/azureblobpayloads/LargePayloadInterceptorTest.java +++ b/azure-blob-payloads/src/test/java/com/microsoft/durabletask/azureblobpayloads/LargePayloadInterceptorTest.java @@ -2,12 +2,32 @@ // Licensed under the MIT License. package com.microsoft.durabletask.azureblobpayloads; +import com.azure.core.http.HttpResponse; import com.azure.storage.blob.models.BlobStorageException; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.google.protobuf.Empty; +import com.google.protobuf.Message; import com.google.protobuf.StringValue; +import com.microsoft.durabletask.EntityInstanceId; +import com.microsoft.durabletask.EntityOperationFailedException; +import com.microsoft.durabletask.FailureDetails; +import com.microsoft.durabletask.OrchestrationRunner; +import com.microsoft.durabletask.TaskOrchestration; import com.microsoft.durabletask.implementation.protobuf.OrchestratorService.*; +import io.grpc.CallOptions; +import io.grpc.Channel; +import io.grpc.ClientCall; +import io.grpc.MethodDescriptor; +import io.grpc.protobuf.ProtoUtils; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.io.IOException; +import java.io.UncheckedIOException; +import java.util.concurrent.atomic.AtomicReference; import static org.junit.jupiter.api.Assertions.*; import static org.mockito.ArgumentMatchers.anyString; @@ -190,6 +210,51 @@ void externalize_OrchestratorResponse_permanentFailure_becomesFailedCompletion() assertTrue(r.getActions(0).getCompleteOrchestration().getFailureDetails().getIsNonRetriable()); } + @ParameterizedTest + @ValueSource(ints = {408, 429, 503}) + void transientWorkerUploadFailures_propagateWithoutSendingACompletion(int status) { + BlobStorageException storageFailure = storageFailure(status); + for (RuntimeException failure : new RuntimeException[] { + storageFailure, new PayloadStorageException("Storage upload failed", storageFailure)}) { + assertWorkerCompletionsNotSent(failure); + } + } + + @Test + void unexpectedWorkerUploadFailures_propagateWithoutSendingACompletion() { + assertWorkerCompletionsNotSent(new UncheckedIOException(new IOException("Connection reset"))); + } + + @ParameterizedTest + @ValueSource(ints = {401, 403, 404}) + void permanentUploadFailures_completeActivitiesAndOrchestrationsButNotEntities(int status) { + BlobStorageException storageFailure = storageFailure(status); + for (RuntimeException failure : new RuntimeException[] { + storageFailure, new PayloadStorageException("Storage upload failed", storageFailure)}) { + doThrow(failure).when(mockStore).upload(anyString()); + ActivityResponse activity = (ActivityResponse) invokeExternalize(ActivityResponse.newBuilder() + .setResult(StringValue.of("large-activity-result")).build()); + assertTrue(activity.getFailureDetails().getIsNonRetriable()); + assertFalse(activity.hasResult()); + OrchestratorResponse orchestration = (OrchestratorResponse) invokeExternalize( + OrchestratorResponse.newBuilder() + .setCustomStatus(StringValue.of("large-custom-status")).build()); + assertEquals(OrchestrationStatus.ORCHESTRATION_STATUS_FAILED, + orchestration.getActions(0).getCompleteOrchestration().getOrchestrationStatus()); + assertTrue(orchestration.getActions(0).getCompleteOrchestration() + .getFailureDetails().getIsNonRetriable()); + assertRequestNotSent(EntityBatchResult.newBuilder() + .setEntityState(StringValue.of("large-entity-state")).build(), failure); + } + } + + @Test + void permanentEntityPayloadSizeFailure_propagatesWithoutSendingACompletion() { + assertRequestNotSentWithFailure(EntityBatchResult.newBuilder() + .setEntityState(StringValue.of("large-entity-state")).build(), + new PayloadStorageException("Payload too large")); + } + @Test void externalize_OrchestratorAction_ScheduleTask_input() { OrchestratorResponse request = OrchestratorResponse.newBuilder() @@ -413,6 +478,115 @@ void resolve_historyEvent_allTypes() { verifyEventResolution(HistoryEvent.newBuilder() .setEntityOperationCompleted(EntityOperationCompletedEvent.newBuilder() .setOutput(StringValue.of("blob:v1:c:n")).build()).build()); + verifyEventResolution(HistoryEvent.newBuilder() + .setEntityOperationFailed(EntityOperationFailedEvent.newBuilder() + .setFailureDetails(TaskFailureDetails.newBuilder() + .setErrorType("EntityFailure") + .setStackTrace(StringValue.of("blob:v1:c:n"))).build()).build()); + } + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + void resolve_entityOperationFailed_resolvesNestedFailureStackTraces(boolean streamedHistory) { + when(mockStore.download("blob:v1:container:outer")).thenReturn("outer stack"); + when(mockStore.download("blob:v1:container:inner")).thenReturn("inner stack"); + + TaskFailureDetails inner = TaskFailureDetails.newBuilder() + .setErrorType("Inner") + .setStackTrace(StringValue.of("blob:v1:container:inner")) + .build(); + TaskFailureDetails outer = TaskFailureDetails.newBuilder() + .setErrorType("Outer") + .setStackTrace(StringValue.of("blob:v1:container:outer")) + .setInnerFailure(inner) + .build(); + HistoryEvent event = HistoryEvent.newBuilder() + .setEntityOperationFailed(EntityOperationFailedEvent.newBuilder().setFailureDetails(outer)) + .build(); + TaskFailureDetails resolved; + if (streamedHistory) { + HistoryChunk chunk = (HistoryChunk) invokeResolve(HistoryChunk.newBuilder().addEvents(event).build()); + resolved = chunk.getEvents(0).getEntityOperationFailed().getFailureDetails(); + } else { + WorkItem workItem = (WorkItem) invokeResolve(WorkItem.newBuilder() + .setOrchestratorRequest(OrchestratorRequest.newBuilder() + .addPastEvents(event).addNewEvents(event)).build()); + resolved = workItem.getOrchestratorRequest().getNewEvents(0) + .getEntityOperationFailed().getFailureDetails(); + assertEquals(resolved, workItem.getOrchestratorRequest().getPastEvents(0) + .getEntityOperationFailed().getFailureDetails()); + } + assertEquals("outer stack", resolved.getStackTrace().getValue()); + assertEquals("inner stack", resolved.getInnerFailure().getStackTrace().getValue()); + } + + @Test + void externalizedEntityFailure_reachesCallerWithOriginalNestedDiagnostics() throws Exception { + options.setThresholdBytes(1024); + interceptor = new LargePayloadInterceptor(mockStore, options); + String outerTrace = "outer trace\n".repeat(200); + String innerTrace = "inner trace\n".repeat(200); + when(mockStore.upload(outerTrace)).thenReturn("blob:v1:container:outer"); + when(mockStore.upload(innerTrace)).thenReturn("blob:v1:container:inner"); + when(mockStore.download("blob:v1:container:outer")).thenReturn(outerTrace); + when(mockStore.download("blob:v1:container:inner")).thenReturn(innerTrace); + TaskFailureDetails failure = TaskFailureDetails.newBuilder() + .setErrorType("Outer").setErrorMessage("operation failed") + .setStackTrace(StringValue.of(outerTrace)) + .setInnerFailure(TaskFailureDetails.newBuilder().setErrorType("Inner") + .setErrorMessage("inner cause").setStackTrace(StringValue.of(innerTrace))) + .build(); + EntityBatchResult externalized = (EntityBatchResult) invokeExternalize(EntityBatchResult.newBuilder() + .addResults(OperationResult.newBuilder().setFailure( + OperationResultFailure.newBuilder().setFailureDetails(failure))).build()); + TaskFailureDetails storedFailure = externalized.getResults(0).getFailure().getFailureDetails(); + assertEquals("blob:v1:container:outer", storedFailure.getStackTrace().getValue()); + assertEquals("blob:v1:container:inner", storedFailure.getInnerFailure().getStackTrace().getValue()); + + AtomicReference caught = new AtomicReference<>(); + TaskOrchestration orchestration = ctx -> { + try { + ctx.callEntity(new EntityInstanceId("Counter", "one"), "get", null, Integer.class).await(); + ctx.complete("unexpected success"); + } catch (EntityOperationFailedException e) { + caught.set(e.getFailureDetails()); + ctx.complete("caught"); + } + }; + HistoryEvent started = HistoryEvent.newBuilder() + .setOrchestratorStarted(OrchestratorStartedEvent.getDefaultInstance()).build(); + HistoryEvent execution = HistoryEvent.newBuilder() + .setExecutionStarted(ExecutionStartedEvent.newBuilder().setName("Caller") + .setOrchestrationInstance(OrchestrationInstance.newBuilder().setInstanceId("caller"))).build(); + HistoryEvent completed = HistoryEvent.newBuilder() + .setOrchestratorCompleted(OrchestratorCompletedEvent.getDefaultInstance()).build(); + OrchestratorRequest initial = OrchestratorRequest.newBuilder().setInstanceId("caller") + .addPastEvents(started).addPastEvents(execution).addNewEvents(completed).build(); + OrchestratorResponse first = OrchestratorResponse.parseFrom( + OrchestrationRunner.loadAndRun(initial.toByteArray(), orchestration)); + String requestId = new ObjectMapper().readTree(first.getActions(0).getSendEvent().getData().getValue()) + .path("id").asText(); + assertFalse(requestId.isEmpty()); + WorkItem replay = WorkItem.newBuilder().setOrchestratorRequest(initial.toBuilder() + .clearNewEvents() + .addPastEvents(HistoryEvent.newBuilder().setEventId(0) + .setEventSent(EventSentEvent.getDefaultInstance())) + .addPastEvents(completed) + .addNewEvents(started) + .addNewEvents(HistoryEvent.newBuilder().setEntityOperationFailed( + EntityOperationFailedEvent.newBuilder().setRequestId(requestId).setFailureDetails(storedFailure))) + .addNewEvents(completed)).build(); + WorkItem resolved = (WorkItem) invokeResolve(replay); + OrchestratorResponse result = OrchestratorResponse.parseFrom( + OrchestrationRunner.loadAndRun(resolved.getOrchestratorRequest().toByteArray(), orchestration)); + + assertNotNull(caught.get(), "The actual entity caller must receive the failure"); + assertEquals("Outer", caught.get().getErrorType()); + assertEquals("operation failed", caught.get().getErrorMessage()); + assertEquals(outerTrace, caught.get().getStackTrace()); + assertEquals(innerTrace, caught.get().getInnerFailure().getStackTrace()); + assertEquals(OrchestrationStatus.ORCHESTRATION_STATUS_COMPLETED, + result.getActions(0).getCompleteOrchestration().getOrchestrationStatus()); } @Test @@ -453,6 +627,64 @@ void isPermanentStorageFailure_runtimeException_returnsFalse() { // ==================== Helpers ==================== + private static BlobStorageException storageFailure(int status) { + HttpResponse response = mock(HttpResponse.class); + when(response.getStatusCode()).thenReturn(status); + return new BlobStorageException("Storage upload failed", response, null); + } + + private void assertWorkerCompletionsNotSent(RuntimeException failure) { + TaskFailureDetails details = TaskFailureDetails.newBuilder() + .setErrorType("OriginalError") + .setStackTrace(StringValue.of("large-failure-stack-trace")) + .build(); + Message[] requests = { + ActivityResponse.newBuilder().setResult(StringValue.of("large-activity-result")).build(), + ActivityResponse.newBuilder().setFailureDetails(details).build(), + OrchestratorResponse.newBuilder().setCustomStatus(StringValue.of("large-custom-status")).build(), + OrchestratorResponse.newBuilder().addActions(OrchestratorAction.newBuilder() + .setScheduleTask(ScheduleTaskAction.newBuilder() + .setInput(StringValue.of("large-activity-input")))).build(), + OrchestratorResponse.newBuilder().addActions(OrchestratorAction.newBuilder() + .setCompleteOrchestration(CompleteOrchestrationAction.newBuilder() + .setFailureDetails(details))).build(), + EntityBatchResult.newBuilder().setEntityState(StringValue.of("large-entity-state")).build(), + EntityBatchResult.newBuilder().addResults(OperationResult.newBuilder() + .setSuccess(OperationResultSuccess.newBuilder() + .setResult(StringValue.of("large-entity-result")))).build(), + EntityBatchResult.newBuilder().addActions(OperationAction.newBuilder() + .setSendSignal(SendSignalAction.newBuilder() + .setInput(StringValue.of("large-entity-signal")))).build(), + EntityBatchResult.newBuilder().setFailureDetails(details).build(), + EntityBatchResult.newBuilder().addResults(OperationResult.newBuilder() + .setFailure(OperationResultFailure.newBuilder().setFailureDetails(details))).build() + }; + for (Message request : requests) { + assertRequestNotSentWithFailure(request, failure); + } + } + + private void assertRequestNotSentWithFailure(ReqT request, RuntimeException failure) { + doThrow(failure).when(mockStore).upload(anyString()); + assertRequestNotSent(request, failure); + } + + private void assertRequestNotSent(ReqT request, RuntimeException failure) { + MethodDescriptor method = MethodDescriptor.newBuilder() + .setType(MethodDescriptor.MethodType.UNARY) + .setFullMethodName("test/CompleteTask") + .setRequestMarshaller(ProtoUtils.marshaller(request)) + .setResponseMarshaller(ProtoUtils.marshaller(Empty.getDefaultInstance())) + .build(); + ClientCall delegate = mock(); + Channel channel = mock(Channel.class); + when(channel.newCall(method, CallOptions.DEFAULT)).thenReturn(delegate); + ClientCall call = interceptor.interceptCall(method, CallOptions.DEFAULT, channel); + + assertSame(failure, assertThrows(RuntimeException.class, () -> call.sendMessage(request))); + verify(delegate, never()).sendMessage(any()); + } + /** * Uses reflection to invoke the private externalizeRequestPayloads method. */ diff --git a/azure-blob-payloads/src/test/java/com/microsoft/durabletask/azureblobpayloads/LargePayloadStorageOptionsTest.java b/azure-blob-payloads/src/test/java/com/microsoft/durabletask/azureblobpayloads/LargePayloadStorageOptionsTest.java index baedb49..5c431c7 100644 --- a/azure-blob-payloads/src/test/java/com/microsoft/durabletask/azureblobpayloads/LargePayloadStorageOptionsTest.java +++ b/azure-blob-payloads/src/test/java/com/microsoft/durabletask/azureblobpayloads/LargePayloadStorageOptionsTest.java @@ -14,7 +14,7 @@ class LargePayloadStorageOptionsTest { @Test void defaults_areCorrect() { LargePayloadStorageOptions options = new LargePayloadStorageOptions(); - assertEquals(900_000, options.getThresholdBytes()); + assertEquals(256 * 1024, options.getThresholdBytes()); assertEquals(10 * 1024 * 1024, options.getMaxPayloadBytes()); assertNull(options.getConnectionString()); assertNull(options.getAccountUri()); diff --git a/azure-blob-payloads/src/test/java/com/microsoft/durabletask/azureblobpayloads/OfflineBlobService.java b/azure-blob-payloads/src/test/java/com/microsoft/durabletask/azureblobpayloads/OfflineBlobService.java new file mode 100644 index 0000000..7607316 --- /dev/null +++ b/azure-blob-payloads/src/test/java/com/microsoft/durabletask/azureblobpayloads/OfflineBlobService.java @@ -0,0 +1,117 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. +package com.microsoft.durabletask.azureblobpayloads; + +import com.azure.core.http.HttpClient; +import com.azure.core.http.HttpHeaders; +import com.azure.core.http.HttpMethod; +import com.azure.core.http.HttpRequest; +import com.azure.core.http.HttpResponse; +import com.azure.core.util.BinaryData; +import com.azure.storage.blob.BlobContainerClientBuilder; +import com.azure.storage.common.policy.RequestRetryOptions; +import com.azure.storage.common.policy.RetryPolicyType; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.Mockito.*; + +/** + * An offline HTTP service beneath the real Azure Blob SDK, including its retry policy. + */ +final class OfflineBlobService implements HttpClient { + private final Map blobs = new HashMap<>(); + private final Map encodings = new HashMap<>(); + final List putUrls = new ArrayList<>(); + final List putBodies = new ArrayList<>(); + boolean loseFirstPutResponse; + boolean changeStoredBytes; + boolean changeStoredEncoding; + int downloadStatus = 200; + int downloads; + + BlobPayloadStore createStore(String container, boolean compressed) { + return new BlobPayloadStore(new BlobContainerClientBuilder() + .endpoint("https://offline.invalid/" + container) + .httpClient(this) + .retryOptions(new RequestRetryOptions(RetryPolicyType.FIXED, 2, 5, 1L, 1L, null)) + .buildClient(), + new LargePayloadStorageOptions().setCompressionEnabled(compressed)); + } + + @Override + public Mono send(HttpRequest request) { + return BinaryData.fromFlux(request.getBody() == null ? Flux.empty() : request.getBody()) + .flatMap(data -> Mono.fromCallable(() -> respond(request, data.toBytes()))); + } + + private synchronized HttpResponse respond(HttpRequest request, byte[] body) throws IOException { + String url = request.getUrl().toString(); + if (request.getHttpMethod() == HttpMethod.PUT && url.contains("restype=container")) { + return response(request, 201, new byte[0], null); + } + String path = request.getUrl().getPath(); + if (request.getHttpMethod() == HttpMethod.PUT) { + assertEquals("*", request.getHeaders().getValue("If-None-Match")); + putUrls.add(url); + putBodies.add(body); + if (blobs.containsKey(path)) { + return response(request, 412, new byte[0], "ConditionNotMet"); + } + blobs.put(path, changeStoredBytes ? "collision".getBytes(StandardCharsets.UTF_8) : body); + String encoding = request.getHeaders().getValue("x-ms-blob-content-encoding"); + encodings.put(path, changeStoredEncoding ? "br" : encoding); + if (loseFirstPutResponse) { + // Persist first, then fail the transport before the caller sees a response. + throw new IOException("Connection lost after committing the PUT"); + } + return response(request, 201, new byte[0], null); + } + if (request.getHttpMethod() == HttpMethod.GET) { + downloads++; + if (downloadStatus != 200) { + return response(request, downloadStatus, new byte[0], "ServerBusy"); + } + byte[] stored = blobs.get(path); + if (stored == null) { + return response(request, 404, new byte[0], "BlobNotFound"); + } + HttpResponse response = response(request, 200, stored, null); + String encoding = encodings.get(path); + if (encoding != null) { + response.getHeaders().set("Content-Encoding", encoding); + } + return response; + } + throw new AssertionError("Unexpected request: " + request.getHttpMethod() + " " + url); + } + + private static HttpResponse response(HttpRequest request, int status, byte[] body, String errorCode) { + HttpHeaders headers = new HttpHeaders() + .set("Content-Length", Integer.toString(body.length)) + .set("ETag", "\"offline-etag\"") + .set("Last-Modified", "Mon, 19 Oct 2026 00:00:00 GMT") + .set("x-ms-request-id", "offline-request"); + if (errorCode != null) { + headers.set("x-ms-error-code", errorCode); + } + HttpResponse response = mock(HttpResponse.class); + when(response.getRequest()).thenReturn(request); + when(response.getStatusCode()).thenReturn(status); + when(response.getHeaders()).thenReturn(headers); + when(response.getHeaderValue(anyString())).thenAnswer(inv -> headers.getValue(inv.getArgument(0))); + when(response.getBody()).thenAnswer(inv -> Flux.just(ByteBuffer.wrap(body))); + when(response.getBodyAsByteArray()).thenReturn(Mono.just(body)); + when(response.getBodyAsString()).thenReturn(Mono.just(new String(body, StandardCharsets.UTF_8))); + return response; + } +} diff --git a/azure-blob-payloads/src/test/java/com/microsoft/durabletask/azureblobpayloads/PayloadTokenTest.java b/azure-blob-payloads/src/test/java/com/microsoft/durabletask/azureblobpayloads/PayloadTokenTest.java index fd25588..e72346b 100644 --- a/azure-blob-payloads/src/test/java/com/microsoft/durabletask/azureblobpayloads/PayloadTokenTest.java +++ b/azure-blob-payloads/src/test/java/com/microsoft/durabletask/azureblobpayloads/PayloadTokenTest.java @@ -3,6 +3,8 @@ package com.microsoft.durabletask.azureblobpayloads; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; import static org.junit.jupiter.api.Assertions.*; @@ -56,6 +58,28 @@ void isKnownPayloadToken_validToken_returnsTrue() { "blob:v1:container:0123456789abcdef0123456789abcdef")); } + @Test + void isKnownPayloadToken_rootContainer_returnsTrue() { + BlobPayloadStore store = createStoreWithDefaults(); + assertTrue(store.isKnownPayloadToken( + "blob:v1:$root:0123456789abcdef0123456789abcdef")); + } + + @ParameterizedTest + @ValueSource(strings = { + "blob:v1:$ROOT:0123456789abcdef0123456789abcdef", + "blob:v1:$root-extra:0123456789abcdef0123456789abcdef", + "blob:v1:$logs:0123456789abcdef0123456789abcdef", + "blob:v1:$root:0123456789abcdef0123456789abcde", + "blob:v1:$root:0123456789abcdef0123456789abcdeg", + "blob:v1:$root:0123456789abcdef0123456789abcdef:extra", + "blob:v1:$root:0123456789abcdef0123456789abcdef\n", + "\"blob:v1:$root:0123456789abcdef0123456789abcdef\"" + }) + void isKnownPayloadToken_rootLookingLiteral_returnsFalse(String literal) { + assertFalse(createStoreWithDefaults().isKnownPayloadToken(literal)); + } + @Test void isKnownPayloadToken_prefixOnly_returnsFalse() { BlobPayloadStore store = createStoreWithDefaults(); diff --git a/azure-blob-payloads/src/test/java/com/microsoft/durabletask/azureblobpayloads/RootContainerRoundTripTest.java b/azure-blob-payloads/src/test/java/com/microsoft/durabletask/azureblobpayloads/RootContainerRoundTripTest.java new file mode 100644 index 0000000..0a678c5 --- /dev/null +++ b/azure-blob-payloads/src/test/java/com/microsoft/durabletask/azureblobpayloads/RootContainerRoundTripTest.java @@ -0,0 +1,153 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. +package com.microsoft.durabletask.azureblobpayloads; + +import com.google.protobuf.StringValue; +import com.microsoft.durabletask.DurableTaskClient; +import com.microsoft.durabletask.DurableTaskGrpcClientBuilder; +import com.microsoft.durabletask.DurableTaskGrpcWorker; +import com.microsoft.durabletask.DurableTaskGrpcWorkerBuilder; +import com.microsoft.durabletask.OrchestrationMetadata; +import com.microsoft.durabletask.TaskActivity; +import com.microsoft.durabletask.TaskActivityFactory; +import com.microsoft.durabletask.history.ExecutionStartedEvent; +import com.microsoft.durabletask.history.GenericEvent; +import com.microsoft.durabletask.history.TaskCompletedEvent; +import com.microsoft.durabletask.implementation.protobuf.OrchestratorService.*; +import com.microsoft.durabletask.implementation.protobuf.TaskHubSidecarServiceGrpc; +import io.grpc.ManagedChannel; +import io.grpc.Server; +import io.grpc.inprocess.InProcessChannelBuilder; +import io.grpc.inprocess.InProcessServerBuilder; +import io.grpc.stub.StreamObserver; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import static org.junit.jupiter.api.Assertions.*; + +class RootContainerRoundTripTest { + @ParameterizedTest + @ValueSource(booleans = {true, false}) + void rootTokens_resolveThroughClientWorkerAndStreamedHistory(boolean compressed) throws Exception { + OfflineBlobService service = new OfflineBlobService(); + BlobPayloadStore store = service.createStore("$root", compressed); + LargePayloadStorageOptions options = new LargePayloadStorageOptions().setThresholdBytes(1024); + String payload = "root payload \u00e9\n".repeat(200); + String literal = "blob:v1:$root:not-a-guid"; + AtomicReference scheduled = new AtomicReference<>(); + AtomicReference completion = new AtomicReference<>(); + AtomicReference activityInput = new AtomicReference<>(); + CountDownLatch completed = new CountDownLatch(1); + String serverName = InProcessServerBuilder.generateName(); + Server server = InProcessServerBuilder.forName(serverName).directExecutor() + .addService(new TaskHubSidecarServiceGrpc.TaskHubSidecarServiceImplBase() { + @Override + public void startInstance(CreateInstanceRequest request, + StreamObserver observer) { + scheduled.set(request); + observer.onNext(CreateInstanceResponse.newBuilder().setInstanceId(request.getInstanceId()).build()); + observer.onCompleted(); + } + + @Override + public void getWorkItems(GetWorkItemsRequest request, StreamObserver observer) { + observer.onNext(WorkItem.newBuilder().setCompletionToken("activity-token") + .setActivityRequest(ActivityRequest.newBuilder().setName("Echo").setTaskId(1) + .setOrchestrationInstance(OrchestrationInstance.newBuilder().setInstanceId("root-test")) + .setInput(scheduled.get().getInput())).build()); + // Keep the stream open until worker shutdown, as a scheduler would. + } + + @Override + public void completeActivityTask(ActivityResponse request, StreamObserver observer) { + completion.set(request); + observer.onNext(CompleteTaskResponse.getDefaultInstance()); + observer.onCompleted(); + completed.countDown(); + } + + @Override + public void getInstance(GetInstanceRequest request, StreamObserver observer) { + observer.onNext(GetInstanceResponse.newBuilder().setExists(true) + .setOrchestrationState(OrchestrationState.newBuilder().setInstanceId("root-test") + .setName("Root").setInput(scheduled.get().getInput()).setOutput(completion.get().getResult()) + .setOrchestrationStatus(OrchestrationStatus.ORCHESTRATION_STATUS_COMPLETED)).build()); + observer.onCompleted(); + } + + @Override + public void streamInstanceHistory(StreamInstanceHistoryRequest request, + StreamObserver observer) { + observer.onNext(HistoryChunk.newBuilder() + .addEvents(HistoryEvent.newBuilder().setExecutionStarted( + com.microsoft.durabletask.implementation.protobuf.OrchestratorService.ExecutionStartedEvent + .newBuilder().setName("Root").setInput(scheduled.get().getInput()))) + .addEvents(HistoryEvent.newBuilder().setTaskCompleted( + com.microsoft.durabletask.implementation.protobuf.OrchestratorService.TaskCompletedEvent + .newBuilder().setTaskScheduledId(1).setResult(completion.get().getResult()))) + .addEvents(HistoryEvent.newBuilder().setGenericEvent( + com.microsoft.durabletask.implementation.protobuf.OrchestratorService.GenericEvent + .newBuilder().setData(StringValue.of(literal)))) + .build()); + observer.onCompleted(); + } + }).build().start(); + ManagedChannel channel = InProcessChannelBuilder.forName(serverName).directExecutor().build(); + DurableTaskGrpcClientBuilder clientBuilder = new DurableTaskGrpcClientBuilder().grpcChannel(channel); + LargePayloadClientExtensions.useExternalizedPayloads(clientBuilder, store, options); + DurableTaskGrpcWorkerBuilder workerBuilder = new DurableTaskGrpcWorkerBuilder().grpcChannel(channel) + .addActivity(new TaskActivityFactory() { + @Override + public String getName() { + return "Echo"; + } + + @Override + public TaskActivity create() { + return ctx -> { + String input = ctx.getInput(String.class); + activityInput.set(input); + return input; + }; + } + }); + LargePayloadWorkerExtensions.useExternalizedPayloads(workerBuilder, store, options); + try (DurableTaskClient client = clientBuilder.build(); + DurableTaskGrpcWorker worker = workerBuilder.build()) { + client.scheduleNewOrchestrationInstance("Root", payload); + String inputToken = scheduled.get().getInput().getValue(); + assertTrue(inputToken.startsWith("blob:v1:$root:")); + assertTrue(store.isKnownPayloadToken(inputToken)); + worker.start(); + assertTrue(completed.await(10, TimeUnit.SECONDS), "Worker must complete the activity"); + assertFalse(completion.get().hasFailureDetails()); + assertEquals(payload, activityInput.get(), "Worker typed input must contain the original payload"); + assertTrue(store.isKnownPayloadToken(completion.get().getResult().getValue())); + OrchestrationMetadata metadata = client.getInstanceMetadata("root-test", true); + assertEquals(payload, metadata.readInputAs(String.class)); + assertEquals(payload, metadata.readOutputAs(String.class)); + int downloadsBeforeHistory = service.downloads; + List history = client.getOrchestrationHistory("root-test"); + assertEquals(3, history.size()); + assertEquals(scheduledPayload(payload), ((ExecutionStartedEvent) history.get(0)).getInput()); + assertEquals(scheduledPayload(payload), ((TaskCompletedEvent) history.get(1)).getResult()); + assertEquals(literal, ((GenericEvent) history.get(2)).getData()); + assertEquals(downloadsBeforeHistory + 2, service.downloads, + "A token-looking literal must not cause a blob download"); + } finally { + channel.shutdownNow(); + server.shutdownNow(); + assertTrue(channel.awaitTermination(5, TimeUnit.SECONDS)); + assertTrue(server.awaitTermination(5, TimeUnit.SECONDS)); + } + } + + private static String scheduledPayload(String payload) { + return new com.microsoft.durabletask.JacksonDataConverter().serialize(payload); + } +} diff --git a/client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcWorker.java b/client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcWorker.java index 086b1cf..87b0b64 100644 --- a/client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcWorker.java +++ b/client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcWorker.java @@ -344,7 +344,7 @@ public void startAndBlock() { .setCompletionToken(workItem.getCompletionToken()) .build(); - this.completeOrchestratorTaskWithChunking(response); + this.completeOrchestratorTaskOrAbandon(response); } else { switch(versioningOptions.getFailureStrategy()) { case FAIL: @@ -366,7 +366,7 @@ public void startAndBlock() { .addActions(action) .build(); - this.completeOrchestratorTaskWithChunking(response); + this.completeOrchestratorTaskOrAbandon(response); break; // Reject and default share the same behavior as it does not change the orchestration to a terminal state. case REJECT: @@ -425,7 +425,18 @@ public void startAndBlock() { responseBuilder.setFailureDetails(failureDetails); } - this.sidecarClient.completeActivityTask(responseBuilder.build()); + try { + this.sidecarClient.completeActivityTask(responseBuilder.build()); + } catch (StatusRuntimeException e) { + throw e; + } catch (RuntimeException e) { + logger.log(Level.WARNING, + "Failed to complete activity for '" + activityInstanceId + + "'. Abandoning work item.", e); + this.sidecarClient.abandonTaskActivityWorkItem(AbandonActivityTaskRequest.newBuilder() + .setCompletionToken(workItem.getCompletionToken()) + .build()); + } } else if (requestType == RequestCase.ENTITYREQUEST) { EntityBatchRequest entityRequest = workItem.getEntityRequest(); this.workItemExecutor.submit(() -> { @@ -559,6 +570,21 @@ else if (requestType == RequestCase.HEALTHPING) } } + private void completeOrchestratorTaskOrAbandon(OrchestratorResponse response) { + try { + this.completeOrchestratorTaskWithChunking(response); + } catch (StatusRuntimeException e) { + throw e; + } catch (RuntimeException e) { + logger.log(Level.WARNING, + "Failed to complete orchestrator for '" + response.getInstanceId() + + "'. Abandoning work item.", e); + this.sidecarClient.abandonTaskOrchestratorWorkItem(AbandonOrchestrationTaskRequest.newBuilder() + .setCompletionToken(response.getCompletionToken()) + .build()); + } + } + /** * Stops the current worker's listen loop, preventing any new orchestrator or activity events from being processed. */ diff --git a/client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcWorkerBuilder.java b/client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcWorkerBuilder.java index 3df0a28..f511f89 100644 --- a/client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcWorkerBuilder.java +++ b/client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcWorkerBuilder.java @@ -47,10 +47,10 @@ public final class DurableTaskGrpcWorkerBuilder { public static final int DEFAULT_MAX_CHUNK_SIZE_BYTES = 4_089_446; int maxChunkSizeBytes = DEFAULT_MAX_CHUNK_SIZE_BYTES; - // Default threshold matches LargePayloadStorageOptions default (900_000 bytes). + // Default threshold matches LargePayloadStorageOptions default (256 KiB). // Used by the worker to estimate post-externalization action sizes during // pre-send validation when supportsLargePayloads=true. - int largePayloadThresholdBytes = 900_000; + int largePayloadThresholdBytes = 256 * 1024; /** * Adds an orchestration factory to be used by the constructed {@link DurableTaskGrpcWorker}. @@ -409,7 +409,7 @@ public DurableTaskGrpcWorkerBuilder setSupportsLargePayloads(boolean enabled) { * post-externalization action sizes during pre-send validation. *

* This should match the {@code thresholdBytes} configured on the large-payload - * interceptor. The default is 900,000 bytes. Fields whose UTF-8 byte length meets + * interceptor. The default is 256 KiB (262,144 bytes). Fields whose UTF-8 byte length meets * or exceeds this threshold are assumed to be replaced with a small blob-token * reference by the interceptor. * diff --git a/client/src/main/java/com/microsoft/durabletask/FailureDetails.java b/client/src/main/java/com/microsoft/durabletask/FailureDetails.java index e9f8fd7..ae7ded3 100644 --- a/client/src/main/java/com/microsoft/durabletask/FailureDetails.java +++ b/client/src/main/java/com/microsoft/durabletask/FailureDetails.java @@ -80,12 +80,21 @@ public static FailureDetails fromException( } FailureDetails(TaskFailureDetails proto) { + this(proto, false); + } + + private FailureDetails(TaskFailureDetails proto, boolean preserveEmptyProperties) { this(proto.getErrorType(), proto.getErrorMessage(), proto.hasStackTrace() ? proto.getStackTrace().getValue() : null, proto.getIsNonRetriable(), - proto.hasInnerFailure() ? new FailureDetails(proto.getInnerFailure()) : null, - convertProtoProperties(proto.getPropertiesMap())); + proto.hasInnerFailure() ? new FailureDetails(proto.getInnerFailure(), preserveEmptyProperties) : null, + convertProtoProperties(proto.getPropertiesMap(), preserveEmptyProperties)); + } + + static FailureDetails fromHistoryProto(TaskFailureDetails proto) { + // The .NET history converter retains empty property dictionaries, including on inner failures. + return new FailureDetails(proto, true); } /** @@ -268,9 +277,10 @@ private static FailureDetails fromExceptionRecursive( } @Nullable - private static Map convertProtoProperties(Map protoProperties) { + private static Map convertProtoProperties( + Map protoProperties, boolean preserveEmptyProperties) { if (protoProperties == null || protoProperties.isEmpty()) { - return null; + return preserveEmptyProperties ? Collections.emptyMap() : null; } Map result = new HashMap<>(); diff --git a/client/src/main/java/com/microsoft/durabletask/HistoryEventConverter.java b/client/src/main/java/com/microsoft/durabletask/HistoryEventConverter.java index b1f94ea..1eaaaad 100644 --- a/client/src/main/java/com/microsoft/durabletask/HistoryEventConverter.java +++ b/client/src/main/java/com/microsoft/durabletask/HistoryEventConverter.java @@ -46,7 +46,7 @@ static HistoryEvent fromProto(OrchestratorService.HistoryEvent proto) { return new ExecutionCompletedEvent(id, ts, OrchestrationRuntimeStatus.fromProtobuf(p.getOrchestrationStatus()), stringOrNull(p.hasResult(), p.getResult()), - p.hasFailureDetails() ? new FailureDetails(p.getFailureDetails()) : null); + p.hasFailureDetails() ? FailureDetails.fromHistoryProto(p.getFailureDetails()) : null); } case EXECUTIONTERMINATED: { OrchestratorService.ExecutionTerminatedEvent p = proto.getExecutionTerminated(); @@ -68,7 +68,7 @@ static HistoryEvent fromProto(OrchestratorService.HistoryEvent proto) { case TASKFAILED: { OrchestratorService.TaskFailedEvent p = proto.getTaskFailed(); return new TaskFailedEvent(id, ts, p.getTaskScheduledId(), - p.hasFailureDetails() ? new FailureDetails(p.getFailureDetails()) : null); + p.hasFailureDetails() ? FailureDetails.fromHistoryProto(p.getFailureDetails()) : null); } case SUBORCHESTRATIONINSTANCECREATED: { OrchestratorService.SubOrchestrationInstanceCreatedEvent p = proto.getSubOrchestrationInstanceCreated(); @@ -89,7 +89,7 @@ static HistoryEvent fromProto(OrchestratorService.HistoryEvent proto) { case SUBORCHESTRATIONINSTANCEFAILED: { OrchestratorService.SubOrchestrationInstanceFailedEvent p = proto.getSubOrchestrationInstanceFailed(); return new SubOrchestrationInstanceFailedEvent(id, ts, p.getTaskScheduledId(), - p.hasFailureDetails() ? new FailureDetails(p.getFailureDetails()) : null); + p.hasFailureDetails() ? FailureDetails.fromHistoryProto(p.getFailureDetails()) : null); } case TIMERCREATED: { OrchestratorService.TimerCreatedEvent p = proto.getTimerCreated(); @@ -161,7 +161,7 @@ static HistoryEvent fromProto(OrchestratorService.HistoryEvent proto) { case ENTITYOPERATIONFAILED: { OrchestratorService.EntityOperationFailedEvent p = proto.getEntityOperationFailed(); return new EntityOperationFailedEvent(id, ts, p.getRequestId(), - p.hasFailureDetails() ? new FailureDetails(p.getFailureDetails()) : null); + p.hasFailureDetails() ? FailureDetails.fromHistoryProto(p.getFailureDetails()) : null); } case ENTITYLOCKREQUESTED: { OrchestratorService.EntityLockRequestedEvent p = proto.getEntityLockRequested(); @@ -238,7 +238,7 @@ private static OrchestrationState toOrchestrationState(OrchestratorService.Orche stringOrNull(p.hasInput(), p.getInput()), stringOrNull(p.hasOutput(), p.getOutput()), stringOrNull(p.hasCustomStatus(), p.getCustomStatus()), - p.hasFailureDetails() ? new FailureDetails(p.getFailureDetails()) : null, + p.hasFailureDetails() ? FailureDetails.fromHistoryProto(p.getFailureDetails()) : null, stringOrNull(p.hasExecutionId(), p.getExecutionId()), stringOrNull(p.hasParentInstanceId(), p.getParentInstanceId()), p.getTagsMap()); diff --git a/client/src/test/java/com/microsoft/durabletask/HistoryEventConverterTest.java b/client/src/test/java/com/microsoft/durabletask/HistoryEventConverterTest.java index 6078123..a6eb2c6 100644 --- a/client/src/test/java/com/microsoft/durabletask/HistoryEventConverterTest.java +++ b/client/src/test/java/com/microsoft/durabletask/HistoryEventConverterTest.java @@ -165,6 +165,25 @@ void convertsExecutionCompletedWithFailureDetails() { assertEquals("boom", completed.getFailureDetails().getErrorMessage()); } + @Test + void historyFailuresPreserveEmptyPropertyMapsWithoutChangingGeneralFailureConversion() { + OrchestratorService.TaskFailureDetails failure = OrchestratorService.TaskFailureDetails.newBuilder() + .setErrorType("Outer") + .setInnerFailure(OrchestratorService.TaskFailureDetails.newBuilder().setErrorType("Inner")) + .build(); + TaskFailedEvent event = (TaskFailedEvent) HistoryEventConverter.fromProto(baseEvent(1) + .setTaskFailed(OrchestratorService.TaskFailedEvent.newBuilder().setFailureDetails(failure)).build()); + FailureDetails converted = event.getFailureDetails(); + + assertNotNull(converted.getProperties()); + assertTrue(converted.getProperties().isEmpty()); + assertNotNull(converted.getInnerFailure().getProperties()); + assertTrue(converted.getInnerFailure().getProperties().isEmpty()); + assertThrows(UnsupportedOperationException.class, () -> converted.getProperties().put("key", "value")); + assertNull(new FailureDetails(failure).getProperties()); + assertNull(new FailureDetails(failure).getInnerFailure().getProperties()); + } + @Test void convertsTaskCompleted() { OrchestratorService.HistoryEvent proto = baseEvent(4) diff --git a/client/src/test/java/com/microsoft/durabletask/OrchestratorChunkingTest.java b/client/src/test/java/com/microsoft/durabletask/OrchestratorChunkingTest.java index 2ff4a64..df88e68 100644 --- a/client/src/test/java/com/microsoft/durabletask/OrchestratorChunkingTest.java +++ b/client/src/test/java/com/microsoft/durabletask/OrchestratorChunkingTest.java @@ -17,6 +17,7 @@ import org.junit.jupiter.api.Test; import java.lang.reflect.Field; +import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Method; import java.util.ArrayList; import java.util.Collections; @@ -36,6 +37,7 @@ public class OrchestratorChunkingTest { */ private final List capturedResponses = Collections.synchronizedList(new ArrayList<>()); + private final List abandonedTokens = new ArrayList<>(); private Server inProcessServer; private ManagedChannel inProcessChannel; @@ -55,6 +57,15 @@ public void completeOrchestratorTask( responseObserver.onNext(CompleteTaskResponse.getDefaultInstance()); responseObserver.onCompleted(); } + + @Override + public void abandonTaskOrchestratorWorkItem( + AbandonOrchestrationTaskRequest request, + StreamObserver responseObserver) { + abandonedTokens.add(request.getCompletionToken()); + responseObserver.onNext(AbandonOrchestrationTaskResponse.getDefaultInstance()); + responseObserver.onCompleted(); + } }) .build() .start(); @@ -76,6 +87,38 @@ void tearDown() { // ==================== validateActionsSize tests ==================== + @Test + void workerBuilder_defaultLargePayloadThreshold_matchesStorageDefault() { + DurableTaskGrpcWorkerBuilder builder = new DurableTaskGrpcWorkerBuilder(); + assertEquals(256 * 1024, builder.largePayloadThresholdBytes); + } + + @Test + void chunking_defaultThreshold_doesNotRejectAggregateMediumPayloadFields() throws Exception { + int chunkLimit = 1_048_576; + String mediumPayload = "X".repeat(400_000); + CompleteOrchestrationAction completion = CompleteOrchestrationAction.newBuilder() + .setOrchestrationStatus(OrchestrationStatus.ORCHESTRATION_STATUS_COMPLETED) + .setResult(StringValue.of(mediumPayload)) + .setDetails(StringValue.of(mediumPayload)) + .addCarryoverEvents(HistoryEvent.newBuilder() + .setEventRaised(EventRaisedEvent.newBuilder().setInput(StringValue.of(mediumPayload)))) + .build(); + OrchestratorResponse response = OrchestratorResponse.newBuilder() + .setInstanceId("medium-fields") + .setCompletionToken("token") + .addActions(OrchestratorAction.newBuilder().setCompleteOrchestration(completion)) + .build(); + assertTrue(response.getActions(0).getSerializedSize() > chunkLimit); + try (DurableTaskGrpcWorker worker = buildWorkerWithChannel(true, chunkLimit)) { + invokeChunkingMethod(worker, response); + } + + assertEquals(1, capturedResponses.size()); + assertEquals(completion, capturedResponses.get(0).getActions(0).getCompleteOrchestration(), + "Fields between 256 KiB and 900 KB will be externalized, not rejected as oversized"); + } + @Test void validateActionsSize_actionUnderLimit_returnsNull() throws Exception { OrchestratorAction smallAction = OrchestratorAction.newBuilder() @@ -488,6 +531,64 @@ void buildGetWorkItemsRequest_lpDisabled_noCapability() throws Exception { "Should NOT include WORKER_CAPABILITY_LARGE_PAYLOADS when LP is disabled"); } + @Test + void completionRuntimeFailure_abandonsOriginalWorkItem() throws Exception { + try (DurableTaskGrpcWorker worker = buildWithCompletionFailure(new IllegalStateException("Upload failed"))) { + invokeCompletionRecovery(worker); + assertEquals(Collections.singletonList("completion-token"), abandonedTokens); + assertTrue(capturedResponses.isEmpty()); + } + } + + @Test + void completionGrpcFailure_propagatesToExistingReconnectLoop() throws Exception { + StatusRuntimeException failure = Status.UNAVAILABLE.withDescription("Disconnected").asRuntimeException(); + try (DurableTaskGrpcWorker worker = buildWithCompletionFailure(failure)) { + InvocationTargetException thrown = assertThrows(InvocationTargetException.class, + () -> invokeCompletionRecovery(worker)); + assertSame(failure, thrown.getCause()); + assertTrue(abandonedTokens.isEmpty()); + assertTrue(capturedResponses.isEmpty()); + } + } + + @Test + void completionSuccess_doesNotAbandon() throws Exception { + try (DurableTaskGrpcWorker worker = buildWorkerWithChannel(false, 4_089_446)) { + invokeCompletionRecovery(worker); + assertEquals(1, capturedResponses.size()); + assertTrue(abandonedTokens.isEmpty()); + } + } + + private DurableTaskGrpcWorker buildWithCompletionFailure(RuntimeException failure) { + return new DurableTaskGrpcWorkerBuilder().grpcChannel(inProcessChannel) + .addInterceptor(new ClientInterceptor() { + @Override + public ClientCall interceptCall( + MethodDescriptor method, CallOptions options, Channel next) { + return new ForwardingClientCall.SimpleForwardingClientCall( + next.newCall(method, options)) { + @Override + public void sendMessage(ReqT message) { + if (method.getFullMethodName().endsWith("/CompleteOrchestratorTask")) { + throw failure; + } + super.sendMessage(message); + } + }; + } + }).build(); + } + + private void invokeCompletionRecovery(DurableTaskGrpcWorker worker) throws Exception { + Method method = DurableTaskGrpcWorker.class.getDeclaredMethod( + "completeOrchestratorTaskOrAbandon", OrchestratorResponse.class); + method.setAccessible(true); + method.invoke(worker, OrchestratorResponse.newBuilder() + .setInstanceId("completion-instance").setCompletionToken("completion-token").build()); + } + // ==================== Helpers ==================== private DurableTaskGrpcWorker buildWorkerWithChannel(boolean supportsLargePayloads, int maxChunkSizeBytes) { diff --git a/exporthistory/README.md b/exporthistory/README.md index 200a7ab..f4ed304 100644 --- a/exporthistory/README.md +++ b/exporthistory/README.md @@ -59,6 +59,8 @@ System.out.println(d.getStatus() + " exported=" + d.getExportedInstances()); - **BATCH** — exports a fixed completion-time window and completes. Requires `completedTimeFrom` and `completedTimeTo` (the upper bound must not be in the future). - **CONTINUOUS** — tails newly-completed terminal instances on a 1-minute idle loop until the job is deleted. + When `completedTimeFrom` is omitted, the lower bound is captured when `ExportJobCreationOptions` is constructed, + matching .NET. It is preserved through job submission, so worker queue delays do not move the window forward. ### Terminal statuses only @@ -74,6 +76,9 @@ supplied, all three are exported. - **Permissions** — the storage credential needs blob write on the container; the DTS credential needs orchestration read. +The worker ensures the export container exists before every upload, recreating it if it was deleted between +exports. The storage credential therefore also needs permission to create the container. + ## Export format Each blob holds the instance's full history using .NET-style history-event JSON. Golden fixtures cover core event serialization and the legacy entity-message encoding produced by the .NET SDK and `Microsoft.Azure.DurableTask.Core`: @@ -82,6 +87,8 @@ Each blob holds the instance's full history using .NET-style history-event JSON. - Each event is `{"eventType": "...", , "eventId": N, "isPlayed": false, "timestamp": "..."}`. - camelCase field names, null event fields omitted, empty maps as `{}`, enum values in PascalCase (e.g. `"Completed"`), timestamps as trimmed ISO-8601 ending in `Z`, and HTML-safe string escaping (`"` → `\u0022`, `& < > ' +` and all non-ASCII → `\uXXXX`). - Sub-millisecond timestamps are preserved when reading history and metadata. Exported timestamps and blob-name timestamps use .NET's 100-nanosecond precision; orchestration replay timestamp behavior is unchanged. +- Ordinary and nested failure details retain custom `properties`, including nested values and date tags projected + to .NET-style timestamps. Failure details read from streamed history retain empty property objects as `{}`. - Entity operations and locks are exported as `EventSent`/`EventRaised` events. Their `input` contains the reference SDK's JSON-encoded entity request or response, including parent orchestration context and nested failure details. This replaces the earlier Java-native `Entity...` event format; consumers of that preview format must update their event handling. The public history API continues returning typed Java entity events. - JSONL uses LF line endings. Identical compressed bytes or identical fields across all SDK versions are not guaranteed. @@ -97,6 +104,11 @@ surfaces (matching .NET). Locally, with the DTS emulator and Azurite: +Run the automated suite with `./gradlew :exporthistory:integrationTest -PskipSigning` after starting the backends. +It checks batch export, JSON and gzipped JSONL failure properties, resolved large-payload history, container +recreation after deletion, and continuous export of an orchestration completed between options construction +and job creation. + 1. Start the backends: ``` docker run --name durabletask-emulator -p 4001:8080 -d mcr.microsoft.com/dts/dts-emulator:latest diff --git a/exporthistory/build.gradle b/exporthistory/build.gradle index 60c122f..131dbe9 100644 --- a/exporthistory/build.gradle +++ b/exporthistory/build.gradle @@ -49,6 +49,7 @@ dependencies { testImplementation 'org.mockito:mockito-junit-jupiter:5.21.0' testImplementation "io.grpc:grpc-inprocess:${grpcVersion}" testImplementation project(':azuremanaged') + testImplementation project(':azure-blob-payloads') } compileJava { diff --git a/exporthistory/src/main/java/com/microsoft/durabletask/exporthistory/BlobExportWriter.java b/exporthistory/src/main/java/com/microsoft/durabletask/exporthistory/BlobExportWriter.java index 614754f..283cc7a 100644 --- a/exporthistory/src/main/java/com/microsoft/durabletask/exporthistory/BlobExportWriter.java +++ b/exporthistory/src/main/java/com/microsoft/durabletask/exporthistory/BlobExportWriter.java @@ -18,25 +18,19 @@ import java.io.UncheckedIOException; import java.nio.charset.StandardCharsets; import java.util.Collections; -import java.util.Set; -import java.util.concurrent.ConcurrentHashMap; import java.util.zip.GZIPOutputStream; /** * Writes serialized orchestration history to Azure Blob Storage. *

* Built once from {@link ExportHistoryStorageOptions} (connection-string or identity auth) and reused across export - * activities. The target container is taken from each {@link ExportDestination}; the container is created on first - * use. + * activities. The target container is taken from each {@link ExportDestination}; its existence is ensured before + * every upload. */ final class BlobExportWriter { private final BlobServiceClient serviceClient; - // Containers already ensured this process-lifetime, so createIfNotExists runs once per container - // rather than once per uploaded blob. The end state is identical; only redundant REST calls are avoided. - private final Set ensuredContainers = ConcurrentHashMap.newKeySet(); - /** * Creates a {@code BlobExportWriter} from storage options. * @@ -103,10 +97,7 @@ void upload(String containerName, String blobPath, String content, ExportFormat throw new IllegalArgumentException("Blob path must not be null or empty."); } BlobContainerClient containerClient = this.serviceClient.getBlobContainerClient(containerName); - if (!this.ensuredContainers.contains(containerName)) { - containerClient.createIfNotExists(); - this.ensuredContainers.add(containerName); - } + containerClient.createIfNotExists(); BlobClient blobClient = containerClient.getBlobClient(blobPath); diff --git a/exporthistory/src/main/java/com/microsoft/durabletask/exporthistory/ExportJobCreationOptions.java b/exporthistory/src/main/java/com/microsoft/durabletask/exporthistory/ExportJobCreationOptions.java index 62a0db5..f3db4fd 100644 --- a/exporthistory/src/main/java/com/microsoft/durabletask/exporthistory/ExportJobCreationOptions.java +++ b/exporthistory/src/main/java/com/microsoft/durabletask/exporthistory/ExportJobCreationOptions.java @@ -3,6 +3,7 @@ package com.microsoft.durabletask.exporthistory; import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonIgnore; import com.fasterxml.jackson.annotation.JsonProperty; import com.microsoft.durabletask.OrchestrationRuntimeStatus; @@ -44,6 +45,8 @@ public final class ExportJobCreationOptions { OrchestrationRuntimeStatus.TERMINATED)); private final String jobId; + @JsonIgnore + private final Instant optionsCreatedAt; private ExportMode mode = ExportMode.BATCH; private Instant completedTimeFrom; private Instant completedTimeTo; @@ -59,7 +62,12 @@ public final class ExportJobCreationOptions { */ @JsonCreator public ExportJobCreationOptions(@JsonProperty("jobId") @Nullable String jobId) { + this(jobId, Instant.now()); + } + + private ExportJobCreationOptions(@Nullable String jobId, Instant optionsCreatedAt) { this.jobId = (jobId == null || jobId.isEmpty()) ? UUID.randomUUID().toString().replace("-", "") : jobId; + this.optionsCreatedAt = optionsCreatedAt; } /** @return the export job ID. */ @@ -86,17 +94,22 @@ public ExportJobCreationOptions setMode(ExportMode mode) { return this; } - /** @return the inclusive completion-time lower bound, or {@code null} if not set. */ + /** + * @return the inclusive completion-time lower bound; defaults to options construction time for + * {@link ExportMode#CONTINUOUS}, or {@code null} if unset for {@link ExportMode#BATCH} + */ @Nullable public Instant getCompletedTimeFrom() { - return this.completedTimeFrom; + return this.completedTimeFrom == null && this.mode == ExportMode.CONTINUOUS + ? this.optionsCreatedAt + : this.completedTimeFrom; } /** * Sets the inclusive completion-time lower bound. Required for {@link ExportMode#BATCH}; for - * {@link ExportMode#CONTINUOUS} it defaults to the job creation time when omitted. + * {@link ExportMode#CONTINUOUS} it defaults to options construction time when omitted, matching .NET. * - * @param completedTimeFrom the lower bound, or {@code null} to clear + * @param completedTimeFrom the lower bound, or {@code null} to restore the mode-specific default * @return this options object */ public ExportJobCreationOptions setCompletedTimeFrom(@Nullable Instant completedTimeFrom) { @@ -252,7 +265,7 @@ public void validateForCreate() { * @return a copy of these options */ ExportJobCreationOptions copy() { - ExportJobCreationOptions c = new ExportJobCreationOptions(this.jobId); + ExportJobCreationOptions c = new ExportJobCreationOptions(this.jobId, this.optionsCreatedAt); c.mode = this.mode; c.completedTimeFrom = this.completedTimeFrom; c.completedTimeTo = this.completedTimeTo; diff --git a/exporthistory/src/main/java/com/microsoft/durabletask/exporthistory/HistoryEventSerializer.java b/exporthistory/src/main/java/com/microsoft/durabletask/exporthistory/HistoryEventSerializer.java index 6087d4d..6a51983 100644 --- a/exporthistory/src/main/java/com/microsoft/durabletask/exporthistory/HistoryEventSerializer.java +++ b/exporthistory/src/main/java/com/microsoft/durabletask/exporthistory/HistoryEventSerializer.java @@ -322,6 +322,7 @@ private static Map failureMap(FailureDetails f) { putIfNotNull(m, "stackTrace", f.getStackTrace()); putIfNotNull(m, "innerFailure", failureMap(f.getInnerFailure())); m.put("isNonRetriable", f.isNonRetriable()); + putIfNotNull(m, "properties", projectFailureProperty(f.getProperties())); return m; } @@ -528,22 +529,22 @@ private static Map entityFailureMap(FailureDetails failure) { result.put("InnerFailure", entityFailureMap(failure.getInnerFailure())); result.put("IsNonRetriable", failure.isNonRetriable()); result.put("Properties", failure.getProperties() == null - ? Collections.emptyMap() : projectEntityFailureProperty(failure.getProperties())); + ? Collections.emptyMap() : projectFailureProperty(failure.getProperties())); return result; } - private static Object projectEntityFailureProperty(Object value) { + private static Object projectFailureProperty(Object value) { if (value instanceof Map) { Map projected = new LinkedHashMap<>(); for (Map.Entry entry : ((Map) value).entrySet()) { - projected.put(entry.getKey(), projectEntityFailureProperty(entry.getValue())); + projected.put(entry.getKey(), projectFailureProperty(entry.getValue())); } return projected; } if (value instanceof List) { List projected = new ArrayList<>(); for (Object item : (List) value) { - projected.add(projectEntityFailureProperty(item)); + projected.add(projectFailureProperty(item)); } return projected; } diff --git a/exporthistory/src/test/java/com/microsoft/durabletask/exporthistory/BlobExportWriterTest.java b/exporthistory/src/test/java/com/microsoft/durabletask/exporthistory/BlobExportWriterTest.java index f71d7ec..c3a3a03 100644 --- a/exporthistory/src/test/java/com/microsoft/durabletask/exporthistory/BlobExportWriterTest.java +++ b/exporthistory/src/test/java/com/microsoft/durabletask/exporthistory/BlobExportWriterTest.java @@ -4,10 +4,12 @@ import com.azure.core.util.BinaryData; import com.azure.core.util.Context; +import com.azure.core.http.HttpResponse; import com.azure.storage.blob.BlobClient; import com.azure.storage.blob.BlobContainerClient; import com.azure.storage.blob.BlobServiceClient; import com.azure.storage.blob.models.BlobHttpHeaders; +import com.azure.storage.blob.models.BlobStorageException; import com.azure.storage.blob.options.BlobParallelUploadOptions; import org.junit.jupiter.api.Test; import org.mockito.ArgumentCaptor; @@ -16,10 +18,14 @@ import java.io.IOException; import java.nio.charset.StandardCharsets; import java.util.List; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.zip.GZIPInputStream; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyMap; import static org.mockito.ArgumentMatchers.eq; @@ -50,8 +56,7 @@ void upload_appliesHeadersMetadataAndOverwriteAtomically() throws IOException { ArgumentCaptor options = ArgumentCaptor.forClass(BlobParallelUploadOptions.class); verify(blobClient, times(2)).uploadWithResponse(options.capture(), isNull(), eq(Context.NONE)); - // The container is ensured once and cached, not re-checked on every blob upload. - verify(containerClient, times(1)).createIfNotExists(); + verify(containerClient, times(2)).createIfNotExists(); verify(blobClient, never()).setHttpHeaders(any(BlobHttpHeaders.class)); verify(blobClient, never()).setMetadata(anyMap()); @@ -60,6 +65,82 @@ void upload_appliesHeadersMetadataAndOverwriteAtomically() throws IOException { assertUpload(uploads.get(1), "second", "instance-2"); } + @Test + void upload_recreatesAContainerDeletedBetweenUploads() { + BlobServiceClient service = mock(BlobServiceClient.class); + BlobContainerClient container = mock(BlobContainerClient.class); + BlobClient blob = mock(BlobClient.class); + when(service.getBlobContainerClient("container")).thenReturn(container); + when(container.getBlobClient("history.json")).thenReturn(blob); + AtomicBoolean exists = new AtomicBoolean(false); + BlobStorageException missing = storageFailure(404); + when(container.createIfNotExists()).thenAnswer(invocation -> { + exists.set(true); + return true; + }); + when(blob.uploadWithResponse(any(BlobParallelUploadOptions.class), isNull(), eq(Context.NONE))) + .thenAnswer(invocation -> { + if (!exists.get()) { + throw missing; + } + return null; + }); + BlobExportWriter writer = new BlobExportWriter(service); + ExportFormat format = new ExportFormat(ExportFormatKind.JSON, "1.0"); + + writer.upload("container", "history.json", "[]", format, "instance-1"); + exists.set(false); + assertDoesNotThrow(() -> writer.upload("container", "history.json", "[]", format, "instance-1")); + verify(container, times(2)).createIfNotExists(); + verify(blob, times(2)).uploadWithResponse(any(BlobParallelUploadOptions.class), isNull(), eq(Context.NONE)); + } + + @Test + void upload_containerCreationFailurePropagatesAndTheNextAttemptRechecks() { + BlobServiceClient service = mock(BlobServiceClient.class); + BlobContainerClient container = mock(BlobContainerClient.class); + BlobClient blob = mock(BlobClient.class); + when(service.getBlobContainerClient("container")).thenReturn(container); + when(container.getBlobClient("history.json")).thenReturn(blob); + BlobStorageException unavailable = storageFailure(503); + when(container.createIfNotExists()).thenThrow(unavailable).thenReturn(true); + BlobExportWriter writer = new BlobExportWriter(service); + ExportFormat format = new ExportFormat(ExportFormatKind.JSON, "1.0"); + + assertSame(unavailable, assertThrows(BlobStorageException.class, + () -> writer.upload("container", "history.json", "[]", format, "instance-1"))); + verify(blob, never()).uploadWithResponse(any(BlobParallelUploadOptions.class), isNull(), eq(Context.NONE)); + assertDoesNotThrow(() -> writer.upload("container", "history.json", "[]", format, "instance-1")); + verify(container, times(2)).createIfNotExists(); + verify(blob).uploadWithResponse(any(BlobParallelUploadOptions.class), isNull(), eq(Context.NONE)); + } + + @Test + void upload_blobUploadFailurePropagatesAndTheRetryEnsuresTheContainerAgain() { + BlobServiceClient service = mock(BlobServiceClient.class); + BlobContainerClient container = mock(BlobContainerClient.class); + BlobClient blob = mock(BlobClient.class); + when(service.getBlobContainerClient("container")).thenReturn(container); + when(container.getBlobClient("history.json")).thenReturn(blob); + BlobStorageException unavailable = storageFailure(503); + when(blob.uploadWithResponse(any(BlobParallelUploadOptions.class), isNull(), eq(Context.NONE))) + .thenThrow(unavailable).thenReturn(null); + BlobExportWriter writer = new BlobExportWriter(service); + ExportFormat format = new ExportFormat(ExportFormatKind.JSON, "1.0"); + + assertSame(unavailable, assertThrows(BlobStorageException.class, + () -> writer.upload("container", "history.json", "[]", format, "instance-1"))); + assertDoesNotThrow(() -> writer.upload("container", "history.json", "[]", format, "instance-1")); + verify(container, times(2)).createIfNotExists(); + verify(blob, times(2)).uploadWithResponse(any(BlobParallelUploadOptions.class), isNull(), eq(Context.NONE)); + } + + private static BlobStorageException storageFailure(int status) { + HttpResponse response = mock(HttpResponse.class); + when(response.getStatusCode()).thenReturn(status); + return new BlobStorageException("Storage failure", response, null); + } + private static void assertUpload( BlobParallelUploadOptions options, String expectedContent, String expectedInstanceId) throws IOException { diff --git a/exporthistory/src/test/java/com/microsoft/durabletask/exporthistory/ExportHistoryIntegrationTest.java b/exporthistory/src/test/java/com/microsoft/durabletask/exporthistory/ExportHistoryIntegrationTest.java index 3f966ff..788597e 100644 --- a/exporthistory/src/test/java/com/microsoft/durabletask/exporthistory/ExportHistoryIntegrationTest.java +++ b/exporthistory/src/test/java/com/microsoft/durabletask/exporthistory/ExportHistoryIntegrationTest.java @@ -7,6 +7,8 @@ import com.azure.storage.blob.BlobServiceClient; import com.azure.storage.blob.BlobServiceClientBuilder; import com.azure.storage.blob.models.BlobProperties; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; import com.microsoft.durabletask.DurableTaskClient; import com.microsoft.durabletask.DurableTaskGrpcClientBuilder; import com.microsoft.durabletask.DurableTaskGrpcWorker; @@ -15,6 +17,13 @@ import com.microsoft.durabletask.OrchestrationRuntimeStatus; import com.microsoft.durabletask.TaskOrchestration; import com.microsoft.durabletask.TaskOrchestrationFactory; +import com.microsoft.durabletask.TaskActivityFactory; +import com.microsoft.durabletask.TaskActivity; +import com.microsoft.durabletask.azureblobpayloads.BlobPayloadStore; +import com.microsoft.durabletask.azureblobpayloads.LargePayloadClientExtensions; +import com.microsoft.durabletask.azureblobpayloads.LargePayloadStorageOptions; +import com.microsoft.durabletask.azureblobpayloads.LargePayloadWorkerExtensions; +import com.microsoft.durabletask.azureblobpayloads.PayloadStore; import com.microsoft.durabletask.azuremanaged.DurableTaskSchedulerClientOptions; import com.microsoft.durabletask.azuremanaged.DurableTaskSchedulerWorkerOptions; @@ -32,6 +41,8 @@ import java.time.Instant; import java.util.ArrayList; import java.util.List; +import java.util.Collections; +import java.util.UUID; import java.util.concurrent.TimeoutException; import java.util.zip.GZIPInputStream; @@ -157,7 +168,7 @@ public TaskOrchestration create() { } @Test - void blobWriter_repeatedUploadAtomicallyReplacesContentAndProperties() throws IOException { + void blobWriter_repeatedUploadAndContainerDeletionPreserveContentAndProperties() throws IOException { String container = "exporthistory-writer-it-" + System.currentTimeMillis(); BlobServiceClient serviceClient = new BlobServiceClientBuilder() .connectionString(AZURITE_CONNECTION_STRING) @@ -174,21 +185,237 @@ void blobWriter_repeatedUploadAtomicallyReplacesContentAndProperties() throws IO writer.upload(container, blobPath, "first", format, "instance-1"); writer.upload(container, blobPath, "second", format, "instance-2"); + assertEquals("second", readGzip(containerClient.getBlobClient(blobPath))); + + containerClient.delete(); + writer.upload(container, blobPath, "after-deletion", format, "instance-3"); BlobClient blobClient = containerClient.getBlobClient(blobPath); BlobProperties properties = blobClient.getProperties(); assertEquals("application/jsonl+gzip", properties.getContentType()); assertEquals("gzip", properties.getContentEncoding()); - assertEquals("instance-2", properties.getMetadata().get("instanceId")); - byte[] payload = blobClient.downloadContent().toBytes(); - try (GZIPInputStream stream = new GZIPInputStream(new ByteArrayInputStream(payload))) { - assertEquals("second", new String(stream.readAllBytes(), StandardCharsets.UTF_8)); - } + assertEquals("instance-3", properties.getMetadata().get("instanceId")); + assertEquals("after-deletion", readGzip(blobClient)); } finally { containerClient.deleteIfExists(); } } + @Test + void batchExport_largePayloadHistoryContainsResolvedInputAndOutput() throws Exception { + String container = "exporthistory-large-" + UUID.randomUUID(); + String payloadContainer = "exporthistory-payload-" + UUID.randomUUID(); + String payload = "abcdef".repeat(250_000); + ExportHistoryStorageOptions storage = storageOptions(container); + LargePayloadStorageOptions payloadOptions = new LargePayloadStorageOptions() + .setConnectionString(AZURITE_CONNECTION_STRING) + .setContainerName(payloadContainer); + PayloadStore store = new BlobPayloadStore(payloadOptions); + DurableTaskGrpcClientBuilder clientBuilder = createClientBuilder(); + LargePayloadClientExtensions.useExternalizedPayloads(clientBuilder, store, payloadOptions); + this.client = clientBuilder.build(); + DurableTaskGrpcWorkerBuilder workerBuilder = createWorkerBuilder(); + LargePayloadWorkerExtensions.useExternalizedPayloads(workerBuilder, store, payloadOptions); + addEchoOrchestration(workerBuilder); + ExportHistoryWorkerExtensions.useExportHistory(workerBuilder, storage, this.client); + this.worker = workerBuilder.build(); + this.worker.start(); + + try { + Instant from = Instant.now(); + String id = this.client.scheduleNewOrchestrationInstance(ECHO_ORCHESTRATION, payload); + OrchestrationMetadata result = this.client.waitForInstanceCompletion(id, DEFAULT_TIMEOUT, true); + assertEquals(OrchestrationRuntimeStatus.COMPLETED, result.getRuntimeStatus()); + assertEquals(payload, result.readOutputAs(String.class)); + BlobContainerClient payloadBlobs = containerClient(payloadContainer); + assertTrue(payloadBlobs.listBlobs().stream().count() >= 2, + "Input and output must actually be externalized to Blob Storage"); + + exportBatch(storage, from, ExportFormatKind.JSON); + JsonNode events = new ObjectMapper().readTree( + readExport(findExportedBlob(container, id), ExportFormatKind.JSON)); + ObjectMapper mapper = new ObjectMapper(); + assertEquals(payload, mapper.readTree(findEvent(events, "ExecutionStarted").path("input").asText()) + .asText()); + assertEquals(payload, mapper.readTree(findEvent(events, "ExecutionCompleted").path("result").asText()) + .asText()); + } finally { + containerClient(container).deleteIfExists(); + containerClient(payloadContainer).deleteIfExists(); + } + } + + @Test + void batchExport_failedActivityPreservesPropertiesInBothFormats() throws Exception { + String container = "exporthistory-failure-" + UUID.randomUUID(); + ExportHistoryStorageOptions storage = storageOptions(container); + this.client = createClientBuilder().build(); + DurableTaskGrpcWorkerBuilder workerBuilder = createWorkerBuilder() + .exceptionPropertiesProvider(ex -> ex instanceof IllegalStateException + ? Collections.singletonMap("diagnosticCode", "e2e-failure") : Collections.emptyMap()); + workerBuilder.addOrchestration(new TaskOrchestrationFactory() { + @Override + public String getName() { return "ExportFailure"; } + + @Override + public TaskOrchestration create() { + return ctx -> ctx.callActivity("ExportFailureActivity", null, String.class).await(); + } + }); + workerBuilder.addActivity(new TaskActivityFactory() { + @Override + public String getName() { return "ExportFailureActivity"; } + + @Override + public TaskActivity create() { + return ctx -> { throw new IllegalStateException("e2e activity failure"); }; + } + }); + ExportHistoryWorkerExtensions.useExportHistory(workerBuilder, storage, this.client); + this.worker = workerBuilder.build(); + this.worker.start(); + try { + Instant from = Instant.now(); + String id = this.client.scheduleNewOrchestrationInstance("ExportFailure"); + assertEquals(OrchestrationRuntimeStatus.FAILED, + this.client.waitForInstanceCompletion(id, DEFAULT_TIMEOUT, false).getRuntimeStatus()); + ObjectMapper mapper = new ObjectMapper(); + for (ExportFormatKind kind : ExportFormatKind.values()) { + containerClient(container).deleteIfExists(); + exportBatch(storage, from, kind); + String content = readExport(findExportedBlob(container, id), kind); + JsonNode events = mapper.readTree(kind == ExportFormatKind.JSON + ? content : "[" + content.trim().replace("\n", ",") + "]"); + JsonNode failure = findEvent(events, "TaskFailed").path("failureDetails"); + assertEquals("e2e activity failure", failure.path("errorMessage").asText()); + assertEquals("e2e-failure", failure.path("properties").path("diagnosticCode").asText()); + assertEquals(mapper.createObjectNode(), + findEvent(events, "ExecutionCompleted").path("failureDetails").path("properties")); + } + } finally { + containerClient(container).deleteIfExists(); + } + } + + @Test + void continuousExport_includesCompletionBetweenOptionsConstructionAndJobCreation() throws Exception { + String container = "exporthistory-continuous-" + UUID.randomUUID(); + ExportHistoryStorageOptions storage = storageOptions(container); + this.client = createClientBuilder().build(); + DurableTaskGrpcWorkerBuilder workerBuilder = createWorkerBuilder(); + addEchoOrchestration(workerBuilder); + ExportHistoryWorkerExtensions.useExportHistory(workerBuilder, storage, this.client); + this.worker = workerBuilder.build(); + this.worker.start(); + ExportHistoryJobClient job = null; + try { + ExportJobCreationOptions options = new ExportJobCreationOptions("continuous-" + UUID.randomUUID()) + .setMode(ExportMode.CONTINUOUS); + Instant submittedFrom = options.getCompletedTimeFrom(); + String id = this.client.scheduleNewOrchestrationInstance(ECHO_ORCHESTRATION, "before-job-creation"); + OrchestrationMetadata completed = this.client.waitForInstanceCompletion(id, DEFAULT_TIMEOUT, false); + assertEquals(OrchestrationRuntimeStatus.COMPLETED, completed.getRuntimeStatus()); + assertTrue(completed.getLastUpdatedAt().isAfter(submittedFrom)); + job = ExportHistoryClientExtensions.useExportHistory(this.client, storage).createJob(options); + ExportJobDescription description = job.describe(); + Instant persistedFrom = description.getConfig().getFilter().getCompletedTimeFrom(); + assertTrue(Duration.between(submittedFrom, persistedFrom).abs().compareTo(Duration.ofNanos(1000)) <= 0, + "Entity-state transport must preserve the construction-time bound within one microsecond"); + assertTrue(description.getCreatedAt().isAfter(completed.getLastUpdatedAt())); + Instant deadline = Instant.now().plusSeconds(60); + BlobClient exported = null; + do { + BlobContainerClient blobs = containerClient(container); + if (blobs.exists()) { + exported = blobs.listBlobs().stream() + .filter(blob -> id.equals(blobs.getBlobClient(blob.getName()) + .getProperties().getMetadata().get("instanceId"))) + .map(blob -> blobs.getBlobClient(blob.getName())).findFirst().orElse(null); + } + if (exported == null) { + Thread.sleep(500); + } + } while (exported == null && Instant.now().isBefore(deadline)); + assertNotNull(exported, "Continuous export must include the completion before job creation"); + assertTrue(readGzip(exported).contains("before-job-creation")); + } finally { + if (job != null) { + job.delete(); + } + containerClient(container).deleteIfExists(); + } + } + + private void exportBatch(ExportHistoryStorageOptions storage, Instant from, ExportFormatKind kind) + throws InterruptedException { + ExportHistoryJobClient job = ExportHistoryClientExtensions.useExportHistory(this.client, storage) + .createJob(new ExportJobCreationOptions("batch-" + UUID.randomUUID()) + .setMode(ExportMode.BATCH) + .setCompletedTimeFrom(from) + .setCompletedTimeTo(Instant.now()) + .setFormat(new ExportFormat(kind, "1.0"))); + ExportJobDescription result = waitForJobCompletion(job, Duration.ofSeconds(60)); + assertEquals(ExportJobStatus.COMPLETED, result.getStatus(), result.getLastError()); + } + + private static ExportHistoryStorageOptions storageOptions(String container) { + return new ExportHistoryStorageOptions().setConnectionString(AZURITE_CONNECTION_STRING) + .setContainerName(container); + } + + private static BlobContainerClient containerClient(String container) { + return new BlobServiceClientBuilder().connectionString(AZURITE_CONNECTION_STRING).buildClient() + .getBlobContainerClient(container); + } + + private static BlobClient findExportedBlob(String container, String instanceId) { + BlobContainerClient blobs = containerClient(container); + return blobs.listBlobs().stream().filter(blob -> instanceId.equals(blobs.getBlobClient(blob.getName()) + .getProperties().getMetadata().get("instanceId"))) + .map(blob -> blobs.getBlobClient(blob.getName())).findFirst() + .orElseThrow(() -> new AssertionError("Missing exported history for " + instanceId)); + } + + private static String readGzip(BlobClient blob) throws IOException { + try (GZIPInputStream stream = new GZIPInputStream( + new ByteArrayInputStream(blob.downloadContent().toBytes()))) { + return new String(stream.readAllBytes(), StandardCharsets.UTF_8); + } + } + + private static String readExport(BlobClient blob, ExportFormatKind kind) throws IOException { + BlobProperties properties = blob.getProperties(); + if (kind == ExportFormatKind.JSON) { + assertEquals("application/json", properties.getContentType()); + assertTrue(properties.getContentEncoding() == null || properties.getContentEncoding().isEmpty()); + return new String(blob.downloadContent().toBytes(), StandardCharsets.UTF_8); + } + assertEquals("application/jsonl+gzip", properties.getContentType()); + assertEquals("gzip", properties.getContentEncoding()); + return readGzip(blob); + } + + private static JsonNode findEvent(JsonNode events, String eventType) { + for (JsonNode event : events) { + if (eventType.equals(event.path("eventType").asText())) { + return event; + } + } + throw new AssertionError("Missing event " + eventType); + } + + private static void addEchoOrchestration(DurableTaskGrpcWorkerBuilder builder) { + builder.addOrchestration(new TaskOrchestrationFactory() { + @Override + public String getName() { return ECHO_ORCHESTRATION; } + + @Override + public TaskOrchestration create() { + return ctx -> ctx.complete(ctx.getInput(String.class)); + } + }); + } + private ExportJobDescription waitForJobCompletion(ExportHistoryJobClient jobClient, Duration timeout) throws InterruptedException { Instant deadline = Instant.now().plus(timeout); diff --git a/exporthistory/src/test/java/com/microsoft/durabletask/exporthistory/ExportJobCreationOptionsTest.java b/exporthistory/src/test/java/com/microsoft/durabletask/exporthistory/ExportJobCreationOptionsTest.java index eb04917..c945648 100644 --- a/exporthistory/src/test/java/com/microsoft/durabletask/exporthistory/ExportJobCreationOptionsTest.java +++ b/exporthistory/src/test/java/com/microsoft/durabletask/exporthistory/ExportJobCreationOptionsTest.java @@ -2,8 +2,10 @@ // Licensed under the MIT License. package com.microsoft.durabletask.exporthistory; +import com.microsoft.durabletask.JacksonDataConverter; import com.microsoft.durabletask.OrchestrationRuntimeStatus; import org.junit.jupiter.api.Test; +import org.mockito.MockedStatic; import java.time.Instant; import java.util.Arrays; @@ -12,9 +14,12 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; 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 static org.mockito.Mockito.CALLS_REAL_METHODS; +import static org.mockito.Mockito.mockStatic; /** * Unit tests for {@link ExportJobCreationOptions}. @@ -26,6 +31,7 @@ void defaults_areCorrect() { ExportJobCreationOptions options = new ExportJobCreationOptions("job-1"); assertEquals("job-1", options.getJobId()); assertEquals(ExportMode.BATCH, options.getMode()); + assertNull(options.getCompletedTimeFrom()); assertEquals(100, options.getMaxInstancesPerBatch()); assertEquals(ExportFormatKind.JSONL, options.getFormat().getKind()); assertEquals( @@ -36,6 +42,66 @@ void defaults_areCorrect() { options.getRuntimeStatus()); } + @Test + void continuousDefault_isCapturedAtConstructionAndPreservedByCopyAndSerialization() { + Instant constructedAt = Instant.parse("2026-07-01T10:00:00Z"); + Instant queuedAt = constructedAt.plusSeconds(300); + JacksonDataConverter converter = new JacksonDataConverter(); + try (MockedStatic clock = mockStatic(Instant.class, CALLS_REAL_METHODS)) { + clock.when(Instant::now).thenReturn(constructedAt); + ExportJobCreationOptions options = new ExportJobCreationOptions("job-1"); + + clock.when(Instant::now).thenReturn(queuedAt); + options.setMode(ExportMode.CONTINUOUS); + assertEquals(constructedAt, options.getCompletedTimeFrom()); + ExportJobCreationOptions copy = options.copy(); + assertEquals(constructedAt, copy.getCompletedTimeFrom()); + + String serialized = converter.serialize(copy); + assertFalse(serialized.contains("optionsCreatedAt")); + ExportJobCreationOptions restored = converter.deserialize(serialized, ExportJobCreationOptions.class); + assertEquals(ExportMode.CONTINUOUS, restored.getMode()); + assertEquals(constructedAt, restored.getCompletedTimeFrom()); + assertEquals(constructedAt, restored.copy().getCompletedTimeFrom()); + } + } + + @Test + void continuousDefault_doesNotSupplyTheRequiredBatchLowerBound() { + ExportJobCreationOptions options = new ExportJobCreationOptions("job-1") + .setMode(ExportMode.CONTINUOUS); + assertNotNull(options.getCompletedTimeFrom()); + options.setMode(ExportMode.BATCH).setCompletedTimeTo(Instant.now().minusSeconds(60)); + assertNull(options.getCompletedTimeFrom()); + assertThrows(IllegalArgumentException.class, options::validateForCreate); + } + + @Test + void continuousExplicitLowerBound_canBeClearedToTheConstructionDefault() { + ExportJobCreationOptions options = new ExportJobCreationOptions("job-1") + .setMode(ExportMode.CONTINUOUS); + Instant defaultFrom = options.getCompletedTimeFrom(); + Instant explicit = Instant.parse("2026-06-01T00:00:00Z"); + options.setCompletedTimeFrom(explicit); + assertEquals(explicit, options.getCompletedTimeFrom()); + assertEquals(explicit, options.copy().getCompletedTimeFrom()); + options.setCompletedTimeFrom(null); + assertEquals(defaultFrom, options.getCompletedTimeFrom()); + } + + @Test + void continuousExplicitLowerBound_beforeModeSelection_survivesSerialization() { + Instant explicit = Instant.parse("2026-06-01T00:00:00Z"); + ExportJobCreationOptions options = new ExportJobCreationOptions("job-1") + .setCompletedTimeFrom(explicit) + .setMode(ExportMode.CONTINUOUS); + JacksonDataConverter converter = new JacksonDataConverter(); + ExportJobCreationOptions restored = converter.deserialize( + converter.serialize(options.copy()), ExportJobCreationOptions.class); + assertEquals(explicit, options.getCompletedTimeFrom()); + assertEquals(explicit, restored.getCompletedTimeFrom()); + } + @Test void nullOrEmptyJobId_generatesId() { assertNotNull(new ExportJobCreationOptions(null).getJobId()); diff --git a/exporthistory/src/test/java/com/microsoft/durabletask/exporthistory/ExportJobTest.java b/exporthistory/src/test/java/com/microsoft/durabletask/exporthistory/ExportJobTest.java index 3df8eb3..135f32a 100644 --- a/exporthistory/src/test/java/com/microsoft/durabletask/exporthistory/ExportJobTest.java +++ b/exporthistory/src/test/java/com/microsoft/durabletask/exporthistory/ExportJobTest.java @@ -3,16 +3,18 @@ package com.microsoft.durabletask.exporthistory; import com.microsoft.durabletask.AbstractTaskEntity; +import com.microsoft.durabletask.JacksonDataConverter; import com.microsoft.durabletask.TaskEntityContext; import org.junit.jupiter.api.Test; +import org.mockito.MockedStatic; import java.lang.reflect.Field; import java.time.Instant; import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertFalse; -import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.mockito.Mockito.CALLS_REAL_METHODS; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.mockStatic; /** * Unit tests for the {@link ExportJob} entity's {@code create} operation. @@ -20,22 +22,24 @@ class ExportJobTest { @Test - void create_continuousWithoutCompletedTimeFrom_defaultsToCreationInstant() throws Exception { + void create_delayedContinuousJob_preservesTheSubmittedLowerBound() throws Exception { ExportJob entity = newEntityWithPendingState(); + Instant submittedAt = Instant.parse("2026-07-01T10:00:00Z"); + Instant processedAt = submittedAt.plusSeconds(300); + JacksonDataConverter converter = new JacksonDataConverter(); + try (MockedStatic clock = mockStatic(Instant.class, CALLS_REAL_METHODS)) { + clock.when(Instant::now).thenReturn(submittedAt); + ExportJobCreationOptions options = new ExportJobCreationOptions("job-continuous") + .setMode(ExportMode.CONTINUOUS) + .setDestination(new ExportDestination("container")); + String request = converter.serialize(options.copy()); - Instant before = Instant.now(); - entity.create(new ExportJobCreationOptions("job-continuous") - .setMode(ExportMode.CONTINUOUS) - .setDestination(new ExportDestination("container"))); - Instant after = Instant.now(); - - Instant completedTimeFrom = entity.get().getConfig().getFilter().getCompletedTimeFrom(); - assertNotNull(completedTimeFrom, - "CONTINUOUS create must default completedTimeFrom so it does not re-export the whole task hub"); - assertFalse(completedTimeFrom.isBefore(before), "defaulted completedTimeFrom must be at/after job creation"); - assertFalse(completedTimeFrom.isAfter(after), "defaulted completedTimeFrom must be at/before job creation"); + clock.when(Instant::now).thenReturn(processedAt); + entity.create(converter.deserialize(request, ExportJobCreationOptions.class)); - assertEquals(entity.get().getCreatedAt(), completedTimeFrom); + assertEquals(submittedAt, entity.get().getConfig().getFilter().getCompletedTimeFrom()); + assertEquals(processedAt, entity.get().getCreatedAt()); + } } @Test diff --git a/exporthistory/src/test/java/com/microsoft/durabletask/exporthistory/HistoryEventSerializerParityTest.java b/exporthistory/src/test/java/com/microsoft/durabletask/exporthistory/HistoryEventSerializerParityTest.java index 9ad9f05..c788627 100644 --- a/exporthistory/src/test/java/com/microsoft/durabletask/exporthistory/HistoryEventSerializerParityTest.java +++ b/exporthistory/src/test/java/com/microsoft/durabletask/exporthistory/HistoryEventSerializerParityTest.java @@ -6,11 +6,16 @@ import com.google.protobuf.Struct; import com.google.protobuf.StringValue; import com.google.protobuf.Value; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; import com.microsoft.durabletask.DataConverter; import com.microsoft.durabletask.DurableTaskClient; import com.microsoft.durabletask.DurableTaskGrpcClientBuilder; import com.microsoft.durabletask.FailureDetails; import com.microsoft.durabletask.OrchestrationRuntimeStatus; +import com.microsoft.durabletask.azureblobpayloads.LargePayloadClientExtensions; +import com.microsoft.durabletask.azureblobpayloads.LargePayloadStorageOptions; +import com.microsoft.durabletask.azureblobpayloads.PayloadStore; import com.microsoft.durabletask.history.ContinueAsNewEvent; import com.microsoft.durabletask.history.EntityLockGrantedEvent; import com.microsoft.durabletask.history.EntityLockRequestedEvent; @@ -243,6 +248,121 @@ private static Value propertyString(String text) { return Value.newBuilder().setStringValue(text).build(); } + @Test + void ordinaryFailuresFromStreamedHistoryPreserveNestedPropertiesInBothFormats() throws Exception { + Value values = Value.newBuilder().setListValue(ListValue.newBuilder() + .addValues(Value.newBuilder().setNumberValue(42.5)) + .addValues(Value.newBuilder().setBoolValue(true)) + .addValues(Value.newBuilder().setNullValueValue(0)) + .addValues(Value.newBuilder().setStructValue(Struct.newBuilder() + .putFields("date", propertyString("dt:2026-06-30T12:00:00.1234000Z"))))) + .build(); + OrchestratorService.TaskFailureDetails failure = OrchestratorService.TaskFailureDetails.newBuilder() + .setErrorType("Outer") + .setErrorMessage("boom") + .putProperties("values", values) + .setInnerFailure(OrchestratorService.TaskFailureDetails.newBuilder() + .setErrorType("Inner") + .setErrorMessage("inner") + .putProperties("date", propertyString("dto:2026-06-30T12:00:00.1000000+00:00"))) + .build(); + ObjectMapper mapper = new ObjectMapper(); + JsonNode expectedProperties = mapper.readTree( + "{\"values\":[42.5,true,null,{\"date\":\"2026-06-30T12:00:00.1234Z\"}]}"); + JsonNode expectedInnerProperties = mapper.readTree("{\"date\":\"2026-06-30T12:00:00.1+00:00\"}"); + for (OrchestratorService.HistoryEvent event : ordinaryFailureEvents(failure)) { + List history = streamHistoryEvent(event); + for (ExportFormat format : Arrays.asList(JSONL, JSON)) { + JsonNode serialized = mapper.readTree(HistoryEventSerializer.serialize(history, format)); + JsonNode exported = format.getKind() == ExportFormatKind.JSON ? serialized.get(0) : serialized; + assertEquals(expectedProperties, exported.path("failureDetails").get("properties")); + assertEquals(expectedInnerProperties, + exported.path("failureDetails").path("innerFailure").get("properties")); + } + } + } + + @Test + void ordinaryFailuresFromStreamedHistoryPreserveEmptyPropertyObjects() throws Exception { + OrchestratorService.TaskFailureDetails failure = OrchestratorService.TaskFailureDetails.newBuilder() + .setErrorType("Outer") + .setErrorMessage("boom") + .setInnerFailure(OrchestratorService.TaskFailureDetails.newBuilder().setErrorType("Inner")) + .build(); + ObjectMapper mapper = new ObjectMapper(); + for (OrchestratorService.HistoryEvent event : ordinaryFailureEvents(failure)) { + List history = streamHistoryEvent(event); + for (ExportFormat format : Arrays.asList(JSONL, JSON)) { + JsonNode serialized = mapper.readTree(HistoryEventSerializer.serialize(history, format)); + JsonNode exported = format.getKind() == ExportFormatKind.JSON ? serialized.get(0) : serialized; + assertEquals(mapper.createObjectNode(), exported.path("failureDetails").get("properties")); + assertEquals(mapper.createObjectNode(), + exported.path("failureDetails").path("innerFailure").get("properties")); + } + } + } + + @Test + void entityOperationFailureExportsResolvedStackTracesInBothFormats() throws Exception { + String outerToken = "blob:v1:$root:0123456789abcdef0123456789abcdef"; + String innerToken = "blob:v1:$root:fedcba9876543210fedcba9876543210"; + String outerTrace = "outer stack\n".repeat(200); + String innerTrace = "inner stack\n".repeat(200); + PayloadStore store = org.mockito.Mockito.mock(PayloadStore.class); + org.mockito.Mockito.when(store.isKnownPayloadToken(outerToken)).thenReturn(true); + org.mockito.Mockito.when(store.isKnownPayloadToken(innerToken)).thenReturn(true); + org.mockito.Mockito.when(store.download(outerToken)).thenReturn(outerTrace); + org.mockito.Mockito.when(store.download(innerToken)).thenReturn(innerTrace); + OrchestratorService.TaskFailureDetails failure = OrchestratorService.TaskFailureDetails.newBuilder() + .setErrorType("Outer") + .setErrorMessage("boom") + .setStackTrace(StringValue.of(outerToken)) + .setInnerFailure(OrchestratorService.TaskFailureDetails.newBuilder() + .setErrorType("Inner") + .setErrorMessage("inner") + .setStackTrace(StringValue.of(innerToken))) + .build(); + OrchestratorService.HistoryEvent event = OrchestratorService.HistoryEvent.newBuilder() + .setEventId(6) + .setTimestamp(DataConverter.getTimestampFromInstant(TS)) + .setEntityOperationFailed(OrchestratorService.EntityOperationFailedEvent.newBuilder() + .setRequestId("req-failed").setFailureDetails(failure)) + .build(); + List history = streamHistoryEvent(event, store); + ObjectMapper mapper = new ObjectMapper(); + for (ExportFormat format : Arrays.asList(JSONL, JSON)) { + String serialized = HistoryEventSerializer.serialize(history, format); + JsonNode document = mapper.readTree(serialized); + JsonNode exported = format.getKind() == ExportFormatKind.JSON ? document.get(0) : document; + assertEquals("EventRaised", exported.path("eventType").asText()); + JsonNode response = mapper.readTree(exported.path("input").asText()); + JsonNode details = response.path("failureDetails"); + assertEquals("Outer", details.path("ErrorType").asText()); + assertEquals("boom", details.path("ErrorMessage").asText()); + assertEquals(outerTrace, details.path("StackTrace").asText()); + assertEquals(innerTrace, details.path("InnerFailure").path("StackTrace").asText()); + assertTrue(!serialized.contains("blob:v1:"), "No unresolved trace references may reach the archive"); + } + org.mockito.Mockito.verify(store).download(outerToken); + org.mockito.Mockito.verify(store).download(innerToken); + } + + private static List ordinaryFailureEvents( + OrchestratorService.TaskFailureDetails failure) { + OrchestratorService.HistoryEvent.Builder event = OrchestratorService.HistoryEvent.newBuilder() + .setEventId(6) + .setTimestamp(DataConverter.getTimestampFromInstant(TS)); + return Arrays.asList( + event.clone().setTaskFailed(OrchestratorService.TaskFailedEvent.newBuilder() + .setTaskScheduledId(1).setFailureDetails(failure)).build(), + event.clone().setSubOrchestrationInstanceFailed( + OrchestratorService.SubOrchestrationInstanceFailedEvent.newBuilder() + .setTaskScheduledId(1).setFailureDetails(failure)).build(), + event.clone().setExecutionCompleted(OrchestratorService.ExecutionCompletedEvent.newBuilder() + .setOrchestrationStatus(OrchestratorService.OrchestrationStatus.ORCHESTRATION_STATUS_FAILED) + .setFailureDetails(failure)).build()); + } + private static List streamEntityFailure( OrchestratorService.TaskFailureDetails failure) throws Exception { OrchestratorService.HistoryEvent event = OrchestratorService.HistoryEvent.newBuilder() @@ -253,6 +373,15 @@ private static List streamEntityFailure( .setRequestId("req-failed") .setFailureDetails(failure)) .build(); + return streamHistoryEvent(event); + } + + private static List streamHistoryEvent(OrchestratorService.HistoryEvent event) throws Exception { + return streamHistoryEvent(event, null); + } + + private static List streamHistoryEvent( + OrchestratorService.HistoryEvent event, PayloadStore store) throws Exception { String serverName = InProcessServerBuilder.generateName(); Server server = InProcessServerBuilder.forName(serverName) .directExecutor() @@ -271,7 +400,12 @@ public void streamInstanceHistory( ManagedChannel channel = InProcessChannelBuilder.forName(serverName).directExecutor().build(); try { server.start(); - try (DurableTaskClient client = new DurableTaskGrpcClientBuilder().grpcChannel(channel).build()) { + DurableTaskGrpcClientBuilder builder = new DurableTaskGrpcClientBuilder().grpcChannel(channel); + if (store != null) { + LargePayloadClientExtensions.useExternalizedPayloads(builder, store, + new LargePayloadStorageOptions().setThresholdBytes(1024)); + } + try (DurableTaskClient client = builder.build()) { List history = client.getOrchestrationHistory("instance-1"); assertEquals(1, history.size()); return history; diff --git a/exporthistory/src/test/java/com/microsoft/durabletask/exporthistory/HistoryEventSerializerTest.java b/exporthistory/src/test/java/com/microsoft/durabletask/exporthistory/HistoryEventSerializerTest.java index 8659664..72620fa 100644 --- a/exporthistory/src/test/java/com/microsoft/durabletask/exporthistory/HistoryEventSerializerTest.java +++ b/exporthistory/src/test/java/com/microsoft/durabletask/exporthistory/HistoryEventSerializerTest.java @@ -3,6 +3,7 @@ package com.microsoft.durabletask.exporthistory; import com.fasterxml.jackson.core.JsonProcessingException; +import com.microsoft.durabletask.FailureDetails; import com.microsoft.durabletask.history.EntityLockRequestedEvent; import com.microsoft.durabletask.history.EntityOperationCalledEvent; import com.microsoft.durabletask.history.EntityOperationFailedEvent; @@ -13,6 +14,7 @@ import com.microsoft.durabletask.history.HistoryEvent; import com.microsoft.durabletask.history.OrchestrationInstance; import com.microsoft.durabletask.history.TaskCompletedEvent; +import com.microsoft.durabletask.history.TaskFailedEvent; import org.junit.jupiter.api.Test; import java.time.Instant; @@ -62,6 +64,21 @@ void jsonl_omitsNullFields() throws JsonProcessingException { assertTrue(result.contains("\"eventId\":1")); } + @Test + void failurePropertiesFromExceptionArePreservedAndNullPropertiesRemainOmitted() { + FailureDetails withProperties = FailureDetails.fromException( + new IllegalStateException("boom"), exception -> Collections.singletonMap("code", 42)); + FailureDetails withoutProperties = FailureDetails.fromException(new IllegalStateException("boom"), null); + ExportFormat format = new ExportFormat(ExportFormatKind.JSONL, "1.0"); + + String with = HistoryEventSerializer.serialize( + Collections.singletonList(new TaskFailedEvent(2, TS, 1, withProperties)), format); + assertTrue(with.contains("\"properties\":{\"code\":42}")); + String without = HistoryEventSerializer.serialize( + Collections.singletonList(new TaskFailedEvent(2, TS, 1, withoutProperties)), format); + assertFalse(without.contains("\"properties\"")); + } + @Test void json_producesArray() throws JsonProcessingException { ExportFormat format = new ExportFormat(ExportFormatKind.JSON, "1.0");