fix(remote-storage/azure): fix re-cache of large remote blobs (#9174) (#9179)

* fix(remote-storage/azure): fix re-cache of large remote blobs (#9174)

ReadFile issued a single DownloadStream for the entire requested byte
range, so a large re-cache (e.g. a 2 GB blob re-fetched on S3 GET after
eviction) had to move the whole range over one HTTP connection within
the SDK's per-try TryTimeout. TryTimeout was set to 10s "to fail faster
on auth issues", which silently broke large reads: every attempt hit
context deadline, the filer's CacheRemoteObjectToLocalCluster returned
an error, and the S3 gateway surfaced it to clients as an ETag-mismatch
on the partial response.

Switch ReadFile to the SDK's parallel block downloader (DownloadBuffer
with 4 MiB blocks) so each individual HTTP GET is small enough to
complete well inside TryTimeout. Expose the parallelism through the
RemoteStorageConcurrentReader interface so callers (FetchAndWriteNeedle)
can tune it per request, matching the S3 backend.

Also restore TryTimeout to 60s. With parallel block transfers it is no
longer on the critical path for large-blob bodies, but it gives metadata
operations and any non-parallel paths more headroom on slow links.

* fix(remote-storage/azure): guard ReadFileWithConcurrency inputs

Addresses review feedback on PR #9179:

- Reject negative size up front instead of panicking inside make([]byte, size).
- Clamp concurrency to math.MaxUint16 before casting to uint16 so an
  oversized caller value can't silently wrap to a small number.

* fix(remote-storage/azure): reject negative offset in ReadFileWithConcurrency

Addresses review feedback on PR #9179. Without this guard, a negative
offset combined with size == 0 would compute `size = ContentLength -
offset` -> a value larger than the blob, then attempt to allocate and
download past the end.

weed/remote_storage/azure/azure_storage_client.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"math"
 	"os"
 	"reflect"
 	"regexp"
@@ -26,14 +27,15 @@ import (
 )
 
 const (
-	defaultBlockSize   = 4 * 1024 * 1024
-	defaultConcurrency = 16
+	defaultBlockSize       = 4 * 1024 * 1024
+	defaultConcurrency     = 16
+	defaultReadConcurrency = 16
 
-	// DefaultAzureOpTimeout is the timeout for individual Azure blob operations.
-	// This should be larger than the maximum time the Azure SDK client will spend
-	// retrying. With MaxRetries=3 (4 total attempts) and TryTimeout=10s, the maximum
-	// time is roughly 4*10s + delays(~7s) = 47s. We use 60s to provide a reasonable
-	// buffer while still failing faster than indefinite hangs.
+	// DefaultAzureOpTimeout is the timeout for short metadata operations
+	// (GetProperties, Delete, conditional updates, etc.). Blob body transfers
+	// (ReadFile / WriteFile) are NOT bounded by this timeout: they use
+	// parallel 4 MiB block transfers and rely on the SDK's per-try TryTimeout
+	// to bound each individual block.
 	DefaultAzureOpTimeout = 60 * time.Second
 )
 
@@ -41,12 +43,21 @@ const (
 // with consistent retry configuration across the application.
 // This centralizes the retry policy to ensure uniform behavior between
 // remote storage and replication sink implementations.
+//
+// TryTimeout bounds a single HTTP attempt (headers + body) in the Azure SDK.
+// With parallel 4 MiB block transfers (DownloadBuffer / UploadStream) a 60s
+// TryTimeout is comfortable. A previous regression set this to 10s "to fail
+// faster on auth issues" — that silently broke large-blob reads: a single
+// non-parallelized Download of more than ~40 MiB over a typical WAN link
+// cannot finish in 10s, so re-caching a large remote object (e.g. a 2 GB
+// blob mounted via `remote.mount` and fetched through the S3 gateway) failed
+// every attempt with a generic context-deadline error.
 func DefaultAzBlobClientOptions() *azblob.ClientOptions {
 	return &azblob.ClientOptions{
 		ClientOptions: azcore.ClientOptions{
 			Retry: policy.RetryOptions{
-				MaxRetries:    3,                // Reasonable retry count - aggressive retries mask configuration errors
-				TryTimeout:    10 * time.Second, // Reduced from 1 minute to fail faster on auth issues
+				MaxRetries:    3,
+				TryTimeout:    60 * time.Second,
 				RetryDelay:    1 * time.Second,
 				MaxRetryDelay: 10 * time.Second,
 			},
@@ -126,6 +137,7 @@ type azureRemoteStorageClient struct {
 }
 
 var _ = remote_storage.RemoteStorageClient(&azureRemoteStorageClient{})
+var _ = remote_storage.RemoteStorageConcurrentReader(&azureRemoteStorageClient{})
 
 func (az *azureRemoteStorageClient) ListDirectory(ctx context.Context, loc *remote_pb.RemoteStorageLocation, visitFn remote_storage.VisitFunc) (err error) {
 	pathKey := loc.Path[1:]
@@ -264,31 +276,67 @@ func (az *azureRemoteStorageClient) Traverse(loc *remote_pb.RemoteStorageLocatio
 }
 
 func (az *azureRemoteStorageClient) ReadFile(loc *remote_pb.RemoteStorageLocation, offset int64, size int64) (data []byte, err error) {
+	return az.ReadFileWithConcurrency(loc, offset, size, defaultReadConcurrency)
+}
+
+// ReadFileWithConcurrency fetches a byte range of a blob using the Azure SDK's
+// parallel block downloader. The blob is split into `defaultBlockSize` (4 MiB)
+// blocks that are fetched in parallel up to `concurrency` at a time. This
+// keeps each individual HTTP GET small enough to complete well within the
+// SDK's per-try TryTimeout, so large-blob reads remain reliable even on
+// slower WAN links.
+func (az *azureRemoteStorageClient) ReadFileWithConcurrency(loc *remote_pb.RemoteStorageLocation, offset int64, size int64, concurrency int) (data []byte, err error) {
+
+	if offset < 0 {
+		return nil, fmt.Errorf("invalid offset %d for %s%s", offset, loc.Bucket, loc.Path)
+	}
+	if size < 0 {
+		return nil, fmt.Errorf("invalid size %d for %s%s", size, loc.Bucket, loc.Path)
+	}
+
+	if concurrency <= 0 {
+		concurrency = defaultReadConcurrency
+	} else if concurrency > math.MaxUint16 {
+		// DownloadBufferOptions.Concurrency is uint16; clamp to avoid wraparound.
+		concurrency = math.MaxUint16
+	}
 
 	key := loc.Path[1:]
 	blobClient := az.client.ServiceClient().NewContainerClient(loc.Bucket).NewBlockBlobClient(key)
 
-	count := size
-	if count == 0 {
-		count = blob.CountToEnd
+	// DownloadBuffer requires a pre-sized destination. If the caller asked for
+	// "read to end" (size == 0), discover the blob length via GetProperties
+	// first so we can allocate accurately.
+	if size == 0 {
+		propsCtx, cancelProps := context.WithTimeout(context.Background(), DefaultAzureOpTimeout)
+		props, propsErr := blobClient.GetProperties(propsCtx, nil)
+		cancelProps()
+		if propsErr != nil {
+			return nil, fmt.Errorf("get properties %s%s: %w", loc.Bucket, loc.Path, propsErr)
+		}
+		if props.ContentLength == nil {
+			return nil, fmt.Errorf("azure %s%s: missing ContentLength", loc.Bucket, loc.Path)
+		}
+		size = *props.ContentLength - offset
+		if size <= 0 {
+			return []byte{}, nil
+		}
 	}
-	downloadResp, err := blobClient.DownloadStream(context.Background(), &blob.DownloadStreamOptions{
+
+	data = make([]byte, size)
+	_, err = blobClient.DownloadBuffer(context.Background(), data, &blob.DownloadBufferOptions{
 		Range: blob.HTTPRange{
 			Offset: offset,
-			Count:  count,
+			Count:  size,
 		},
+		BlockSize:   defaultBlockSize,
+		Concurrency: uint16(concurrency),
 	})
 	if err != nil {
 		return nil, fmt.Errorf("failed to download file %s%s: %w", loc.Bucket, loc.Path, err)
 	}
-	defer downloadResp.Body.Close()
 
-	data, err = io.ReadAll(downloadResp.Body)
-	if err != nil {
-		return nil, fmt.Errorf("failed to read download stream %s%s: %w", loc.Bucket, loc.Path, err)
-	}
-
-	return
+	return data, nil
 }
 
 func (az *azureRemoteStorageClient) WriteDirectory(loc *remote_pb.RemoteStorageLocation, entry *filer_pb.Entry) (err error) {