Generate audit and saga load against a ServiceControl.Audit instance (#5969)

* Add direct audit-queue writer and saga episode generation to the testing tool

Extends the testing tool to drive ServiceControl.Audit ingestion, not just
error ingestion. The new writer sends audit envelopes straight to the audit
queue, including whole saga episodes (New/Updated/Completed messages plus
their snapshots) linked via InvokedSagas/SagaStateChange headers so the
audit instance can resolve saga history and drill-through.

Also wires audit instances in the AppHost to the same SQL Server/PostgreSQL
backend as the primary via EF persistence, with ingestion tuning, OTel
export and custom-check reporting to the primary, and extends the Grafana
dashboard and docs to cover the new audit load paths.

* Bump Aspire to 13.6.1 and default audit instances to 1

Updates Aspire.Hosting.* packages and the AppHost SDK to 13.6.1, enables
the CLI bundle, and adds appsettings.json for logging configuration.
Also changes the default audit-instances option from 0 to 1 so audit
load is exercised out of the box.

* Fix audit instance creation & remove warning

* Default audit to 1 instance, change dashboard calc

* Pin pacakge with CVE

* retry folder delete to avoid locking problem on windows ci builds

* Add cancellation support

* Update test

---------

Co-authored-by: Rhys Bevilaqua <rhys.bevilaqua@particular.net>

Author: johnsimons

src/ServiceControlInstaller.Engine.UnitTests/Validation/PackagedRavenDBTests.cs

@@ -77,10 +77,7 @@ await store.Maintenance.SendAsync(new PutIndexesOperation(new IndexDefinition
             finally
             {
                 await StopServer();
-                if (Directory.Exists(root))
-                {
-                    FileUtils.DeleteDirectory(root, recursive: true, contentsOnly: false);
-                }
+                await TryDeleteDirectory(root, CancellationToken.None);
             }
 
             async Task StartServer()
@@ -143,9 +140,9 @@ async Task StopServer()
                         server.Kill(entireProcessTree: true);
                     }
 
-                    await server.WaitForExitAsync(CancellationToken.None).WaitAsync(TimeSpan.FromSeconds(15));
-                    TestContext.Out.WriteLine(await standardOutput.WaitAsync(TimeSpan.FromSeconds(15)));
-                    TestContext.Out.WriteLine(await standardError.WaitAsync(TimeSpan.FromSeconds(15)));
+                    await server.WaitForExitAsync(CancellationToken.None).WaitAsync(TimeSpan.FromSeconds(15), cancellationToken);
+                    await TestContext.Out.WriteLineAsync(await standardOutput.WaitAsync(TimeSpan.FromSeconds(15), cancellationToken));
+                    await TestContext.Out.WriteLineAsync(await standardError.WaitAsync(TimeSpan.FromSeconds(15), cancellationToken));
                 }
                 finally
                 {
@@ -155,6 +152,30 @@ async Task StopServer()
             }
         }
 
+        // Cleanup of the temp folder must not fail the test; Directory.Delete keeps the path stable across retries.
+        static async Task TryDeleteDirectory(string root, CancellationToken cancellationToken, int attempts = 5, TimeSpan? delay = null)
+        {
+            var retryDelay = delay ?? TimeSpan.FromMilliseconds(500);
+            for (var attempt = 1; Directory.Exists(root); attempt++)
+            {
+                try
+                {
+                    Directory.Delete(root, recursive: true);
+                    return;
+                }
+                catch (Exception ex) when (ex is IOException or UnauthorizedAccessException)
+                {
+                    if (attempt == attempts)
+                    {
+                        await TestContext.Out.WriteLineAsync($"WARNING: Could not delete '{root}' after {attempts} attempts: {ex}");
+                        return;
+                    }
+
+                    await Task.Delay(retryDelay * attempt, cancellationToken);
+                }
+            }
+        }
+
         static async Task AssertDocumentAndIndex(IDocumentStore store, CancellationToken cancellationToken)
         {
             using var session = store.OpenAsyncSession();

tools/testing-tool/README.md

@@ -4,7 +4,7 @@ A stateless, horizontally-scalable .NET 10 service that generates error load and
 scenarios against a test ServiceControl instance to validate its error-ingestion performance, with
 OpenTelemetry observability and a simple web UI for manual scenario control.
 
-The tool targets the error ingestion path only — audit testing is out of scope.
+The tool drives both ingestion paths: failed messages into the primary ServiceControl instance, and audited messages and saga snapshots into a ServiceControl.Audit instance.
 
 ## What it does
 
@@ -73,6 +73,28 @@ failed message and groups it like the equivalent handler failure. Control via:
 - `POST /api/bypass/stop`
 - `GET /api/bypass/status`
 
+### Direct audit-queue writer
+
+The audit-queue counterpart of the bypass writer. It writes audit envelopes straight to the ServiceControl.Audit instance's audit queue, so audit ingestion can be loaded without any endpoint ever processing a message. Control via:
+
+- `POST /api/audit-bypass/start` — `{ "rate": 100, "parallelism": 4, "sagaSnapshotRatio": 0.2, "durationSeconds": 60 }`
+- `POST /api/audit-bypass/stop`
+- `GET /api/audit-bypass/status`
+
+Most claims emit one processed-message envelope carrying the headers a real endpoint would have stamped: `NServiceBus.ProcessingEndpoint`, the host ids, and `TimeSent`, `ProcessingStarted` and `ProcessingEnded` spread so delivery, processing and critical times are not zero. They rotate across a pool of simulated endpoints, message types and conversations.
+
+A share of claims instead emit a whole **saga episode**: the three messages that drove one saga through New, Updated and Completed, each followed by the snapshot it produced. The episode is written by a single worker so the phases reach the queue in order, and the pieces are linked the way a real deployment links them:
+
+- each driving message carries `NServiceBus.InvokedSagas` (`SagaType:SagaId`) and `ServiceControl.SagaStateChange` (`SagaId:New|Updated|Completed`), which is the only thing that lets the audit instance attach saga information to an audited message
+- each snapshot's `Initiator.InitiatingMessageId` is the id of the message directly before it, so drill-through from the saga view resolves to a real audited message
+- the snapshot body is written in exact CLR casing, because ServiceControl deserializes it with a source-generated, case-sensitive context, and its `EnclosedMessageTypes` header is the bare unqualified type name, which is what the audit instance compares against to route a message to its saga branch
+
+`sagaSnapshotRatio` is saga snapshots per processed message; the default `0.2` emits two saga snapshots for every ten processed messages. `0` disables saga episodes.
+
+Audit ingestion needs a ServiceControl.Audit instance on SQL Server or PostgreSQL, so pair `--audit-instances` with `--persistence:PostgreSql` or `--persistence:SqlServer`. Under `--persistence:RavenDb` the audit instance runs on RavenDB and the EF path is not exercised.
+
+The testing tool's own endpoint also audits its successfully handled load messages to the same queue, which is what makes its installers create the queue before the writer first sends into it.
+
 ### Release-test scenario presets
 
 The tool ships with presets mapped from [docs/testing-scenarios.md](/docs/testing-scenarios.md) so
@@ -162,11 +184,11 @@ separator) or `--name:value` (colon separator):
 
 | Flag | Default | Values | Description |
 |---|---|---|---|
-| `--persistence` | `RavenDb` | `RavenDb`, `SqlServer`, `PostgreSql` | Persistence backend for the ServiceControl error instance |
+| `--persistence` | `RavenDb` | `RavenDb`, `SqlServer`, `PostgreSql` | Persistence backend for the ServiceControl error instance and for the audit instances |
 | `--transport` | `RabbitMq` | `RabbitMq`, `SqlServer` | Transport shared by ServiceControl and the testing tool. `SqlServer` reuses the persistence SQL Server container with a separate `Transport` database |
 | `--tag` | *(none — uses the current build's image)* | any image tag, e.g. `pr-1234` or `6.3.1` | Override the ServiceControl container image tag (useful for testing PR-based prereleases) |
 | `--error-ingestion-scale-unit` | `0` | non-negative integer | Number of additional error-ingestion-only scale-out instances to spin up alongside the primary error instance (each runs with `--error-ingestion-only`) |
-| `--audit-instances` | `0` | non-negative integer | Number of ServiceControl audit instances to add |
+| `--audit-instances` | `1` | non-negative integer | Number of ServiceControl.Audit instances to add. They take the same `--persistence` backend, report custom checks to the primary and export OTel to the collector |
 
 The Aspire dashboard provides allocated ports for each service. The testing tool automatically
 connects to ServiceControl via the platform's transport and REST API URL, and sends its OTLP
@@ -242,7 +264,8 @@ All configuration is via environment variables (no files, no database). Settings
 | `TestingTool__CustomCheckInterval` | `00:00:30` | Default interval for the custom-check-failures job |
 | `TestingTool__CustomCheckHost` | `ServiceControl` | `Host` field on injected custom-check reports |
 | `TestingTool__CustomCheckFailureProbability` | `0.4` | Probability (0–1) a given check is reported failed each cycle |
-| `TestingTool__AuditQueueName` | `audit` | NServiceBus audit queue for processed messages |
+| `TestingTool__AuditQueueName` | `audit` | NServiceBus audit queue the ServiceControl.Audit instance drains |
+| `TestingTool__AuditSagaSnapshotRatio` | `0.2` | Default saga snapshots per processed message for the audit-queue writer |
 | `TestingTool__MonitoringQueueName` | `Particular.Monitoring` | ServiceControl monitoring instance queue that endpoint metrics are sent to |
 | `TestingTool__ErrorQueueName` | `error` | NServiceBus error queue (ServiceControl monitors this) |
 | `TestingTool__AutoStartBackgroundNoise` | `false` | Auto-start the background-noise scenario on startup |

tools/testing-tool/TestingTool.AppHost/AppHost.cs

@@ -3,6 +3,7 @@
 
 var options = CliOptions.Parse(args);
 var persistenceType = options.GetValue("persistence", PersistenceType.RavenDb);
+var imageTag = options.GetValueOrDefault("tag");
 Console.WriteLine($"Using persistence type: {persistenceType}");
 
 var builder = DistributedApplication.CreateBuilder(args);
@@ -51,10 +52,25 @@
 
 platform.AddServicePulse("pulse", primaryErrorInstance, platform.AddServiceControlMonitoringInstance("monitoring"));
 
-for (int i = 0; i < options.GetValue("audit-instances", 0); i++)
+// EF persistence lives in the audit instance itself, so audit load needs a real audit container.
+// The platform persistence argument stays RavenDB because the Aspire package only knows about
+// platform persistences; the SQL wiring is layered on with WithPersistenceType.
+var auditInstanceCount = options.GetValue("audit-instances", 0);
+
+for (int i = 0; i < auditInstanceCount; i++)
 {
+    //note: these audit instances are currently independant and are not scale out nodes like the error ingestion ones.
     platform.AddServiceControlAuditInstance("audit" + i, primaryErrorInstance, raven)
-        .WithEnvironment("INSTANCENAME", "Audit-" + i);
+        .WithEnvironment("SERVICECONTROL_AUDIT_INSTANCENAME", "Audit-" + i)
+        .WithEnvironment("OTEL_EXPORTER_OTLP_ENDPOINT", observability.Collector.GetEndpoint("otlp-grpc"))
+        // Reports the audit ingestion and failed-import custom checks to the primary, which is the
+        // only instance ServicePulse asks. Without it a failed audit import is invisible outside
+        // the audit container's own log.
+        .WithEnvironment("SERVICECONTROL_AUDIT_SERVICECONTROLQUEUEADDRESS", "Particular.ServiceControl")
+        .WithAuditQueueName(i == 0 ? "audit" : "audit" + i)
+        .WithRunMode(PlatformRunMode.SetupAndRun)
+        .WithIngestionTuning()
+        .WithPersistenceType(persistenceType);
 }
 
 // --- Testing tool ---
@@ -69,6 +85,6 @@
     .WaitFor(observability.Collector);
 
 // --- Optional: override ServiceControl image tag for prerelease testing ---
-builder.UseServiceControlImageTag(options.GetValueOrDefault("tag"));
+builder.UseServiceControlImageTag(imageTag);
 
 builder.Build().Run();

tools/testing-tool/TestingTool.AppHost/HostBuilderExtensions.cs

@@ -70,6 +70,52 @@ public static IResourceBuilder<ServiceControlErrorInstanceResource> WithPersiste
 
 
 
+    /// <summary>
+    /// Points a ServiceControl.Audit instance at the same SQL Server or PostgreSQL container the
+    /// primary uses. The keys are namespaced to the audit instance, and there is no body storage:
+    /// the audit EF persisters store bodies inline in the audit row, capped by MaxBodySizeToStore.
+    /// </summary>
+    /// <remarks>
+    /// The setting root namespace is "ServiceControl.Audit", and the settings reader rewrites both
+    /// the dot and the slash to an underscore, so ServiceControl.Audit/Database/ConnectionString is
+    /// SERVICECONTROL_AUDIT_DATABASE_CONNECTIONSTRING. The unprefixed names the audit container's
+    /// Dockerfile bakes in (PersistenceType, AuditRetentionPeriod) are later in the lookup order,
+    /// so the namespaced form wins.
+    /// </remarks>
+    public static IResourceBuilder<ServiceControlAuditInstanceResource> WithPersistenceType(
+        this IResourceBuilder<ServiceControlAuditInstanceResource> audit, PersistenceType type)
+    {
+        if (type == PersistenceType.RavenDb)
+        {
+            return audit;
+        }
+
+        var db = type switch
+        {
+            PersistenceType.SqlServer => audit.ApplicationBuilder.AddSqlServerPersistence("ServiceControl"),
+            PersistenceType.PostgreSql => audit.ApplicationBuilder.AddPostgresPersistence("servicecontrol"),
+            _ => throw new ArgumentOutOfRangeException(nameof(type), type, null)
+        };
+
+        return audit
+            .WaitFor(db)
+            .WithEnvironment("SERVICECONTROL_AUDIT_PERSISTENCETYPE", PersistenceTypeName(type))
+            .WithEnvironment("SERVICECONTROL_AUDIT_DATABASE_CONNECTIONSTRING", db);
+    }
+
+    /// <summary>
+    /// The audit equivalent of the ingestion tuning the primary carries. The setting names are
+    /// AuditIngestion*, and they are read under the audit namespace, so the primary's
+    /// SERVICECONTROL_ERRORINGESTION* values have no effect here.
+    /// </summary>
+    public static IResourceBuilder<ServiceControlAuditInstanceResource> WithIngestionTuning(
+        this IResourceBuilder<ServiceControlAuditInstanceResource> audit) =>
+        audit
+            .WithEnvironment("SERVICECONTROL_AUDIT_MAXIMUMCONCURRENCYLEVEL", "100")
+            .WithEnvironment("SERVICECONTROL_AUDIT_AUDITINGESTIONBATCHSIZE", "25")
+            .WithEnvironment("SERVICECONTROL_AUDIT_AUDITINGESTIONMAXPARALLELWRITERS", "4")
+            .WithEnvironment("SERVICECONTROL_AUDIT_AUDITINGESTIONBATCHTIMEOUT", "00:00:00.100");
+
     public static IResourceBuilder<IResourceWithConnectionString> AddTransport(
         this IResourceBuilder<ParticularPlatformResource> platform, TransportType type)
     {

tools/testing-tool/TestingTool.AppHost/TestingTool.AppHost.csproj

@@ -4,6 +4,7 @@
     <Nullable>enable</Nullable>
     <ImplicitUsings>enable</ImplicitUsings>
     <OutputType>Exe</OutputType>
+    <AspireUseCliBundle>true</AspireUseCliBundle>
   </PropertyGroup>
 
   <ItemGroup>
@@ -18,4 +19,8 @@
     <ProjectReference Include="..\TestingTool\TestingTool.csproj" />
   </ItemGroup>
 
+  <ItemGroup Label="Direct references to transitive dependencies to avoid versions with CVE">
+    <PackageReference Include="System.Security.Cryptography.Pkcs" Version="10.0.13" />
+  </ItemGroup>
+
 </Project>

tools/testing-tool/TestingTool.AppHost/appsettings.json

@@ -0,0 +1,11 @@
+{
+  "Logging": {
+    "LogLevel": {
+      "Default": "Information",
+      "Aspire": "Information",
+      "Aspire.Hosting.Dcp": "Warning",
+      "Microsoft": "Warning",
+      "Particular": "Information"
+    }
+  }
+}