Fix smart retry for structured message validation to restart from beginning

The issue was that structured messages cannot be decoded from arbitrary offsets - they must be decoded sequentially from the start. When a network interruption occurs during a download with structured message validation, the retry must restart from offset 0, not from the interrupted offset.

Changes to BlobAsyncClientBase:
- Modified retry logic to check if structured message validation is enabled
- When enabled, retries restart from the initial offset (beginning of download)
- When disabled, normal smart retry behavior resumes from interrupted offset
- This fixes "Unexpected segment number" errors that occurred when trying to decode from middle offsets

Changes to tests:
- Updated test assertions to verify all requests start from offset 0 with structured message validation
- Updated test comments to reflect correct behavior (retries restart from beginning)
- Tests now properly validate that structured message downloads succeed despite interruptions

Co-authored-by: gunjansingh-msft <[email protected]>

Author: Copilot

sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/specialized/BlobAsyncClientBase.java

@@ -1389,26 +1389,26 @@ Mono<BlobDownloadAsyncResponse> downloadStreamWithResponse(BlobRange range, Down
                     }
 
                     try {
-                        // For retry context, preserve decoder state if structured message validation is enabled
+                        // For retry context, determine the retry behavior based on validation options
                         Context retryContext = firstRangeContext;
+                        BlobRange retryRange;
 
-                        // If structured message decoding is enabled, we need to include the decoder state
-                        // so the retry can continue from where we left off
+                        // If structured message decoding is enabled, we must restart from the beginning
+                        // because structured messages cannot be decoded from arbitrary offsets
                         if (contentValidationOptions != null
                             && contentValidationOptions.isStructuredMessageValidationEnabled()) {
-                            // The decoder state will be set by the policy during processing
-                            // We preserve it in the context for the retry request
-                            Object decoderState
-                                = firstRangeContext.getData(Constants.STRUCTURED_MESSAGE_DECODER_STATE_CONTEXT_KEY)
-                                    .orElse(null);
-                            if (decoderState != null) {
-                                retryContext = retryContext
-                                    .addData(Constants.STRUCTURED_MESSAGE_DECODER_STATE_CONTEXT_KEY, decoderState);
-                            }
+                            // Structured messages require sequential decoding from the start.
+                            // We cannot resume from middle, so restart the entire download.
+                            // Clear the decoder state to start fresh.
+                            retryContext
+                                = retryContext.addData(Constants.STRUCTURED_MESSAGE_DECODER_STATE_CONTEXT_KEY, null);
+                            retryRange = new BlobRange(initialOffset, finalCount);
+                        } else {
+                            // For non-structured downloads, use smart retry from the interrupted offset
+                            retryRange = new BlobRange(initialOffset + offset, newCount);
                         }
 
-                        return downloadRange(new BlobRange(initialOffset + offset, newCount), finalRequestConditions,
-                            eTag, finalGetMD5, retryContext);
+                        return downloadRange(retryRange, finalRequestConditions, eTag, finalGetMD5, retryContext);
                     } catch (Exception e) {
                         return Mono.error(e);
                     }

sdk/storage/azure-storage-blob/src/test/java/com/azure/storage/blob/specialized/BlobMessageDecoderDownloadTests.java

@@ -227,9 +227,10 @@ public void downloadStreamWithResponseContentValidationVeryLargeBlob() throws IO
     public void downloadStreamWithResponseContentValidationSmartRetry() throws IOException {
         // Test smart retry functionality with structured message validation
         // This test simulates network interruptions and verifies that:
-        // 1. The decoder validates checksums for all received data before retry
-        // 2. The decoder state is preserved across retries
-        // 3. The SDK continues from the correct offset after interruption
+        // 1. The decoder validates checksums for all received data
+        // 2. Retries restart from the beginning (offset 0) since structured messages
+        //    cannot be decoded from arbitrary offsets
+        // 3. The download eventually succeeds despite multiple interruptions
 
         byte[] randomData = getRandomByteArray(Constants.KB);
         StructuredMessageEncoder encoder
@@ -272,21 +273,20 @@ public void downloadStreamWithResponseContentValidationSmartRetry() throws IOExc
         List<String> rangeHeaders = mockPolicy.getRangeHeaders();
         assertTrue(rangeHeaders.size() > 0, "Expected range headers for retries");
 
-        // The first request should be for the entire blob
-        assertTrue(rangeHeaders.get(0).startsWith("bytes=0-"), "First request should start from offset 0");
-
-        // Subsequent requests should continue from where we left off (after the first byte)
-        for (int i = 1; i < rangeHeaders.size(); i++) {
+        // With structured message validation, all requests (including retries) must start from offset 0
+        // because structured messages cannot be decoded from arbitrary offsets
+        for (int i = 0; i < rangeHeaders.size(); i++) {
             String rangeHeader = rangeHeaders.get(i);
-            assertTrue(rangeHeader.startsWith("bytes=" + i + "-"),
-                "Retry request " + i + " should start from offset " + i + " but was: " + rangeHeader);
+            assertTrue(rangeHeader.startsWith("bytes=0-"), "Request " + i
+                + " should start from offset 0 for structured message validation, but was: " + rangeHeader);
         }
     }
 
     @Test
     public void downloadStreamWithResponseContentValidationSmartRetryMultipleSegments() throws IOException {
         // Test smart retry with multiple segments to ensure checksum validation
-        // works correctly across segment boundaries during network interruptions
+        // works correctly across segment boundaries. Retries restart from the beginning
+        // to properly validate all segments sequentially.
 
         byte[] randomData = getRandomByteArray(2 * Constants.KB);
         StructuredMessageEncoder encoder
@@ -328,12 +328,19 @@ public void downloadStreamWithResponseContentValidationSmartRetryMultipleSegment
         List<String> rangeHeaders = mockPolicy.getRangeHeaders();
         assertTrue(rangeHeaders.size() >= 4,
             "Expected at least 4 range headers for retries, got: " + rangeHeaders.size());
+
+        // With structured message validation, all requests must start from offset 0
+        for (int i = 0; i < rangeHeaders.size(); i++) {
+            String rangeHeader = rangeHeaders.get(i);
+            assertTrue(rangeHeader.startsWith("bytes=0-"), "Request " + i
+                + " should start from offset 0 for structured message validation, but was: " + rangeHeader);
+        }
     }
 
     @Test
     public void downloadStreamWithResponseContentValidationSmartRetryLargeBlob() throws IOException {
-        // Test smart retry with a larger blob to ensure decoder state
-        // is correctly maintained across retries with more data
+        // Test smart retry with a larger blob to ensure retries restart from
+        // the beginning and successfully validate all data
 
         byte[] randomData = getRandomByteArray(5 * Constants.KB);
         StructuredMessageEncoder encoder
@@ -370,5 +377,13 @@ public void downloadStreamWithResponseContentValidationSmartRetryLargeBlob() thr
 
         // Verify that retries occurred
         assertEquals(0, mockPolicy.getTriesRemaining());
+
+        // Verify that all requests start from offset 0 with structured message validation
+        List<String> rangeHeaders = mockPolicy.getRangeHeaders();
+        for (int i = 0; i < rangeHeaders.size(); i++) {
+            String rangeHeader = rangeHeaders.get(i);
+            assertTrue(rangeHeader.startsWith("bytes=0-"), "Request " + i
+                + " should start from offset 0 for structured message validation, but was: " + rangeHeader);
+        }
     }
 }