Add external body storage for audit on sql

docs/README.md

@@ -24,6 +24,7 @@ This section points to sources that explain why ServiceControl is designed the w
 
 - [Ingestion pipeline](ingestion-pipeline.md) — why batch parallelism is a storage decision, not an instance decision
 - [Error ingestion design](error-ingestion-design.md) — why the relational persisters write failed messages with hand-written SQL
+- [Message body storage](message-body-storage.md) — where the Audit SQL Server and PostgreSQL persisters keep message bodies, how retention removes them, and which settings are optional
 - [Bulk retries design](bulk-retries-design.md) — how ServiceControl retries failed messages in bulk
 - [Retries over Azure Storage Queues transport](retries-asq-transport.md) — how retries behave with one or several storage accounts
 - [Data versioning design](data-versioning-design.md) — which invariant keeps a cached API response from going stale

docs/load-testing.md

@@ -21,6 +21,11 @@ Useful flags, passed after `--`:
 | `--tag <tag>` | Test a specific ServiceControl image tag, for example a PR prerelease like `pr-1234` |
 | `--persistence <type>` | Persistence for the error instance: `RavenDb` (default), `SqlServer`, or `PostgreSql` |
 | `--error-ingestion-scale-unit <n>` | Start `n` additional error-ingestion-only ServiceControl instances |
+| `--azure-blob-body-storage "<connection string>"` | Store message bodies in Azure Blob Storage instead of the container file system. Applies only to `SqlServer` and `PostgreSql`. The error instance and its scale-out units share one container, and each audit instance gets its own |
+| `--s3-body-storage <bucket>` | Store message bodies in S3 instead of the container file system. Applies only to `SqlServer` and `PostgreSql`, and is mutually exclusive with `--azure-blob-body-storage`. The error instance and its scale-out units share the `error-bodies/` prefix, and each audit instance gets its own |
+| `--s3-region <region>` | Region for `--s3-body-storage` |
+| `--s3-service-url <url>` | S3-compatible endpoint, such as MinIO or LocalStack. It must be reachable from inside the containers, so `localhost` does not work |
+| `--s3-access-key-id <key>`, `--s3-secret-access-key <secret>` | Static credentials, set both or neither. Without them, the containers use the default AWS credential chain; host credentials are not forwarded |
 
 ## Run against an existing ServiceControl
 
@@ -35,7 +40,7 @@ Open http://localhost:5290. The tool defaults to a ServiceControl instance at `h
 
 Everything is controllable from the web UI, or via the HTTP API:
 
-- Scenarios: the five built-in failure scenarios (`third-party-outage`, `timeout-spike`, `poison-message`, `deserialization-failure`, `background-noise`), each producing errors that group naturally. Start with `POST /api/scenarios/{name}/start` (optional `rate` and `durationSeconds`).
+- Scenarios: the six built-in scenarios (`third-party-outage`, `timeout-spike`, `poison-message`, `deserialization-failure`, `background-noise`, `large-messages`), each producing errors that group naturally. `large-messages` starts at 3 messages per second unless `rate` is set, sends bodies of about 256 KiB (the wire envelope is slightly larger), fails 20% of them to the error queue, and audits the rest. The bypass writer fails every message by design, so it sends the large bodies only as failures. Start with `POST /api/scenarios/{name}/start` (optional `rate` and `durationSeconds`).
 - Bypass writer: high-throughput load that writes failed-message envelopes directly to the ServiceControl error queue, skipping the message handler (`POST /api/bypass/start`).
 - Jobs: retry, archive, search, retention sweep, and custom-check-failure jobs exercise ServiceControl's recoverability, FTS search, and retention pipelines against the load. Jobs do not start automatically. Start them from the UI or `/api/jobs`.
 - Release-test presets: named presets mapped from the [testing scenarios](testing-scenarios.md), for example `POST /api/release-tests/ingestion-load/start`.

docs/message-body-storage.md

@@ -0,0 +1,105 @@
+# Message body storage
+
+The SQL Server and PostgreSQL persisters for ServiceControl.Audit can keep every message body. A small valid text body stays in the database row. With external storage configured, any other body goes there: a directory, an Azure Blob container, or an S3 bucket. Without it, the instance discards bodies over the size limit, as RavenDB does, and bodies that cannot be stored as text. The Audit providers live in `src/ServiceControl.Audit.Persistence.EFCore/Implementation/BodyStorage`. The Error (primary) persisters have their own, in `src/ServiceControl.Persistence.EFCore/Implementation/BodyStorage`, and the two do not share code.
+
+## Placement rules
+
+`MaxBodySizeToStore` (default 100 KB) is the inline threshold. With external storage configured, it does not limit the size of a stored body. The table describes that case; the next section describes the case without it.
+
+| Body | Where it goes | `BodyState` | Inline `BodyText` |
+| --- | --- | --- | --- |
+| Empty | Nowhere | `None` | none |
+| Valid UTF-8 text, at most `MaxBodySizeToStore` bytes | Database row | `Stored` | whole body |
+| Valid UTF-8 text, over `MaxBodySizeToStore` bytes | External storage | `External` | first `MaxBodySizeToStore` bytes, as a search prefix |
+| Binary content type, `Content-Encoding` header, invalid UTF-8, or a NUL character | External storage | `External` | none |
+
+External storage always holds the complete body. Only the inline prefix is searchable, as before.
+
+Rows written before external storage existed keep their meaning: `TooLarge` answers 204 and `NotText` answers 404. The bodies behind those rows were discarded at ingestion and cannot be recovered. `External` is a new value of the existing integer column, so there is no schema change.
+
+## Without external storage
+
+When `MessageBody/StorageType` is not set, nothing is written outside the database and startup needs no provider. Valid UTF-8 text within `MaxBodySizeToStore` stays inline as above. Text over the limit is dropped with the same inline search prefix (`TooLarge`, answers 204). Other bodies are dropped (`NotText`, answers 404; a binary body over the limit is `TooLarge`).
+
+Removing the storage settings from an instance that used them leaves its `External` rows in place. Their bodies answer 204, and retention no longer deletes them from the old location. Clean that location manually.
+
+## Ordering and failures
+
+Ingestion writes external bodies before the batch transaction that inserts the rows. A failed write fails the batch, and no row is inserted. A failed insert can leave an unreferenced body.
+
+Audit stores a body through `IAuditBodyStoragePersistence` under the UTC ingestion hour of its batch and the unique message id: `yyyyMMddHH/{unique message id}`. The file system store writes `{StoragePath}/yyyyMMddHH/{unique message id}.body`, the Azure Blob store uses the blob name `yyyyMMddHH/{unique message id}`, and the S3 store prepends its key prefix. Callers pass the hour explicitly and the stores do not parse it from the id, so reads use the `CreatedOn` of the row. Storage keeps the first write for an hour and id and ignores later writes.
+
+The body URL keeps the audit body id, `yyyyMMddHH-{unique message id}`, which routes the request to the row. The Error (primary) persisters use `IBodyStoragePersistence` with the body id alone; their file system layout remains `{StoragePath}/{body id}.body`.
+
+Reading a body streams it from storage with the stored content type, size, and a version derived from the row. A body missing from storage answers 204.
+
+## Compression
+
+The Audit and Error external storage providers use Brotli compression when the original body size is at least `MessageBody/MinCompressionSize` bytes (default 4096). Smaller bodies are stored uncompressed. This setting applies only to external bodies, not inline `BodyText` or its searchable prefix. Compression does not depend on whether the result is smaller than the original body.
+
+The file system provider uses `CompressionLevel.Fastest`; Azure Blob and S3 use Brotli quality 1 with a 22-bit window. If the cloud encoder reports that compression failed, those providers store the original bytes instead. A file system compression error fails the write.
+
+Each provider records the compression flag and original body size in the file header or object metadata. Reads decompress the body as a stream and return its original bytes, content type, and size. API clients do not need to decompress the storage format.
+
+## Encryption
+
+The external storage providers do not encrypt bodies before writing them. They set no server-side encryption options on Azure Blob or S3 uploads and expose no encryption-key settings. Compression does not protect body contents, and storage authentication does not configure encryption.
+
+Configure encryption at rest and key management in the underlying storage account, S3 bucket, or file system volume. Configure database encryption separately for inline bodies and searchable prefixes, and transport encryption for remote storage and database connections. Whether data is encrypted depends on those systems' settings, not the body storage provider.
+
+## Retention
+
+Retention removes bodies before it removes rows. For each expired unit it calls `IAuditPartitionManager.DeleteBodies`, and it drops the rows only after that call succeeds. If storage fails, the rows and partitions stay and the next sweep retries.
+
+The unit differs by provider, so the delete call is part of `IAuditPartitionManager`:
+
+- SQL Server deletes rows one hour at a time and calls `DeleteBodiesForHour`.
+- PostgreSQL drops whole day partitions without scanning rows and calls `DeleteBodiesForDay`, which covers every hour of that day. A day is only dropped once all of it is older than the retention period, so no body is removed earlier than its row would be.
+
+The file system store deletes the hour directory recursively, so it also removes leftover temporary files in that directory. The day call deletes the 24 hour directories of that day without listing the storage root. The Azure Blob store lists the blobs of the hour or day and deletes them in batches of at most 256. A batch is not atomic, so the store checks the status of every delete, treats a missing blob as deleted, and fails the call on any other error. The S3 store deletes objects in batches of up to 1,000 and checks for per-object failures.
+
+SQL Server only visits hours that have rows, so an unreferenced body in an hour with no rows stays until that location is cleaned manually. PostgreSQL visits every expired day partition, including pre-created partitions without rows, so it also cleans unreferenced bodies of those days.
+
+## Configuration
+
+External storage is optional for Audit instances using SQL Server or PostgreSQL. An Audit instance without `MessageBody/StorageType` starts without it. External storage remains required for Error (primary) instances using these persisters; they fail to start without `ServiceControl/MessageBody/StorageType`. See [Error body placement](error-ingestion-design.md#body-placement).
+
+An invalid value, or missing settings for the chosen type, still fails at startup. Setup creates the directory, container, or bucket.
+
+Audit settings use the `ServiceControl.Audit/` prefix in the application settings file. Provider-specific validation matches the Error instance, whose settings use the `ServiceControl/` prefix.
+
+| Key | Applies to | Default |
+| --- | --- | --- |
+| `MessageBody/StorageType` | all, optional: `FileSystem`, `AzureBlob`, or `S3` | none |
+| `MessageBody/MinCompressionSize` | all external stores, minimum original body size in bytes to compress | 4096 |
+| `MessageBody/FileSystem/StoragePath` | `FileSystem`, required | none |
+| `MessageBody/FileSystem/DataSpaceRemainingThreshold` | `FileSystem`, free space percentage below which the custom check fails | 15 |
+| `MessageBody/Azure/ConnectionString` | `AzureBlob`, shared key or SAS | none |
+| `MessageBody/Azure/ServiceUri` | `AzureBlob`, managed identity | none |
+| `MessageBody/Azure/ManagedIdentityClientId` | `AzureBlob`, managed identity | none |
+| `MessageBody/Azure/AuthorityHost` | `AzureBlob`, managed identity | SDK default |
+| `MessageBody/Azure/ContainerName` | `AzureBlob` | `audit-bodies` |
+| `MessageBody/S3/BucketName` | `S3`, required | none |
+| `MessageBody/S3/KeyPrefix` | `S3` | `audit-bodies/` |
+| `MessageBody/S3/Region` | `S3` | SDK default |
+| `MessageBody/S3/ServiceUrl` | `S3` | SDK default |
+| `MessageBody/S3/AccessKeyId`, `MessageBody/S3/SecretAccessKey` | `S3`, set both or neither | default AWS credential chain |
+
+Azure requires exactly one of the connection string and the service URI. `MaxBodySizeToStore` keeps its existing top-level setting.
+
+```xml
+<add key="ServiceControl.Audit/MessageBody/StorageType" value="FileSystem" />
+<add key="ServiceControl.Audit/MessageBody/FileSystem/StoragePath" value="D:\ServiceControl.Audit\bodies" />
+```
+
+```xml
+<add key="ServiceControl.Audit/MessageBody/StorageType" value="S3" />
+<add key="ServiceControl.Audit/MessageBody/S3/BucketName" value="my-bodies" />
+<add key="ServiceControl.Audit/MessageBody/S3/KeyPrefix" value="audit-instance-1/" />
+```
+
+## One location per instance
+
+Give each Audit instance its own directory, container, or key prefix, and never share one with an Error instance. Retention deletes whole hour or day groups, so a shared location lets one instance delete bodies that another still references. The Audit defaults differ from the Error defaults (`error-bodies`, `error-bodies/`) for this reason, but they are the same for every Audit instance, so a second Audit instance in the same account must set its own container name or key prefix.
+
+With `FileSystem` storage, an instance registers the `ServiceControl body storage` custom check that reports low free space on the drive.

src/Directory.Packages.props

@@ -11,6 +11,7 @@
     <PackageVersion Include="Azure.ResourceManager.Monitor" Version="1.3.1" />
     <PackageVersion Include="Azure.ResourceManager.ServiceBus" Version="1.2.0" />
     <PackageVersion Include="Azure.Storage.Blobs" Version="12.29.2" />
+    <PackageVersion Include="Azure.Storage.Blobs.Batch" Version="12.26.1" />
     <PackageVersion Include="ByteSize" Version="2.1.2" />
     <PackageVersion Include="Caliburn.Micro" Version="5.0.258" />
     <PackageVersion Include="DnsClient" Version="1.8.0" />

src/ServiceControl.AcceptanceTests/ClockRegistrationTests.cs

@@ -3,10 +3,11 @@ namespace ServiceControl.AcceptanceTests
     using System;
     using System.Runtime.Loader;
     using System.Threading.Tasks;
+    using Microsoft.AspNetCore.Builder;
     using Microsoft.Extensions.DependencyInjection;
-    using Microsoft.Extensions.Hosting;
     using NServiceBus;
     using NUnit.Framework;
+    using global::ServiceControl.Infrastructure.WebApi;
     using Particular.ServiceControl;
     using ServiceBus.Management.Infrastructure.Settings;
 
@@ -41,9 +42,10 @@ public void Should_keep_the_clock_the_host_registered()
             var endpointConfiguration = new EndpointConfiguration(settings.InstanceName);
             endpointConfiguration.AssemblyScanner().Disable = true;
 
-            var hostBuilder = Host.CreateApplicationBuilder();
+            var hostBuilder = WebApplication.CreateBuilder();
             hostBuilder.Services.AddSingleton(hostClock);
             hostBuilder.AddServiceControl(settings, endpointConfiguration);
+            hostBuilder.AddServiceControlApi(settings.CorsSettings);
 
             using var host = hostBuilder.Build();
 

src/ServiceControl.Audit.AcceptanceTests.PostgreSql/AcceptanceTestStorageConfiguration.cs

@@ -3,6 +3,7 @@ namespace ServiceControl.Audit.AcceptanceTests.PostgreSql;
 using System;
 using System.Collections.Concurrent;
 using System.Collections.Generic;
+using System.IO;
 using System.Threading;
 using System.Threading.Tasks;
 using ServiceControl.Audit.AcceptanceTests.TestSupport;
@@ -22,18 +23,39 @@ public async Task<IDictionary<string, string>> CustomizeSettings(CancellationTok
 
         schemas.Add(schema);
 
+        var bodyStoragePath = Directory.CreateTempSubdirectory("sc_audit_at_bodies_").FullName;
+        bodyStoragePaths.Add(bodyStoragePath);
+
         return new Dictionary<string, string>
         {
             [EFPersistenceConfigurationBase.ConnectionStringKey] = connectionString,
-            [EFPersistenceConfigurationBase.SchemaKey] = schema
+            [EFPersistenceConfigurationBase.SchemaKey] = schema,
+            ["MessageBody/StorageType"] = "FileSystem",
+            ["MessageBody/FileSystem/StoragePath"] = bodyStoragePath
         };
     }
 
     public async Task Cleanup(CancellationToken cancellationToken = default)
     {
-        while (schemas.TryTake(out var schema))
+        try
+        {
+            while (schemas.TryTake(out var schema))
+            {
+                await TestSchema.Drop(connectionString, schema, cancellationToken);
+            }
+        }
+        finally
         {
-            await TestSchema.Drop(connectionString, schema, cancellationToken);
+            while (bodyStoragePaths.TryTake(out var bodyStoragePath))
+            {
+                try
+                {
+                    Directory.Delete(bodyStoragePath, recursive: true);
+                }
+                catch (DirectoryNotFoundException)
+                {
+                }
+            }
         }
     }
 
@@ -51,5 +73,6 @@ sealed class Release : IDisposable
 
     static readonly SemaphoreSlim initializationLock = new(1, 1);
     readonly ConcurrentBag<string> schemas = [];
+    readonly ConcurrentBag<string> bodyStoragePaths = [];
     string connectionString;
 }

src/ServiceControl.Audit.AcceptanceTests.PostgreSql/ServiceControl.Audit.AcceptanceTests.PostgreSql.csproj

@@ -30,12 +30,16 @@
 
   <ItemGroup>
     <Compile Include="..\ServiceControl.Audit.AcceptanceTests\**\*.cs" Exclude="..\ServiceControl.Audit.AcceptanceTests\obj\**\*.*;..\ServiceControl.Audit.AcceptanceTests\bin\**\*.*" LinkBase="Shared" />
+    <Compile Include="..\ServiceControl.Audit.AcceptanceTests.SqlServer\When_messages_with_big_bodies_are_ingested.ExternalBody.cs" />
     <Compile Include="..\ServiceControl.Persistence.Tests.PostgreSql\PostgreSqlSharedContainer.cs" />
     <Compile Include="..\ServiceControl.Persistence.Tests.PostgreSql\TestSchema.cs" />
     <Compile Include="..\ServiceControl.Persistence.Tests.PostgreSql\StopSharedPostgreSql.cs" />
 
     <!-- The EF Core persisters always index bodies for full text search, whatever EnableFullTextSearchOnBodies says. -->
     <Compile Remove="..\ServiceControl.Audit.AcceptanceTests\Auditing\When_processed_message_searched_by_body_content.FullTextSearchDisabled.cs" />
+
+    <!-- These tests configure external body storage, which keeps every body, so a body over MaxBodySizeToStore is served rather than answering no content. -->
+    <Compile Remove="..\ServiceControl.Audit.AcceptanceTests\Auditing\When_messages_with_big_bodies_are_ingested.MaxBodySize.cs" />
   </ItemGroup>
 
 </Project>