Storage - Content Validation Decoder Implementation with pipeline policy

sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/implementation/util/BuilderHelper.java

@@ -39,6 +39,7 @@
 import com.azure.storage.common.policy.ResponseValidationPolicyBuilder;
 import com.azure.storage.common.policy.ScrubEtagPolicy;
 import com.azure.storage.common.policy.StorageBearerTokenChallengeAuthorizationPolicy;
+import com.azure.storage.common.policy.StorageContentValidationDecoderPolicy;
 import com.azure.storage.common.policy.StorageSharedKeyCredentialPolicy;
 
 import java.net.MalformedURLException;
@@ -96,6 +97,8 @@ public static HttpPipeline buildPipeline(StorageSharedKeyCredential storageShare
         // Closest to API goes first, closest to wire goes last.
         List<HttpPipelinePolicy> policies = new ArrayList<>();
 
+        policies.add(new StorageContentValidationDecoderPolicy());
+
         policies.add(getUserAgentPolicy(configuration, logOptions, clientOptions));
         policies.add(new RequestIdPolicy());
 

sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/implementation/util/ChunkedDownloadUtils.java

@@ -34,7 +34,7 @@ public class ChunkedDownloadUtils {
     public static Mono<Tuple3<Long, BlobRequestConditions, BlobDownloadAsyncResponse>> downloadFirstChunk(
         BlobRange range, ParallelTransferOptions parallelTransferOptions, BlobRequestConditions requestConditions,
         BiFunction<BlobRange, BlobRequestConditions, Mono<BlobDownloadAsyncResponse>> downloader, boolean eTagLock) {
-        return downloadFirstChunk(range, parallelTransferOptions, requestConditions, downloader, eTagLock, null);
+        return downloadFirstChunk(range, parallelTransferOptions, requestConditions, downloader, null, eTagLock, null);
     }
 
     /*
@@ -48,6 +48,22 @@ public static Mono<Tuple3<Long, BlobRequestConditions, BlobDownloadAsyncResponse
         BlobRange range, ParallelTransferOptions parallelTransferOptions, BlobRequestConditions requestConditions,
         BiFunction<BlobRange, BlobRequestConditions, Mono<BlobDownloadAsyncResponse>> downloader, boolean eTagLock,
         Context context) {
+        return downloadFirstChunk(range, parallelTransferOptions, requestConditions, downloader, null, eTagLock,
+            context);
+    }
+
+    /*
+    Has a context value for additional download adjustments and an optional fallback downloader for empty blobs.
+
+    Download the first chunk. Construct a Mono which will emit the total count for calculating the number of chunks,
+    access conditions containing the etag to lock on, and the response from downloading the first chunk.
+     */
+    @SuppressWarnings("unchecked")
+    public static Mono<Tuple3<Long, BlobRequestConditions, BlobDownloadAsyncResponse>> downloadFirstChunk(
+        BlobRange range, ParallelTransferOptions parallelTransferOptions, BlobRequestConditions requestConditions,
+        BiFunction<BlobRange, BlobRequestConditions, Mono<BlobDownloadAsyncResponse>> downloader,
+        BiFunction<BlobRange, BlobRequestConditions, Mono<BlobDownloadAsyncResponse>> emptyBlobDownloader,
+        boolean eTagLock, Context context) {
         // We will scope our initial download to either be one chunk or the total size.
         long initialChunkSize
             = range.getCount() != null && range.getCount() < parallelTransferOptions.getBlockSizeLong()
@@ -69,7 +85,12 @@ public static Mono<Tuple3<Long, BlobRequestConditions, BlobDownloadAsyncResponse
                     : requestConditions;
 
                 // Extract the total length of the blob from the contentRange header. e.g. "bytes 1-6/7"
-                long totalLength = extractTotalBlobLength(response.getDeserializedHeaders().getContentRange());
+                // Fall back to contentLength when contentRange isn't present (e.g., 304 responses).
+                String contentRange = response.getDeserializedHeaders().getContentRange();
+                Long contentLength = response.getDeserializedHeaders().getContentLength();
+                long totalLength = contentRange != null
+                    ? extractTotalBlobLength(contentRange)
+                    : (contentLength == null ? 0L : contentLength);
 
                 if (context != null) {
                     Optional<Object> contextAdjustment = context.getData(Constants.ADJUSTED_BLOB_LENGTH_KEY);
@@ -99,7 +120,9 @@ public static Mono<Tuple3<Long, BlobRequestConditions, BlobDownloadAsyncResponse
                     && extractTotalBlobLength(
                         blobStorageException.getResponse().getHeaders().getValue(HttpHeaderName.CONTENT_RANGE)) == 0) {
 
-                    return downloader.apply(new BlobRange(0, 0L), requestConditions)
+                    BiFunction<BlobRange, BlobRequestConditions, Mono<BlobDownloadAsyncResponse>> fallbackDownloader
+                        = emptyBlobDownloader != null ? emptyBlobDownloader : downloader;
+                    return fallbackDownloader.apply(new BlobRange(0, 0L), requestConditions)
                         // Subscribe on boundElastic instead of elastic as elastic is deprecated and boundElastic
                         // provided the same functionality with the added benefit that it won't infinitely create
                         // threads if needed and will instead queue.
@@ -143,7 +166,7 @@ public static <T> Flux<T> downloadChunk(Integer chunkNum, BlobDownloadAsyncRespo
     private static BlobRequestConditions setEtag(BlobRequestConditions requestConditions, String etag) {
         // We don't want to modify the user's object, so we'll create a duplicate and set the retrieved etag.
         return new BlobRequestConditions().setIfModifiedSince(requestConditions.getIfModifiedSince())
-            .setIfUnmodifiedSince(requestConditions.getIfModifiedSince())
+            .setIfUnmodifiedSince(requestConditions.getIfUnmodifiedSince())
             .setIfMatch(etag)
             .setIfNoneMatch(requestConditions.getIfNoneMatch())
             .setLeaseId(requestConditions.getLeaseId());

sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/implementation/util/DownloadValidationUtils.java

@@ -0,0 +1,50 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.storage.blob.implementation.util;
+
+import com.azure.core.util.Context;
+import com.azure.storage.common.StorageChecksumAlgorithm;
+import com.azure.storage.common.implementation.Constants;
+
+/**
+ * Centralizes download content validation decisions based on {@link StorageChecksumAlgorithm}.
+ * <p>
+ * Mirrors the pattern established by {@code ContentValidationModeResolver} for uploads.
+ * <p>
+ * RESERVED FOR INTERNAL USE.
+ */
+public final class DownloadValidationUtils {
+
+    private DownloadValidationUtils() {
+    }
+
+    /**
+     * Whether the algorithm requires structured message decoding (CRC64 / AUTO).
+     */
+    public static boolean isStructuredMessageAlgorithm(StorageChecksumAlgorithm algorithm) {
+        return algorithm == StorageChecksumAlgorithm.CRC64 || algorithm == StorageChecksumAlgorithm.AUTO;
+    }
+
+    /**
+     * Resolves the effective algorithm, defaulting null to NONE.
+     */
+    public static StorageChecksumAlgorithm resolveAlgorithm(StorageChecksumAlgorithm algorithm) {
+        return algorithm != null ? algorithm : StorageChecksumAlgorithm.NONE;
+    }
+
+    /**
+     * Adds structured message decoding context key when CRC64/AUTO validation is active.
+     *
+     * @param context The base context to augment. Null is treated as {@link Context#NONE}.
+     * @param algorithm The resolved checksum algorithm.
+     * @return The augmented context.
+     */
+    public static Context applyStructuredMessageContext(Context context, StorageChecksumAlgorithm algorithm) {
+        Context base = context == null ? Context.NONE : context;
+        if (!isStructuredMessageAlgorithm(algorithm)) {
+            return base;
+        }
+        return base.addData(Constants.STRUCTURED_MESSAGE_DECODING_CONTEXT_KEY, true);
+    }
+}

sdk/storage/azure-storage-blob/src/main/java/com/azure/storage/blob/models/BlobDownloadAsyncResponse.java

@@ -13,6 +13,7 @@
 import com.azure.storage.blob.implementation.accesshelpers.BlobDownloadAsyncResponseConstructorProxy;
 import com.azure.storage.blob.implementation.models.BlobsDownloadHeaders;
 import com.azure.storage.blob.implementation.util.ModelHelper;
+import com.azure.core.util.logging.ClientLogger;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
@@ -29,8 +30,18 @@
 public final class BlobDownloadAsyncResponse extends ResponseBase<BlobDownloadHeaders, Flux<ByteBuffer>>
     implements Closeable {
 
+    private static final ClientLogger LOGGER = new ClientLogger(BlobDownloadAsyncResponse.class);
+
     static {
-        BlobDownloadAsyncResponseConstructorProxy.setAccessor(BlobDownloadAsyncResponse::new);
+        BlobDownloadAsyncResponseConstructorProxy
+            .setAccessor(new BlobDownloadAsyncResponseConstructorProxy.BlobDownloadAsyncResponseConstructorAccessor() {
+                @Override
+                public BlobDownloadAsyncResponse create(StreamResponse sourceResponse,
+                    BiFunction<Throwable, Long, Mono<StreamResponse>> onErrorResume,
+                    DownloadRetryOptions retryOptions) {
+                    return new BlobDownloadAsyncResponse(sourceResponse, onErrorResume, retryOptions);
+                }
+            });
     }
 
     private static final ByteBuffer EMPTY_BUFFER = ByteBuffer.allocate(0);
@@ -56,13 +67,6 @@ public BlobDownloadAsyncResponse(HttpRequest request, int statusCode, HttpHeader
         this.retryOptions = null;
     }
 
-    /**
-     * Constructs a {@link BlobDownloadAsyncResponse}.
-     *
-     * @param sourceResponse The initial Stream Response
-     * @param onErrorResume Function used to resume.
-     * @param retryOptions Retry options.
-     */
     BlobDownloadAsyncResponse(StreamResponse sourceResponse,
         BiFunction<Throwable, Long, Mono<StreamResponse>> onErrorResume, DownloadRetryOptions retryOptions) {
         super(sourceResponse.getRequest(), sourceResponse.getStatusCode(), sourceResponse.getHeaders(),
@@ -95,7 +99,12 @@ private static Flux<ByteBuffer> createResponseFlux(StreamResponse sourceResponse
      */
     public Mono<Void> writeValueToAsync(AsynchronousByteChannel channel, ProgressReporter progressReporter) {
         Objects.requireNonNull(channel, "'channel' must not be null");
+        LOGGER.atVerbose()
+            .addKeyValue("thread", Thread.currentThread().getName())
+            .log("BlobDownloadAsyncResponse.writeValueToAsync entry");
         if (sourceResponse != null) {
+            LOGGER.atVerbose()
+                .log("BlobDownloadAsyncResponse.writeValueToAsync using sourceResponse (IOUtils.transfer)");
             return IOUtils.transferStreamResponseToAsynchronousByteChannel(channel, sourceResponse, onErrorResume,
                 progressReporter, retryOptions.getMaxRetryRequests());
         } else if (super.getValue() != null) {

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

@@ -45,6 +45,7 @@
 import com.azure.storage.blob.implementation.util.BlobRequestConditionProperty;
 import com.azure.storage.blob.implementation.util.BlobSasImplUtil;
 import com.azure.storage.blob.implementation.util.ChunkedDownloadUtils;
+import com.azure.storage.blob.implementation.util.DownloadValidationUtils;
 import com.azure.storage.blob.implementation.util.ModelHelper;
 import com.azure.storage.blob.models.AccessTier;
 import com.azure.storage.blob.models.BlobBeginCopySourceRequestConditions;
@@ -1259,27 +1260,29 @@ Mono<BlobDownloadAsyncResponse> downloadStreamWithResponseInternal(BlobRange ran
         BlobRequestConditions requestConditions, boolean getRangeContentMd5,
         StorageChecksumAlgorithm responseChecksumAlgorithm, Context context) {
         BlobRange finalRange = range == null ? new BlobRange(0) : range;
-        Boolean getMD5 = getRangeContentMd5 ? getRangeContentMd5 : null;
+
+        final StorageChecksumAlgorithm algorithm = DownloadValidationUtils.resolveAlgorithm(responseChecksumAlgorithm);
+        final boolean isStructuredMessageEnabled = DownloadValidationUtils.isStructuredMessageAlgorithm(algorithm);
+
+        final Boolean finalGetMD5 = (!isStructuredMessageEnabled && getRangeContentMd5) ? true : null;
+
         BlobRequestConditions finalRequestConditions
             = requestConditions == null ? new BlobRequestConditions() : requestConditions;
         DownloadRetryOptions finalOptions = (options == null) ? new DownloadRetryOptions() : options;
 
-        // The first range should eagerly convert headers as they'll be used to create response types.
-        Context firstRangeContext = context == null
+        final Context baseContext = context == null
             ? new Context("azure-eagerly-convert-headers", true)
             : context.addData("azure-eagerly-convert-headers", true);
 
-        return downloadRange(finalRange, finalRequestConditions, finalRequestConditions.getIfMatch(), getMD5,
-            firstRangeContext).map(response -> {
+        final Context downloadContext = DownloadValidationUtils.applyStructuredMessageContext(baseContext, algorithm);
+
+        return downloadRange(finalRange, finalRequestConditions, finalRequestConditions.getIfMatch(), finalGetMD5,
+            downloadContext).map(response -> {
                 BlobsDownloadHeaders blobsDownloadHeaders = new BlobsDownloadHeaders(response.getHeaders());
                 String eTag = blobsDownloadHeaders.getETag();
                 BlobDownloadHeaders blobDownloadHeaders = ModelHelper.populateBlobDownloadHeaders(blobsDownloadHeaders,
                     ModelHelper.getErrorCode(response.getHeaders()));
 
-                /*
-                 * If the customer did not specify a count, they are reading to the end of the blob. Extract this value
-                 * from the response for better book-keeping towards the end.
-                 */
                 long finalCount;
                 long initialOffset = finalRange.getOffset();
                 if (finalRange.getCount() == null) {
@@ -1289,38 +1292,38 @@ Mono<BlobDownloadAsyncResponse> downloadStreamWithResponseInternal(BlobRange ran
                     finalCount = finalRange.getCount();
                 }
 
-                // The resume function takes throwable and offset at the destination.
-                // I.e. offset is relative to the starting point.
                 BiFunction<Throwable, Long, Mono<StreamResponse>> onDownloadErrorResume = (throwable, offset) -> {
                     if (!(throwable instanceof IOException || throwable instanceof TimeoutException)) {
                         return Mono.error(throwable);
                     }
 
-                    long newCount = finalCount - offset;
-
-                    /*
-                     * It's possible that the network stream will throw an error after emitting all data but before
-                     * completing. Issuing a retry at this stage would leave the download in a bad state with
-                     * incorrect count and offset values. Because we have read the intended amount of data, we can
-                     * ignore the error at the end of the stream.
-                     */
-                    if (newCount == 0) {
-                        LOGGER.warning("Exception encountered in ReliableDownload after all data read from the network "
-                            + "but before stream signaled completion. Returning success as all data was downloaded. "
-                            + "Exception message: " + throwable.getMessage());
-                        return Mono.empty();
-                    }
-
                     try {
-                        return downloadRange(new BlobRange(initialOffset + offset, newCount), finalRequestConditions,
-                            eTag, getMD5, context);
+                        long newCount = finalCount - offset;
+
+                        /*
+                         * It's possible that the network stream will throw an error after emitting all data but
+                         * before completing. Issuing a retry at this stage would leave the download in a bad
+                         * state with incorrect count and offset values. Because we have read the intended amount
+                         * of data, we can ignore the error at the end of the stream.
+                         */
+                        if (newCount == 0) {
+                            LOGGER.warning(
+                                "Exception encountered in ReliableDownload after all data read from the network "
+                                    + "but before stream signaled completion. Returning success as all data was "
+                                    + "downloaded. Exception message: " + throwable.getMessage());
+                            return Mono.empty();
+                        }
+
+                        BlobRange retryRange = new BlobRange(initialOffset + offset, newCount);
+                        return downloadRange(retryRange, finalRequestConditions, eTag, finalGetMD5, downloadContext);
                     } catch (Exception e) {
                         return Mono.error(e);
                     }
                 };
 
                 return BlobDownloadAsyncResponseConstructorProxy.create(response, onDownloadErrorResume, finalOptions);
             });
+
     }
 
     private Mono<StreamResponse> downloadRange(BlobRange range, BlobRequestConditions requestConditions, String eTag,
@@ -1536,7 +1539,6 @@ Mono<Response<BlobProperties>> downloadToFileWithResponse(BlobDownloadToFileOpti
         BlobRequestConditions finalConditions
             = options.getRequestConditions() == null ? new BlobRequestConditions() : options.getRequestConditions();
 
-        // Default behavior is not to overwrite
         Set<OpenOption> openOptions = options.getOpenOptions();
         if (openOptions == null) {
             openOptions = DEFAULT_OPEN_OPTIONS_SET;