Skip to content

Commit c6b2027

Browse files
bachuvbachuv
committed
addressed pr feedback
1 parent 6cc9040 commit c6b2027

9 files changed

Lines changed: 197 additions & 9 deletions

File tree

‎CHANGELOG.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
## Unreleased
22
* Change the default large-payload externalization threshold from 900,000 bytes to 256 KiB, matching .NET.
3+
* 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.
34
* 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.
45
* Ensure the export container exists before every upload, allowing export retries to recover containers deleted between uploads.
56
* Preserve custom and empty property objects in exported failure details, including nested failures, matching .NET's history conversion.

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

Lines changed: 52 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@
88
import com.azure.storage.blob.BlobServiceClient;
99
import com.azure.storage.blob.BlobServiceClientBuilder;
1010
import com.azure.storage.blob.models.BlobDownloadResponse;
11+
import com.azure.storage.blob.models.BlobDownloadHeaders;
1112
import com.azure.storage.blob.models.BlobErrorCode;
1213
import com.azure.storage.blob.models.BlobHttpHeaders;
1314
import com.azure.storage.blob.models.BlobRequestConditions;
@@ -20,6 +21,7 @@
2021
import java.io.IOException;
2122
import java.io.InputStream;
2223
import java.nio.charset.StandardCharsets;
24+
import java.util.Arrays;
2325
import java.util.UUID;
2426
import java.util.concurrent.atomic.AtomicReference;
2527
import java.util.concurrent.locks.ReentrantLock;
@@ -41,10 +43,10 @@ public final class BlobPayloadStore extends PayloadStore {
4143

4244
// Blob name is UUID.randomUUID().toString().replace("-", ""): exactly 32 lowercase hex chars.
4345
// Container name follows Azure rules: 3-63 chars, lowercase alphanumerics and single hyphens,
44-
// must start and end with alphanumeric (see isValidContainerName).
46+
// must start and end with alphanumeric, or the special $root container.
4547
// Full token grammar: blob:v1:<container>:<32-lowercase-hex>
4648
private static final Pattern TOKEN_PATTERN = Pattern.compile(
47-
"^blob:v1:[a-z0-9](?:[a-z0-9]|-(?=[a-z0-9])){1,61}[a-z0-9]:[0-9a-f]{32}$");
49+
"^blob:v1:(?:\\$root|[a-z0-9](?:[a-z0-9]|-(?=[a-z0-9])){1,61}[a-z0-9]):[0-9a-f]{32}$");
4850

4951
private final BlobContainerClient containerClient;
5052
private final LargePayloadStorageOptions options;
@@ -160,14 +162,58 @@ private void uploadBlob(BlobClient blob, byte[] payloadBytes) throws IOException
160162
byte[] compressedBytes = compressedBuffer.toByteArray();
161163
BlobHttpHeaders headers = new BlobHttpHeaders().setContentEncoding(CONTENT_ENCODING_GZIP);
162164
try (InputStream stream = new ByteArrayInputStream(compressedBytes)) {
163-
blob.uploadWithResponse(
164-
stream, compressedBytes.length, null, headers, null, null, conditions, null, Context.NONE);
165+
try {
166+
blob.uploadWithResponse(
167+
stream, compressedBytes.length, null, headers, null, null, conditions, null, Context.NONE);
168+
} catch (BlobStorageException e) {
169+
if (isPreconditionFailure(e) && blobContentMatches(blob, compressedBytes, CONTENT_ENCODING_GZIP)) {
170+
return;
171+
}
172+
throw e;
173+
}
165174
}
166175
} else {
167176
try (InputStream stream = new ByteArrayInputStream(payloadBytes)) {
168-
blob.uploadWithResponse(
169-
stream, payloadBytes.length, null, null, null, null, conditions, null, Context.NONE);
177+
try {
178+
blob.uploadWithResponse(
179+
stream, payloadBytes.length, null, null, null, null, conditions, null, Context.NONE);
180+
} catch (BlobStorageException e) {
181+
if (isPreconditionFailure(e) && blobContentMatches(blob, payloadBytes, null)) {
182+
return;
183+
}
184+
throw e;
185+
}
186+
}
187+
}
188+
}
189+
190+
private static boolean isPreconditionFailure(BlobStorageException e) {
191+
return e.getStatusCode() == 412;
192+
}
193+
194+
private static boolean blobContentMatches(BlobClient blob, byte[] expectedBytes, String expectedContentEncoding) {
195+
try {
196+
ByteArrayOutputStream output = new ByteArrayOutputStream();
197+
BlobDownloadResponse response = blob.downloadStreamWithResponse(
198+
output,
199+
null,
200+
null,
201+
null,
202+
false,
203+
null,
204+
Context.NONE);
205+
BlobDownloadHeaders headers = response.getDeserializedHeaders();
206+
String actualEncoding = headers == null ? null : headers.getContentEncoding();
207+
if (expectedContentEncoding == null) {
208+
if (actualEncoding != null && !actualEncoding.isEmpty()) {
209+
return false;
210+
}
211+
} else if (!expectedContentEncoding.equalsIgnoreCase(actualEncoding)) {
212+
return false;
170213
}
214+
return Arrays.equals(expectedBytes, output.toByteArray());
215+
} catch (RuntimeException e) {
216+
return false;
171217
}
172218
}
173219

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

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -847,6 +847,16 @@ private HistoryEvent resolveEventPayloads(HistoryEvent e) {
847847
}
848848
}
849849
break;
850+
case ENTITYOPERATIONFAILED:
851+
if (e.hasEntityOperationFailed() && e.getEntityOperationFailed().hasFailureDetails()) {
852+
EntityOperationFailedEvent efailed = e.getEntityOperationFailed();
853+
TaskFailureDetails fd = resolveFailureDetails(efailed.getFailureDetails());
854+
if (fd != efailed.getFailureDetails()) {
855+
return e.toBuilder().setEntityOperationFailed(
856+
efailed.toBuilder().setFailureDetails(fd)).build();
857+
}
858+
}
859+
break;
850860
case HISTORYSTATE:
851861
if (e.hasHistoryState()) {
852862
HistoryStateEvent hs = e.getHistoryState();

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

Lines changed: 64 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -129,6 +129,39 @@ void upload_preconditionFailed_throwsPayloadStorageException() {
129129
"Exception message should indicate a blob-name collision: " + ex.getMessage());
130130
}
131131

132+
@ParameterizedTest
133+
@ValueSource(booleans = {true, false})
134+
void upload_preconditionFailedAfterCommittedPut_recoversWhenStoredContentMatches(boolean compressed) {
135+
options.setCompressionEnabled(compressed);
136+
BlobStorageException preconditionFailed = mock(BlobStorageException.class);
137+
when(preconditionFailed.getStatusCode()).thenReturn(412);
138+
final byte[][] attemptedUpload = new byte[1][];
139+
doAnswer(invocation -> {
140+
InputStream stream = invocation.getArgument(0);
141+
attemptedUpload[0] = stream.readAllBytes();
142+
throw preconditionFailed;
143+
}).when(mockBlobClient).uploadWithResponse(
144+
any(InputStream.class), anyLong(), isNull(),
145+
nullable(BlobHttpHeaders.class), isNull(), isNull(), any(BlobRequestConditions.class), isNull(), any());
146+
doAnswer(invocation -> {
147+
OutputStream out = invocation.getArgument(0);
148+
out.write(attemptedUpload[0]);
149+
BlobDownloadHeaders downloadHeaders = mock(BlobDownloadHeaders.class);
150+
when(downloadHeaders.getContentEncoding()).thenReturn(compressed ? "gzip" : null);
151+
BlobDownloadResponse downloadResponse = mock(BlobDownloadResponse.class);
152+
when(downloadResponse.getDeserializedHeaders()).thenReturn(downloadHeaders);
153+
return downloadResponse;
154+
}).when(mockBlobClient).downloadStreamWithResponse(
155+
any(OutputStream.class), isNull(), isNull(), isNull(), eq(false), isNull(), any());
156+
157+
BlobPayloadStore store = new BlobPayloadStore(mockContainerClient, options);
158+
String token = store.upload("payload committed before retry");
159+
160+
assertTrue(store.isKnownPayloadToken(token));
161+
verify(mockBlobClient).downloadStreamWithResponse(
162+
any(OutputStream.class), isNull(), isNull(), isNull(), eq(false), isNull(), any());
163+
}
164+
132165
@Test
133166
void upload_tokenContainsGuidBlobName() {
134167
BlobPayloadStore store = new BlobPayloadStore(mockContainerClient, options);
@@ -142,6 +175,30 @@ void upload_tokenContainsGuidBlobName() {
142175
assertTrue(parts[1].matches("[0-9a-f]{32}"), "Blob name should be a GUID without dashes: " + parts[1]);
143176
}
144177

178+
@ParameterizedTest
179+
@ValueSource(booleans = {true, false})
180+
void upload_download_roundTrip_rootContainerTokenIsRecognized(boolean compressed) throws Exception {
181+
when(mockContainerClient.getBlobContainerName()).thenReturn("$root");
182+
options.setCompressionEnabled(compressed);
183+
String originalPayload = "root container payload";
184+
final byte[][] capturedBytes = new byte[1][];
185+
doAnswer(inv -> {
186+
InputStream stream = inv.getArgument(0);
187+
capturedBytes[0] = stream.readAllBytes();
188+
return null;
189+
}).when(mockBlobClient).uploadWithResponse(
190+
any(InputStream.class), anyLong(), isNull(),
191+
nullable(BlobHttpHeaders.class), isNull(), isNull(), any(BlobRequestConditions.class), isNull(), any());
192+
193+
BlobPayloadStore store = new BlobPayloadStore(mockContainerClient, options);
194+
String token = store.upload(originalPayload);
195+
196+
assertTrue(token.startsWith("blob:v1:$root:"));
197+
assertTrue(store.isKnownPayloadToken(token));
198+
mockDownloadResponse(capturedBytes[0], compressed ? "gzip" : null);
199+
assertEquals(originalPayload, store.download(token));
200+
}
201+
145202
@Test
146203
void upload_containerAlreadyExists_succeeds() {
147204
BlobStorageException conflict = mock(BlobStorageException.class);
@@ -454,6 +511,13 @@ void isKnownPayloadToken_validToken_returnsTrue() {
454511
"blob:v1:container:0123456789abcdef0123456789abcdef"));
455512
}
456513

514+
@Test
515+
void isKnownPayloadToken_rootContainer_returnsTrue() {
516+
BlobPayloadStore store = new BlobPayloadStore(mockContainerClient, options);
517+
assertTrue(store.isKnownPayloadToken(
518+
"blob:v1:$root:0123456789abcdef0123456789abcdef"));
519+
}
520+
457521
@Test
458522
void isKnownPayloadToken_nonToken_returnsFalse() {
459523
BlobPayloadStore store = new BlobPayloadStore(mockContainerClient, options);

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

Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -471,6 +471,40 @@ void resolve_historyEvent_allTypes() {
471471
verifyEventResolution(HistoryEvent.newBuilder()
472472
.setEntityOperationCompleted(EntityOperationCompletedEvent.newBuilder()
473473
.setOutput(StringValue.of("blob:v1:c:n")).build()).build());
474+
verifyEventResolution(HistoryEvent.newBuilder()
475+
.setEntityOperationFailed(EntityOperationFailedEvent.newBuilder()
476+
.setFailureDetails(TaskFailureDetails.newBuilder()
477+
.setErrorType("EntityFailure")
478+
.setStackTrace(StringValue.of("blob:v1:c:n"))).build()).build());
479+
}
480+
481+
@Test
482+
void resolve_HistoryChunk_entityOperationFailed_resolvesNestedFailureStackTraces() {
483+
when(mockStore.download("blob:v1:container:outer")).thenReturn("outer stack");
484+
when(mockStore.download("blob:v1:container:inner")).thenReturn("inner stack");
485+
486+
TaskFailureDetails inner = TaskFailureDetails.newBuilder()
487+
.setErrorType("Inner")
488+
.setStackTrace(StringValue.of("blob:v1:container:inner"))
489+
.build();
490+
TaskFailureDetails outer = TaskFailureDetails.newBuilder()
491+
.setErrorType("Outer")
492+
.setStackTrace(StringValue.of("blob:v1:container:outer"))
493+
.setInnerFailure(inner)
494+
.build();
495+
HistoryChunk chunk = HistoryChunk.newBuilder()
496+
.addEvents(HistoryEvent.newBuilder()
497+
.setEntityOperationFailed(EntityOperationFailedEvent.newBuilder()
498+
.setFailureDetails(outer)))
499+
.build();
500+
501+
Object result = invokeResolve(chunk);
502+
503+
assertTrue(result instanceof HistoryChunk);
504+
TaskFailureDetails resolved = ((HistoryChunk) result).getEvents(0)
505+
.getEntityOperationFailed().getFailureDetails();
506+
assertEquals("outer stack", resolved.getStackTrace().getValue());
507+
assertEquals("inner stack", resolved.getInnerFailure().getStackTrace().getValue());
474508
}
475509

476510
@Test

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

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -56,6 +56,13 @@ void isKnownPayloadToken_validToken_returnsTrue() {
5656
"blob:v1:container:0123456789abcdef0123456789abcdef"));
5757
}
5858

59+
@Test
60+
void isKnownPayloadToken_rootContainer_returnsTrue() {
61+
BlobPayloadStore store = createStoreWithDefaults();
62+
assertTrue(store.isKnownPayloadToken(
63+
"blob:v1:$root:0123456789abcdef0123456789abcdef"));
64+
}
65+
5966
@Test
6067
void isKnownPayloadToken_prefixOnly_returnsFalse() {
6168
BlobPayloadStore store = createStoreWithDefaults();

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

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -47,10 +47,10 @@ public final class DurableTaskGrpcWorkerBuilder {
4747
public static final int DEFAULT_MAX_CHUNK_SIZE_BYTES = 4_089_446;
4848

4949
int maxChunkSizeBytes = DEFAULT_MAX_CHUNK_SIZE_BYTES;
50-
// Default threshold matches LargePayloadStorageOptions default (900_000 bytes).
50+
// Default threshold matches LargePayloadStorageOptions default (256 KiB).
5151
// Used by the worker to estimate post-externalization action sizes during
5252
// pre-send validation when supportsLargePayloads=true.
53-
int largePayloadThresholdBytes = 900_000;
53+
int largePayloadThresholdBytes = 256 * 1024;
5454

5555
/**
5656
* Adds an orchestration factory to be used by the constructed {@link DurableTaskGrpcWorker}.
@@ -409,7 +409,7 @@ public DurableTaskGrpcWorkerBuilder setSupportsLargePayloads(boolean enabled) {
409409
* post-externalization action sizes during pre-send validation.
410410
* <p>
411411
* This should match the {@code thresholdBytes} configured on the large-payload
412-
* interceptor. The default is 900,000 bytes. Fields whose UTF-8 byte length meets
412+
* interceptor. The default is 256 KiB (262,144 bytes). Fields whose UTF-8 byte length meets
413413
* or exceeds this threshold are assumed to be replaced with a small blob-token
414414
* reference by the interceptor.
415415
*

‎client/src/test/java/com/microsoft/durabletask/OrchestratorChunkingTest.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -87,6 +87,12 @@ void tearDown() {
8787

8888
// ==================== validateActionsSize tests ====================
8989

90+
@Test
91+
void workerBuilder_defaultLargePayloadThreshold_matchesStorageDefault() {
92+
DurableTaskGrpcWorkerBuilder builder = new DurableTaskGrpcWorkerBuilder();
93+
assertEquals(256 * 1024, builder.largePayloadThresholdBytes);
94+
}
95+
9096
@Test
9197
void validateActionsSize_actionUnderLimit_returnsNull() throws Exception {
9298
OrchestratorAction smallAction = OrchestratorAction.newBuilder()

‎exporthistory/src/test/java/com/microsoft/durabletask/exporthistory/HistoryEventSerializerParityTest.java‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -299,6 +299,26 @@ void ordinaryFailuresFromStreamedHistoryPreserveEmptyPropertyObjects() throws Ex
299299
}
300300
}
301301

302+
@Test
303+
void entityOperationFailureExportsResolvedStackTracesInBothFormats() throws Exception {
304+
OrchestratorService.TaskFailureDetails failure = OrchestratorService.TaskFailureDetails.newBuilder()
305+
.setErrorType("Outer")
306+
.setErrorMessage("boom")
307+
.setStackTrace(StringValue.of("outer stack"))
308+
.setInnerFailure(OrchestratorService.TaskFailureDetails.newBuilder()
309+
.setErrorType("Inner")
310+
.setErrorMessage("inner")
311+
.setStackTrace(StringValue.of("inner stack")))
312+
.build();
313+
List<HistoryEvent> history = streamEntityFailure(failure);
314+
for (ExportFormat format : Arrays.asList(JSONL, JSON)) {
315+
String serialized = HistoryEventSerializer.serialize(history, format);
316+
assertTrue(serialized.contains("outer stack"));
317+
assertTrue(serialized.contains("inner stack"));
318+
assertTrue(!serialized.contains("blob:v1:"));
319+
}
320+
}
321+
302322
private static List<OrchestratorService.HistoryEvent> ordinaryFailureEvents(
303323
OrchestratorService.TaskFailureDetails failure) {
304324
OrchestratorService.HistoryEvent.Builder event = OrchestratorService.HistoryEvent.newBuilder()

0 commit comments

Comments
 (0)