Add external body storage for audit

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 required
 - [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 |
+| `--azureblob-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 `--azureblob-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,79 @@
+# Message body storage
+
+The SQL Server and PostgreSQL persisters for ServiceControl.Audit keep every message body. A small valid text body stays in the database row. Any other body goes to external storage: a directory, an Azure Blob container, or an S3 bucket. The providers live in `src/ServiceControl.Persistence.BodyStorage` and are shared with the Error (primary) persisters.
+
+## Placement rules
+
+`MaxBodySizeToStore` (default 100 KB) is the inline threshold. It does not limit the size of a stored body.
+
+| 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.
+
+## 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.
+
+A body is stored under its audit body id, `yyyyMMddHH-{unique message id}`, the same id used in the body URL. Storage keeps the first write of an id and ignores later writes.
+
+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.
+
+## Retention
+
+Retention removes bodies before it removes rows. For each expired unit it calls `IBodyStoragePersistence.DeleteBodiesWithPrefix`, 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 prefix is part of `IAuditPartitionManager`:
+
+- SQL Server deletes rows one hour at a time. The prefix is the hour, `yyyyMMddHH-`.
+- PostgreSQL drops whole day partitions without scanning rows. The prefix is the day, `yyyyMMdd`, and it matches 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.
+
+Prefixes must not be empty and must not contain wildcard or path characters. 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 required. The instance fails to start when `MessageBody/StorageType` is missing, which includes existing SQL Server and PostgreSQL Audit instances that upgrade without adding it. Setup creates the directory, container, or bucket.
+
+Settings use the `ServiceControl.Audit/` prefix in the application settings file. The validation matches the Error instance, whose settings use the `ServiceControl/` prefix.
+
+| Key | Applies to | Default |
+| --- | --- | --- |
+| `MessageBody/StorageType` | all, required: `FileSystem`, `AzureBlob`, or `S3` | none |
+| `MessageBody/MinCompressionSize` | all, minimum 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 by id prefix, and the prefixes of two instances can match the same hours. An instance that shares a location can delete bodies that another instance 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/ServiceControl.AcceptanceTests.PostgreSql/AcceptanceTestStorageConfiguration.cs

@@ -1,5 +1,6 @@
 namespace ServiceControl.AcceptanceTests.PostgreSql;
 
+using ServiceControl.Persistence.BodyStorage;
 using System;
 using System.Collections.Concurrent;
 using System.IO;

src/ServiceControl.AcceptanceTests.SqlServer/AcceptanceTestStorageConfiguration.cs

@@ -1,5 +1,6 @@
 namespace ServiceControl.AcceptanceTests.SqlServer;
 
+using ServiceControl.Persistence.BodyStorage;
 using System;
 using System.Collections.Concurrent;
 using System.IO;

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" />
+
+    <!-- The EF Core persisters keep 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>

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

@@ -3,6 +3,7 @@ namespace ServiceControl.Audit.AcceptanceTests.SqlServer;
 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.SqlServer/ServiceControl.Audit.AcceptanceTests.SqlServer.csproj

@@ -36,6 +36,9 @@
 
     <!-- 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" />
+
+    <!-- The EF Core persisters keep 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>

src/ServiceControl.Audit.AcceptanceTests.SqlServer/When_messages_with_big_bodies_are_ingested.ExternalBody.cs

@@ -0,0 +1,49 @@
+namespace ServiceControl.Audit.AcceptanceTests.Auditing
+{
+    using System.Text.Json;
+    using System.Threading.Tasks;
+    using AcceptanceTesting;
+    using AcceptanceTesting.EndpointTemplates;
+    using Audit.Auditing.MessagesView;
+    using NServiceBus;
+    using NServiceBus.AcceptanceTesting;
+    using NUnit.Framework;
+
+    partial class When_messages_with_big_bodies_are_ingested
+    {
+        [Test]
+        public async Task Should_serve_the_full_body_when_it_is_larger_than_MaxBodySizeToStore()
+        {
+            SetSettings = settings => settings.MaxBodySizeToStore = MAX_BODY_SIZE;
+
+            var expected = new byte[MAX_BODY_SIZE + 1000];
+            byte[] body = null;
+
+            await Define<Context>()
+                .WithEndpoint<FatMessageEndpoint>(c => c.When(b => b.SendLocal(new BigFatMessage { BigFatBody = expected })))
+                .Done(
+                    async c =>
+                    {
+                        if (c.MessageId == null)
+                        {
+                            return false;
+                        }
+
+                        var result = await this.TryGetSingle<MessagesView>("/api/messages", r => r.MessageId == c.MessageId);
+                        MessagesView auditMessage = result;
+                        if (!result)
+                        {
+                            return false;
+                        }
+
+                        body = await this.DownloadData($"/api{auditMessage.BodyUrl}");
+
+                        return true;
+                    })
+                .Run();
+
+            using var document = JsonDocument.Parse(body);
+            Assert.That(document.RootElement.GetProperty(nameof(BigFatMessage.BigFatBody)).GetBytesFromBase64(), Is.EqualTo(expected));
+        }
+    }
+}

src/ServiceControl.Audit.AcceptanceTests/Auditing/When_messages_with_big_bodies_are_ingested.MaxBodySize.cs

@@ -0,0 +1,55 @@
+namespace ServiceControl.Audit.AcceptanceTests.Auditing
+{
+    using System.Net;
+    using System.Threading.Tasks;
+    using AcceptanceTesting;
+    using AcceptanceTesting.EndpointTemplates;
+    using Audit.Auditing.MessagesView;
+    using NServiceBus;
+    using NServiceBus.AcceptanceTesting;
+    using NUnit.Framework;
+
+    partial class When_messages_with_big_bodies_are_ingested
+    {
+        [Test]
+        public async Task Should_get_an_empty_audit_message_body_when_configured_MaxBodySizeToStore_is_less_then_message_size()
+        {
+            //Arrange
+            SetSettings = settings => settings.MaxBodySizeToStore = MAX_BODY_SIZE;
+
+            byte[] body = null;
+
+            //Act
+            await Define<Context>()
+                .WithEndpoint<FatMessageEndpoint>(c => c.When(b => b.SendLocal(
+                    new BigFatMessage // An endpoint that is configured for audit
+                    {
+                        BigFatBody = new byte[MAX_BODY_SIZE + 1000]
+                    }))
+                )
+                .Done(
+                    async c =>
+                    {
+                        if (c.MessageId == null)
+                        {
+                            return false;
+                        }
+
+                        var result = await this.TryGetSingle<MessagesView>("/api/messages", r => r.MessageId == c.MessageId);
+                        MessagesView auditMessage = result;
+                        if (!result)
+                        {
+                            return false;
+                        }
+
+                        body = await this.DownloadData($"/api{auditMessage.BodyUrl}", HttpStatusCode.NoContent);
+
+                        return true;
+                    })
+                .Run();
+
+            //Assert
+            Assert.That(body, Is.Empty);
+        }
+    }
+}