addressed copilot pr comments

Author: bachuv

.github/workflows/build-validation.yml

@@ -122,12 +122,15 @@ jobs:
       continue-on-error: true
 
     - name: Kill Durable Task Emulator
+      if: always()
       run: docker kill durabletask-emulator
 
     - name: Kill Azurite
+      if: always()
       run: docker kill azurite
 
     - name: Upload Durable Task Emulator Logs
+      if: always()
       uses: actions/upload-artifact@v4
       with:
         name: Durable Task Emulator Logs

azure-blob-payloads/src/main/java/com/microsoft/durabletask/azureblobpayloads/BlobPayloadStore.java

@@ -20,7 +20,9 @@
 import java.io.InputStream;
 import java.nio.charset.StandardCharsets;
 import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.regex.Pattern;
 import java.util.zip.GZIPInputStream;
 import java.util.zip.GZIPOutputStream;
@@ -45,7 +47,13 @@ public final class BlobPayloadStore extends PayloadStore {
 
     private final BlobContainerClient containerClient;
     private final LargePayloadStorageOptions options;
-    private final AtomicBoolean containerVerified = new AtomicBoolean(false);
+
+    // Container-creation guard. The first thread to call ensureContainerExists() creates the
+    // latch and performs the RPC. Concurrent callers await the latch so they don't race ahead
+    // and upload to a container that hasn't been created yet. On failure the reference is
+    // reset to null so a subsequent call can retry.
+    private final AtomicReference<CountDownLatch> containerLatch = new AtomicReference<>();
+    private volatile boolean containerVerified;
 
     /**
      * Creates a new {@code BlobPayloadStore} from the given options.
@@ -115,25 +123,9 @@ public String upload(String payload) {
 
         byte[] payloadBytes = payload.getBytes(StandardCharsets.UTF_8);
 
-        // Ensure container exists (idempotent) — skip after first successful check.
-        // compareAndSet lets only one concurrent caller perform the RPC; others skip.
-        // On failure we reset the flag so a later call can retry.
-        if (this.containerVerified.compareAndSet(false, true)) {
-            try {
-                this.containerClient.createIfNotExists();
-            } catch (BlobStorageException e) {
-                // 409 Conflict means it already exists — safe to ignore, leave flag set.
-                if (e.getStatusCode() != 409) {
-                    this.containerVerified.set(false); // allow a future upload to retry
-                    throw new PayloadStorageException(
-                        "Failed to create blob container '" + this.containerClient.getBlobContainerName() + "'.", e);
-                }
-            } catch (RuntimeException e) {
-                // Any other transport/SDK failure: also allow retry on next upload.
-                this.containerVerified.set(false);
-                throw e;
-            }
-        }
+        // Ensure container exists before uploading. Thread-safe: the first caller creates
+        // the container while concurrent callers wait for it to complete.
+        ensureContainerExists();
 
         try {
             // Defense-in-depth: require the blob to not already exist (If-None-Match: *).
@@ -193,6 +185,54 @@ public String upload(String payload) {
         return encodeToken(this.containerClient.getBlobContainerName(), blobName);
     }
 
+    /**
+     * Ensures the blob container exists, creating it if necessary. Thread-safe: the first
+     * caller performs the RPC while concurrent callers wait for it to complete. On success
+     * the check is skipped on all future calls. On failure the guard is reset so a later
+     * call can retry.
+     */
+    private void ensureContainerExists() {
+        if (this.containerVerified) {
+            return;
+        }
+
+        CountDownLatch latch = new CountDownLatch(1);
+        CountDownLatch existing = this.containerLatch.compareAndExchange(null, latch);
+        if (existing != null) {
+            // Another thread is already creating the container — wait for it.
+            try {
+                existing.await();
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                throw new PayloadStorageException("Interrupted while waiting for container creation.", e);
+            }
+            // If the creating thread failed, containerVerified is still false; the next
+            // upload attempt will retry. For now, return and let the upload proceed
+            // (it will fail fast with a clear error if the container doesn't exist).
+            return;
+        }
+
+        // This thread is responsible for creating the container.
+        try {
+            this.containerClient.createIfNotExists();
+            this.containerVerified = true;
+        } catch (BlobStorageException e) {
+            if (e.getStatusCode() == 409) {
+                // 409 Conflict means it already exists — safe to ignore.
+                this.containerVerified = true;
+            } else {
+                this.containerLatch.set(null); // allow a future call to retry
+                throw new PayloadStorageException(
+                    "Failed to create blob container '" + this.containerClient.getBlobContainerName() + "'.", e);
+            }
+        } catch (RuntimeException e) {
+            this.containerLatch.set(null); // allow a future call to retry
+            throw e;
+        } finally {
+            latch.countDown(); // unblock waiting threads
+        }
+    }
+
     @Override
     public String download(String token) {
         String[] decoded = decodeToken(token);

azure-blob-payloads/src/main/java/com/microsoft/durabletask/azureblobpayloads/LargePayloadWorkerExtensions.java

@@ -36,6 +36,7 @@ public static DurableTaskGrpcWorkerBuilder useExternalizedPayloads(
         PayloadStore store = new BlobPayloadStore(options);
         builder.addInterceptor(new LargePayloadInterceptor(store, options));
         builder.setSupportsLargePayloads(true);
+        builder.setLargePayloadThresholdBytes(options.getThresholdBytes());
         return builder;
     }
 
@@ -59,6 +60,7 @@ public static DurableTaskGrpcWorkerBuilder useExternalizedPayloads(
         Objects.requireNonNull(options, "options must not be null");
         builder.addInterceptor(new LargePayloadInterceptor(store, options));
         builder.setSupportsLargePayloads(true);
+        builder.setLargePayloadThresholdBytes(options.getThresholdBytes());
         return builder;
     }
 }

client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcWorker.java

@@ -613,10 +613,14 @@ private static TaskFailureDetails validateActionsSize(
             if (effectiveSize > maxChunkBytes) {
                 double maxMB = maxChunkBytes / 1024.0 / 1024.0;
                 double actionMB = effectiveSize / 1024.0 / 1024.0;
+                String advice = largePayloadsEnabled
+                    ? "Large-payload externalization is enabled but the action still exceeds the limit after " +
+                      "estimated externalization of string payload fields. Reduce non-string field sizes or " +
+                      "increase the maximum chunk size."
+                    : "Enable large-payload externalization to Azure Blob Storage to support oversized actions.";
                 String errorMessage = String.format(
-                    "A single orchestrator action of type %s with id %d exceeds the %.2fMB limit: %.2fMB. " +
-                    "Enable large-payload externalization to Azure Blob Storage to support oversized actions.",
-                    action.getOrchestratorActionTypeCase(), action.getId(), maxMB, actionMB);
+                    "A single orchestrator action of type %s with id %d exceeds the %.2fMB limit: %.2fMB. %s",
+                    action.getOrchestratorActionTypeCase(), action.getId(), maxMB, actionMB, advice);
                 return TaskFailureDetails.newBuilder()
                     .setErrorType("java.lang.IllegalStateException")
                     .setErrorMessage(errorMessage)
@@ -794,22 +798,46 @@ private static int failureDetailsSavings(
     }
 
     /**
-     * Returns the bytes saved if a StringValue with {@code value.length() >= thresholdBytes}
-     * is replaced by a fixed-size blob-reference token. Zero if the value is null, empty,
-     * or below the threshold.
+     * Returns the bytes saved if a StringValue whose UTF-8 byte length meets or exceeds
+     * {@code thresholdBytes} is replaced by a fixed-size blob-reference token.
+     * Zero if the value is null, empty, or below the threshold.
+     * <p>
+     * Uses the same UTF-8 byte-counting logic as {@code LargePayloadInterceptor.utf8ByteLength}
+     * to ensure the estimation threshold and the interceptor's externalization threshold
+     * are compared in consistent units.
      */
     private static int stringValueSavings(StringValue sv, int thresholdBytes) {
         if (sv == null) return 0;
         String v = sv.getValue();
         if (v == null || v.isEmpty()) return 0;
-        // Use String.length() rather than UTF-8 byte length — this is a lower-bound
-        // approximation that is slightly conservative (under-estimates savings), which
-        // means we err toward false positives not false negatives.
-        int n = v.length();
+        int n = utf8ByteLength(v);
         if (n < thresholdBytes) return 0;
         return Math.max(0, n - APPROX_TOKEN_WIRE_SIZE);
     }
 
+    /**
+     * Computes the UTF-8 encoded byte length of a string without allocating a byte array.
+     * Mirrors {@code LargePayloadInterceptor.utf8ByteLength} to keep estimation consistent
+     * with actual externalization decisions.
+     */
+    private static int utf8ByteLength(String s) {
+        int count = 0;
+        for (int i = 0; i < s.length(); i++) {
+            char c = s.charAt(i);
+            if (c <= 0x7F) {
+                count++;
+            } else if (c <= 0x7FF) {
+                count += 2;
+            } else if (Character.isHighSurrogate(c)) {
+                count += 4;
+                i++; // skip the low surrogate
+            } else {
+                count += 3;
+            }
+        }
+        return count;
+    }
+
     /**
      * Completes an orchestrator task with automatic chunking if the response exceeds the maximum chunk size.
      * Matches the .NET SDK's CompleteOrchestratorTaskWithChunkingAsync behavior.

client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcWorkerBuilder.java

@@ -390,8 +390,11 @@ public DurableTaskGrpcWorkerBuilder addInterceptor(ClientInterceptor interceptor
      * Indicates that this worker supports large payload externalization.
      * <p>
      * When enabled, the worker announces the {@code WORKER_CAPABILITY_LARGE_PAYLOADS} capability
-     * to the sidecar and skips the pre-send action size validation (since the gRPC interceptor
-     * will externalize oversized payloads before they hit the wire).
+     * to the sidecar. Pre-send action size validation still runs but uses an estimated
+     * post-externalization size that accounts for payload fields the gRPC interceptor will
+     * replace with small blob-reference tokens. Use
+     * {@link #setLargePayloadThresholdBytes(int)} to align the estimation threshold with
+     * the interceptor's externalization threshold.
      *
      * @param enabled whether large payload support is enabled
      * @return this builder object