[SPIKE] Add plan for host audit ingestion in the primary instance

docs/audit-ingestion-in-the-primary.md

@@ -0,0 +1,114 @@
+# Audit ingestion in the primary instance
+
+## Overview
+
+Storage that advertises `SupportsAuditIngestion` in its `persistence.manifest` can hold audit data
+alongside the primary's own data, which lets the primary ServiceControl process ingest the audit
+queue itself instead of relying on a separate ServiceControl.Audit instance.
+
+The standalone RavenDB audit instance is unaffected. RavenDB does not advertise audit support, does
+not gain combined hosting, and keeps its own executable, settings, API and installers.
+
+No shipped persister advertises audit support yet, so on every existing deployment the audit
+component registers nothing and behavior is unchanged.
+
+## Deployment modes
+
+| Mode | How | What it runs |
+| --- | --- | --- |
+| Normal primary, audit ingestion on | Default where the persister advertises audit support | The audit receiver, the audit capabilities, the primary API and everything a normal primary runs |
+| Normal primary, audit ingestion off | `ServiceControl/IngestAuditMessages=false` | Everything above except the audit receiver. Local audit queries, failed audit tooling and `/api/connection` stay active, because other processes may still be ingesting |
+| Audit ingestion only | `ServiceControl.exe --audit-ingestion-only` | The audit receiver, the endpoint monitor it depends on, this node's custom checks, and the health endpoints. No NServiceBus endpoint, no API, no retention, no licensing |
+
+`--audit-ingestion-only` and `--error-ingestion-only` cannot be combined. Each queue gets its own
+worker pool so the two can be scaled independently, so run one process per mode.
+
+## Settings
+
+The primary reads the audit settings under the same key names the audit instance uses, so an audit
+capable primary is configured exactly the way an audit instance is configured today.
+
+| Setting | Default | Notes |
+| --- | --- | --- |
+| `ServiceControl/IngestAuditMessages` | `true` | Applies to the normal primary only. Always on under `--audit-ingestion-only`, and has no effect where the persister does not support audit |
+| `ServiceBus/AuditQueue` | `audit` | The queue this instance drains |
+| `ServiceBus/AuditLogQueue` | the subscoped audit queue name | Only used when forwarding is on |
+| `ServiceControl/ForwardAuditMessages` | `false` | |
+| `ServiceControl/AuditRetentionPeriod` | unset | Already existed. Validated between 1 hour and 365 days |
+| `ServiceControl/MaximumAuditIngestionConcurrencyLevel` | `32` | Independent of the primary endpoint's concurrency, which is what `MaximumConcurrencyLevel` sets |
+| `ServiceControl/TimeToRestartAuditIngestionAfterFailure` | 60 seconds | Mirrors the error equivalent |
+| `ServiceControl/OtlpEndpointUrl` | unset | Enables the OpenTelemetry metrics exporter |
+| `ServiceControl/MessageBody/FileSystem/PathIsShared` | `false` | Required by both ingestion only modes when body storage is the file system |
+
+### Setting collisions
+
+`ServiceControl` and `ServiceControl.Audit` settings can both be set by bare environment variable
+name, and `ServiceBus/AuditQueue` is literally the same key for both processes. A combined primary
+and a standalone audit instance sharing one environment file therefore collide on
+`INGESTAUDITMESSAGES`, `AUDITRETENTIONPERIOD`, `FORWARDAUDITMESSAGES` and `SERVICEBUS_AUDITQUEUE`.
+
+That combination is unsupported. The primary logs a warning at startup when it has audit ingestion
+enabled and audit remotes configured at the same time, because that is the shape most likely to hit
+the collision.
+
+## Queue ownership
+
+The setup path creates the audit queue, and the audit forwarding queue when forwarding is enabled.
+Ingestion only workers run no installers: they never create queues, never apply database migrations
+and never provision body storage. Run setup from a normal instance before starting any worker.
+
+Transport operations remain in the audit ingestion path for two reasons only:
+
+- **Forwarding**, when `ForwardAuditMessages` is on.
+- **Retry acknowledgements**. `ServiceControl.Retry.AcknowledgementQueue` is stamped by whichever
+  instance issued the retry, so the acknowledgement cannot be short-circuited into the local
+  database. In a combined host it is dispatched to the local error queue and comes straight back in
+  through local error ingestion, which is exactly what happens today.
+
+Endpoints detected from audit headers are written straight to the shared `KnownEndpoints` table
+through the ingestion unit of work, rather than sent to the primary's input queue.
+
+## Body storage
+
+Audit and failed message bodies share one store, and each owns a prefixed keyspace, so an edited
+message's failed body and its audited body do not collide. `IBodyStorage.TryFetch` resolves in a
+fixed order: failed message by `UniqueMessageId`, then failed message by `MessageId`, then audit
+message by `UniqueMessageId`.
+
+Every ingesting process must write bodies somewhere every host can read. Blob and S3 storage
+qualify. File system storage qualifies only if the path is a shared mount, which nothing in the
+settings can detect, so both ingestion only modes refuse to start unless
+`ServiceControl/MessageBody/FileSystem/PathIsShared` asserts it.
+
+## Health endpoints
+
+Both ingestion only hosts map the same two routes, anonymously, returning JSON:
+
+- `/health` is liveness. It answers "is this process still serving" and is what a container health
+  check should restart on.
+- `/health/ready` additionally reports whether the ingestion this host exists to do is happening.
+  An audit ingestion only host answers for `audit-ingestion` and not for `error-ingestion`.
+
+## Querying
+
+Local audit data is served through the existing primary routes under their existing policies:
+`/api/messages` and its variants on `error:messages:view`, `/api/sagas/{id}` on
+`error:sagas:view`, and `endpoints/{endpoint}/audit-count` on `error:messages:view`. A primary
+configured with an audit remote already serves that remote's audit data under those policies today,
+so nothing about the `my/routes` manifest or ServicePulse navigation changes.
+
+Additional audit remotes keep working. The scatter gather runs the local query first and merges the
+remotes after, so a primary can hold audit data locally, query remotes, or both.
+
+Where one local result set contains both failed and audited messages, three rules apply, and
+`LocalMessagesView.Merge` implements them for any persister:
+
+1. **Precedence.** A message that both failed and was audited shows as failed.
+2. **Paging.** The local result is already at most one page, after deduplication.
+3. **Counting.** A message that both failed and was audited is counted once.
+
+## Packaging
+
+The audit runtime ships inside the existing primary artifact. There is no new assembly and no new
+deployment unit. The primary gains three OpenTelemetry package references, which the copied
+ingestion metrics use, exported only when `OtlpEndpointUrl` is set.

src/Particular.LicensingComponent.UnitTests/AuditQuery_Tests.cs

@@ -74,6 +74,43 @@ public async Task Should_return_audit_remotes()
         }
     }
 
+    [Test]
+    public async Task Should_return_the_local_audit_source_alongside_the_remotes()
+    {
+        //Arrange
+        var auditQuery = new AuditQuery(NullLogger<AuditQuery>.Instance, new FakeEndpointApi(), new FakeAuditCountApi(),
+            new ConfigurationApi_ReturningOneValidAuditConfig(), new LocalAuditSource_ForThisInstance());
+
+        //Act
+        var remotes = await auditQuery.GetAuditRemotes();
+
+        //Assert
+        Assert.That(remotes, Has.Count.EqualTo(2), "The local audit source and the remote should both be reported");
+
+        var local = remotes.Single(remote => remote.ApiUri == "http://localhost:33333/api/");
+
+        using (Assert.EnterMultipleScope())
+        {
+            Assert.That(local.Queues, Does.Contain("audit"), "the local audit queue must be recognised as a platform endpoint");
+            Assert.That(local.Queues, Does.Contain("audit.log"));
+            Assert.That(local.Transport, Is.EqualTo("LearningTransport"), "the report's audit service metadata is built from this");
+        }
+    }
+
+    [Test]
+    public async Task Should_not_report_a_local_audit_source_that_is_disabled()
+    {
+        //Arrange
+        var auditQuery = new AuditQuery(NullLogger<AuditQuery>.Instance, new FakeEndpointApi(), new FakeAuditCountApi(),
+            new ConfigurationApi_ReturningOneValidAuditConfig(), new LocalAuditSource_Disabled());
+
+        //Act
+        var remotes = await auditQuery.GetAuditRemotes();
+
+        //Assert
+        Assert.That(remotes, Has.Count.EqualTo(1));
+    }
+
     [Test]
     public async Task Should_return_successful_audit_connection_if_instances_exist_and_are_online()
     {
@@ -204,6 +241,29 @@ public Task<List<Endpoint>> GetEndpoints(CancellationToken cancellationToken = d
 
     }
 
+    class LocalAuditSource_ForThisInstance : ILocalAuditSource
+    {
+        public bool Enabled => true;
+
+        public RemoteInstanceInformation Describe() => new()
+        {
+            ApiUri = "http://localhost:33333/api/",
+            VersionString = "6.0.0",
+            SemanticVersion = new NuGet.Versioning.SemanticVersion(6, 0, 0),
+            Status = "online",
+            Retention = TimeSpan.FromDays(10),
+            Queues = ["audit", "audit.log"],
+            Transport = "LearningTransport"
+        };
+    }
+
+    class LocalAuditSource_Disabled : ILocalAuditSource
+    {
+        public bool Enabled => false;
+
+        public RemoteInstanceInformation Describe() => throw new InvalidOperationException("Describe must not be called when the source is disabled.");
+    }
+
     class AuditCountApi_ReturningThreeAuditCounts : IAuditCountApi
     {
         public async Task<IList<AuditCount>> GetEndpointAuditCounts(string endpoint, CancellationToken cancellationToken = default)

src/Particular.LicensingComponent/AuditThroughput/AuditQuery.cs

@@ -7,7 +7,7 @@
     using ServiceControl.Api;
     using AuditCount = Contracts.AuditCount;
 
-    public class AuditQuery(ILogger<AuditQuery> logger, IEndpointsApi endpointsApi, IAuditCountApi auditCountApi, IConfigurationApi configurationApi) : IAuditQuery
+    public class AuditQuery(ILogger<AuditQuery> logger, IEndpointsApi endpointsApi, IAuditCountApi auditCountApi, IConfigurationApi configurationApi, ILocalAuditSource? localAuditSource = null) : IAuditQuery
     {
         // Customers are expected to run at least version 4.29 for their Audit instances
         public SemanticVersion MinAuditCountsVersion => new(4, 29, 0);
@@ -45,6 +45,11 @@ public async Task<List<RemoteInstanceInformation>> GetAuditRemotes(CancellationT
                 var remotes = await configurationApi.GetRemoteConfigs(cancellationToken);
                 var remotesInfo = new List<RemoteInstanceInformation>();
 
+                if (localAuditSource is { Enabled: true })
+                {
+                    remotesInfo.Add(localAuditSource.Describe());
+                }
+
                 if (remotes.Any())
                 {
                     List<string> queues = [];

src/Particular.LicensingComponent/AuditThroughput/ILocalAuditSource.cs

@@ -0,0 +1,15 @@
+namespace Particular.LicensingComponent.AuditThroughput;
+
+using Particular.LicensingComponent.Contracts;
+
+/// <summary>
+/// Audit throughput collection is driven entirely by audit remotes. A primary that holds audit data
+/// itself has no remote to describe it, so without this its own audit queues are counted as customer
+/// endpoints and the audit service metadata in the licensing report is blank.
+/// </summary>
+public interface ILocalAuditSource
+{
+    bool Enabled { get; }
+
+    RemoteInstanceInformation Describe();
+}

src/Particular.LicensingComponent/LicensingComponentHostBuilderExtensions.cs

@@ -1,4 +1,4 @@
-namespace Particular.LicensingComponent;
+namespace Particular.LicensingComponent;
 
 using AuditThroughput;
 using BrokerThroughput;

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

@@ -12,6 +12,7 @@
   <ItemGroup>
     <ProjectReference Include="..\ServiceControl.AcceptanceTesting\ServiceControl.AcceptanceTesting.csproj" />
     <ProjectReference Include="..\ServiceControl.Persistence.EFCore.PostgreSql\ServiceControl.Persistence.EFCore.PostgreSql.csproj" />
+    <ProjectReference Include="..\ServiceControl.Persistence.Tests.AuditCapable\ServiceControl.Persistence.Tests.AuditCapable.csproj" />
     <ProjectReference Include="..\ServiceControl.Transports.Learning\ServiceControl.Transports.Learning.csproj" />
     <ProjectReference Include="..\ServiceControl\ServiceControl.csproj" />
     <ProjectReference Include="..\TestHelper\TestHelper.csproj" />

src/ServiceControl.AcceptanceTests.RavenDB/ServiceControl.AcceptanceTests.RavenDB.csproj

@@ -36,6 +36,8 @@
 
     <!-- Error ingestion only mode is supported on SQL Server and PostgreSQL storage only. -->
     <Compile Remove="..\ServiceControl.AcceptanceTests\Recoverability\When_hosting_error_ingestion_only.cs" />
+    <Compile Remove="..\ServiceControl.AcceptanceTests\Auditing\When_composing_audit_ingestion_in_the_primary.cs" />
+    <Compile Remove="..\ServiceControl.AcceptanceTests\Auditing\When_hosting_audit_ingestion_only.cs" />
   </ItemGroup>
 
   <ItemGroup>

src/ServiceControl.AcceptanceTests.SqlServer/ServiceControl.AcceptanceTests.SqlServer.csproj

@@ -12,6 +12,7 @@
   <ItemGroup>
     <ProjectReference Include="..\ServiceControl.AcceptanceTesting\ServiceControl.AcceptanceTesting.csproj" />
     <ProjectReference Include="..\ServiceControl.Persistence.EFCore.SqlServer\ServiceControl.Persistence.EFCore.SqlServer.csproj" />
+    <ProjectReference Include="..\ServiceControl.Persistence.Tests.AuditCapable\ServiceControl.Persistence.Tests.AuditCapable.csproj" />
     <ProjectReference Include="..\ServiceControl.Transports.Learning\ServiceControl.Transports.Learning.csproj" />
     <ProjectReference Include="..\ServiceControl\ServiceControl.csproj" />
     <ProjectReference Include="..\TestHelper\TestHelper.csproj" />