Skip to content

Commit d645751

Browse files
KSemenenkoKSemenenko
committed
fix: expose atomic object storage capabilities and persist empty VFS files
1 parent a43488e commit d645751

12 files changed

Lines changed: 491 additions & 7 deletions

File tree

‎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.0.8</Version>
33-
<PackageVersion>10.0.8</PackageVersion>
32+
<Version>10.0.9</Version>
33+
<PackageVersion>10.0.9</PackageVersion>
3434

3535
</PropertyGroup>
3636

‎ManagedCode.Storage.Core/Objects/IObjectStorage.cs‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,31 @@
1+
using System;
2+
using System.Collections.Generic;
3+
using System.IO;
4+
using System.Threading;
5+
using System.Threading.Tasks;
6+
7+
namespace ManagedCode.Storage.Core.Objects;
8+
9+
/// <summary>Optional atomic object operations. Providers must enforce conditions server-side.</summary>
10+
public interface IObjectStorage
11+
{
12+
Uri ContainerUri { get; }
13+
Task<StorageContainerInfo> GetContainerInfoAsync(CancellationToken cancellationToken = default);
14+
Task CreatePrivateContainerAsync(IReadOnlyDictionary<string, string>? metadata = null, CancellationToken cancellationToken = default);
15+
Task SetContainerMetadataAsync(IReadOnlyDictionary<string, string> metadata, CancellationToken cancellationToken = default);
16+
Task<bool> DeleteContainerIfExistsAsync(CancellationToken cancellationToken = default);
17+
Task<StorageObjectInfo> GetObjectInfoAsync(string path, CancellationToken cancellationToken = default);
18+
Task<bool> ObjectExistsAsync(string path, CancellationToken cancellationToken = default);
19+
Task<Stream> OpenObjectReadAsync(string path, StorageReadOptions? options = null, CancellationToken cancellationToken = default);
20+
Task<StorageObjectInfo> WriteObjectAsync(string path, Stream content, StorageWriteOptions? options = null, CancellationToken cancellationToken = default);
21+
Task SetObjectMetadataAsync(string path, IReadOnlyDictionary<string, string> metadata, string? ifMatch = null, CancellationToken cancellationToken = default);
22+
Task<bool> DeleteObjectIfExistsAsync(string path, bool includeSnapshots = false, CancellationToken cancellationToken = default);
23+
Task<StorageObjectPage> ListObjectsAsync(string? prefix = null, string? continuationToken = null, int pageSize = 100, CancellationToken cancellationToken = default);
24+
}
25+
26+
/// <summary>Optional resumable multipart writes. Staged parts stay invisible until committed.</summary>
27+
public interface IMultipartObjectStorage : IObjectStorage
28+
{
29+
Task StagePartAsync(string path, string partId, Stream content, CancellationToken cancellationToken = default);
30+
Task<StorageObjectInfo> CommitPartsAsync(string path, IReadOnlyList<string> partIds, StorageWriteOptions? options = null, CancellationToken cancellationToken = default);
31+
}

‎ManagedCode.Storage.Core/Objects/StorageObjectModels.cs‎

Lines changed: 46 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,46 @@
1+
using System;
2+
using System.Collections.Generic;
3+
4+
namespace ManagedCode.Storage.Core.Objects;
5+
6+
public sealed record StorageContainerInfo(string ETag, bool IsPrivate, IReadOnlyDictionary<string, string> Metadata);
7+
8+
public sealed record StorageObjectInfo(string Path, string ETag, long Length, string? ContentType,
9+
string? ContentEncoding, IReadOnlyDictionary<string, string> Metadata, DateTimeOffset? LastModified = null);
10+
11+
public sealed record StorageObjectPage(IReadOnlyList<StorageObjectInfo> Items, string? ContinuationToken);
12+
13+
public sealed record StorageReadOptions
14+
{
15+
public string? IfMatch { get; init; }
16+
public long Offset { get; init; }
17+
public long? Length { get; init; }
18+
}
19+
20+
public sealed record StorageWriteOptions
21+
{
22+
public bool IfAbsent { get; init; }
23+
public string? IfMatch { get; init; }
24+
public string? ContentType { get; init; }
25+
public string? ContentEncoding { get; init; }
26+
public IReadOnlyDictionary<string, string>? Metadata { get; init; }
27+
}
28+
29+
/// <summary>A provider-independent storage failure; cancellation is never translated.</summary>
30+
public sealed class StorageOperationException(string message, int statusCode, Exception innerException, string? errorCode = null)
31+
: System.IO.IOException(message, innerException)
32+
{
33+
public int StatusCode { get; } = statusCode;
34+
public string? ErrorCode { get; } = errorCode;
35+
public bool IsNotFound => StatusCode == 404;
36+
public bool IsConflict => StatusCode is 409 or 412;
37+
}
38+
39+
public static class ObjectStorageCapabilities
40+
{
41+
public static IObjectStorage RequireObjectStorage(this IStorage storage) =>
42+
storage as IObjectStorage ?? throw new NotSupportedException("The storage provider does not support atomic object operations.");
43+
44+
public static IMultipartObjectStorage RequireMultipartStorage(this IStorage storage) =>
45+
storage as IMultipartObjectStorage ?? throw new NotSupportedException("The storage provider does not support resumable multipart operations.");
46+
}

‎ManagedCode.Storage.VirtualFileSystem/Streaming/VfsWriteStream.cs‎

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -132,8 +132,6 @@ public override async ValueTask DisposeAsync()
132132

133133
private async Task UploadBufferedDataAsync()
134134
{
135-
if (_bufferStream.Length == 0) { _logger.LogDebug("No data to upload for: {BlobKey}", _blobKey); return; }
136-
137135
_logger.LogDebug("Uploading buffered data: {BlobKey}, size: {Size}", _blobKey, _bufferStream.Length);
138136

139137
try

‎Storages/ManagedCode.Storage.Azure/AzureObjectOperations.cs‎

Lines changed: 150 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,150 @@
1+
using System;
2+
using System.Collections.Generic;
3+
using System.IO;
4+
using System.Linq;
5+
using System.Threading;
6+
using System.Threading.Tasks;
7+
using Azure;
8+
using Azure.Storage.Blobs;
9+
using Azure.Storage.Blobs.Models;
10+
using Azure.Storage.Blobs.Specialized;
11+
using ManagedCode.Storage.Core.Objects;
12+
13+
namespace ManagedCode.Storage.Azure;
14+
15+
internal sealed class AzureObjectOperations(BlobContainerClient container) : IMultipartObjectStorage
16+
{
17+
public Uri ContainerUri => container.Uri;
18+
19+
public Task<StorageContainerInfo> GetContainerInfoAsync(CancellationToken cancellationToken = default) => ExecuteAsync(async () =>
20+
{
21+
var value = (await container.GetPropertiesAsync(cancellationToken: cancellationToken)).Value;
22+
return new StorageContainerInfo(value.ETag.ToString(), value.PublicAccess == PublicAccessType.None,
23+
new Dictionary<string, string>(value.Metadata));
24+
});
25+
26+
public Task CreatePrivateContainerAsync(IReadOnlyDictionary<string, string>? metadata = null, CancellationToken cancellationToken = default) => ExecuteAsync(async () =>
27+
{
28+
await container.CreateIfNotExistsAsync(PublicAccessType.None, Copy(metadata), cancellationToken: cancellationToken);
29+
return true;
30+
});
31+
32+
public Task SetContainerMetadataAsync(IReadOnlyDictionary<string, string> metadata, CancellationToken cancellationToken = default) => ExecuteAsync(async () =>
33+
{
34+
await container.SetMetadataAsync(Copy(metadata), cancellationToken: cancellationToken);
35+
return true;
36+
});
37+
38+
public Task<bool> DeleteContainerIfExistsAsync(CancellationToken cancellationToken = default) => ExecuteAsync(async () =>
39+
(await container.DeleteIfExistsAsync(cancellationToken: cancellationToken)).Value);
40+
41+
public Task<StorageObjectInfo> GetObjectInfoAsync(string path, CancellationToken cancellationToken = default) => ExecuteAsync(async () =>
42+
Info(path, (await container.GetBlobClient(path).GetPropertiesAsync(cancellationToken: cancellationToken)).Value));
43+
44+
public Task<bool> ObjectExistsAsync(string path, CancellationToken cancellationToken = default) => ExecuteAsync(async () =>
45+
(await container.GetBlobClient(path).ExistsAsync(cancellationToken)).Value);
46+
47+
public Task<Stream> OpenObjectReadAsync(string path, StorageReadOptions? options = null, CancellationToken cancellationToken = default) => ExecuteAsync(async () =>
48+
{
49+
options ??= new StorageReadOptions();
50+
ArgumentOutOfRangeException.ThrowIfNegative(options.Offset);
51+
if (options.Length is <= 0) throw new ArgumentOutOfRangeException(nameof(options));
52+
var blob = container.GetBlobClient(path);
53+
var etag = options.IfMatch ?? (await blob.GetPropertiesAsync(cancellationToken: cancellationToken)).Value.ETag.ToString();
54+
if (options.Length is null)
55+
{
56+
return await blob.OpenReadAsync(new BlobOpenReadOptions(false)
57+
{
58+
Position = options.Offset,
59+
Conditions = new BlobRequestConditions { IfMatch = new ETag(etag) }
60+
}, cancellationToken);
61+
}
62+
var download = await blob.DownloadStreamingAsync(new BlobDownloadOptions
63+
{
64+
Range = new HttpRange(options.Offset, options.Length),
65+
Conditions = new BlobRequestConditions { IfMatch = new ETag(etag) }
66+
}, cancellationToken);
67+
return download.Value.Content;
68+
});
69+
70+
public Task<StorageObjectInfo> WriteObjectAsync(string path, Stream content, StorageWriteOptions? options = null, CancellationToken cancellationToken = default) => ExecuteAsync(async () =>
71+
{
72+
options ??= new StorageWriteOptions();
73+
var response = await container.GetBlobClient(path).UploadAsync(content, new BlobUploadOptions
74+
{
75+
Conditions = Conditions(options),
76+
HttpHeaders = Headers(options),
77+
Metadata = Copy(options.Metadata),
78+
TransferOptions = new global::Azure.Storage.StorageTransferOptions { MaximumConcurrency = 1, InitialTransferSize = 4 * 1024 * 1024, MaximumTransferSize = 4 * 1024 * 1024 }
79+
}, cancellationToken);
80+
return await ReadWrittenInfoAsync(path, response.Value.ETag, cancellationToken);
81+
});
82+
83+
public Task SetObjectMetadataAsync(string path, IReadOnlyDictionary<string, string> metadata, string? ifMatch = null, CancellationToken cancellationToken = default) => ExecuteAsync(async () =>
84+
{
85+
await container.GetBlobClient(path).SetMetadataAsync(Copy(metadata), new BlobRequestConditions { IfMatch = ETagOrNull(ifMatch) }, cancellationToken);
86+
return true;
87+
});
88+
89+
public Task<bool> DeleteObjectIfExistsAsync(string path, bool includeSnapshots = false, CancellationToken cancellationToken = default) => ExecuteAsync(async () =>
90+
(await container.GetBlobClient(path).DeleteIfExistsAsync(includeSnapshots ? DeleteSnapshotsOption.IncludeSnapshots : DeleteSnapshotsOption.None,
91+
cancellationToken: cancellationToken)).Value);
92+
93+
public Task<StorageObjectPage> ListObjectsAsync(string? prefix = null, string? continuationToken = null, int pageSize = 100, CancellationToken cancellationToken = default) => ExecuteAsync(async () =>
94+
{
95+
ArgumentOutOfRangeException.ThrowIfLessThan(pageSize, 1);
96+
ArgumentOutOfRangeException.ThrowIfGreaterThan(pageSize, 5000);
97+
await foreach (var page in container.GetBlobsAsync(new GetBlobsOptions { Prefix = prefix, Traits = BlobTraits.Metadata }, cancellationToken)
98+
.AsPages(continuationToken, pageSize))
99+
{
100+
return new StorageObjectPage(page.Values.Select(item => new StorageObjectInfo(item.Name,
101+
item.Properties.ETag?.ToString() ?? throw new InvalidDataException("Object listing returned no ETag."),
102+
item.Properties.ContentLength ?? throw new InvalidDataException("Object listing returned no length."),
103+
item.Properties.ContentType, item.Properties.ContentEncoding, new Dictionary<string, string>(item.Metadata), item.Properties.LastModified)).ToArray(), page.ContinuationToken);
104+
}
105+
return new StorageObjectPage([], null);
106+
});
107+
108+
public Task StagePartAsync(string path, string partId, Stream content, CancellationToken cancellationToken = default) => ExecuteAsync(async () =>
109+
{
110+
await container.GetBlockBlobClient(path).StageBlockAsync(partId, content, cancellationToken: cancellationToken);
111+
return true;
112+
});
113+
114+
public Task<StorageObjectInfo> CommitPartsAsync(string path, IReadOnlyList<string> partIds, StorageWriteOptions? options = null, CancellationToken cancellationToken = default) => ExecuteAsync(async () =>
115+
{
116+
options ??= new StorageWriteOptions();
117+
var response = await container.GetBlockBlobClient(path).CommitBlockListAsync(partIds, new CommitBlockListOptions
118+
{
119+
Conditions = Conditions(options),
120+
HttpHeaders = Headers(options),
121+
Metadata = Copy(options.Metadata)
122+
}, cancellationToken);
123+
return await ReadWrittenInfoAsync(path, response.Value.ETag, cancellationToken);
124+
});
125+
126+
private async Task<StorageObjectInfo> ReadWrittenInfoAsync(string path, ETag etag, CancellationToken cancellationToken) =>
127+
Info(path, (await container.GetBlobClient(path).GetPropertiesAsync(new BlobRequestConditions { IfMatch = etag }, cancellationToken)).Value);
128+
129+
private static StorageObjectInfo Info(string path, BlobProperties value) => new(path, value.ETag.ToString(), value.ContentLength,
130+
value.ContentType, value.ContentEncoding, new Dictionary<string, string>(value.Metadata), value.LastModified);
131+
132+
private static BlobRequestConditions Conditions(StorageWriteOptions options)
133+
{
134+
if (options.IfAbsent && options.IfMatch is not null) throw new ArgumentException("IfAbsent and IfMatch are mutually exclusive.");
135+
return new BlobRequestConditions { IfMatch = ETagOrNull(options.IfMatch), IfNoneMatch = options.IfAbsent ? ETag.All : null };
136+
}
137+
138+
private static BlobHttpHeaders Headers(StorageWriteOptions options) => new() { ContentType = options.ContentType, ContentEncoding = options.ContentEncoding };
139+
private static ETag? ETagOrNull(string? value) => value is null ? null : new ETag(value);
140+
private static Dictionary<string, string>? Copy(IReadOnlyDictionary<string, string>? value) => value is null ? null : new(value);
141+
142+
private static async Task<T> ExecuteAsync<T>(Func<Task<T>> operation)
143+
{
144+
try { return await operation(); }
145+
catch (RequestFailedException exception)
146+
{
147+
throw new StorageOperationException("Object storage operation failed.", exception.Status, exception, exception.ErrorCode);
148+
}
149+
}
150+
}

‎Storages/ManagedCode.Storage.Azure/AzureStorage.ObjectOperations.cs‎

Lines changed: 53 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,53 @@
1+
using System;
2+
using System.Collections.Generic;
3+
using System.IO;
4+
using System.Threading;
5+
using System.Threading.Tasks;
6+
using ManagedCode.Storage.Core.Objects;
7+
8+
namespace ManagedCode.Storage.Azure;
9+
10+
public partial class AzureStorage
11+
{
12+
private AzureObjectOperations ObjectOperations => new(StorageClient);
13+
public Uri ContainerUri => ObjectOperations.ContainerUri;
14+
15+
public Task<StorageContainerInfo> GetContainerInfoAsync(CancellationToken cancellationToken = default) =>
16+
ObjectOperations.GetContainerInfoAsync(cancellationToken);
17+
18+
public Task CreatePrivateContainerAsync(IReadOnlyDictionary<string, string>? metadata = null, CancellationToken cancellationToken = default) =>
19+
ObjectOperations.CreatePrivateContainerAsync(metadata, cancellationToken);
20+
21+
public Task SetContainerMetadataAsync(IReadOnlyDictionary<string, string> metadata, CancellationToken cancellationToken = default) =>
22+
ObjectOperations.SetContainerMetadataAsync(metadata, cancellationToken);
23+
24+
public Task<bool> DeleteContainerIfExistsAsync(CancellationToken cancellationToken = default) =>
25+
ObjectOperations.DeleteContainerIfExistsAsync(cancellationToken);
26+
27+
public Task<StorageObjectInfo> GetObjectInfoAsync(string path, CancellationToken cancellationToken = default) =>
28+
ObjectOperations.GetObjectInfoAsync(path, cancellationToken);
29+
30+
public Task<bool> ObjectExistsAsync(string path, CancellationToken cancellationToken = default) =>
31+
ObjectOperations.ObjectExistsAsync(path, cancellationToken);
32+
33+
public Task<Stream> OpenObjectReadAsync(string path, StorageReadOptions? options = null, CancellationToken cancellationToken = default) =>
34+
ObjectOperations.OpenObjectReadAsync(path, options, cancellationToken);
35+
36+
public Task<StorageObjectInfo> WriteObjectAsync(string path, Stream content, StorageWriteOptions? options = null, CancellationToken cancellationToken = default) =>
37+
ObjectOperations.WriteObjectAsync(path, content, options, cancellationToken);
38+
39+
public Task SetObjectMetadataAsync(string path, IReadOnlyDictionary<string, string> metadata, string? ifMatch = null, CancellationToken cancellationToken = default) =>
40+
ObjectOperations.SetObjectMetadataAsync(path, metadata, ifMatch, cancellationToken);
41+
42+
public Task<bool> DeleteObjectIfExistsAsync(string path, bool includeSnapshots = false, CancellationToken cancellationToken = default) =>
43+
ObjectOperations.DeleteObjectIfExistsAsync(path, includeSnapshots, cancellationToken);
44+
45+
public Task<StorageObjectPage> ListObjectsAsync(string? prefix = null, string? continuationToken = null, int pageSize = 100, CancellationToken cancellationToken = default) =>
46+
ObjectOperations.ListObjectsAsync(prefix, continuationToken, pageSize, cancellationToken);
47+
48+
public Task StagePartAsync(string path, string partId, Stream content, CancellationToken cancellationToken = default) =>
49+
ObjectOperations.StagePartAsync(path, partId, content, cancellationToken);
50+
51+
public Task<StorageObjectInfo> CommitPartsAsync(string path, IReadOnlyList<string> partIds, StorageWriteOptions? options = null, CancellationToken cancellationToken = default) =>
52+
ObjectOperations.CommitPartsAsync(path, partIds, options, cancellationToken);
53+
}

‎Storages/ManagedCode.Storage.Azure/AzureStorage.cs‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -19,8 +19,8 @@
1919

2020
namespace ManagedCode.Storage.Azure;
2121

22-
public class AzureStorage(IAzureStorageOptions options, ILogger<AzureStorage>? logger = default)
23-
: BaseStorage<BlobContainerClient, IAzureStorageOptions>(options), IAzureStorage
22+
public partial class AzureStorage(IAzureStorageOptions options, ILogger<AzureStorage>? logger = default)
23+
: BaseStorage<BlobContainerClient, IAzureStorageOptions>(options), IAzureStorage, ManagedCode.Storage.Core.Objects.IMultipartObjectStorage
2424
{
2525
private static readonly StorageTransferOptions DefaultUploadTransferOptions = new()
2626
{
@@ -169,7 +169,7 @@ protected override BlobContainerClient CreateStorageClient()
169169
azureStorageOptions.OriginalOptions),
170170

171171
AzureStorageCredentialsOptions azureStorageCredentialsOptions => new BlobContainerClient(
172-
new Uri($"https://{azureStorageCredentialsOptions.AccountName}.blob.core.windows.net/{azureStorageCredentialsOptions.ContainerName}"),
172+
new Uri(azureStorageCredentialsOptions.ServiceUri ?? new Uri($"https://{azureStorageCredentialsOptions.AccountName}.blob.core.windows.net/"), azureStorageCredentialsOptions.ContainerName),
173173
azureStorageCredentialsOptions.Credentials ?? throw new InvalidOperationException($"{nameof(AzureStorageCredentialsOptions.Credentials)} must be provided."),
174174
azureStorageCredentialsOptions.OriginalOptions),
175175

‎Storages/ManagedCode.Storage.Azure/AzureStorageConnection.cs‎

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,39 @@
1+
using System;
2+
using Azure.Core;
3+
using ManagedCode.Storage.Azure.Options;
4+
using ManagedCode.Storage.Core;
5+
6+
namespace ManagedCode.Storage.Azure;
7+
8+
/// <summary>Creates container-scoped storage without exposing native storage clients.</summary>
9+
public static class AzureStorageConnection
10+
{
11+
public static IStorage Create(string connection, string containerName, TokenCredential? credential = null)
12+
{
13+
ArgumentException.ThrowIfNullOrWhiteSpace(connection);
14+
ArgumentException.ThrowIfNullOrWhiteSpace(containerName);
15+
if (!Uri.TryCreate(connection, UriKind.Absolute, out var endpoint))
16+
{
17+
return new AzureStorage(new AzureStorageOptions
18+
{
19+
ConnectionString = connection,
20+
Container = containerName,
21+
CreateContainerIfNotExists = false
22+
});
23+
}
24+
25+
ArgumentNullException.ThrowIfNull(credential);
26+
if (endpoint.Scheme != Uri.UriSchemeHttps || endpoint.AbsolutePath != "/" ||
27+
!string.IsNullOrEmpty(endpoint.Query) || !string.IsNullOrEmpty(endpoint.Fragment))
28+
throw new ArgumentException("Storage service endpoint must be an HTTPS origin.", nameof(connection));
29+
30+
return new AzureStorage(new AzureStorageCredentialsOptions
31+
{
32+
ServiceUri = endpoint,
33+
ContainerName = containerName,
34+
Container = containerName,
35+
Credentials = credential,
36+
CreateContainerIfNotExists = false
37+
});
38+
}
39+
}

0 commit comments

Comments
 (0)