Release ManagedCode.Storage 10.1.12

Directory.Build.props

@@ -29,8 +29,8 @@
         <RepositoryUrl>https://github.com/managedcode/Storage</RepositoryUrl>
         <PackageProjectUrl>https://github.com/managedcode/Storage</PackageProjectUrl>
         <Product>Managed Code - Storage</Product>
-        <Version>10.1.11</Version>
-        <PackageVersion>10.1.11</PackageVersion>
+        <Version>10.1.12</Version>
+        <PackageVersion>10.1.12</PackageVersion>
 
     </PropertyGroup>
 

Integraions/ManagedCode.Storage.Server/ChunkUpload/ChunkUploadFiles.cs

@@ -13,7 +13,8 @@ internal static class ChunkUploadFiles
 {
     private const int CopyBufferSize = 64 * 1024;
 
-    internal static async Task StageChunkAsync(ChunkUploadSession session, FileUploadPayload payload, CancellationToken cancellationToken)
+    internal static async Task StageChunkAsync(ChunkUploadSession session, FileUploadPayload payload, long length,
+        CancellationToken cancellationToken)
     {
         var chunkIndex = payload.Payload.ChunkIndex;
         var chunkFilePath = Path.Combine(session.WorkingDirectory, $"{chunkIndex:D6}.part");
@@ -24,11 +25,13 @@ internal static async Task StageChunkAsync(ChunkUploadSession session, FileUploa
             await using (var source = payload.File.OpenReadStream())
             {
                 await source.CopyToAsync(target, CopyBufferSize, cancellationToken).ConfigureAwait(false);
+                if (target.Position != length)
+                    throw new InvalidDataException("The staged chunk differs from its form file length.");
             }
 
             cancellationToken.ThrowIfCancellationRequested();
             File.Move(stagingPath, chunkFilePath, overwrite: true);
-            session.RegisterChunk(chunkIndex, chunkFilePath);
+            session.RegisterChunk(chunkIndex, chunkFilePath, length);
         }
         finally
         {

Integraions/ManagedCode.Storage.Server/ChunkUpload/ChunkUploadOptions.cs

@@ -14,7 +14,8 @@ public class ChunkUploadOptions
     public string TempPath { get; set; } = Path.Combine(Path.GetTempPath(), "managedcode-storage", "chunks");
 
     /// <summary>
-    /// How long chunks are kept on disk after the last write. Expired sessions are cleaned up on completion or abort.
+    /// Idle time after which an upload session expires. Append and completion requests clean up expired sessions.
+    /// A nonpositive value disables expiration.
     /// </summary>
     public TimeSpan SessionTtl { get; set; } = TimeSpan.FromHours(1);
 

Integraions/ManagedCode.Storage.Server/ChunkUpload/ChunkUploadService.cs

@@ -1,10 +1,8 @@
 using System;
 using System.Collections.Concurrent;
 using System.IO;
-using System.Net;
-using System.Security.Cryptography;
-using System.Text;
 using System.Linq;
+using System.Net;
 using System.Threading;
 using System.Threading.Tasks;
 using ManagedCode.Communication;
@@ -48,10 +46,13 @@ public async Task<Result> AppendChunkAsync(FileUploadPayload payload, Cancellati
 
         var descriptor = payload.Payload;
         var uploadId = ValidateUploadId(ChunkUploadDescriptor.ResolveUploadId(descriptor));
+        var chunkLength = payload.File.Length;
         if (descriptor.ChunkIndex <= 0 || descriptor.TotalChunks < 0 ||
             (descriptor.TotalChunks > 0 && descriptor.ChunkIndex > descriptor.TotalChunks) ||
-            descriptor.ChunkSize <= 0 || descriptor.FileSize < 0 || payload.File.Length > descriptor.ChunkSize)
+            descriptor.ChunkSize <= 0 || descriptor.FileSize < 0 || chunkLength < 0 || chunkLength > descriptor.ChunkSize)
             return Result.Fail(HttpStatusCode.BadRequest, "Invalid chunk index or declared size");
+        if (chunkLength > ChunkUploadSession.ResolveExpectedLength(descriptor.TotalChunks, descriptor.FileSize))
+            return Result.Fail(HttpStatusCode.BadRequest, "Upload exceeds its declared length");
 
         RemoveExpiredSessions();
         var session = AdmitSession(uploadId, descriptor);
@@ -63,10 +64,14 @@ public async Task<Result> AppendChunkAsync(FileUploadPayload payload, Cancellati
         {
             if (session.Closed)
                 return Result.Fail("Upload session is closed");
+            if (TryExpireSession(session, DateTimeOffset.UtcNow))
+                return Result.Fail("Upload session has expired");
             if (session.TotalChunks != descriptor.TotalChunks || session.FileSize != descriptor.FileSize)
                 return Result.Fail(HttpStatusCode.BadRequest, "Chunk does not match the upload session");
+            if (!session.CanAcceptChunk(descriptor.ChunkIndex, chunkLength))
+                return Result.Fail(HttpStatusCode.BadRequest, "Upload exceeds its declared length");
 
-            await ChunkUploadFiles.StageChunkAsync(session, payload, cancellationToken).ConfigureAwait(false);
+            await ChunkUploadFiles.StageChunkAsync(session, payload, chunkLength, cancellationToken).ConfigureAwait(false);
             return Result.Succeed();
         }
         finally
@@ -84,8 +89,7 @@ public async Task<Result> AppendChunkAsync(FileUploadPayload payload, Cancellati
             if (_options.MaxActiveSessions > 0 && _sessions.Count >= _options.MaxActiveSessions)
                 return null;
 
-            var directoryKey = Convert.ToHexStringLower(SHA256.HashData(Encoding.UTF8.GetBytes(uploadId)));
-            var directory = Path.Combine(_options.TempPath, directoryKey);
+            var directory = Path.Combine(_options.TempPath, Guid.NewGuid().ToString("N"));
             Directory.CreateDirectory(directory);
             var session = new ChunkUploadSession(uploadId, descriptor.FileName ?? descriptor.UploadId,
                 descriptor.ContentType, descriptor.TotalChunks, descriptor.ChunkSize, descriptor.FileSize, directory);
@@ -118,17 +122,18 @@ public async Task<Result<ChunkUploadCompleteResponse>> CompleteAsync(ChunkUpload
             acquired = true;
             if (session.Closed)
                 return Result<ChunkUploadCompleteResponse>.Fail("Upload session is closed");
+            if (TryExpireSession(session, DateTimeOffset.UtcNow))
+                return Result<ChunkUploadCompleteResponse>.Fail("Upload session has expired");
 
             session.Touch();
             session.EnsureAllChunksPresent();
             var orderedChunks = session.ChunkFiles
                 .OrderBy(x => x.Key)
-                .Select(x => x.Value)
+                .Select(x => x.Value.Path)
                 .ToArray();
 
             var mergedFilePath = Path.Combine(session.WorkingDirectory, "merged.bin");
-            var expectedLength = session.TotalChunks == 0 && session.FileSize == 0 ? null : session.FileSize;
-            var crc = await ChunkUploadFiles.MergeChunksAsync(mergedFilePath, orderedChunks, expectedLength, cancellationToken).ConfigureAwait(false);
+            var crc = await ChunkUploadFiles.MergeChunksAsync(mergedFilePath, orderedChunks, session.ExpectedLength, cancellationToken).ConfigureAwait(false);
 
             BlobMetadata? metadata = null;
             if (request.CommitToStorage)
@@ -211,21 +216,31 @@ private void RemoveExpiredSessions()
             return;
         }
 
-        var expirationThreshold = DateTimeOffset.UtcNow - _options.SessionTtl;
+        var now = DateTimeOffset.UtcNow;
         foreach (var (_, session) in _sessions)
         {
             if (!session.Gate.Wait(0))
                 continue;
 
             try
             {
-                if (!session.Closed && session.LastTouchedUtc < expirationThreshold)
-                    CloseSession(session);
+                if (!session.Closed)
+                    TryExpireSession(session, now);
             }
             finally
             {
                 session.Gate.Release();
             }
         }
     }
+
+    private bool TryExpireSession(ChunkUploadSession session, DateTimeOffset now)
+    {
+        var ttl = _options.SessionTtl;
+        if (ttl <= TimeSpan.Zero || now - session.LastTouchedUtc < ttl)
+            return false;
+
+        CloseSession(session);
+        return true;
+    }
 }

Integraions/ManagedCode.Storage.Server/ChunkUpload/ChunkUploadSession.cs

@@ -1,5 +1,4 @@
 using System;
-using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.IO;
 using System.Threading;
@@ -8,7 +7,8 @@ namespace ManagedCode.Storage.Server.ChunkUpload;
 
 internal sealed class ChunkUploadSession
 {
-    private readonly ConcurrentDictionary<int, string> _chunkFiles = new();
+    private readonly Dictionary<int, (string Path, long Length)> _chunkFiles = new();
+    private long _acceptedBytes;
 
     public ChunkUploadSession(string uploadId, string fileName, string? contentType, int totalChunks, int chunkSize, long? fileSize, string workingDirectory)
     {
@@ -34,6 +34,8 @@ public ChunkUploadSession(string uploadId, string fileName, string? contentType,
 
     public long? FileSize { get; }
 
+    public long? ExpectedLength => ResolveExpectedLength(TotalChunks, FileSize);
+
     public string WorkingDirectory { get; }
 
     public SemaphoreSlim Gate { get; } = new(1, 1);
@@ -42,18 +44,24 @@ public ChunkUploadSession(string uploadId, string fileName, string? contentType,
 
     public DateTimeOffset LastTouchedUtc { get; private set; }
 
-    public IReadOnlyDictionary<int, string> ChunkFiles => _chunkFiles;
+    public IReadOnlyDictionary<int, (string Path, long Length)> ChunkFiles => _chunkFiles;
+
+    public static long? ResolveExpectedLength(int totalChunks, long? fileSize) =>
+        totalChunks == 0 && fileSize == 0 ? null : fileSize;
 
     public void Touch()
     {
         LastTouchedUtc = DateTimeOffset.UtcNow;
     }
 
-    public string RegisterChunk(int index, string path)
+    public bool CanAcceptChunk(int index, long length) =>
+        ExpectedLength is not long expected || length <= expected - _acceptedBytes + _chunkFiles.GetValueOrDefault(index).Length;
+
+    public void RegisterChunk(int index, string path, long length)
     {
-        _chunkFiles[index] = path;
+        _acceptedBytes += length - _chunkFiles.GetValueOrDefault(index).Length;
+        _chunkFiles[index] = (path, length);
         Touch();
-        return path;
     }
 
     public void EnsureAllChunksPresent()
@@ -71,7 +79,7 @@ public void EnsureAllChunksPresent()
 
     public void Cleanup(bool keepMergedFile = false)
     {
-        foreach (var (_, path) in _chunkFiles)
+        foreach (var (path, _) in _chunkFiles.Values)
         {
             if (File.Exists(path))
             {

Integraions/ManagedCode.Storage.Server/Hubs/StorageHubBase.cs

@@ -252,15 +252,34 @@ public virtual async IAsyncEnumerable<byte[]> DownloadStreamAsync(string blobNam
             blobName, null, cancellationToken);
         byte[]? buffer = null;
         Exception? failure = null;
-        var wasCanceled = false;
 
         var token = registration.Cancellation.Token;
 
         try
         {
-            await using var sourceStream = await OpenDownloadStreamAsync(blobName, token).ConfigureAwait(false);
-            registration.Status.TotalBytes = sourceStream.CanSeek ? sourceStream.Length : null;
-            buffer = ArrayPool<byte>.Shared.Rent(_options.StreamBufferSize);
+            Stream openedStream;
+            try
+            {
+                openedStream = await OpenDownloadStreamAsync(blobName, token).ConfigureAwait(false);
+            }
+            catch (Exception ex)
+            {
+                _logger.LogDebug(ex, "DownloadStreamAsync could not open {TransferId}", transferId);
+                failure = ex;
+                throw;
+            }
+
+            await using var sourceStream = openedStream;
+            try
+            {
+                registration.Status.TotalBytes = sourceStream.CanSeek ? sourceStream.Length : null;
+                buffer = ArrayPool<byte>.Shared.Rent(_options.StreamBufferSize);
+            }
+            catch (Exception ex)
+            {
+                failure = ex;
+                throw;
+            }
 
             while (true)
             {
@@ -272,7 +291,6 @@ public virtual async IAsyncEnumerable<byte[]> DownloadStreamAsync(string blobNam
                 catch (OperationCanceledException oce)
                 {
                     failure = oce;
-                    wasCanceled = true;
                     throw;
                 }
                 catch (Exception ex)
@@ -293,55 +311,28 @@ public virtual async IAsyncEnumerable<byte[]> DownloadStreamAsync(string blobNam
                 registration.Touch();
 
                 var progressSnapshot = CreateStatusSnapshot(registration.Status);
-                try
-                {
-                    await NotifyClientAsync(StorageHubEvents.TransferProgress, progressSnapshot, token).ConfigureAwait(false);
-                }
-                catch (OperationCanceledException oce)
-                {
-                    failure = oce;
-                    wasCanceled = true;
-                    throw;
-                }
-                catch (HubException)
-                {
-                    throw;
-                }
+                await NotifyClientAsync(StorageHubEvents.TransferProgress, progressSnapshot, token).ConfigureAwait(false);
                 yield return chunk;
             }
 
+            token.ThrowIfCancellationRequested();
             registration.Status.IsCompleted = true;
             var completionSnapshot = CreateStatusSnapshot(registration.Status);
-            try
-            {
-                await NotifyClientAsync(StorageHubEvents.TransferCompleted, completionSnapshot, token).ConfigureAwait(false);
-            }
-            catch (OperationCanceledException oce)
-            {
-                failure = oce;
-                wasCanceled = true;
-                throw;
-            }
+            await NotifyClientAsync(StorageHubEvents.TransferCompleted, completionSnapshot, token).ConfigureAwait(false);
         }
         finally
         {
             if (buffer is not null)
                 ArrayPool<byte>.Shared.Return(buffer);
             Transfers.Remove(registration);
 
-            if (failure is not null)
+            if (!registration.Status.IsCompleted && (failure is not null || token.IsCancellationRequested))
             {
-                if (wasCanceled)
-                {
-                    registration.Status.IsCanceled = true;
-                    registration.Status.Error ??= "Transfer canceled";
-                    _ = NotifyClientAsync(StorageHubEvents.TransferCanceled, CreateStatusSnapshot(registration.Status), CancellationToken.None);
-                }
-                else
-                {
-                    registration.Status.Error = failure.Message;
-                    _ = NotifyClientAsync(StorageHubEvents.TransferFaulted, CreateStatusSnapshot(registration.Status), CancellationToken.None);
-                }
+                var canceled = failure is OperationCanceledException || token.IsCancellationRequested;
+                registration.Status.IsCanceled = canceled;
+                registration.Status.Error = canceled ? "Transfer canceled" : "Download failed";
+                var eventName = canceled ? StorageHubEvents.TransferCanceled : StorageHubEvents.TransferFaulted;
+                await NotifyClientAsync(eventName, CreateStatusSnapshot(registration.Status), CancellationToken.None).ConfigureAwait(false);
             }
         }
     }

ManagedCode.Storage.VirtualFileSystem/Extensions/ServiceCollectionExtensions.cs

@@ -146,10 +146,12 @@ public override async Task SetVfsMetadataAsync(
         string blobName,
         CancellationToken cancellationToken = default)
     {
-        var metadata = await GetBlobInfoAsync(blobName, cancellationToken);
-        return metadata?.Metadata is { } values ? ParseVfsMetadata(values) : null;
+        return DecodeVfsMetadata(await GetBlobInfoAsync(blobName, cancellationToken));
     }
 
+    internal VfsMetadata? DecodeVfsMetadata(BlobMetadata? metadata) =>
+        metadata?.Metadata is { } values ? ParseVfsMetadata(values) : null;
+
     public override async Task<IReadOnlyDictionary<string, string>> GetCustomMetadataAsync(
         string blobName,
         CancellationToken cancellationToken = default)