Fix byte tracking to count received bytes not consumed bytes

The issue was that totalEncodedBytesProcessed was tracking consumed bytes
(what the decoder processed), but it should track received bytes (what came
from the network). This ensures retry requests fetch new data after what
was already received, while the pending buffer holds unconsumed bytes.

On retry: request starts after all received bytes, decoder continues from
its internal state with pending bytes, new bytes appended after pending.

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

Author: Copilot

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

@@ -85,8 +85,9 @@ public Mono<HttpResponse> process(HttpPipelineCallContext context, HttpPipelineN
      */
     private Flux<ByteBuffer> decodeStream(Flux<ByteBuffer> encodedFlux, DecoderState state) {
         return encodedFlux.concatMap(encodedBuffer -> {
-            // Track how many bytes were pending before we process
-            int previousPendingBytes = (state.pendingBuffer != null) ? state.pendingBuffer.remaining() : 0;
+            // Track the NEW bytes received from the network (before combining with pending)
+            int newBytesReceived = encodedBuffer.remaining();
+            state.totalEncodedBytesProcessed.addAndGet(newBytesReceived);
 
             // Combine with pending data if any
             ByteBuffer dataToProcess = state.combineWithPending(encodedBuffer);
@@ -105,14 +106,6 @@ private Flux<ByteBuffer> decodeStream(Flux<ByteBuffer> encodedFlux, DecoderState
                 int bytesConsumed = duplicateForDecode.position() - initialPosition;
                 int bytesRemaining = availableSize - bytesConsumed;
 
-                // Track the newly consumed encoded bytes (excluding previously pending bytes)
-                // The consumed bytes include both old pending bytes and new bytes from this buffer
-                // We only want to add the NEW bytes that were consumed
-                int newBytesConsumed = bytesConsumed - previousPendingBytes;
-                if (newBytesConsumed > 0) {
-                    state.totalEncodedBytesProcessed.addAndGet(newBytesConsumed);
-                }
-
                 // Save only unconsumed portion to pending
                 if (bytesRemaining > 0) {
                     // Position the original buffer to skip consumed bytes, then slice to get unconsumed