Skip to content

Commit 5691801

Browse files
KSemenenkoKSemenenko
committed
harden storage reads and uploads and release 10.1.12
1 parent be5d88b commit 5691801

35 files changed

Lines changed: 1141 additions & 261 deletions

‎Directory.Build.props‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -29,8 +29,8 @@
2929
<RepositoryUrl>https://github.com/managedcode/Storage</RepositoryUrl>
3030
<PackageProjectUrl>https://github.com/managedcode/Storage</PackageProjectUrl>
3131
<Product>Managed Code - Storage</Product>
32-
<Version>10.1.11</Version>
33-
<PackageVersion>10.1.11</PackageVersion>
32+
<Version>10.1.12</Version>
33+
<PackageVersion>10.1.12</PackageVersion>
3434

3535
</PropertyGroup>
3636

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

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,8 @@ internal static class ChunkUploadFiles
1313
{
1414
private const int CopyBufferSize = 64 * 1024;
1515

16-
internal static async Task StageChunkAsync(ChunkUploadSession session, FileUploadPayload payload, CancellationToken cancellationToken)
16+
internal static async Task StageChunkAsync(ChunkUploadSession session, FileUploadPayload payload, long length,
17+
CancellationToken cancellationToken)
1718
{
1819
var chunkIndex = payload.Payload.ChunkIndex;
1920
var chunkFilePath = Path.Combine(session.WorkingDirectory, $"{chunkIndex:D6}.part");
@@ -24,11 +25,13 @@ internal static async Task StageChunkAsync(ChunkUploadSession session, FileUploa
2425
await using (var source = payload.File.OpenReadStream())
2526
{
2627
await source.CopyToAsync(target, CopyBufferSize, cancellationToken).ConfigureAwait(false);
28+
if (target.Position != length)
29+
throw new InvalidDataException("The staged chunk differs from its form file length.");
2730
}
2831

2932
cancellationToken.ThrowIfCancellationRequested();
3033
File.Move(stagingPath, chunkFilePath, overwrite: true);
31-
session.RegisterChunk(chunkIndex, chunkFilePath);
34+
session.RegisterChunk(chunkIndex, chunkFilePath, length);
3235
}
3336
finally
3437
{

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

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,8 @@ public class ChunkUploadOptions
1414
public string TempPath { get; set; } = Path.Combine(Path.GetTempPath(), "managedcode-storage", "chunks");
1515

1616
/// <summary>
17-
/// How long chunks are kept on disk after the last write. Expired sessions are cleaned up on completion or abort.
17+
/// Idle time after which an upload session expires. Append and completion requests clean up expired sessions.
18+
/// A nonpositive value disables expiration.
1819
/// </summary>
1920
public TimeSpan SessionTtl { get; set; } = TimeSpan.FromHours(1);
2021

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

Lines changed: 28 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,8 @@
11
using System;
22
using System.Collections.Concurrent;
33
using System.IO;
4-
using System.Net;
5-
using System.Security.Cryptography;
6-
using System.Text;
74
using System.Linq;
5+
using System.Net;
86
using System.Threading;
97
using System.Threading.Tasks;
108
using ManagedCode.Communication;
@@ -48,10 +46,13 @@ public async Task<Result> AppendChunkAsync(FileUploadPayload payload, Cancellati
4846

4947
var descriptor = payload.Payload;
5048
var uploadId = ValidateUploadId(ChunkUploadDescriptor.ResolveUploadId(descriptor));
49+
var chunkLength = payload.File.Length;
5150
if (descriptor.ChunkIndex <= 0 || descriptor.TotalChunks < 0 ||
5251
(descriptor.TotalChunks > 0 && descriptor.ChunkIndex > descriptor.TotalChunks) ||
53-
descriptor.ChunkSize <= 0 || descriptor.FileSize < 0 || payload.File.Length > descriptor.ChunkSize)
52+
descriptor.ChunkSize <= 0 || descriptor.FileSize < 0 || chunkLength < 0 || chunkLength > descriptor.ChunkSize)
5453
return Result.Fail(HttpStatusCode.BadRequest, "Invalid chunk index or declared size");
54+
if (chunkLength > ChunkUploadSession.ResolveExpectedLength(descriptor.TotalChunks, descriptor.FileSize))
55+
return Result.Fail(HttpStatusCode.BadRequest, "Upload exceeds its declared length");
5556

5657
RemoveExpiredSessions();
5758
var session = AdmitSession(uploadId, descriptor);
@@ -63,10 +64,14 @@ public async Task<Result> AppendChunkAsync(FileUploadPayload payload, Cancellati
6364
{
6465
if (session.Closed)
6566
return Result.Fail("Upload session is closed");
67+
if (TryExpireSession(session, DateTimeOffset.UtcNow))
68+
return Result.Fail("Upload session has expired");
6669
if (session.TotalChunks != descriptor.TotalChunks || session.FileSize != descriptor.FileSize)
6770
return Result.Fail(HttpStatusCode.BadRequest, "Chunk does not match the upload session");
71+
if (!session.CanAcceptChunk(descriptor.ChunkIndex, chunkLength))
72+
return Result.Fail(HttpStatusCode.BadRequest, "Upload exceeds its declared length");
6873

69-
await ChunkUploadFiles.StageChunkAsync(session, payload, cancellationToken).ConfigureAwait(false);
74+
await ChunkUploadFiles.StageChunkAsync(session, payload, chunkLength, cancellationToken).ConfigureAwait(false);
7075
return Result.Succeed();
7176
}
7277
finally
@@ -84,8 +89,7 @@ public async Task<Result> AppendChunkAsync(FileUploadPayload payload, Cancellati
8489
if (_options.MaxActiveSessions > 0 && _sessions.Count >= _options.MaxActiveSessions)
8590
return null;
8691

87-
var directoryKey = Convert.ToHexStringLower(SHA256.HashData(Encoding.UTF8.GetBytes(uploadId)));
88-
var directory = Path.Combine(_options.TempPath, directoryKey);
92+
var directory = Path.Combine(_options.TempPath, Guid.NewGuid().ToString("N"));
8993
Directory.CreateDirectory(directory);
9094
var session = new ChunkUploadSession(uploadId, descriptor.FileName ?? descriptor.UploadId,
9195
descriptor.ContentType, descriptor.TotalChunks, descriptor.ChunkSize, descriptor.FileSize, directory);
@@ -118,17 +122,18 @@ public async Task<Result<ChunkUploadCompleteResponse>> CompleteAsync(ChunkUpload
118122
acquired = true;
119123
if (session.Closed)
120124
return Result<ChunkUploadCompleteResponse>.Fail("Upload session is closed");
125+
if (TryExpireSession(session, DateTimeOffset.UtcNow))
126+
return Result<ChunkUploadCompleteResponse>.Fail("Upload session has expired");
121127

122128
session.Touch();
123129
session.EnsureAllChunksPresent();
124130
var orderedChunks = session.ChunkFiles
125131
.OrderBy(x => x.Key)
126-
.Select(x => x.Value)
132+
.Select(x => x.Value.Path)
127133
.ToArray();
128134

129135
var mergedFilePath = Path.Combine(session.WorkingDirectory, "merged.bin");
130-
var expectedLength = session.TotalChunks == 0 && session.FileSize == 0 ? null : session.FileSize;
131-
var crc = await ChunkUploadFiles.MergeChunksAsync(mergedFilePath, orderedChunks, expectedLength, cancellationToken).ConfigureAwait(false);
136+
var crc = await ChunkUploadFiles.MergeChunksAsync(mergedFilePath, orderedChunks, session.ExpectedLength, cancellationToken).ConfigureAwait(false);
132137

133138
BlobMetadata? metadata = null;
134139
if (request.CommitToStorage)
@@ -211,21 +216,31 @@ private void RemoveExpiredSessions()
211216
return;
212217
}
213218

214-
var expirationThreshold = DateTimeOffset.UtcNow - _options.SessionTtl;
219+
var now = DateTimeOffset.UtcNow;
215220
foreach (var (_, session) in _sessions)
216221
{
217222
if (!session.Gate.Wait(0))
218223
continue;
219224

220225
try
221226
{
222-
if (!session.Closed && session.LastTouchedUtc < expirationThreshold)
223-
CloseSession(session);
227+
if (!session.Closed)
228+
TryExpireSession(session, now);
224229
}
225230
finally
226231
{
227232
session.Gate.Release();
228233
}
229234
}
230235
}
236+
237+
private bool TryExpireSession(ChunkUploadSession session, DateTimeOffset now)
238+
{
239+
var ttl = _options.SessionTtl;
240+
if (ttl <= TimeSpan.Zero || now - session.LastTouchedUtc < ttl)
241+
return false;
242+
243+
CloseSession(session);
244+
return true;
245+
}
231246
}

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

Lines changed: 15 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,4 @@
11
using System;
2-
using System.Collections.Concurrent;
32
using System.Collections.Generic;
43
using System.IO;
54
using System.Threading;
@@ -8,7 +7,8 @@ namespace ManagedCode.Storage.Server.ChunkUpload;
87

98
internal sealed class ChunkUploadSession
109
{
11-
private readonly ConcurrentDictionary<int, string> _chunkFiles = new();
10+
private readonly Dictionary<int, (string Path, long Length)> _chunkFiles = new();
11+
private long _acceptedBytes;
1212

1313
public ChunkUploadSession(string uploadId, string fileName, string? contentType, int totalChunks, int chunkSize, long? fileSize, string workingDirectory)
1414
{
@@ -34,6 +34,8 @@ public ChunkUploadSession(string uploadId, string fileName, string? contentType,
3434

3535
public long? FileSize { get; }
3636

37+
public long? ExpectedLength => ResolveExpectedLength(TotalChunks, FileSize);
38+
3739
public string WorkingDirectory { get; }
3840

3941
public SemaphoreSlim Gate { get; } = new(1, 1);
@@ -42,18 +44,24 @@ public ChunkUploadSession(string uploadId, string fileName, string? contentType,
4244

4345
public DateTimeOffset LastTouchedUtc { get; private set; }
4446

45-
public IReadOnlyDictionary<int, string> ChunkFiles => _chunkFiles;
47+
public IReadOnlyDictionary<int, (string Path, long Length)> ChunkFiles => _chunkFiles;
48+
49+
public static long? ResolveExpectedLength(int totalChunks, long? fileSize) =>
50+
totalChunks == 0 && fileSize == 0 ? null : fileSize;
4651

4752
public void Touch()
4853
{
4954
LastTouchedUtc = DateTimeOffset.UtcNow;
5055
}
5156

52-
public string RegisterChunk(int index, string path)
57+
public bool CanAcceptChunk(int index, long length) =>
58+
ExpectedLength is not long expected || length <= expected - _acceptedBytes + _chunkFiles.GetValueOrDefault(index).Length;
59+
60+
public void RegisterChunk(int index, string path, long length)
5361
{
54-
_chunkFiles[index] = path;
62+
_acceptedBytes += length - _chunkFiles.GetValueOrDefault(index).Length;
63+
_chunkFiles[index] = (path, length);
5564
Touch();
56-
return path;
5765
}
5866

5967
public void EnsureAllChunksPresent()
@@ -71,7 +79,7 @@ public void EnsureAllChunksPresent()
7179

7280
public void Cleanup(bool keepMergedFile = false)
7381
{
74-
foreach (var (_, path) in _chunkFiles)
82+
foreach (var (path, _) in _chunkFiles.Values)
7583
{
7684
if (File.Exists(path))
7785
{

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

Lines changed: 32 additions & 41 deletions
Original file line numberDiff line numberDiff line change
@@ -252,15 +252,34 @@ public virtual async IAsyncEnumerable<byte[]> DownloadStreamAsync(string blobNam
252252
blobName, null, cancellationToken);
253253
byte[]? buffer = null;
254254
Exception? failure = null;
255-
var wasCanceled = false;
256255

257256
var token = registration.Cancellation.Token;
258257

259258
try
260259
{
261-
await using var sourceStream = await OpenDownloadStreamAsync(blobName, token).ConfigureAwait(false);
262-
registration.Status.TotalBytes = sourceStream.CanSeek ? sourceStream.Length : null;
263-
buffer = ArrayPool<byte>.Shared.Rent(_options.StreamBufferSize);
260+
Stream openedStream;
261+
try
262+
{
263+
openedStream = await OpenDownloadStreamAsync(blobName, token).ConfigureAwait(false);
264+
}
265+
catch (Exception ex)
266+
{
267+
_logger.LogDebug(ex, "DownloadStreamAsync could not open {TransferId}", transferId);
268+
failure = ex;
269+
throw;
270+
}
271+
272+
await using var sourceStream = openedStream;
273+
try
274+
{
275+
registration.Status.TotalBytes = sourceStream.CanSeek ? sourceStream.Length : null;
276+
buffer = ArrayPool<byte>.Shared.Rent(_options.StreamBufferSize);
277+
}
278+
catch (Exception ex)
279+
{
280+
failure = ex;
281+
throw;
282+
}
264283

265284
while (true)
266285
{
@@ -272,7 +291,6 @@ public virtual async IAsyncEnumerable<byte[]> DownloadStreamAsync(string blobNam
272291
catch (OperationCanceledException oce)
273292
{
274293
failure = oce;
275-
wasCanceled = true;
276294
throw;
277295
}
278296
catch (Exception ex)
@@ -293,55 +311,28 @@ public virtual async IAsyncEnumerable<byte[]> DownloadStreamAsync(string blobNam
293311
registration.Touch();
294312

295313
var progressSnapshot = CreateStatusSnapshot(registration.Status);
296-
try
297-
{
298-
await NotifyClientAsync(StorageHubEvents.TransferProgress, progressSnapshot, token).ConfigureAwait(false);
299-
}
300-
catch (OperationCanceledException oce)
301-
{
302-
failure = oce;
303-
wasCanceled = true;
304-
throw;
305-
}
306-
catch (HubException)
307-
{
308-
throw;
309-
}
314+
await NotifyClientAsync(StorageHubEvents.TransferProgress, progressSnapshot, token).ConfigureAwait(false);
310315
yield return chunk;
311316
}
312317

318+
token.ThrowIfCancellationRequested();
313319
registration.Status.IsCompleted = true;
314320
var completionSnapshot = CreateStatusSnapshot(registration.Status);
315-
try
316-
{
317-
await NotifyClientAsync(StorageHubEvents.TransferCompleted, completionSnapshot, token).ConfigureAwait(false);
318-
}
319-
catch (OperationCanceledException oce)
320-
{
321-
failure = oce;
322-
wasCanceled = true;
323-
throw;
324-
}
321+
await NotifyClientAsync(StorageHubEvents.TransferCompleted, completionSnapshot, token).ConfigureAwait(false);
325322
}
326323
finally
327324
{
328325
if (buffer is not null)
329326
ArrayPool<byte>.Shared.Return(buffer);
330327
Transfers.Remove(registration);
331328

332-
if (failure is not null)
329+
if (!registration.Status.IsCompleted && (failure is not null || token.IsCancellationRequested))
333330
{
334-
if (wasCanceled)
335-
{
336-
registration.Status.IsCanceled = true;
337-
registration.Status.Error ??= "Transfer canceled";
338-
_ = NotifyClientAsync(StorageHubEvents.TransferCanceled, CreateStatusSnapshot(registration.Status), CancellationToken.None);
339-
}
340-
else
341-
{
342-
registration.Status.Error = failure.Message;
343-
_ = NotifyClientAsync(StorageHubEvents.TransferFaulted, CreateStatusSnapshot(registration.Status), CancellationToken.None);
344-
}
331+
var canceled = failure is OperationCanceledException || token.IsCancellationRequested;
332+
registration.Status.IsCanceled = canceled;
333+
registration.Status.Error = canceled ? "Transfer canceled" : "Download failed";
334+
var eventName = canceled ? StorageHubEvents.TransferCanceled : StorageHubEvents.TransferFaulted;
335+
await NotifyClientAsync(eventName, CreateStatusSnapshot(registration.Status), CancellationToken.None).ConfigureAwait(false);
345336
}
346337
}
347338
}

‎ManagedCode.Storage.VirtualFileSystem/Extensions/ServiceCollectionExtensions.cs‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -146,10 +146,12 @@ public override async Task SetVfsMetadataAsync(
146146
string blobName,
147147
CancellationToken cancellationToken = default)
148148
{
149-
var metadata = await GetBlobInfoAsync(blobName, cancellationToken);
150-
return metadata?.Metadata is { } values ? ParseVfsMetadata(values) : null;
149+
return DecodeVfsMetadata(await GetBlobInfoAsync(blobName, cancellationToken));
151150
}
152151

152+
internal VfsMetadata? DecodeVfsMetadata(BlobMetadata? metadata) =>
153+
metadata?.Metadata is { } values ? ParseVfsMetadata(values) : null;
154+
153155
public override async Task<IReadOnlyDictionary<string, string>> GetCustomMetadataAsync(
154156
string blobName,
155157
CancellationToken cancellationToken = default)

0 commit comments

Comments
 (0)