Align large payloads and history export with .NET behavior

CHANGELOG.md

@@ -1,4 +1,12 @@
 ## Unreleased
+* Recover verified ambiguous payload uploads on HTTP 409 `BlobAlreadyExists` as well as HTTP 412, while rejecting unrelated conflicts.
+* 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.

azure-blob-payloads/README.md

@@ -0,0 +1,101 @@
+# 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.
+
+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 or HTTP 409 `BlobAlreadyExists`. Other HTTP 409 errors are not recovered.
+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.

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')
 }
 

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,17 +162,60 @@ 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 (isPossibleCommittedUpload(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 (isPossibleCommittedUpload(e) && blobContentMatches(blob, payloadBytes, null)) {
+                        return;
+                    }
+                    throw e;
+                }
             }
         }
     }
 
+    private static boolean isPossibleCommittedUpload(BlobStorageException e) {
+        return e.getStatusCode() == 412
+            || (e.getStatusCode() == 409 && BlobErrorCode.BLOB_ALREADY_EXISTS.equals(e.getErrorCode()));
+    }
+
+    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 conflict.
+        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) {

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