diff --git a/src/Directory.Packages.props b/src/Directory.Packages.props index 30d13cbed0..cfdce8ef0a 100644 --- a/src/Directory.Packages.props +++ b/src/Directory.Packages.props @@ -6,9 +6,11 @@ + + @@ -76,6 +78,8 @@ + + diff --git a/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlPersistence.cs b/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlPersistence.cs index dabe754e30..9e807fc358 100644 --- a/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlPersistence.cs +++ b/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlPersistence.cs @@ -12,7 +12,7 @@ public void AddPersistence(IServiceCollection services) { RegisterSettings(services); ConfigureDbContext(services); - RegisterDataStores(services); + RegisterDataStores(services, settings); services.AddSingleton(); } @@ -23,6 +23,7 @@ public void AddInstaller(IServiceCollection services) ConfigureDbContext(services); services.AddScoped(); + RegisterBodyStorageInstaller(services, settings); } void RegisterSettings(IServiceCollection services) diff --git a/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlPersistenceConfiguration.cs b/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlPersistenceConfiguration.cs index d6e8705dd3..2418cef18e 100644 --- a/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlPersistenceConfiguration.cs +++ b/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlPersistenceConfiguration.cs @@ -15,6 +15,6 @@ public override IPersistence Create(PersistenceSettings settings) return new PostgreSqlPersistence((PostgreSqlPersisterSettings)settings); } - protected override EFPersisterSettings CreateSettings(string connectionString) => - new PostgreSqlPersisterSettings { ConnectionString = connectionString }; + protected override EFPersisterSettings CreateSettings(string connectionString, BodyStorageSettings bodyStorage) => + new PostgreSqlPersisterSettings { ConnectionString = connectionString, BodyStorage = bodyStorage }; } diff --git a/src/ServiceControl.Persistence.EFCore.PostgreSql/persistence.manifest b/src/ServiceControl.Persistence.EFCore.PostgreSql/persistence.manifest index 925712ab78..8e667275c6 100644 --- a/src/ServiceControl.Persistence.EFCore.PostgreSql/persistence.manifest +++ b/src/ServiceControl.Persistence.EFCore.PostgreSql/persistence.manifest @@ -13,10 +13,58 @@ "Name": "ServiceControl/Database/CommandTimeout", "Mandatory": false }, + { + "Name": "ServiceControl/MessageBody/StorageType", + "Mandatory": true + }, { "Name": "ServiceControl/MessageBody/StoragePath", "Mandatory": false }, + { + "Name": "ServiceControl/MessageBody/Azure/ConnectionString", + "Mandatory": false + }, + { + "Name": "ServiceControl/MessageBody/Azure/ServiceUri", + "Mandatory": false + }, + { + "Name": "ServiceControl/MessageBody/Azure/ManagedIdentityClientId", + "Mandatory": false + }, + { + "Name": "ServiceControl/MessageBody/Azure/AuthorityHost", + "Mandatory": false + }, + { + "Name": "ServiceControl/MessageBody/Azure/ContainerName", + "Mandatory": false + }, + { + "Name": "ServiceControl/MessageBody/S3/BucketName", + "Mandatory": false + }, + { + "Name": "ServiceControl/MessageBody/S3/KeyPrefix", + "Mandatory": false + }, + { + "Name": "ServiceControl/MessageBody/S3/Region", + "Mandatory": false + }, + { + "Name": "ServiceControl/MessageBody/S3/ServiceUrl", + "Mandatory": false + }, + { + "Name": "ServiceControl/MessageBody/S3/AccessKeyId", + "Mandatory": false + }, + { + "Name": "ServiceControl/MessageBody/S3/SecretAccessKey", + "Mandatory": false + }, { "Name": "ServiceControl/MessageBody/MinCompressionSize", "Mandatory": false diff --git a/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerPersistence.cs b/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerPersistence.cs index 5e9336e747..3c6d1384b3 100644 --- a/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerPersistence.cs +++ b/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerPersistence.cs @@ -12,7 +12,7 @@ public void AddPersistence(IServiceCollection services) { RegisterSettings(services); ConfigureDbContext(services); - RegisterDataStores(services); + RegisterDataStores(services, settings); services.AddSingleton(); } @@ -23,6 +23,7 @@ public void AddInstaller(IServiceCollection services) ConfigureDbContext(services); services.AddScoped(); + RegisterBodyStorageInstaller(services, settings); } void RegisterSettings(IServiceCollection services) diff --git a/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerPersistenceConfiguration.cs b/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerPersistenceConfiguration.cs index 49d5d8e415..eb1d9e3d18 100644 --- a/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerPersistenceConfiguration.cs +++ b/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerPersistenceConfiguration.cs @@ -15,6 +15,6 @@ public override IPersistence Create(PersistenceSettings settings) return new SqlServerPersistence((SqlServerPersisterSettings)settings); } - protected override EFPersisterSettings CreateSettings(string connectionString) => - new SqlServerPersisterSettings { ConnectionString = connectionString }; + protected override EFPersisterSettings CreateSettings(string connectionString, BodyStorageSettings bodyStorage) => + new SqlServerPersisterSettings { ConnectionString = connectionString, BodyStorage = bodyStorage }; } diff --git a/src/ServiceControl.Persistence.EFCore.SqlServer/persistence.manifest b/src/ServiceControl.Persistence.EFCore.SqlServer/persistence.manifest index aa1b57b88f..51de2f2ee8 100644 --- a/src/ServiceControl.Persistence.EFCore.SqlServer/persistence.manifest +++ b/src/ServiceControl.Persistence.EFCore.SqlServer/persistence.manifest @@ -13,10 +13,58 @@ "Name": "ServiceControl/Database/CommandTimeout", "Mandatory": false }, + { + "Name": "ServiceControl/MessageBody/StorageType", + "Mandatory": true + }, { "Name": "ServiceControl/MessageBody/StoragePath", "Mandatory": false }, + { + "Name": "ServiceControl/MessageBody/Azure/ConnectionString", + "Mandatory": false + }, + { + "Name": "ServiceControl/MessageBody/Azure/ServiceUri", + "Mandatory": false + }, + { + "Name": "ServiceControl/MessageBody/Azure/ManagedIdentityClientId", + "Mandatory": false + }, + { + "Name": "ServiceControl/MessageBody/Azure/AuthorityHost", + "Mandatory": false + }, + { + "Name": "ServiceControl/MessageBody/Azure/ContainerName", + "Mandatory": false + }, + { + "Name": "ServiceControl/MessageBody/S3/BucketName", + "Mandatory": false + }, + { + "Name": "ServiceControl/MessageBody/S3/KeyPrefix", + "Mandatory": false + }, + { + "Name": "ServiceControl/MessageBody/S3/Region", + "Mandatory": false + }, + { + "Name": "ServiceControl/MessageBody/S3/ServiceUrl", + "Mandatory": false + }, + { + "Name": "ServiceControl/MessageBody/S3/AccessKeyId", + "Mandatory": false + }, + { + "Name": "ServiceControl/MessageBody/S3/SecretAccessKey", + "Mandatory": false + }, { "Name": "ServiceControl/MessageBody/MinCompressionSize", "Mandatory": false diff --git a/src/ServiceControl.Persistence.EFCore/Abstractions/AzureBlobBodyStorageSettings.cs b/src/ServiceControl.Persistence.EFCore/Abstractions/AzureBlobBodyStorageSettings.cs new file mode 100644 index 0000000000..0d7aec07ac --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Abstractions/AzureBlobBodyStorageSettings.cs @@ -0,0 +1,28 @@ +namespace ServiceControl.Persistence.EFCore.Abstractions; + +public sealed class AzureBlobBodyStorageSettings : BodyStorageSettings +{ + public const string DefaultContainerName = "error-bodies"; + + public required AzureBlobAuthentication Authentication { get; set; } + public string ContainerName { get; set; } = DefaultContainerName; +} + +// Shared-key and managed-identity auth are mutually exclusive, and the managed identity options are +// meaningless alongside a connection string. +public abstract class AzureBlobAuthentication; + +public sealed class AzureBlobSharedKeyAuthentication : AzureBlobAuthentication +{ + public required string ConnectionString { get; set; } +} + +public sealed class AzureBlobManagedIdentityAuthentication : AzureBlobAuthentication +{ + public required Uri ServiceUri { get; set; } + public string? ClientId { get; set; } + + // Steers the login endpoint for sovereign clouds; when unset the SDK honours the + // AZURE_AUTHORITY_HOST environment variable. + public Uri? AuthorityHost { get; set; } +} diff --git a/src/ServiceControl.Persistence.EFCore/Abstractions/BasePersistence.cs b/src/ServiceControl.Persistence.EFCore/Abstractions/BasePersistence.cs index c4d67e60dc..ef5bd0aa00 100644 --- a/src/ServiceControl.Persistence.EFCore/Abstractions/BasePersistence.cs +++ b/src/ServiceControl.Persistence.EFCore/Abstractions/BasePersistence.cs @@ -1,10 +1,12 @@ namespace ServiceControl.Persistence.EFCore.Abstractions; using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.DependencyInjection.Extensions; using NServiceBus.Unicast.Subscriptions.MessageDrivenSubscriptions; using Particular.LicensingComponent.Persistence; using ServiceControl.Operations.BodyStorage; using ServiceControl.Persistence.EFCore.Implementation; +using ServiceControl.Persistence.EFCore.Implementation.BodyStorage; using ServiceControl.Persistence.EFCore.Implementation.UnitOfWork; using ServiceControl.Persistence.EFCore.Infrastructure; using ServiceControl.Persistence.MessageRedirects; @@ -13,7 +15,7 @@ namespace ServiceControl.Persistence.EFCore.Abstractions; public abstract class BasePersistence { - protected static void RegisterDataStores(IServiceCollection services) + protected static void RegisterDataStores(IServiceCollection services, EFPersisterSettings settings) { services.AddSingleton(TimeProvider.System); services.AddSingleton(); @@ -51,6 +53,50 @@ protected static void RegisterDataStores(IServiceCollection services) services.AddSingleton(); - services.AddSingleton(); + RegisterBodyStorage(services, settings); + } + + // Settings are registered under their concrete type so each store resolves only what it can act on. + static void RegisterBodyStorage(IServiceCollection services, EFPersisterSettings settings) + { + switch (settings.BodyStorage) + { + case FileSystemBodyStorageSettings fileSystem: + services.TryAddSingleton(fileSystem); + services.AddSingleton(); + break; + case AzureBlobBodyStorageSettings azureBlob: + services.TryAddSingleton(azureBlob); + services.AddSingleton(); + break; + case S3BodyStorageSettings s3: + services.TryAddSingleton(s3); + services.AddSingleton(); + break; + default: + throw new ArgumentOutOfRangeException(nameof(settings), settings.BodyStorage, "Unknown body storage type."); + } + } + + // Only stores needing setup-time provisioning register an installer; SetupCommand skips when none is. + protected static void RegisterBodyStorageInstaller(IServiceCollection services, EFPersisterSettings settings) + { + switch (settings.BodyStorage) + { + case FileSystemBodyStorageSettings fileSystem: + services.TryAddSingleton(fileSystem); + services.AddScoped(); + break; + case AzureBlobBodyStorageSettings azureBlob: + services.TryAddSingleton(azureBlob); + services.AddScoped(); + break; + case S3BodyStorageSettings s3: + services.TryAddSingleton(s3); + services.AddScoped(); + break; + default: + throw new ArgumentOutOfRangeException(nameof(settings), settings.BodyStorage, "Unknown body storage type."); + } } } diff --git a/src/ServiceControl.Persistence.EFCore/Abstractions/BodyStorageSettings.cs b/src/ServiceControl.Persistence.EFCore/Abstractions/BodyStorageSettings.cs new file mode 100644 index 0000000000..dd7ecb9e67 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Abstractions/BodyStorageSettings.cs @@ -0,0 +1,17 @@ +namespace ServiceControl.Persistence.EFCore.Abstractions; + +/// +/// Settings for the selected body storage type. +/// +/// +/// One subclass per storage type, so a store only ever receives the settings it can act on and the +/// configuration layer's validation is carried by the type rather than re-asserted at the point of use. +/// +public abstract class BodyStorageSettings +{ + public const int DefaultMinCompressionSize = 4096; + public const int DefaultMaxBodySizeToStore = 102400; // 100 kb + + public int MinCompressionSize { get; set; } = DefaultMinCompressionSize; + public int MaxBodySizeToStore { get; set; } = DefaultMaxBodySizeToStore; +} diff --git a/src/ServiceControl.Persistence.EFCore/Abstractions/BodyStorageType.cs b/src/ServiceControl.Persistence.EFCore/Abstractions/BodyStorageType.cs new file mode 100644 index 0000000000..9485cab036 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Abstractions/BodyStorageType.cs @@ -0,0 +1,8 @@ +namespace ServiceControl.Persistence.EFCore.Abstractions; + +public enum BodyStorageType +{ + FileSystem, + AzureBlob, + S3 +} diff --git a/src/ServiceControl.Persistence.EFCore/Abstractions/EFPersistenceConfigurationBase.cs b/src/ServiceControl.Persistence.EFCore/Abstractions/EFPersistenceConfigurationBase.cs index fba0f10e3d..bb15945cb0 100644 --- a/src/ServiceControl.Persistence.EFCore/Abstractions/EFPersistenceConfigurationBase.cs +++ b/src/ServiceControl.Persistence.EFCore/Abstractions/EFPersistenceConfigurationBase.cs @@ -8,7 +8,19 @@ public abstract class EFPersistenceConfigurationBase : IPersistenceConfiguration { const string ConnectionStringKey = "Database/ConnectionString"; const string CommandTimeoutKey = "Database/CommandTimeout"; + const string BodyStorageTypeKey = "MessageBody/StorageType"; const string MessageBodyStoragePathKey = "MessageBody/StoragePath"; + const string AzureConnectionStringKey = "MessageBody/Azure/ConnectionString"; + const string AzureServiceUriKey = "MessageBody/Azure/ServiceUri"; + const string AzureManagedIdentityClientIdKey = "MessageBody/Azure/ManagedIdentityClientId"; + const string AzureAuthorityHostKey = "MessageBody/Azure/AuthorityHost"; + const string AzureContainerNameKey = "MessageBody/Azure/ContainerName"; + const string S3BucketNameKey = "MessageBody/S3/BucketName"; + const string S3KeyPrefixKey = "MessageBody/S3/KeyPrefix"; + const string S3RegionKey = "MessageBody/S3/Region"; + const string S3ServiceUrlKey = "MessageBody/S3/ServiceUrl"; + const string S3AccessKeyIdKey = "MessageBody/S3/AccessKeyId"; + const string S3SecretAccessKeyKey = "MessageBody/S3/SecretAccessKey"; const string MinBodySizeForCompressionKey = "MessageBody/MinCompressionSize"; const string MaxBodySizeToStoreKey = "MaxBodySizeToStore"; const string ErrorRetentionPeriodKey = "ErrorRetentionPeriod"; @@ -16,12 +28,11 @@ public abstract class EFPersistenceConfigurationBase : IPersistenceConfiguration public PersistenceSettings CreateSettings(SettingsRootNamespace settingsRootNamespace) { - var settings = CreateSettings(GetRequiredSetting(settingsRootNamespace, ConnectionStringKey)); + var settings = CreateSettings( + GetRequiredSetting(settingsRootNamespace, ConnectionStringKey), + CreateBodyStorageSettings(settingsRootNamespace)); settings.CommandTimeout = SettingsReader.Read(settingsRootNamespace, CommandTimeoutKey, EFPersisterSettings.DefaultCommandTimeout); - settings.MessageBodyStoragePath = SettingsReader.Read(settingsRootNamespace, MessageBodyStoragePathKey); - settings.MinBodySizeForCompression = SettingsReader.Read(settingsRootNamespace, MinBodySizeForCompressionKey, EFPersisterSettings.DefaultMinBodySizeForCompression); - settings.MaxBodySizeToStore = ReadMaxBodySizeToStore(settingsRootNamespace); settings.ErrorRetentionPeriod = GetRequiredSetting(settingsRootNamespace, ErrorRetentionPeriodKey); settings.EnableFullTextSearchOnBodies = SettingsReader.Read(settingsRootNamespace, EnableFullTextSearchOnBodiesKey, true); @@ -30,18 +41,135 @@ public PersistenceSettings CreateSettings(SettingsRootNamespace settingsRootName public abstract IPersistence Create(PersistenceSettings settings); - protected abstract EFPersisterSettings CreateSettings(string connectionString); + protected abstract EFPersisterSettings CreateSettings(string connectionString, BodyStorageSettings bodyStorage); + + static BodyStorageSettings CreateBodyStorageSettings(SettingsRootNamespace settingsRootNamespace) + { + var storageType = ReadBodyStorageType(settingsRootNamespace); + + BodyStorageSettings bodyStorage = storageType switch + { + BodyStorageType.FileSystem => new FileSystemBodyStorageSettings + { + StoragePath = GetRequiredSetting(settingsRootNamespace, MessageBodyStoragePathKey) + }, + BodyStorageType.AzureBlob => CreateAzureBlobSettings(settingsRootNamespace), + BodyStorageType.S3 => CreateS3Settings(settingsRootNamespace), + _ => throw new ArgumentOutOfRangeException(nameof(settingsRootNamespace), storageType, "Unknown body storage type.") + }; + + bodyStorage.MinCompressionSize = SettingsReader.Read(settingsRootNamespace, MinBodySizeForCompressionKey, BodyStorageSettings.DefaultMinCompressionSize); + bodyStorage.MaxBodySizeToStore = ReadMaxBodySizeToStore(settingsRootNamespace); + + return bodyStorage; + } + + static S3BodyStorageSettings CreateS3Settings(SettingsRootNamespace settingsRootNamespace) => + new() + { + BucketName = GetRequiredSetting(settingsRootNamespace, S3BucketNameKey), + KeyPrefix = SettingsReader.Read(settingsRootNamespace, S3KeyPrefixKey, S3BodyStorageSettings.DefaultKeyPrefix), + Region = SettingsReader.Read(settingsRootNamespace, S3RegionKey), + ServiceUrl = SettingsReader.Read(settingsRootNamespace, S3ServiceUrlKey), + Credentials = ReadS3Credentials(settingsRootNamespace) + }; + + static S3StaticCredentials? ReadS3Credentials(SettingsRootNamespace settingsRootNamespace) + { + var accessKeyId = SettingsReader.Read(settingsRootNamespace, S3AccessKeyIdKey); + var secretAccessKey = SettingsReader.Read(settingsRootNamespace, S3SecretAccessKeyKey); + var hasAccessKeyId = !string.IsNullOrWhiteSpace(accessKeyId); + var hasSecretAccessKey = !string.IsNullOrWhiteSpace(secretAccessKey); + + if (hasAccessKeyId != hasSecretAccessKey) + { + throw new Exception($"S3 body storage requires both {S3AccessKeyIdKey} and {S3SecretAccessKeyKey} together, or neither to use the default AWS credential chain (IAM role)."); + } + + return hasAccessKeyId + ? new S3StaticCredentials { AccessKeyId = accessKeyId, SecretAccessKey = secretAccessKey } + : null; + } + + static AzureBlobBodyStorageSettings CreateAzureBlobSettings(SettingsRootNamespace settingsRootNamespace) => + new() + { + Authentication = ReadAzureBlobAuthentication(settingsRootNamespace), + ContainerName = SettingsReader.Read(settingsRootNamespace, AzureContainerNameKey, AzureBlobBodyStorageSettings.DefaultContainerName) + }; + + static AzureBlobAuthentication ReadAzureBlobAuthentication(SettingsRootNamespace settingsRootNamespace) + { + var connectionString = SettingsReader.Read(settingsRootNamespace, AzureConnectionStringKey); + var serviceUri = SettingsReader.Read(settingsRootNamespace, AzureServiceUriKey); + + var hasConnectionString = !string.IsNullOrWhiteSpace(connectionString); + var hasServiceUri = !string.IsNullOrWhiteSpace(serviceUri); + + // A connection string carries shared-key/SAS auth; a service URI uses managed identity. They + // are mutually exclusive, and exactly one must be provided. + if (hasConnectionString == hasServiceUri) + { + throw new Exception($"Azure Blob body storage requires exactly one of {AzureConnectionStringKey} (shared key / SAS) or {AzureServiceUriKey} (managed identity). {(hasConnectionString ? "Both were set." : "Neither was set.")}"); + } + + if (hasConnectionString) + { + return new AzureBlobSharedKeyAuthentication { ConnectionString = connectionString }; + } + + return new AzureBlobManagedIdentityAuthentication + { + ServiceUri = ReadAbsoluteUri(serviceUri, AzureServiceUriKey), + ClientId = SettingsReader.Read(settingsRootNamespace, AzureManagedIdentityClientIdKey), + AuthorityHost = ReadAzureAuthorityHost(settingsRootNamespace) + }; + } + + static Uri? ReadAzureAuthorityHost(SettingsRootNamespace settingsRootNamespace) + { + var raw = SettingsReader.Read(settingsRootNamespace, AzureAuthorityHostKey); + + return string.IsNullOrWhiteSpace(raw) ? null : ReadAbsoluteUri(raw, AzureAuthorityHostKey); + } + + static Uri ReadAbsoluteUri(string raw, string key) + { + if (!Uri.TryCreate(raw, UriKind.Absolute, out var uri)) + { + throw new Exception($"Setting {key} value '{raw}' is not a valid absolute URI."); + } + + return uri; + } + + static BodyStorageType ReadBodyStorageType(SettingsRootNamespace settingsRootNamespace) + { + var raw = SettingsReader.Read(settingsRootNamespace, BodyStorageTypeKey); + + if (string.IsNullOrWhiteSpace(raw)) + { + throw new Exception($"Setting {BodyStorageTypeKey} is required. Valid values: {string.Join(", ", Enum.GetNames())}."); + } + + if (!Enum.TryParse(raw, ignoreCase: true, out var value) || !Enum.IsDefined(value)) + { + throw new Exception($"Setting {BodyStorageTypeKey} value '{raw}' is not valid. Valid values: {string.Join(", ", Enum.GetNames())}."); + } + + return value; + } static int ReadMaxBodySizeToStore(SettingsRootNamespace settingsRootNamespace) { - var maxBodySizeToStore = SettingsReader.Read(settingsRootNamespace, MaxBodySizeToStoreKey, EFPersisterSettings.DefaultMaxBodySizeToStore); + var maxBodySizeToStore = SettingsReader.Read(settingsRootNamespace, MaxBodySizeToStoreKey, BodyStorageSettings.DefaultMaxBodySizeToStore); if (maxBodySizeToStore <= 0) { LoggerUtil.CreateStaticLogger() - .LogError("MaxBodySizeToStore setting is invalid, 1 is the minimum value. Defaulting to {MaxBodySizeToStoreDefault}", EFPersisterSettings.DefaultMaxBodySizeToStore); + .LogError("MaxBodySizeToStore setting is invalid, 1 is the minimum value. Defaulting to {MaxBodySizeToStoreDefault}", BodyStorageSettings.DefaultMaxBodySizeToStore); - return EFPersisterSettings.DefaultMaxBodySizeToStore; + return BodyStorageSettings.DefaultMaxBodySizeToStore; } return maxBodySizeToStore; diff --git a/src/ServiceControl.Persistence.EFCore/Abstractions/EFPersisterSettings.cs b/src/ServiceControl.Persistence.EFCore/Abstractions/EFPersisterSettings.cs index 3b919453c4..1311e327e2 100644 --- a/src/ServiceControl.Persistence.EFCore/Abstractions/EFPersisterSettings.cs +++ b/src/ServiceControl.Persistence.EFCore/Abstractions/EFPersisterSettings.cs @@ -5,15 +5,11 @@ public abstract class EFPersisterSettings : PersistenceSettings public static readonly TimeSpan MigrationCommandTimeout = TimeSpan.FromMinutes(40); public const int DefaultCommandTimeout = 30; - public const int DefaultMinBodySizeForCompression = 4096; - public const int DefaultMaxBodySizeToStore = 102400; // 100 kb public required string ConnectionString { get; set; } public int CommandTimeout { get; set; } = DefaultCommandTimeout; public TimeSpan ErrorRetentionPeriod { get; set; } - public string? MessageBodyStoragePath { get; set; } - public int MinBodySizeForCompression { get; set; } = DefaultMinBodySizeForCompression; - public int MaxBodySizeToStore { get; set; } = DefaultMaxBodySizeToStore; + public required BodyStorageSettings BodyStorage { get; set; } public int MaxRetryCount { get; set; } = 5; public int MaxRetryDelayInSeconds { get; set; } = 30; public bool EnableSensitiveDataLogging { get; set; } diff --git a/src/ServiceControl.Persistence.EFCore/Abstractions/FileSystemBodyStorageSettings.cs b/src/ServiceControl.Persistence.EFCore/Abstractions/FileSystemBodyStorageSettings.cs new file mode 100644 index 0000000000..6045cfd7a5 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Abstractions/FileSystemBodyStorageSettings.cs @@ -0,0 +1,6 @@ +namespace ServiceControl.Persistence.EFCore.Abstractions; + +public sealed class FileSystemBodyStorageSettings : BodyStorageSettings +{ + public required string StoragePath { get; set; } +} diff --git a/src/ServiceControl.Persistence.EFCore/Abstractions/S3BodyStorageSettings.cs b/src/ServiceControl.Persistence.EFCore/Abstractions/S3BodyStorageSettings.cs new file mode 100644 index 0000000000..97af894a07 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Abstractions/S3BodyStorageSettings.cs @@ -0,0 +1,20 @@ +namespace ServiceControl.Persistence.EFCore.Abstractions; + +public sealed class S3BodyStorageSettings : BodyStorageSettings +{ + public const string DefaultKeyPrefix = "error-bodies/"; + + public required string BucketName { get; set; } + public string KeyPrefix { get; set; } = DefaultKeyPrefix; + public string? Region { get; set; } + public string? ServiceUrl { get; set; } + + // Null resolves the ambient IAM role through the SDK's default credential chain. + public S3StaticCredentials? Credentials { get; set; } +} + +public sealed class S3StaticCredentials +{ + public required string AccessKeyId { get; set; } + public required string SecretAccessKey { get; set; } +} diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/BodyStorage.cs b/src/ServiceControl.Persistence.EFCore/Implementation/BodyStorage.cs deleted file mode 100644 index e1caa893a7..0000000000 --- a/src/ServiceControl.Persistence.EFCore/Implementation/BodyStorage.cs +++ /dev/null @@ -1,9 +0,0 @@ -namespace ServiceControl.Persistence.EFCore.Implementation; - -using ServiceControl.Operations.BodyStorage; - -public class BodyStorage : IBodyStorage -{ - public Task TryFetch(string bodyId) => - throw new NotImplementedException(); -} diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/BodyStorage/AzureBlobBodyStorageInstaller.cs b/src/ServiceControl.Persistence.EFCore/Implementation/BodyStorage/AzureBlobBodyStorageInstaller.cs new file mode 100644 index 0000000000..d0aa969eb7 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Implementation/BodyStorage/AzureBlobBodyStorageInstaller.cs @@ -0,0 +1,10 @@ +namespace ServiceControl.Persistence.EFCore.Implementation.BodyStorage; + +using ServiceControl.Persistence; +using ServiceControl.Persistence.EFCore.Abstractions; + +public class AzureBlobBodyStorageInstaller(AzureBlobBodyStorageSettings settings) : IBodyStorageInstaller +{ + public Task Provision(CancellationToken cancellationToken = default) => + AzureBlobClientFactory.CreateContainerClient(settings).CreateIfNotExistsAsync(cancellationToken: cancellationToken); +} diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/BodyStorage/AzureBlobBodyStoragePersistence.cs b/src/ServiceControl.Persistence.EFCore/Implementation/BodyStorage/AzureBlobBodyStoragePersistence.cs new file mode 100644 index 0000000000..7744152903 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Implementation/BodyStorage/AzureBlobBodyStoragePersistence.cs @@ -0,0 +1,98 @@ +namespace ServiceControl.Persistence.EFCore.Implementation.BodyStorage; + +using System.IO.Compression; +using System.Net; +using Azure; +using Azure.Storage.Blobs; +using Azure.Storage.Blobs.Models; +using ServiceControl.Persistence.EFCore.Abstractions; +using ServiceControl.Persistence.EFCore.Infrastructure; + +public class AzureBlobBodyStoragePersistence : IBodyStoragePersistence +{ + const string FormatVersion = "1"; + + readonly BlobContainerClient container; + readonly int minBodySizeForCompression; + + public AzureBlobBodyStoragePersistence(AzureBlobBodyStorageSettings settings) + { + container = AzureBlobClientFactory.CreateContainerClient(settings); + minBodySizeForCompression = settings.MinCompressionSize; + } + + public async Task WriteBody(string bodyId, ReadOnlyMemory body, string contentType, CancellationToken cancellationToken = default) + { + var blob = container.GetBlobClient(bodyId); + + var compressed = body.Length >= minBodySizeForCompression ? BodyCompression.TryCompress(body) : null; + var data = compressed is null ? BinaryData.FromBytes(body) : BinaryData.FromBytes(compressed); + + var options = new BlobUploadOptions + { + // Bodies are immutable, so create the blob only if it does not already exist. + Conditions = new BlobRequestConditions { IfNoneMatch = ETag.All }, + Metadata = new Dictionary + { + ["FormatVersion"] = FormatVersion, + ["ContentType"] = Uri.EscapeDataString(contentType), + ["BodySize"] = body.Length.ToString(), + ["IsCompressed"] = (compressed is not null).ToString() + } + }; + + try + { + await blob.UploadAsync(data, options, cancellationToken); + } + catch (RequestFailedException ex) when (ex.Status is (int)HttpStatusCode.Conflict or (int)HttpStatusCode.PreconditionFailed) + { + // A blob for this immutable body already exists. + } + } + + public async Task ReadBody(string bodyId, CancellationToken cancellationToken = default) + { + var blob = container.GetBlobClient(bodyId); + + try + { + var content = (await blob.DownloadStreamingAsync(cancellationToken: cancellationToken)).Value; + try + { + var metadata = content.Details.Metadata; + + if (metadata.TryGetValue("FormatVersion", out var version) && version != FormatVersion) + { + throw new InvalidOperationException($"Unsupported blob format version {version} for {bodyId}."); + } + + var contentType = metadata.TryGetValue("ContentType", out var ct) ? Uri.UnescapeDataString(ct) : "application/octet-stream"; + var bodySize = metadata.TryGetValue("BodySize", out var sizeText) && int.TryParse(sizeText, out var size) ? size : 0; + var isCompressed = metadata.TryGetValue("IsCompressed", out var compressedText) && bool.TryParse(compressedText, out var compressed) && compressed; + Stream stream = isCompressed + ? new ExpectedLengthStream(new BrotliStream(content.Content, CompressionMode.Decompress), bodySize) + : content.Content; + + return new MessageBodyFileResult + { + Stream = stream, + ContentType = contentType, + BodySize = bodySize + }; + } + catch + { + content.Dispose(); + throw; + } + } + catch (RequestFailedException ex) when (ex.Status == (int)HttpStatusCode.NotFound) + { + return null; + } + } + + public Task DeleteBody(string bodyId, CancellationToken cancellationToken = default) => + container.GetBlobClient(bodyId).DeleteIfExistsAsync(cancellationToken: cancellationToken); +} diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/BodyStorage/AzureBlobClientFactory.cs b/src/ServiceControl.Persistence.EFCore/Implementation/BodyStorage/AzureBlobClientFactory.cs new file mode 100644 index 0000000000..6c2ed6308c --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Implementation/BodyStorage/AzureBlobClientFactory.cs @@ -0,0 +1,38 @@ +namespace ServiceControl.Persistence.EFCore.Implementation.BodyStorage; + +using Azure.Core; +using Azure.Identity; +using Azure.Storage.Blobs; +using ServiceControl.Persistence.EFCore.Abstractions; + +static class AzureBlobClientFactory +{ + public static BlobContainerClient CreateContainerClient(AzureBlobBodyStorageSettings settings) + { + var serviceClient = settings.Authentication switch + { + AzureBlobSharedKeyAuthentication sharedKey => new BlobServiceClient(sharedKey.ConnectionString), + AzureBlobManagedIdentityAuthentication managedIdentity => new BlobServiceClient(managedIdentity.ServiceUri, CreateCredential(managedIdentity)), + _ => throw new ArgumentOutOfRangeException(nameof(settings), settings.Authentication, "Unknown Azure Blob authentication.") + }; + + return serviceClient.GetBlobContainerClient(settings.ContainerName); + } + + static TokenCredential CreateCredential(AzureBlobManagedIdentityAuthentication authentication) + { + var options = new DefaultAzureCredentialOptions(); + + if (authentication.AuthorityHost is { } authorityHost) + { + options.AuthorityHost = authorityHost; + } + + if (authentication.ClientId is { Length: > 0 } clientId) + { + options.ManagedIdentityClientId = clientId; + } + + return new DefaultAzureCredential(options); + } +} diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/BodyStorage/BodyCompression.cs b/src/ServiceControl.Persistence.EFCore/Implementation/BodyStorage/BodyCompression.cs new file mode 100644 index 0000000000..2946391dec --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Implementation/BodyStorage/BodyCompression.cs @@ -0,0 +1,39 @@ +namespace ServiceControl.Persistence.EFCore.Implementation.BodyStorage; + +using System.Buffers; +using System.IO.Compression; + +/// +/// Compresses message bodies on their way into cloud storage. +/// +/// +/// Brotli ships in the BCL, so it costs no extra dependency, and at quality 1 it compresses about as +/// quickly as gzip or deflate while producing a smaller payload. The quality is pinned deliberately +/// low because this runs on the ingestion path, where throughput matters more than the last few +/// percent of ratio; window 22 is the .NET default. +/// produces the same format through BrotliStream, which maps CompressionLevel.Fastest onto exactly +/// these parameters, so the two paths stay interchangeable without sharing this buffer-based helper. +/// Decompression is not here on purpose: every read path streams, so nothing needs the whole body in +/// memory at once. +/// +static class BodyCompression +{ + /// + /// Returns the Brotli-compressed bytes, or null when compression fails and the caller should + /// store the body uncompressed. + /// + public static byte[]? TryCompress(ReadOnlyMemory body) + { + var buffer = ArrayPool.Shared.Rent(BrotliEncoder.GetMaxCompressedLength(body.Length)); + try + { + return BrotliEncoder.TryCompress(body.Span, buffer, out var written, quality: 1, window: 22) + ? buffer.AsSpan(0, written).ToArray() + : null; + } + finally + { + ArrayPool.Shared.Return(buffer); + } + } +} diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/BodyStorage/BodyStorage.cs b/src/ServiceControl.Persistence.EFCore/Implementation/BodyStorage/BodyStorage.cs new file mode 100644 index 0000000000..e62ac98259 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Implementation/BodyStorage/BodyStorage.cs @@ -0,0 +1,102 @@ +namespace ServiceControl.Persistence.EFCore.Implementation.BodyStorage; + +using System.Linq.Expressions; +using System.Text; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.DependencyInjection; +using ServiceControl.Operations.BodyStorage; +using ServiceControl.Persistence.EFCore.DbContexts; +using ServiceControl.Persistence.EFCore.Entities; +using ServiceControl.Persistence.EFCore.Infrastructure; + +/// +/// Resolves a message body from wherever it was stored. +/// +/// +/// A body is stored inline in BodyText (small text) or in external storage (binary, or large text +/// where BodyText keeps only a search prefix). External storage is authoritative, so check it first. +/// bodyId is usually a UniqueMessageId (a Guid) but may be a plain MessageId. +/// +public class BodyStorage(IServiceScopeFactory scopeFactory, IBodyStoragePersistence storagePersistence) : DataStoreBase(scopeFactory), IBodyStorage +{ + public async Task TryFetch(string bodyId) + { + var row = await ExecuteWithDbContext(dbContext => ResolveBody(dbContext, bodyId)); + + if (row == null) + { + return null; // No such message: the API turns this into a 404. + } + + // Bodies are immutable per message, so the id is a stable ETag. + var uniqueMessageId = row.UniqueMessageId.ToString(); + + if (row.BodyStoredExternally) + { + var external = await storagePersistence.ReadBody(uniqueMessageId); + + return external == null + ? new MessageBodyStreamResult { HasResult = false } + : new MessageBodyStreamResult + { + HasResult = true, + Stream = external.Stream, + ContentType = external.ContentType, + BodySize = external.BodySize, + Etag = uniqueMessageId + }; + } + + if (row.BodyText != null) + { + var bytes = Encoding.UTF8.GetBytes(row.BodyText); + + return new MessageBodyStreamResult + { + HasResult = true, + Stream = new MemoryStream(bytes, writable: false), + ContentType = row.BodyContentType, + BodySize = bytes.Length, + Etag = uniqueMessageId + }; + } + + return new MessageBodyStreamResult { HasResult = false }; // Message exists but carries no body. + } + + static async Task ResolveBody(ServiceControlDbContext dbContext, string bodyId) + { + if (Guid.TryParse(bodyId, out var uniqueMessageId)) + { + var byUniqueId = await Query(dbContext, message => message.UniqueMessageId == uniqueMessageId); + if (byUniqueId != null) + { + return byUniqueId; + } + } + + return await Query(dbContext, message => message.MessageId == bodyId); + } + + static Task Query(ServiceControlDbContext dbContext, Expression> predicate) => + dbContext.FailedMessages + .AsNoTracking() + .Where(predicate) + .OrderBy(message => message.UniqueMessageId) + .Select(message => new BodyRow + { + UniqueMessageId = message.UniqueMessageId, + BodyText = message.BodyText, + BodyStoredExternally = message.BodyStoredExternally, + BodyContentType = message.BodyContentType + }) + .FirstOrDefaultAsync(); + + sealed class BodyRow + { + public Guid UniqueMessageId { get; init; } + public string? BodyText { get; init; } + public bool BodyStoredExternally { get; init; } + public string? BodyContentType { get; init; } + } +} diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/BodyStorage/FileSystemBodyStorageInstaller.cs b/src/ServiceControl.Persistence.EFCore/Implementation/BodyStorage/FileSystemBodyStorageInstaller.cs new file mode 100644 index 0000000000..2766e08e3d --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Implementation/BodyStorage/FileSystemBodyStorageInstaller.cs @@ -0,0 +1,14 @@ +namespace ServiceControl.Persistence.EFCore.Implementation.BodyStorage; + +using ServiceControl.Persistence; +using ServiceControl.Persistence.EFCore.Abstractions; + +public class FileSystemBodyStorageInstaller(FileSystemBodyStorageSettings settings) : IBodyStorageInstaller +{ + public Task Provision(CancellationToken cancellationToken = default) + { + Directory.CreateDirectory(settings.StoragePath); + + return Task.CompletedTask; + } +} diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/BodyStorage/FileSystemBodyStoragePersistence.cs b/src/ServiceControl.Persistence.EFCore/Implementation/BodyStorage/FileSystemBodyStoragePersistence.cs new file mode 100644 index 0000000000..062534f742 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Implementation/BodyStorage/FileSystemBodyStoragePersistence.cs @@ -0,0 +1,154 @@ +namespace ServiceControl.Persistence.EFCore.Implementation.BodyStorage; + +using System.IO.Compression; +using System.Text; +using ServiceControl.Persistence.EFCore.Abstractions; +using ServiceControl.Persistence.EFCore.Infrastructure; + +/// +/// Stores message bodies as files, each one a small header followed by the body. +/// +/// +/// Bodies are immutable and keyed by bodyId alone, so a re-failure resolves to the same file and an +/// existing one is left untouched. +/// +public class FileSystemBodyStoragePersistence(FileSystemBodyStorageSettings settings) : IBodyStoragePersistence +{ + const int FormatVersion = 1; + + string StoragePath => settings.StoragePath; + + public async Task WriteBody(string bodyId, ReadOnlyMemory body, string contentType, CancellationToken cancellationToken = default) + { + var filePath = GetBodyFilePath(bodyId); + + if (File.Exists(filePath)) + { + return; + } + + // A unique temp name lets concurrent writers of the same body race without clobbering. + var tempFilePath = $"{filePath}.{Guid.NewGuid():N}.tmp"; + + try + { + var fileStream = new FileStream(tempFilePath, FileMode.CreateNew, FileAccess.Write, FileShare.None, bufferSize: 4096, useAsync: true); + await using (fileStream.ConfigureAwait(false)) + { + var shouldCompress = body.Length >= settings.MinCompressionSize; + + using (var writer = new BinaryWriter(fileStream, Encoding.UTF8, leaveOpen: true)) + { + writer.Write(FormatVersion); + writer.Write(contentType); + writer.Write(body.Length); + writer.Write(shouldCompress); + writer.Flush(); + } + + if (shouldCompress) + { + var brotliStream = new BrotliStream(fileStream, CompressionLevel.Fastest, leaveOpen: true); + await using (brotliStream.ConfigureAwait(false)) + { + await brotliStream.WriteAsync(body, cancellationToken).ConfigureAwait(false); + } + } + else + { + await fileStream.WriteAsync(body, cancellationToken).ConfigureAwait(false); + } + + await fileStream.FlushAsync(cancellationToken).ConfigureAwait(false); + } + + try + { + File.Move(tempFilePath, filePath, overwrite: false); + } + catch (IOException) when (File.Exists(filePath)) + { + // A concurrent writer already produced the (immutable) body; discard our copy. + TryDelete(tempFilePath); + } + } + catch + { + TryDelete(tempFilePath); + throw; + } + } + + public Task ReadBody(string bodyId, CancellationToken cancellationToken = default) + { + var filePath = GetBodyFilePath(bodyId); + + if (!File.Exists(filePath)) + { + return Task.FromResult(null); + } + + FileStream? fileStream = null; + try + { + fileStream = new FileStream(filePath, FileMode.Open, FileAccess.Read, FileShare.Read, bufferSize: 4096, useAsync: true); + + var reader = new BinaryReader(fileStream, Encoding.UTF8, leaveOpen: true); + + var formatVersion = reader.ReadInt32(); + if (formatVersion != FormatVersion) + { + throw new InvalidOperationException($"Unsupported body file format version {formatVersion} for {bodyId}."); + } + + var contentType = reader.ReadString(); + var bodySize = reader.ReadInt32(); + var isCompressed = reader.ReadBoolean(); + + // The returned stream owns fileStream and is disposed by the caller. Both branches are + // wrapped so the body is never exposed as a seekable stream: fileStream is positioned + // past the header, but its Length still counts the header, and ASP.NET Core would set + // Content-Length from that and then write only the bytes after the current position. + Stream bodyStream = isCompressed + ? new ExpectedLengthStream(new BrotliStream(fileStream, CompressionMode.Decompress, leaveOpen: false), bodySize) + : new ExpectedLengthStream(fileStream, bodySize); + + return Task.FromResult(new MessageBodyFileResult + { + Stream = bodyStream, + ContentType = contentType, + BodySize = bodySize + }); + } + catch (FileNotFoundException) + { + fileStream?.Dispose(); + return Task.FromResult(null); + } + catch + { + fileStream?.Dispose(); + throw; + } + } + + public Task DeleteBody(string bodyId, CancellationToken cancellationToken = default) + { + TryDelete(GetBodyFilePath(bodyId)); + return Task.CompletedTask; + } + + string GetBodyFilePath(string bodyId) => Path.Combine(StoragePath, $"{bodyId}.body"); + + static void TryDelete(string filePath) + { + try + { + File.Delete(filePath); + } + catch (DirectoryNotFoundException) + { + // Nothing to delete. + } + } +} diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/BodyStorage/S3BodyStorageInstaller.cs b/src/ServiceControl.Persistence.EFCore/Implementation/BodyStorage/S3BodyStorageInstaller.cs new file mode 100644 index 0000000000..767fbda461 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Implementation/BodyStorage/S3BodyStorageInstaller.cs @@ -0,0 +1,19 @@ +namespace ServiceControl.Persistence.EFCore.Implementation.BodyStorage; + +using Amazon.S3.Model; +using Amazon.S3.Util; +using ServiceControl.Persistence; +using ServiceControl.Persistence.EFCore.Abstractions; + +public class S3BodyStorageInstaller(S3BodyStorageSettings settings) : IBodyStorageInstaller +{ + public async Task Provision(CancellationToken cancellationToken = default) + { + using var client = S3ClientFactory.Create(settings); + + if (!await AmazonS3Util.DoesS3BucketExistV2Async(client, settings.BucketName)) + { + await client.PutBucketAsync(new PutBucketRequest { BucketName = settings.BucketName }, cancellationToken); + } + } +} diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/BodyStorage/S3BodyStoragePersistence.cs b/src/ServiceControl.Persistence.EFCore/Implementation/BodyStorage/S3BodyStoragePersistence.cs new file mode 100644 index 0000000000..a6065177e4 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Implementation/BodyStorage/S3BodyStoragePersistence.cs @@ -0,0 +1,113 @@ +namespace ServiceControl.Persistence.EFCore.Implementation.BodyStorage; + +using System.IO.Compression; +using System.Net; +using Amazon.S3; +using Amazon.S3.Model; +using ServiceControl.Infrastructure; +using ServiceControl.Persistence.EFCore.Abstractions; +using ServiceControl.Persistence.EFCore.Infrastructure; + +public class S3BodyStoragePersistence : IBodyStoragePersistence +{ + const string FormatVersion = "1"; + + readonly IAmazonS3 client; + readonly string bucketName; + readonly string keyPrefix; + readonly int minBodySizeForCompression; + + public S3BodyStoragePersistence(S3BodyStorageSettings settings) + { + client = S3ClientFactory.Create(settings); + bucketName = settings.BucketName; + keyPrefix = settings.KeyPrefix; + minBodySizeForCompression = settings.MinCompressionSize; + } + + public async Task WriteBody(string bodyId, ReadOnlyMemory body, string contentType, CancellationToken cancellationToken = default) + { + var key = Key(bodyId); + + // Bodies are immutable, so an existing object is already correct. + if (await Exists(key, cancellationToken)) + { + return; + } + + var compressed = body.Length >= minBodySizeForCompression ? BodyCompression.TryCompress(body) : null; + + var payload = compressed is null ? body : compressed.AsMemory(); + var request = new PutObjectRequest + { + BucketName = bucketName, + Key = key, + InputStream = new ReadOnlyStream(payload), + ContentType = contentType + }; + request.Metadata.Add("format-version", FormatVersion); + request.Metadata.Add("body-size", body.Length.ToString()); + request.Metadata.Add("is-compressed", (compressed is not null).ToString()); + + await client.PutObjectAsync(request, cancellationToken); + } + + public async Task ReadBody(string bodyId, CancellationToken cancellationToken = default) + { + try + { + var response = await client.GetObjectAsync(bucketName, Key(bodyId), cancellationToken); + try + { + var metadata = response.Metadata; + + var version = metadata["format-version"]; + if (!string.IsNullOrEmpty(version) && version != FormatVersion) + { + throw new InvalidOperationException($"Unsupported object format version {version} for {bodyId}."); + } + + var bodySize = int.TryParse(metadata["body-size"], out var size) ? size : 0; + var isCompressed = bool.TryParse(metadata["is-compressed"], out var compressed) && compressed; + var contentType = response.Headers.ContentType ?? "application/octet-stream"; + Stream stream = isCompressed + ? new ExpectedLengthStream(new BrotliStream(response.ResponseStream, CompressionMode.Decompress), bodySize) + : response.ResponseStream; + + return new MessageBodyFileResult + { + Stream = new OwnedStream(stream, response), + ContentType = contentType, + BodySize = bodySize + }; + } + catch + { + response.Dispose(); + throw; + } + } + catch (AmazonS3Exception ex) when (ex.StatusCode == HttpStatusCode.NotFound) + { + return null; + } + } + + public Task DeleteBody(string bodyId, CancellationToken cancellationToken = default) => + client.DeleteObjectAsync(bucketName, Key(bodyId), cancellationToken); + + async Task Exists(string key, CancellationToken cancellationToken) + { + try + { + await client.GetObjectMetadataAsync(bucketName, key, cancellationToken); + return true; + } + catch (AmazonS3Exception ex) when (ex.StatusCode == HttpStatusCode.NotFound) + { + return false; + } + } + + string Key(string bodyId) => $"{keyPrefix}{bodyId}"; +} diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/BodyStorage/S3ClientFactory.cs b/src/ServiceControl.Persistence.EFCore/Implementation/BodyStorage/S3ClientFactory.cs new file mode 100644 index 0000000000..42aa2ae45c --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Implementation/BodyStorage/S3ClientFactory.cs @@ -0,0 +1,39 @@ +namespace ServiceControl.Persistence.EFCore.Implementation.BodyStorage; + +using Amazon; +using Amazon.Runtime; +using Amazon.S3; +using ServiceControl.Persistence.EFCore.Abstractions; + +static class S3ClientFactory +{ + public static IAmazonS3 Create(S3BodyStorageSettings settings) + { + var config = new AmazonS3Config(); + + var serviceUrl = settings.ServiceUrl; + if (!string.IsNullOrEmpty(serviceUrl)) + { + config.ServiceURL = serviceUrl; + config.ForcePathStyle = true; // Required for S3-compatible endpoints (MinIO, LocalStack). + } + + var region = settings.Region; + if (!string.IsNullOrEmpty(region)) + { + if (!string.IsNullOrEmpty(serviceUrl)) + { + config.AuthenticationRegion = region; + } + else + { + config.RegionEndpoint = RegionEndpoint.GetBySystemName(region); + } + } + + // With no static keys the SDK's default credential chain resolves the ambient IAM role. + return settings.Credentials is { } credentials + ? new AmazonS3Client(new BasicAWSCredentials(credentials.AccessKeyId, credentials.SecretAccessKey), config) + : new AmazonS3Client(config); + } +} diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/FakeBodyStoragePersistence.cs b/src/ServiceControl.Persistence.EFCore/Implementation/FakeBodyStoragePersistence.cs deleted file mode 100644 index c4fe41a092..0000000000 --- a/src/ServiceControl.Persistence.EFCore/Implementation/FakeBodyStoragePersistence.cs +++ /dev/null @@ -1,10 +0,0 @@ -namespace ServiceControl.Persistence.EFCore.Implementation; - -using ServiceControl.Persistence.EFCore.Infrastructure; - -public class FakeBodyStoragePersistence : IBodyStoragePersistence -{ - public Task DeleteBody(string bodyId, CancellationToken cancellationToken = default) => throw new NotImplementedException(); - public Task ReadBody(string bodyId, CancellationToken cancellationToken = default) => throw new NotImplementedException(); - public Task WriteBody(string bodyId, ReadOnlyMemory body, string contentType, CancellationToken cancellationToken = default) => throw new NotImplementedException(); -} diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFRecoverabilityIngestionUnitOfWork.cs b/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFRecoverabilityIngestionUnitOfWork.cs index d33a539cdc..4755af15bf 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFRecoverabilityIngestionUnitOfWork.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFRecoverabilityIngestionUnitOfWork.cs @@ -19,7 +19,7 @@ public Task RecordFailedProcessingAttempt(MessageContext context, var uniqueMessageId = context.Headers.UniqueId(); var contentType = context.Headers.GetValueOrDefault(Headers.ContentType, "text/plain"); var bodySize = context.Body.Length; - var (bodyText, storeExternally) = MessageBodyClassifier.Classify(context.Headers, context.Body, settings.MaxBodySizeToStore); + var (bodyText, storeExternally) = MessageBodyClassifier.Classify(context.Headers, context.Body, settings.BodyStorage.MaxBodySizeToStore); if (storeExternally) { diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/MessageBodyClassifier.cs b/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/MessageBodyClassifier.cs index 48dc630ac2..22f620a47a 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/MessageBodyClassifier.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/MessageBodyClassifier.cs @@ -3,8 +3,13 @@ namespace ServiceControl.Persistence.EFCore.Implementation.UnitOfWork; using System.Text; using ServiceControl.Persistence.Infrastructure; -// Bodies are always stored. MaxBodySizeToStore only decides where: inline in BodyText, or in -// external storage with at most a search prefix left inline. +/// +/// Decides whether a body lives inline in the database or in external storage. +/// +/// +/// Bodies are always stored. MaxBodySizeToStore only decides where: inline in BodyText, or in +/// external storage with at most a search prefix left inline. +/// static class MessageBodyClassifier { public static (string? BodyText, bool StoreExternally) Classify(IReadOnlyDictionary headers, ReadOnlyMemory body, int maxBodySizeToStore) diff --git a/src/ServiceControl.Persistence.EFCore/Infrastructure/ExpectedLengthStream.cs b/src/ServiceControl.Persistence.EFCore/Infrastructure/ExpectedLengthStream.cs new file mode 100644 index 0000000000..914c66995c --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Infrastructure/ExpectedLengthStream.cs @@ -0,0 +1,65 @@ +namespace ServiceControl.Persistence.EFCore.Infrastructure; + +/// +/// Validates the byte count of a forward-only stream without buffering its contents. +/// +/// +/// Compressed body objects record their uncompressed length in metadata. Before cloud reads were +/// streamed, eager decompression verified that length before returning the body. This wrapper +/// preserves that integrity check at end of stream. A consumer that stops reading early does not +/// perform the final length check, which is necessary to keep response streaming and cancellation. +/// +public sealed class ExpectedLengthStream(Stream stream, long expectedLength) : Stream +{ + long bytesRead; + + public override bool CanRead => stream.CanRead; + public override bool CanSeek => false; + public override bool CanWrite => false; + public override long Length => throw new NotSupportedException(); + + public override long Position + { + get => bytesRead; + set => throw new NotSupportedException(); + } + + public override void Flush() => throw new NotSupportedException(); + + public override int Read(byte[] buffer, int offset, int count) => ValidateRead(stream.Read(buffer, offset, count)); + + public override int Read(Span buffer) => ValidateRead(stream.Read(buffer)); + + public override async Task ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) => + ValidateRead(await stream.ReadAsync(buffer, offset, count, cancellationToken)); + + public override async ValueTask ReadAsync(Memory buffer, CancellationToken cancellationToken = default) => + ValidateRead(await stream.ReadAsync(buffer, cancellationToken)); + + public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException(); + + public override void SetLength(long value) => throw new NotSupportedException(); + + public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException(); + + int ValidateRead(int count) + { + bytesRead += count; + if (bytesRead > expectedLength || (count == 0 && bytesRead != expectedLength)) + { + throw new InvalidOperationException("Decompressed body size does not match its metadata."); + } + + return count; + } + + protected override void Dispose(bool disposing) + { + if (disposing) + { + stream.Dispose(); + } + + base.Dispose(disposing); + } +} \ No newline at end of file diff --git a/src/ServiceControl.Persistence.EFCore/Infrastructure/IBodyStoragePersistence.cs b/src/ServiceControl.Persistence.EFCore/Infrastructure/IBodyStoragePersistence.cs index c089275bc1..2427951fae 100644 --- a/src/ServiceControl.Persistence.EFCore/Infrastructure/IBodyStoragePersistence.cs +++ b/src/ServiceControl.Persistence.EFCore/Infrastructure/IBodyStoragePersistence.cs @@ -4,8 +4,13 @@ namespace ServiceControl.Persistence.EFCore.Infrastructure; using System.Threading; using System.Threading.Tasks; -// Bodies are immutable and addressed by bodyId (the UniqueMessageId) alone, so re-failures of the -// same message resolve to the same stored body and writes can be skipped when it already exists. +/// +/// External storage for message bodies too large or too binary to sit inline in the database. +/// +/// +/// Bodies are immutable and addressed by bodyId (the UniqueMessageId) alone, so re-failures of the +/// same message resolve to the same stored body and writes can be skipped when it already exists. +/// public interface IBodyStoragePersistence { Task WriteBody(string bodyId, ReadOnlyMemory body, string contentType, CancellationToken cancellationToken = default); diff --git a/src/ServiceControl.Persistence.EFCore/Infrastructure/MessageBodyFileResult.cs b/src/ServiceControl.Persistence.EFCore/Infrastructure/MessageBodyFileResult.cs index 531eba1f58..81ebf87a23 100644 --- a/src/ServiceControl.Persistence.EFCore/Infrastructure/MessageBodyFileResult.cs +++ b/src/ServiceControl.Persistence.EFCore/Infrastructure/MessageBodyFileResult.cs @@ -7,5 +7,4 @@ public class MessageBodyFileResult public required Stream Stream { get; set; } public required string ContentType { get; set; } public required int BodySize { get; set; } - public required string Etag { get; set; } } \ No newline at end of file diff --git a/src/ServiceControl.Persistence.EFCore/Infrastructure/OwnedStream.cs b/src/ServiceControl.Persistence.EFCore/Infrastructure/OwnedStream.cs new file mode 100644 index 0000000000..be265364fb --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Infrastructure/OwnedStream.cs @@ -0,0 +1,68 @@ +namespace ServiceControl.Persistence.EFCore.Infrastructure; + +/// +/// Couples a returned stream to another resource that must remain alive while the stream is read. +/// +/// +/// S3 returns a response whose response stream is consumed later by ASP.NET Core, after the +/// persistence method has returned. Disposing this wrapper disposes both the exposed stream and +/// the owning response, releasing the underlying HTTP connection. +/// +public sealed class OwnedStream(Stream stream, IDisposable owner) : Stream +{ + public override bool CanRead => stream.CanRead; + public override bool CanSeek => stream.CanSeek; + public override bool CanWrite => stream.CanWrite; + public override long Length => stream.Length; + + public override long Position + { + get => stream.Position; + set => stream.Position = value; + } + + public override void Flush() => stream.Flush(); + + public override Task FlushAsync(CancellationToken cancellationToken) => stream.FlushAsync(cancellationToken); + + public override int Read(byte[] buffer, int offset, int count) => stream.Read(buffer, offset, count); + + public override int Read(Span buffer) => stream.Read(buffer); + + public override Task ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) => + stream.ReadAsync(buffer, offset, count, cancellationToken); + + public override ValueTask ReadAsync(Memory buffer, CancellationToken cancellationToken = default) => + stream.ReadAsync(buffer, cancellationToken); + + public override long Seek(long offset, SeekOrigin origin) => stream.Seek(offset, origin); + + public override void SetLength(long value) => stream.SetLength(value); + + public override void Write(byte[] buffer, int offset, int count) => stream.Write(buffer, offset, count); + + public override void Write(ReadOnlySpan buffer) => stream.Write(buffer); + + public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) => + stream.WriteAsync(buffer, offset, count, cancellationToken); + + public override ValueTask WriteAsync(ReadOnlyMemory buffer, CancellationToken cancellationToken = default) => + stream.WriteAsync(buffer, cancellationToken); + + protected override void Dispose(bool disposing) + { + if (disposing) + { + try + { + stream.Dispose(); + } + finally + { + owner.Dispose(); + } + } + + base.Dispose(disposing); + } +} diff --git a/src/ServiceControl.Persistence.EFCore/ServiceControl.Persistence.EFCore.csproj b/src/ServiceControl.Persistence.EFCore/ServiceControl.Persistence.EFCore.csproj index f80428271a..5aca192107 100644 --- a/src/ServiceControl.Persistence.EFCore/ServiceControl.Persistence.EFCore.csproj +++ b/src/ServiceControl.Persistence.EFCore/ServiceControl.Persistence.EFCore.csproj @@ -19,6 +19,9 @@ + + + diff --git a/src/ServiceControl.Persistence.Tests.PostgreSql/PersistenceTestsContext.cs b/src/ServiceControl.Persistence.Tests.PostgreSql/PersistenceTestsContext.cs index a104f188eb..5536688218 100644 --- a/src/ServiceControl.Persistence.Tests.PostgreSql/PersistenceTestsContext.cs +++ b/src/ServiceControl.Persistence.Tests.PostgreSql/PersistenceTestsContext.cs @@ -2,6 +2,7 @@ namespace ServiceControl.Persistence.Tests; using System; +using System.IO; using System.Linq; using System.Threading.Tasks; using EFCore.PostgreSql; @@ -10,12 +11,14 @@ namespace ServiceControl.Persistence.Tests; using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Time.Testing; using Npgsql; +using ServiceControl.Persistence.EFCore.Abstractions; using ServiceControl.Persistence.EFCore.Infrastructure; public class PersistenceTestsContext : IPersistenceTestsContext { IHost host; string databaseName; + string bodyStoragePath; public FakeTimeProvider FakeTime { get; } = new(); @@ -28,9 +31,12 @@ public async Task Setup(IHostApplicationBuilder hostBuilder) Database = databaseName }; + bodyStoragePath = Directory.CreateTempSubdirectory("sc_test_bodies_").FullName; + PersistenceSettings = new PostgreSqlPersisterSettings { - ConnectionString = connectionStringBuilder.ConnectionString + ConnectionString = connectionStringBuilder.ConnectionString, + BodyStorage = new FileSystemBodyStorageSettings { StoragePath = bodyStoragePath } }; var persistence = new PostgreSqlPersistenceConfiguration().Create(PersistenceSettings); @@ -52,6 +58,8 @@ public async Task PostSetup(IHost host) public async Task TearDown() { + DeleteBodyStorage(); + await using var connection = new NpgsqlConnection(await PostgreSqlSharedContainer.GetConnectionStringAsync()); await connection.OpenAsync(); await using var command = connection.CreateCommand(); @@ -72,4 +80,15 @@ public async Task CompleteDatabaseOperation() public PersistenceSettings PersistenceSettings { get; set; } public string GenerateFailedMessageRecordId(string messageId) => messageId; + + void DeleteBodyStorage() + { + try + { + Directory.Delete(bodyStoragePath, recursive: true); + } + catch (DirectoryNotFoundException) + { + } + } } diff --git a/src/ServiceControl.Persistence.Tests.PostgreSql/ServiceControl.Persistence.Tests.PostgreSql.csproj b/src/ServiceControl.Persistence.Tests.PostgreSql/ServiceControl.Persistence.Tests.PostgreSql.csproj index 59de3364fa..3c63181a00 100644 --- a/src/ServiceControl.Persistence.Tests.PostgreSql/ServiceControl.Persistence.Tests.PostgreSql.csproj +++ b/src/ServiceControl.Persistence.Tests.PostgreSql/ServiceControl.Persistence.Tests.PostgreSql.csproj @@ -22,6 +22,8 @@ + + diff --git a/src/ServiceControl.Persistence.Tests.SqlServer/PersistenceTestsContext.cs b/src/ServiceControl.Persistence.Tests.SqlServer/PersistenceTestsContext.cs index 9af5dc6de0..96fbcd2634 100644 --- a/src/ServiceControl.Persistence.Tests.SqlServer/PersistenceTestsContext.cs +++ b/src/ServiceControl.Persistence.Tests.SqlServer/PersistenceTestsContext.cs @@ -2,6 +2,7 @@ namespace ServiceControl.Persistence.Tests; using System; +using System.IO; using System.Linq; using System.Threading.Tasks; using EFCore.SqlServer; @@ -10,12 +11,14 @@ namespace ServiceControl.Persistence.Tests; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Time.Testing; +using ServiceControl.Persistence.EFCore.Abstractions; using ServiceControl.Persistence.EFCore.Infrastructure; public class PersistenceTestsContext : IPersistenceTestsContext { IHost host; string databaseName; + string bodyStoragePath; public FakeTimeProvider FakeTime { get; } = new(); @@ -28,9 +31,12 @@ public async Task Setup(IHostApplicationBuilder hostBuilder) InitialCatalog = databaseName }; + bodyStoragePath = Directory.CreateTempSubdirectory("sc_test_bodies_").FullName; + PersistenceSettings = new SqlServerPersisterSettings { - ConnectionString = connectionStringBuilder.ConnectionString + ConnectionString = connectionStringBuilder.ConnectionString, + BodyStorage = new FileSystemBodyStorageSettings { StoragePath = bodyStoragePath } }; var persistence = new SqlServerPersistenceConfiguration().Create(PersistenceSettings); @@ -52,6 +58,8 @@ public async Task PostSetup(IHost host) public async Task TearDown() { + DeleteBodyStorage(); + await using var connection = new SqlConnection(await SqlServerSharedContainer.GetConnectionStringAsync()); await connection.OpenAsync(); await using var command = connection.CreateCommand(); @@ -78,4 +86,15 @@ public async Task CompleteDatabaseOperation() public PersistenceSettings PersistenceSettings { get; set; } public string GenerateFailedMessageRecordId(string messageId) => messageId; + + void DeleteBodyStorage() + { + try + { + Directory.Delete(bodyStoragePath, recursive: true); + } + catch (DirectoryNotFoundException) + { + } + } } \ No newline at end of file diff --git a/src/ServiceControl.Persistence.Tests.SqlServer/ServiceControl.Persistence.Tests.SqlServer.csproj b/src/ServiceControl.Persistence.Tests.SqlServer/ServiceControl.Persistence.Tests.SqlServer.csproj index 186518974d..c29c79b2e3 100644 --- a/src/ServiceControl.Persistence.Tests.SqlServer/ServiceControl.Persistence.Tests.SqlServer.csproj +++ b/src/ServiceControl.Persistence.Tests.SqlServer/ServiceControl.Persistence.Tests.SqlServer.csproj @@ -22,6 +22,8 @@ + + diff --git a/src/ServiceControl.Persistence.Tests/EFCore/AzureBlobBodyStorageTests.cs b/src/ServiceControl.Persistence.Tests/EFCore/AzureBlobBodyStorageTests.cs new file mode 100644 index 0000000000..66f06d1881 --- /dev/null +++ b/src/ServiceControl.Persistence.Tests/EFCore/AzureBlobBodyStorageTests.cs @@ -0,0 +1,140 @@ +namespace ServiceControl.Persistence.Tests; + +using System; +using System.IO; +using System.Text; +using System.Threading.Tasks; +using NUnit.Framework; +using ServiceControl.Persistence.EFCore.Abstractions; +using ServiceControl.Persistence.EFCore.Implementation.BodyStorage; +using Testcontainers.Azurite; + +[TestFixture] +[Platform(Exclude = "Win", Reason = "Azurite has no Windows container image")] +class AzureBlobBodyStorageTests +{ + AzuriteContainer azurite; + AzureBlobBodyStoragePersistence store; + + [OneTimeSetUp] + public async Task StartAzurite() + { + // Azurite 3.36.0 does not implement the service version the current Azure.Storage.Blobs + // package negotiates, so the API version check has to be skipped. Support is milestoned for + // Azurite 3.37.0 (https://github.com/Azure/Azurite/issues/2623); when that ships, move the + // tag forward and drop --skipApiVersionCheck together. + azurite = new AzuriteBuilder("mcr.microsoft.com/azure-storage/azurite:3.36.0") + .WithCommand("--skipApiVersionCheck") + .Build(); + await azurite.StartAsync(); + } + + [OneTimeTearDown] + public async Task StopAzurite() + { + if (azurite != null) + { + await azurite.DisposeAsync(); + } + } + + [SetUp] + public async Task CreateContainer() + { + var settings = new AzureBlobBodyStorageSettings + { + MinCompressionSize = 64, + Authentication = new AzureBlobSharedKeyAuthentication { ConnectionString = azurite.GetConnectionString() }, + ContainerName = $"bodies-{Guid.NewGuid():n}" + }; + + await new AzureBlobBodyStorageInstaller(settings).Provision(); + store = new AzureBlobBodyStoragePersistence(settings); + } + + [Test] + public async Task Round_trips_a_small_uncompressed_body() + { + var bodyId = Guid.NewGuid().ToString(); + var body = Encoding.UTF8.GetBytes("hello world"); + + await store.WriteBody(bodyId, body, "text/plain"); + + var result = await store.ReadBody(bodyId); + + Assert.That(result, Is.Not.Null); + using (result.Stream) + { + Assert.That(ReadAll(result.Stream), Is.EqualTo(body)); + } + + Assert.Multiple(() => + { + Assert.That(result.ContentType, Is.EqualTo("text/plain")); + Assert.That(result.BodySize, Is.EqualTo(body.Length)); + }); + } + + [Test] + public async Task Round_trips_a_large_body_over_the_compression_threshold() + { + var bodyId = Guid.NewGuid().ToString(); + var body = Encoding.UTF8.GetBytes(new string('a', 100_000)); + + await store.WriteBody(bodyId, body, "application/json"); + + var result = await store.ReadBody(bodyId); + + Assert.That(result, Is.Not.Null); + using (result.Stream) + { + Assert.That(ReadAll(result.Stream), Is.EqualTo(body)); + } + + Assert.That(result.BodySize, Is.EqualTo(body.Length)); + } + + [Test] + public async Task Returns_null_for_a_missing_body() => + Assert.That(await store.ReadBody(Guid.NewGuid().ToString()), Is.Null); + + [Test] + public async Task Delete_removes_the_body() + { + var bodyId = Guid.NewGuid().ToString(); + await store.WriteBody(bodyId, Encoding.UTF8.GetBytes("payload"), "text/plain"); + + await store.DeleteBody(bodyId); + + Assert.That(await store.ReadBody(bodyId), Is.Null); + } + + [Test] + public void Delete_of_a_missing_body_does_not_throw() => + Assert.DoesNotThrowAsync(() => store.DeleteBody(Guid.NewGuid().ToString())); + + [Test] + public async Task Rewriting_an_existing_body_keeps_the_first_write() + { + var bodyId = Guid.NewGuid().ToString(); + var original = Encoding.UTF8.GetBytes("original"); + + await store.WriteBody(bodyId, original, "text/plain"); + await store.WriteBody(bodyId, Encoding.UTF8.GetBytes("different"), "text/plain"); + + var result = await store.ReadBody(bodyId); + + Assert.That(result, Is.Not.Null); + using (result.Stream) + { + Assert.That(ReadAll(result.Stream), Is.EqualTo(original), "bodies are immutable, so the first write wins"); + } + } + + static byte[] ReadAll(Stream stream) + { + using var buffer = new MemoryStream(); + stream.CopyTo(buffer); + return buffer.ToArray(); + } +} diff --git a/src/ServiceControl.Persistence.Tests/EFCore/BodyReadTests.cs b/src/ServiceControl.Persistence.Tests/EFCore/BodyReadTests.cs new file mode 100644 index 0000000000..3223dc8323 --- /dev/null +++ b/src/ServiceControl.Persistence.Tests/EFCore/BodyReadTests.cs @@ -0,0 +1,118 @@ +namespace ServiceControl.Persistence.Tests; + +using System; +using System.IO; +using System.Text; +using System.Threading.Tasks; +using Microsoft.Extensions.DependencyInjection; +using NUnit.Framework; +using ServiceControl.Operations.BodyStorage; + +class BodyReadTests : ErrorIngestionTestBase +{ + const int Cap = 64; + + [SetUp] + public void ShrinkTheBodyCap() => EFSettings.BodyStorage.MaxBodySizeToStore = Cap; + + [Test] + public async Task Fetches_an_inline_text_body() + { + var failure = new IngestedFailure { ContentType = "text/xml", Body = Encoding.UTF8.GetBytes("1") }; + await Ingest(failure); + + var result = await Fetch(failure.UniqueMessageIdString); + + Assert.That(result, Is.Not.Null); + using (Assert.EnterMultipleScope()) + { + Assert.That(result.HasResult, Is.True); + Assert.That(result.ContentType, Is.EqualTo("text/xml")); + Assert.That(Encoding.UTF8.GetString(ReadAll(result.Stream)), Is.EqualTo("1")); + } + } + + [Test] + public async Task Fetches_an_external_binary_body() + { + var body = BitConverter.GetBytes(0xDEADBEEF); + var failure = new IngestedFailure { ContentType = "application/octet-stream", Body = body }; + await Ingest(failure); + + var result = await Fetch(failure.UniqueMessageIdString); + + Assert.That(result, Is.Not.Null); + using (Assert.EnterMultipleScope()) + { + Assert.That(result.HasResult, Is.True); + Assert.That(result.ContentType, Is.EqualTo("application/octet-stream")); + Assert.That(ReadAll(result.Stream), Is.EqualTo(body)); + } + } + + [Test] + public async Task Fetches_the_whole_body_for_large_text_not_the_inline_prefix() + { + var body = Encoding.UTF8.GetBytes(new string('a', Cap * 2)); + var failure = new IngestedFailure { Body = body }; + await Ingest(failure); + + var result = await Fetch(failure.UniqueMessageIdString); + + Assert.That(result, Is.Not.Null); + using (Assert.EnterMultipleScope()) + { + Assert.That(result.HasResult, Is.True); + Assert.That(ReadAll(result.Stream), Is.EqualTo(body), "external storage is authoritative, not the inline search prefix"); + } + } + + [Test] + public async Task Fetches_by_message_id() + { + var failure = new IngestedFailure { Body = Encoding.UTF8.GetBytes("1") }; + await Ingest(failure); + + var result = await Fetch(failure.MessageId); + + Assert.That(result, Is.Not.Null); + Assert.That(result.HasResult, Is.True); + } + + [Test] + public async Task Reports_no_body_for_an_empty_body() + { + var failure = new IngestedFailure { Body = [] }; + await Ingest(failure); + + var result = await Fetch(failure.UniqueMessageIdString); + + Assert.That(result, Is.Not.Null); + Assert.That(result.HasResult, Is.False); + } + + [Test] + public async Task Returns_null_for_an_unknown_message() + { + var result = await Fetch(Guid.NewGuid().ToString()); + + Assert.That(result, Is.Null); + } + + async Task Fetch(string bodyId) + { + using var scope = ServiceProvider.CreateScope(); + var bodyStorage = scope.ServiceProvider.GetRequiredService(); + return await bodyStorage.TryFetch(bodyId); + } + + static byte[] ReadAll(Stream stream) + { + using (stream) + { + using var buffer = new MemoryStream(); + stream.CopyTo(buffer); + return buffer.ToArray(); + } + } +} diff --git a/src/ServiceControl.Persistence.Tests/EFCore/BodyStorageConfigurationTests.cs b/src/ServiceControl.Persistence.Tests/EFCore/BodyStorageConfigurationTests.cs new file mode 100644 index 0000000000..ffb0f507a7 --- /dev/null +++ b/src/ServiceControl.Persistence.Tests/EFCore/BodyStorageConfigurationTests.cs @@ -0,0 +1,213 @@ +namespace ServiceControl.Persistence.Tests; + +using System; +using System.Collections.Generic; +using NUnit.Framework; +using ServiceControl.Configuration; +using ServiceControl.Persistence.EFCore.Abstractions; + +// Covers the settings-reader validation that lets the storage settings expose non-nullable members, +// so the stores never have to re-check what configuration already guaranteed. +[TestFixture] +[NonParallelizable] +class BodyStorageConfigurationTests +{ + static readonly SettingsRootNamespace TestNamespace = new("ServiceControl"); + + static readonly string[] Keys = + [ + "SERVICECONTROL_DATABASE_CONNECTIONSTRING", + "SERVICECONTROL_ERRORRETENTIONPERIOD", + "SERVICECONTROL_MESSAGEBODY_STORAGETYPE", + "SERVICECONTROL_MESSAGEBODY_STORAGEPATH", + "SERVICECONTROL_MESSAGEBODY_MINCOMPRESSIONSIZE", + "SERVICECONTROL_MESSAGEBODY_AZURE_CONNECTIONSTRING", + "SERVICECONTROL_MESSAGEBODY_AZURE_SERVICEURI", + "SERVICECONTROL_MESSAGEBODY_AZURE_MANAGEDIDENTITYCLIENTID", + "SERVICECONTROL_MESSAGEBODY_AZURE_AUTHORITYHOST", + "SERVICECONTROL_MESSAGEBODY_AZURE_CONTAINERNAME", + "SERVICECONTROL_MESSAGEBODY_S3_BUCKETNAME", + "SERVICECONTROL_MESSAGEBODY_S3_KEYPREFIX", + "SERVICECONTROL_MESSAGEBODY_S3_REGION", + "SERVICECONTROL_MESSAGEBODY_S3_ACCESSKEYID", + "SERVICECONTROL_MESSAGEBODY_S3_SECRETACCESSKEY" + ]; + + [SetUp] + public void SetUp() + { + ClearKeys(); + Set("SERVICECONTROL_DATABASE_CONNECTIONSTRING", "Server=nowhere"); + Set("SERVICECONTROL_ERRORRETENTIONPERIOD", "10.00:00:00"); + } + + [TearDown] + public void TearDown() => ClearKeys(); + + [Test] + public void File_system_storage_yields_a_populated_path() + { + Set("SERVICECONTROL_MESSAGEBODY_STORAGETYPE", "FileSystem"); + Set("SERVICECONTROL_MESSAGEBODY_STORAGEPATH", "/var/bodies"); + + var bodyStorage = CreateBodyStorageSettings(); + + Assert.That(bodyStorage, Is.TypeOf()); + Assert.That(((FileSystemBodyStorageSettings)bodyStorage).StoragePath, Is.EqualTo("/var/bodies")); + } + + [Test] + public void Azure_connection_string_yields_shared_key_authentication() + { + Set("SERVICECONTROL_MESSAGEBODY_STORAGETYPE", "AzureBlob"); + Set("SERVICECONTROL_MESSAGEBODY_AZURE_CONNECTIONSTRING", "UseDevelopmentStorage=true"); + + var azure = (AzureBlobBodyStorageSettings)CreateBodyStorageSettings(); + + Assert.That(azure.Authentication, Is.TypeOf()); + Assert.That(((AzureBlobSharedKeyAuthentication)azure.Authentication).ConnectionString, Is.EqualTo("UseDevelopmentStorage=true")); + } + + [Test] + public void Azure_service_uri_yields_managed_identity_authentication() + { + Set("SERVICECONTROL_MESSAGEBODY_STORAGETYPE", "AzureBlob"); + Set("SERVICECONTROL_MESSAGEBODY_AZURE_SERVICEURI", "https://account.blob.core.windows.net"); + Set("SERVICECONTROL_MESSAGEBODY_AZURE_MANAGEDIDENTITYCLIENTID", "client-id"); + + var azure = (AzureBlobBodyStorageSettings)CreateBodyStorageSettings(); + + var managedIdentity = azure.Authentication as AzureBlobManagedIdentityAuthentication; + Assert.That(managedIdentity, Is.Not.Null); + using (Assert.EnterMultipleScope()) + { + Assert.That(managedIdentity.ServiceUri, Is.EqualTo(new Uri("https://account.blob.core.windows.net"))); + Assert.That(managedIdentity.ClientId, Is.EqualTo("client-id")); + Assert.That(managedIdentity.AuthorityHost, Is.Null); + } + } + + [Test] + public void S3_without_keys_falls_back_to_the_default_credential_chain() + { + Set("SERVICECONTROL_MESSAGEBODY_STORAGETYPE", "S3"); + Set("SERVICECONTROL_MESSAGEBODY_S3_BUCKETNAME", "bodies"); + + var s3 = (S3BodyStorageSettings)CreateBodyStorageSettings(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(s3.BucketName, Is.EqualTo("bodies")); + Assert.That(s3.Credentials, Is.Null); + } + } + + [Test] + public void S3_with_both_keys_yields_static_credentials() + { + Set("SERVICECONTROL_MESSAGEBODY_STORAGETYPE", "S3"); + Set("SERVICECONTROL_MESSAGEBODY_S3_BUCKETNAME", "bodies"); + Set("SERVICECONTROL_MESSAGEBODY_S3_ACCESSKEYID", "key"); + Set("SERVICECONTROL_MESSAGEBODY_S3_SECRETACCESSKEY", "secret"); + + var s3 = (S3BodyStorageSettings)CreateBodyStorageSettings(); + + Assert.That(s3.Credentials, Is.Not.Null); + using (Assert.EnterMultipleScope()) + { + Assert.That(s3.Credentials.AccessKeyId, Is.EqualTo("key")); + Assert.That(s3.Credentials.SecretAccessKey, Is.EqualTo("secret")); + } + } + + [Test] + public void Missing_storage_type_is_rejected() => + Assert.That(CreateBodyStorageSettings, Throws.Exception.With.Message.Contains("MessageBody/StorageType")); + + [Test] + public void Unknown_storage_type_is_rejected() + { + Set("SERVICECONTROL_MESSAGEBODY_STORAGETYPE", "Dropbox"); + + Assert.That(CreateBodyStorageSettings, Throws.Exception.With.Message.Contains("is not valid")); + } + + [Test] + public void File_system_storage_without_a_path_is_rejected() + { + Set("SERVICECONTROL_MESSAGEBODY_STORAGETYPE", "FileSystem"); + + Assert.That(CreateBodyStorageSettings, Throws.Exception.With.Message.Contains("MessageBody/StoragePath")); + } + + [Test] + public void S3_storage_without_a_bucket_is_rejected() + { + Set("SERVICECONTROL_MESSAGEBODY_STORAGETYPE", "S3"); + + Assert.That(CreateBodyStorageSettings, Throws.Exception.With.Message.Contains("MessageBody/S3/BucketName")); + } + + [TestCase("SERVICECONTROL_MESSAGEBODY_S3_ACCESSKEYID")] + [TestCase("SERVICECONTROL_MESSAGEBODY_S3_SECRETACCESSKEY")] + public void A_lone_S3_credential_half_is_rejected(string key) + { + Set("SERVICECONTROL_MESSAGEBODY_STORAGETYPE", "S3"); + Set("SERVICECONTROL_MESSAGEBODY_S3_BUCKETNAME", "bodies"); + Set(key, "value"); + + Assert.That(CreateBodyStorageSettings, Throws.Exception.With.Message.Contains("together")); + } + + [Test] + public void Azure_storage_without_any_credential_is_rejected() + { + Set("SERVICECONTROL_MESSAGEBODY_STORAGETYPE", "AzureBlob"); + + Assert.That(CreateBodyStorageSettings, Throws.Exception.With.Message.Contains("Neither was set.")); + } + + [Test] + public void Azure_storage_with_both_credentials_is_rejected() + { + Set("SERVICECONTROL_MESSAGEBODY_STORAGETYPE", "AzureBlob"); + Set("SERVICECONTROL_MESSAGEBODY_AZURE_CONNECTIONSTRING", "UseDevelopmentStorage=true"); + Set("SERVICECONTROL_MESSAGEBODY_AZURE_SERVICEURI", "https://account.blob.core.windows.net"); + + Assert.That(CreateBodyStorageSettings, Throws.Exception.With.Message.Contains("Both were set.")); + } + + [TestCase("SERVICECONTROL_MESSAGEBODY_AZURE_SERVICEURI", "not-a-uri")] + [TestCase("SERVICECONTROL_MESSAGEBODY_AZURE_AUTHORITYHOST", "not-a-uri")] + public void A_non_absolute_Azure_uri_is_rejected(string key, string value) + { + Set("SERVICECONTROL_MESSAGEBODY_STORAGETYPE", "AzureBlob"); + Set("SERVICECONTROL_MESSAGEBODY_AZURE_SERVICEURI", "https://account.blob.core.windows.net"); + Set(key, value); + + Assert.That(CreateBodyStorageSettings, Throws.Exception.With.Message.Contains("is not a valid absolute URI")); + } + + static BodyStorageSettings CreateBodyStorageSettings() => + ((EFPersisterSettings)new TestPersistenceConfiguration().CreateSettings(TestNamespace)).BodyStorage; + + static void Set(string key, string value) => Environment.SetEnvironmentVariable(key, value); + + static void ClearKeys() + { + foreach (var key in Keys) + { + Environment.SetEnvironmentVariable(key, null); + } + } + + sealed class TestPersistenceConfiguration : EFPersistenceConfigurationBase + { + public override IPersistence Create(PersistenceSettings settings) => throw new NotSupportedException(); + + protected override EFPersisterSettings CreateSettings(string connectionString, BodyStorageSettings bodyStorage) => + new TestPersisterSettings { ConnectionString = connectionString, BodyStorage = bodyStorage }; + } + + sealed class TestPersisterSettings : EFPersisterSettings; +} diff --git a/src/ServiceControl.Persistence.Tests/EFCore/BodyStoragePersistenceTests.cs b/src/ServiceControl.Persistence.Tests/EFCore/BodyStoragePersistenceTests.cs new file mode 100644 index 0000000000..2c90b22912 --- /dev/null +++ b/src/ServiceControl.Persistence.Tests/EFCore/BodyStoragePersistenceTests.cs @@ -0,0 +1,189 @@ +namespace ServiceControl.Persistence.Tests; + +using System; +using System.IO; +using System.Text; +using System.Threading.Tasks; +using NUnit.Framework; +using ServiceControl.Persistence.EFCore.Abstractions; +using ServiceControl.Persistence.EFCore.Implementation.BodyStorage; +using ServiceControl.Persistence.EFCore.Infrastructure; + +[TestFixture] +class BodyStoragePersistenceTests +{ + const string FileSystem = nameof(FileSystem); + const string InMemory = nameof(InMemory); + + string tempDir; + + [SetUp] + public void SetUp() => tempDir = Directory.CreateTempSubdirectory("sc-body-tests-").FullName; + + [TearDown] + public void TearDown() + { + try + { + Directory.Delete(tempDir, recursive: true); + } + catch (DirectoryNotFoundException) + { + } + } + + IBodyStoragePersistence CreateStore(string kind) => kind switch + { + InMemory => new InMemoryBodyStoragePersistence(), + FileSystem => new FileSystemBodyStoragePersistence(new FileSystemBodyStorageSettings + { + StoragePath = tempDir, + MinCompressionSize = 64 + }), + _ => throw new ArgumentOutOfRangeException(nameof(kind)) + }; + + [TestCase(InMemory)] + [TestCase(FileSystem)] + public async Task Round_trips_a_small_uncompressed_body(string kind) + { + var store = CreateStore(kind); + var bodyId = Guid.NewGuid().ToString(); + var body = Encoding.UTF8.GetBytes("hello world"); + + await store.WriteBody(bodyId, body, "text/plain"); + + var result = await store.ReadBody(bodyId); + + Assert.That(result, Is.Not.Null); + using (result.Stream) + { + Assert.That(ReadAll(result.Stream), Is.EqualTo(body)); + } + + Assert.Multiple(() => + { + Assert.That(result.ContentType, Is.EqualTo("text/plain")); + Assert.That(result.BodySize, Is.EqualTo(body.Length)); + }); + } + + [TestCase(InMemory)] + [TestCase(FileSystem)] + public async Task Round_trips_a_large_body_over_the_compression_threshold(string kind) + { + var store = CreateStore(kind); + var bodyId = Guid.NewGuid().ToString(); + var body = Encoding.UTF8.GetBytes(new string('a', 100_000)); + + await store.WriteBody(bodyId, body, "application/json"); + + var result = await store.ReadBody(bodyId); + + Assert.That(result, Is.Not.Null); + using (result.Stream) + { + Assert.That(ReadAll(result.Stream), Is.EqualTo(body)); + } + + Assert.That(result.BodySize, Is.EqualTo(body.Length)); + } + + [TestCase(InMemory)] + [TestCase(FileSystem)] + public async Task Returns_null_for_a_missing_body(string kind) + { + var store = CreateStore(kind); + + Assert.That(await store.ReadBody(Guid.NewGuid().ToString()), Is.Null); + } + + [TestCase(InMemory)] + [TestCase(FileSystem)] + public async Task Delete_removes_the_body(string kind) + { + var store = CreateStore(kind); + var bodyId = Guid.NewGuid().ToString(); + await store.WriteBody(bodyId, Encoding.UTF8.GetBytes("payload"), "text/plain"); + + await store.DeleteBody(bodyId); + + Assert.That(await store.ReadBody(bodyId), Is.Null); + } + + [TestCase(InMemory)] + [TestCase(FileSystem)] + public void Delete_of_a_missing_body_does_not_throw(string kind) + { + var store = CreateStore(kind); + + Assert.DoesNotThrowAsync(() => store.DeleteBody(Guid.NewGuid().ToString())); + } + + [TestCase(InMemory)] + [TestCase(FileSystem)] + public async Task Rewriting_an_existing_body_keeps_the_first_write(string kind) + { + var store = CreateStore(kind); + var bodyId = Guid.NewGuid().ToString(); + var original = Encoding.UTF8.GetBytes("original"); + + await store.WriteBody(bodyId, original, "text/plain"); + await store.WriteBody(bodyId, Encoding.UTF8.GetBytes("different"), "text/plain"); + + var result = await store.ReadBody(bodyId); + + Assert.That(result, Is.Not.Null); + using (result.Stream) + { + Assert.That(ReadAll(result.Stream), Is.EqualTo(original), "bodies are immutable, so the first write wins"); + } + } + + [TestCase(InMemory, 11)] + [TestCase(InMemory, 100_000)] + [TestCase(FileSystem, 11)] + [TestCase(FileSystem, 100_000)] + public async Task Returned_stream_never_counts_bytes_it_will_not_yield(string kind, int bodyLength) + { + var store = CreateStore(kind); + var bodyId = Guid.NewGuid().ToString(); + var body = Encoding.UTF8.GetBytes(new string('a', bodyLength)); + + await store.WriteBody(bodyId, body, "text/plain"); + + var result = await store.ReadBody(bodyId); + + Assert.That(result, Is.Not.Null); + using (result.Stream) + { + if (result.Stream.CanSeek) + { + // ASP.NET Core sets the response Content-Length from Stream.Length outright, but + // copies only from the current position onwards, so anything ahead of the position + // is promised to the client and never sent. + Assert.That(result.Stream.Length, Is.EqualTo(body.Length)); + } + + Assert.That(ReadAll(result.Stream), Is.EqualTo(body)); + } + } + + [Test] + public async Task Provisioning_creates_the_filesystem_storage_directory() + { + var path = Path.Combine(tempDir, "nested", "bodies"); + Assert.That(Directory.Exists(path), Is.False); + + await new FileSystemBodyStorageInstaller(new FileSystemBodyStorageSettings { StoragePath = path }).Provision(); + + Assert.That(Directory.Exists(path), Is.True); + } + + static byte[] ReadAll(Stream stream) + { + using var buffer = new MemoryStream(); + stream.CopyTo(buffer); + return buffer.ToArray(); + } +} diff --git a/src/ServiceControl.Persistence.Tests/EFCore/ErrorIngestionBodyTests.cs b/src/ServiceControl.Persistence.Tests/EFCore/ErrorIngestionBodyTests.cs index d2bd359688..e474c112de 100644 --- a/src/ServiceControl.Persistence.Tests/EFCore/ErrorIngestionBodyTests.cs +++ b/src/ServiceControl.Persistence.Tests/EFCore/ErrorIngestionBodyTests.cs @@ -13,7 +13,7 @@ class ErrorIngestionBodyTests : ErrorIngestionTestBase const int Cap = 64; [SetUp] - public void ShrinkTheBodyCap() => EFSettings.MaxBodySizeToStore = Cap; + public void ShrinkTheBodyCap() => EFSettings.BodyStorage.MaxBodySizeToStore = Cap; [Test] public async Task Text_within_the_cap_is_stored_inline() diff --git a/src/ServiceControl.Persistence.Tests/EFCore/ErrorIngestionTestBase.cs b/src/ServiceControl.Persistence.Tests/EFCore/ErrorIngestionTestBase.cs index 32bc18e917..8d5e50e9f7 100644 --- a/src/ServiceControl.Persistence.Tests/EFCore/ErrorIngestionTestBase.cs +++ b/src/ServiceControl.Persistence.Tests/EFCore/ErrorIngestionTestBase.cs @@ -3,7 +3,6 @@ namespace ServiceControl.Persistence.Tests; using System; using System.Collections.Generic; using System.Linq; -using System.Threading; using System.Threading.Tasks; using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.DependencyInjection; @@ -20,7 +19,7 @@ abstract class ErrorIngestionTestBase : PersistenceTestBase protected ErrorIngestionTestBase() => RegisterServices = services => services.AddSingleton(RecordedBodies); - protected RecordingBodyStoragePersistence RecordedBodies { get; } = new(); + protected InMemoryBodyStoragePersistence RecordedBodies { get; } = new(); protected EFPersisterSettings EFSettings => (EFPersisterSettings)PersistenceSettings; @@ -97,65 +96,4 @@ async Task Query(Func> query) return await query(dbContext); } - - protected class RecordingBodyStoragePersistence : IBodyStoragePersistence - { - readonly List written = []; - readonly List deleted = []; - - // Body ids whose deletion should throw, to exercise the sweep's tolerate-missing handling. - public HashSet FailDeleteFor { get; } = []; - - public IReadOnlyList Written - { - get - { - lock (written) - { - return [.. written]; - } - } - } - - public IReadOnlyList Deleted - { - get - { - lock (deleted) - { - return [.. deleted]; - } - } - } - - public Task WriteBody(string bodyId, ReadOnlyMemory body, string contentType, CancellationToken cancellationToken = default) - { - lock (written) - { - written.Add(new StoredBody(bodyId, body.ToArray(), contentType)); - } - - return Task.CompletedTask; - } - - public Task ReadBody(string bodyId, CancellationToken cancellationToken = default) => - throw new NotImplementedException(); - - public Task DeleteBody(string bodyId, CancellationToken cancellationToken = default) - { - if (FailDeleteFor.Contains(bodyId)) - { - throw new InvalidOperationException($"Simulated missing body for {bodyId}"); - } - - lock (deleted) - { - deleted.Add(bodyId); - } - - return Task.CompletedTask; - } - - public record StoredBody(string BodyId, byte[] Body, string ContentType); - } } diff --git a/src/ServiceControl.Persistence.Tests/EFCore/ExpectedLengthStreamTests.cs b/src/ServiceControl.Persistence.Tests/EFCore/ExpectedLengthStreamTests.cs new file mode 100644 index 0000000000..67570b2add --- /dev/null +++ b/src/ServiceControl.Persistence.Tests/EFCore/ExpectedLengthStreamTests.cs @@ -0,0 +1,40 @@ +namespace ServiceControl.Persistence.Tests; + +using System.IO; +using NUnit.Framework; +using ServiceControl.Persistence.EFCore.Infrastructure; + +[TestFixture] +class ExpectedLengthStreamTests +{ + [Test] + public void Allows_a_stream_with_the_expected_length() + { + using var stream = new ExpectedLengthStream(new MemoryStream([1, 2, 3]), 3); + + Assert.That(ReadAll(stream), Is.EqualTo(new byte[] { 1, 2, 3 })); + } + + [Test] + public void Throws_when_a_stream_ends_before_the_expected_length() + { + using var stream = new ExpectedLengthStream(new MemoryStream([1, 2]), 3); + + Assert.That(() => ReadAll(stream), Throws.InvalidOperationException); + } + + [Test] + public void Throws_when_a_stream_exceeds_the_expected_length() + { + using var stream = new ExpectedLengthStream(new MemoryStream([1, 2, 3]), 2); + + Assert.That(() => ReadAll(stream), Throws.InvalidOperationException); + } + + static byte[] ReadAll(Stream stream) + { + using var buffer = new MemoryStream(); + stream.CopyTo(buffer); + return buffer.ToArray(); + } +} \ No newline at end of file diff --git a/src/ServiceControl.Persistence.Tests/EFCore/InMemoryBodyStoragePersistence.cs b/src/ServiceControl.Persistence.Tests/EFCore/InMemoryBodyStoragePersistence.cs new file mode 100644 index 0000000000..f3b2c7c0cc --- /dev/null +++ b/src/ServiceControl.Persistence.Tests/EFCore/InMemoryBodyStoragePersistence.cs @@ -0,0 +1,91 @@ +namespace ServiceControl.Persistence.Tests; + +using System; +using System.Collections.Generic; +using System.IO; +using System.Threading; +using System.Threading.Tasks; +using ServiceControl.Persistence.EFCore.Infrastructure; + +class InMemoryBodyStoragePersistence : IBodyStoragePersistence +{ + readonly object gate = new(); + readonly List written = []; + readonly List deleted = []; + readonly Dictionary store = []; + + public HashSet FailDeleteFor { get; } = []; + + public IReadOnlyList Written + { + get + { + lock (gate) + { + return [.. written]; + } + } + } + + public IReadOnlyList Deleted + { + get + { + lock (gate) + { + return [.. deleted]; + } + } + } + + public Task WriteBody(string bodyId, ReadOnlyMemory body, string contentType, CancellationToken cancellationToken = default) + { + var entry = new StoredBody(bodyId, body.ToArray(), contentType); + + lock (gate) + { + written.Add(entry); + store.TryAdd(bodyId, entry); // First write wins, like the real stores. + } + + return Task.CompletedTask; + } + + public Task ReadBody(string bodyId, CancellationToken cancellationToken = default) + { + StoredBody entry; + lock (gate) + { + entry = store.GetValueOrDefault(bodyId); + } + + var result = entry is null + ? null + : new MessageBodyFileResult + { + Stream = new MemoryStream(entry.Body, writable: false), + ContentType = entry.ContentType, + BodySize = entry.Body.Length + }; + + return Task.FromResult(result); + } + + public Task DeleteBody(string bodyId, CancellationToken cancellationToken = default) + { + if (FailDeleteFor.Contains(bodyId)) + { + throw new InvalidOperationException($"Simulated missing body for {bodyId}"); + } + + lock (gate) + { + deleted.Add(bodyId); + store.Remove(bodyId); + } + + return Task.CompletedTask; + } + + public record StoredBody(string BodyId, byte[] Body, string ContentType); +} diff --git a/src/ServiceControl.Persistence.Tests/EFCore/S3BodyStorageTests.cs b/src/ServiceControl.Persistence.Tests/EFCore/S3BodyStorageTests.cs new file mode 100644 index 0000000000..f2a0679f90 --- /dev/null +++ b/src/ServiceControl.Persistence.Tests/EFCore/S3BodyStorageTests.cs @@ -0,0 +1,137 @@ +namespace ServiceControl.Persistence.Tests; + +using System; +using System.IO; +using System.Text; +using System.Threading.Tasks; +using NUnit.Framework; +using ServiceControl.Persistence.EFCore.Abstractions; +using ServiceControl.Persistence.EFCore.Implementation.BodyStorage; +using Testcontainers.LocalStack; + +[TestFixture] +[Platform(Exclude = "Win", Reason = "LocalStack has no Windows container image")] +class S3BodyStorageTests +{ + LocalStackContainer localStack; + S3BodyStoragePersistence store; + + [OneTimeSetUp] + public async Task StartLocalStack() + { + // We need to pin the tag to 4.4.0 because after that version LocalStack has become a paid service for commercial use. + localStack = new LocalStackBuilder("localstack/localstack:4.4.0").Build(); + await localStack.StartAsync(); + } + + [OneTimeTearDown] + public async Task StopLocalStack() + { + if (localStack != null) + { + await localStack.DisposeAsync(); + } + } + + [SetUp] + public async Task CreateBucket() + { + var settings = new S3BodyStorageSettings + { + MinCompressionSize = 64, + ServiceUrl = localStack.GetConnectionString(), + Region = "us-east-1", + Credentials = new S3StaticCredentials { AccessKeyId = "test", SecretAccessKey = "test" }, + BucketName = $"bodies-{Guid.NewGuid():n}" + }; + + await new S3BodyStorageInstaller(settings).Provision(); + store = new S3BodyStoragePersistence(settings); + } + + [Test] + public async Task Round_trips_a_small_uncompressed_body() + { + var bodyId = Guid.NewGuid().ToString(); + var body = Encoding.UTF8.GetBytes("hello world"); + + await store.WriteBody(bodyId, body, "text/plain"); + + var result = await store.ReadBody(bodyId); + + Assert.That(result, Is.Not.Null); + using (result.Stream) + { + Assert.That(ReadAll(result.Stream), Is.EqualTo(body)); + } + + Assert.Multiple(() => + { + Assert.That(result.ContentType, Is.EqualTo("text/plain")); + Assert.That(result.BodySize, Is.EqualTo(body.Length)); + }); + } + + [Test] + public async Task Round_trips_a_large_body_over_the_compression_threshold() + { + var bodyId = Guid.NewGuid().ToString(); + var body = Encoding.UTF8.GetBytes(new string('a', 100_000)); + + await store.WriteBody(bodyId, body, "application/json"); + + var result = await store.ReadBody(bodyId); + + Assert.That(result, Is.Not.Null); + using (result.Stream) + { + Assert.That(ReadAll(result.Stream), Is.EqualTo(body)); + } + + Assert.That(result.BodySize, Is.EqualTo(body.Length)); + } + + [Test] + public async Task Returns_null_for_a_missing_body() => + Assert.That(await store.ReadBody(Guid.NewGuid().ToString()), Is.Null); + + [Test] + public async Task Delete_removes_the_body() + { + var bodyId = Guid.NewGuid().ToString(); + await store.WriteBody(bodyId, Encoding.UTF8.GetBytes("payload"), "text/plain"); + + await store.DeleteBody(bodyId); + + Assert.That(await store.ReadBody(bodyId), Is.Null); + } + + [Test] + public void Delete_of_a_missing_body_does_not_throw() => + Assert.DoesNotThrowAsync(() => store.DeleteBody(Guid.NewGuid().ToString())); + + [Test] + public async Task Rewriting_an_existing_body_keeps_the_first_write() + { + var bodyId = Guid.NewGuid().ToString(); + var original = Encoding.UTF8.GetBytes("original"); + + await store.WriteBody(bodyId, original, "text/plain"); + await store.WriteBody(bodyId, Encoding.UTF8.GetBytes("different"), "text/plain"); + + var result = await store.ReadBody(bodyId); + + Assert.That(result, Is.Not.Null); + using (result.Stream) + { + Assert.That(ReadAll(result.Stream), Is.EqualTo(original), "bodies are immutable, so the first write wins"); + } + } + + static byte[] ReadAll(Stream stream) + { + using var buffer = new MemoryStream(); + stream.CopyTo(buffer); + return buffer.ToArray(); + } +} diff --git a/src/ServiceControl.Persistence/IBodyStorageInstaller.cs b/src/ServiceControl.Persistence/IBodyStorageInstaller.cs new file mode 100644 index 0000000000..c16abbd6be --- /dev/null +++ b/src/ServiceControl.Persistence/IBodyStorageInstaller.cs @@ -0,0 +1,9 @@ +namespace ServiceControl.Persistence; + +using System.Threading; +using System.Threading.Tasks; + +public interface IBodyStorageInstaller +{ + Task Provision(CancellationToken cancellationToken = default); +} diff --git a/src/ServiceControl/Hosting/Commands/SetupCommand.cs b/src/ServiceControl/Hosting/Commands/SetupCommand.cs index 8cfd79d05e..fd3d8ebbe2 100644 --- a/src/ServiceControl/Hosting/Commands/SetupCommand.cs +++ b/src/ServiceControl/Hosting/Commands/SetupCommand.cs @@ -55,6 +55,11 @@ public override async Task Execute(HostArguments args, Settings settings) { await databaseMigrator.ApplyMigrations(); } + + if (scope.ServiceProvider.GetService() is { } bodyStorageInstaller) + { + await bodyStorageInstaller.Provision(); + } } await host.StopAsync();