Align large payloads and history export with .NET behavior

CHANGELOG.md

@@ -1,4 +1,10 @@
 ## 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.
+* 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.

azure-blob-payloads/README.md

@@ -0,0 +1,83 @@
+# 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://<account>.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.
+
+## 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.

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:<container>:<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,14 +162,58 @@ 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) {
+        try {
+            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());
+        } catch (RuntimeException e) {
+            return false;
         }
     }
 

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();

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
      */

azure-blob-payloads/src/test/java/com/microsoft/durabletask/azureblobpayloads/BlobPayloadStoreTest.java

@@ -129,6 +129,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 +175,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 +511,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);