Implement smart retry with proper duplicate byte skipping for structured messages

Enhanced StorageContentValidationDecoderPolicy to properly handle retries from arbitrary mid-segment offsets by detecting and skipping duplicate bytes.

Changes to StorageContentValidationDecoderPolicy:
- Added getResponseStartOffset() to parse Content-Range header and determine retry offset
- Enhanced DecoderState to track currentResponseStartOffset and currentResponseBytesRead
- Modified decodeStream() to calculate absolute buffer offsets and skip duplicate bytes
- When a retry occurs, the policy now:
  1. Detects the starting offset from Content-Range header
  2. Calculates which bytes in each buffer are duplicates
  3. Skips duplicate bytes before feeding to decoder
  4. Only processes new bytes, maintaining continuous decoder state

This implementation allows the decoder to receive a continuous stream of non-duplicate encoded bytes, even when retries occur from arbitrary offsets. The decoder doesn't need to be aware of retries - the policy handles all offset calculations and duplicate filtering.

Compatible with range downloads and works for arbitrary mid-segment offset resumption.

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

Author: Copilot

sdk/storage/azure-storage-common/src/main/java/com/azure/storage/common/policy/StorageContentValidationDecoderPolicy.java

@@ -63,6 +63,10 @@ public Mono<HttpResponse> process(HttpPipelineCallContext context, HttpPipelineN
                 // Get or create decoder with state tracking
                 DecoderState decoderState = getOrCreateDecoderState(context, contentLength);
 
+                // Determine the starting offset of this response (for retry handling)
+                long responseStartOffset = getResponseStartOffset(httpResponse);
+                decoderState.setCurrentResponseStartOffset(responseStartOffset);
+
                 // Decode using the stateful decoder
                 Flux<ByteBuffer> decodedStream = decodeStream(httpResponse.getBody(), decoderState);
 
@@ -86,15 +90,42 @@ public Mono<HttpResponse> process(HttpPipelineCallContext context, HttpPipelineN
     private Flux<ByteBuffer> decodeStream(Flux<ByteBuffer> encodedFlux, DecoderState state) {
         return encodedFlux.concatMap(encodedBuffer -> {
             try {
-                // Combine with pending data if any
-                ByteBuffer dataToProcess = state.combineWithPending(encodedBuffer);
+                // Calculate the absolute offset of this buffer in the encoded stream
+                long bufferAbsoluteOffset = state.currentResponseStartOffset + state.currentResponseBytesRead;
+                int bufferSize = encodedBuffer.remaining();
+                state.currentResponseBytesRead += bufferSize;
+
+                // Determine how many bytes to skip (if this is overlapping with already-processed data)
+                long bytesToSkip = state.totalEncodedBytesProcessed.get() - bufferAbsoluteOffset;
+
+                ByteBuffer bufferToProcess;
+                if (bytesToSkip > 0) {
+                    // Skip bytes that have already been processed
+                    if (bytesToSkip >= bufferSize) {
+                        // Entire buffer is duplicate, skip it
+                        return Flux.empty();
+                    }
+                    // Skip partial buffer
+                    ByteBuffer skippedBuffer = encodedBuffer.duplicate();
+                    skippedBuffer.position(skippedBuffer.position() + (int) bytesToSkip);
+                    bufferToProcess = skippedBuffer;
+                    // Only count the new bytes
+                    state.totalEncodedBytesProcessed.addAndGet(bufferSize - (int) bytesToSkip);
+                } else {
+                    // All bytes are new
+                    bufferToProcess = encodedBuffer;
+                    state.totalEncodedBytesProcessed.addAndGet(bufferSize);
+                }
 
-                // Track encoded bytes
-                int encodedBytesInBuffer = encodedBuffer.remaining();
-                state.totalEncodedBytesProcessed.addAndGet(encodedBytesInBuffer);
+                // Combine with pending data if any
+                ByteBuffer dataToProcess = state.combineWithPending(bufferToProcess);
 
                 // Try to decode what we have - decoder handles partial data
                 int availableSize = dataToProcess.remaining();
+                if (availableSize == 0) {
+                    return Flux.empty();
+                }
+
                 ByteBuffer decodedData = state.decoder.decode(dataToProcess.duplicate(), availableSize);
 
                 // Track decoded bytes
@@ -186,6 +217,30 @@ private DecoderState getOrCreateDecoderState(HttpPipelineCallContext context, lo
             .orElseGet(() -> new DecoderState(contentLength));
     }
 
+    /**
+     * Gets the starting offset of the response by parsing the Content-Range header.
+     *
+     * @param httpResponse The HTTP response.
+     * @return The starting offset, or 0 if not a range response.
+     */
+    private long getResponseStartOffset(HttpResponse httpResponse) {
+        String contentRange = httpResponse.getHeaderValue(HttpHeaderName.CONTENT_RANGE);
+        if (contentRange != null && contentRange.startsWith("bytes ")) {
+            try {
+                // Format: "bytes start-end/total" or "bytes start-end/*"
+                String range = contentRange.substring(6); // Skip "bytes "
+                int dashIndex = range.indexOf('-');
+                if (dashIndex > 0) {
+                    String startStr = range.substring(0, dashIndex);
+                    return Long.parseLong(startStr);
+                }
+            } catch (Exception e) {
+                LOGGER.warning("Failed to parse Content-Range header: " + contentRange, e);
+            }
+        }
+        return 0;
+    }
+
     /**
      * Checks if the response is a download response.
      *
@@ -207,6 +262,8 @@ public static class DecoderState {
         private final AtomicLong totalBytesDecoded;
         private final AtomicLong totalEncodedBytesProcessed;
         private ByteBuffer pendingBuffer;
+        private long currentResponseStartOffset = 0;
+        private long currentResponseBytesRead = 0;
 
         /**
          * Creates a new decoder state.
@@ -221,6 +278,16 @@ public DecoderState(long expectedContentLength) {
             this.pendingBuffer = null;
         }
 
+        /**
+         * Sets the starting offset of the current response (for retry handling).
+         *
+         * @param offset The starting offset.
+         */
+        public void setCurrentResponseStartOffset(long offset) {
+            this.currentResponseStartOffset = offset;
+            this.currentResponseBytesRead = 0;
+        }
+
         /**
          * Combines pending buffer with new data.
          *