Merge branch 'master' into zombie_queues

# Conflicts:
#	src/ServiceControl.Audit.Persistence.Tests/AuditCountingTests.cs

Author: PhilBastian

docs/coding-and-design-guidelines.md

@@ -40,6 +40,23 @@ There are a few things that are still registered using convention. Note that the
 Additionally, because NServiceBus does a type-scan at startup it will automatically register any implementations of `Feature` and `IHandleMessage<>`. We have chosen to leave this alone as we would be fighting with NServiceBus in order to turn this off. 
 
 
+## Prefer explicit persistence operations
+
+Use a direct data-store method when all inputs for a persistence operation fit in one method call. The method should own and dispose its EF Core scope/context or RavenDB session, accept a `CancellationToken`, and commit before returning. Returned entities are detached snapshots; callers must not be required to mutate tracked entities as an implicit persistence command.
+
+Atomic operations that can encounter concurrency conflicts should document their provider guarantees and translate expected provider exceptions into explicit domain outcomes. For example, a unique-key or optimistic-concurrency conflict should not escape when contention is part of the operation's normal contract.
+
+Reserve a specialized unit of work for cases where a caller genuinely composes several writes into one atomic batch. New unit-of-work APIs should consistently provide:
+
+- an `I...UnitOfWorkFactory`;
+- a `StartNew` factory method;
+- a `Complete(CancellationToken)` commit method;
+- `IAsyncDisposable` lifetime ownership;
+- explicit operation-recording methods rather than mutation of tracked return values;
+- documented commit, abandon, repeated-completion, and concurrency behavior.
+
+Do not introduce generic `IDataSessionManager`-style abstractions or persistence managers with hidden call-order protocols. During review, prefer one explicit store operation unless caller-composed atomicity requires a unit of work.
+
 ## Avoid property injection
 
 Although the Autofac container can be configured to allow property injection, we prefer to avoid it. There is no way to specify property injection using the Microsoft DI abstractions, and the default `IServiceProvider` implementation does not support it. Where possible, use constructor injection instead.

src/Directory.Packages.props

@@ -22,6 +22,7 @@
     <PackageVersion Include="Microsoft.AspNetCore.Authentication.JwtBearer" Version="$(RuntimeFrameworkVersion)" />
     <PackageVersion Include="Microsoft.AspNetCore.Authentication.OpenIdConnect" Version="$(RuntimeFrameworkVersion)" />
     <PackageVersion Include="Microsoft.AspNetCore.Mvc.Testing" Version="$(RuntimeFrameworkVersion)" />
+    <PackageVersion Include="Microsoft.Data.SqlClient.Extensions.Azure" Version="7.0.2" />
     <PackageVersion Include="Microsoft.Extensions.DependencyInjection" Version="$(RuntimeFrameworkVersion)" />
     <PackageVersion Include="Microsoft.Extensions.DependencyModel" Version="$(RuntimeFrameworkVersion)" />
     <PackageVersion Include="Microsoft.Extensions.Hosting" Version="$(RuntimeFrameworkVersion)" />
@@ -77,10 +78,10 @@
     <PackageVersion Include="PropertyChanging.Fody" Version="1.31.0" />
     <PackageVersion Include="PublicApiGenerator" Version="11.5.4" />
     <PackageVersion Include="RavenDB.Embedded" Version="6.2.17" />
-    <PackageVersion Include="Testcontainers.Azurite" Version="4.13.0" />
-    <PackageVersion Include="Testcontainers.LocalStack" Version="4.13.0" />
-    <PackageVersion Include="Testcontainers.MsSql" Version="4.13.0" />
-    <PackageVersion Include="Testcontainers.PostgreSql" Version="4.13.0" />
+    <PackageVersion Include="Testcontainers.Azurite" Version="4.14.0" />
+    <PackageVersion Include="Testcontainers.LocalStack" Version="4.14.0" />
+    <PackageVersion Include="Testcontainers.MsSql" Version="4.14.0" />
+    <PackageVersion Include="Testcontainers.PostgreSql" Version="4.14.0" />
     <PackageVersion Include="ReactiveUI.WPF" Version="22.3.1" />
     <PackageVersion Include="Seq.Extensions.Logging" Version="9.0.0" />
     <PackageVersion Include="ServiceControl.Contracts" Version="5.1.0" />

src/Particular.LicensingComponent.UnitTests/.editorconfig

@@ -2,8 +2,4 @@
 
 # Justification: Test project
 dotnet_diagnostic.CA2007.severity = none
-
-# Cancellation analyzer debt. These fire because Particular.Analyzers is no longer pinned to 0.9.0.
-# They are scheduled work, not accepted exceptions: remove a line once this project has no
-# violations of that rule left, and never add a rule back to this list.
 dotnet_diagnostic.PS0018.severity = none

src/Particular.LicensingComponent.UnitTests/MonitoringService_Tests.cs

@@ -1,6 +1,7 @@
 namespace Particular.LicensingComponent.UnitTests;
 
 using System;
+using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Collections.ObjectModel;
 using System.Linq;
@@ -13,6 +14,7 @@ namespace Particular.LicensingComponent.UnitTests;
 using Infrastructure;
 using MonitoringThroughput;
 using NUnit.Framework;
+using Persistence;
 using ServiceControl.Transports.BrokerThroughput;
 
 [TestFixture]
@@ -159,6 +161,110 @@ public async Task Should_return_error_monitoring_connection_and_diagnostics_if_n
         Approver.Verify(connectionSettingsResult.Diagnostics);
     }
 
+    [Test]
+    public async Task Should_record_throughput_for_every_endpoint_in_the_message()
+    {
+        // Arrange
+        var dataStore = new YieldingLicensingDataStore();
+        var monitoringService = new MonitoringService(dataStore);
+
+        string[] endpointNames = [.. Enumerable.Range(1, 20).Select(i => $"Endpoint{i}")];
+        var message = new RecordEndpointThroughputData
+        {
+            StartDateTime = DateTime.UtcNow.AddMinutes(-5),
+            EndDateTime = DateTime.UtcNow,
+            EndpointThroughputData = [.. endpointNames.Select(name => new EndpointThroughputData { Name = name, Throughput = 15 })]
+        };
+
+        // Act
+        byte[] messageBytes = JsonSerializer.SerializeToUtf8Bytes(message);
+        await monitoringService.RecordMonitoringThroughput(messageBytes);
+
+        // Assert
+        string[] endpointsWithoutThroughput = [.. endpointNames.Where(name => !dataStore.RecordedThroughput.ContainsKey(name))];
+        string[] unsavedEndpoints = [.. endpointNames.Where(name => !dataStore.SavedEndpoints.ContainsKey(name))];
+
+        using (Assert.EnterMultipleScope())
+        {
+            Assert.That(unsavedEndpoints, Is.Empty,
+                $"{unsavedEndpoints.Length} of {endpointNames.Length} endpoints were not saved when RecordMonitoringThroughput returned: {string.Join(", ", unsavedEndpoints)}");
+            Assert.That(endpointsWithoutThroughput, Is.Empty,
+                $"{endpointsWithoutThroughput.Length} of {endpointNames.Length} endpoints had no throughput recorded when RecordMonitoringThroughput returned: {string.Join(", ", endpointsWithoutThroughput)}");
+        }
+
+        Assert.That(dataStore.RecordedThroughput.Values, Is.All.EqualTo(15L), "Expected a throughput of 15 for every endpoint");
+    }
+
+    // Records only after suspending, so any write the service starts without awaiting is still
+    // outstanding when RecordMonitoringThroughput returns.
+    class YieldingLicensingDataStore : ILicensingDataStore
+    {
+        public ConcurrentDictionary<string, Endpoint> SavedEndpoints { get; } = new();
+
+        public ConcurrentDictionary<string, long> RecordedThroughput { get; } = new();
+
+        public async Task<Endpoint> GetEndpoint(EndpointIdentifier id, CancellationToken cancellationToken = default)
+        {
+            await Task.Yield();
+            return SavedEndpoints.TryGetValue(id.Name, out Endpoint endpoint) ? endpoint : null;
+        }
+
+        public async Task SaveEndpoint(Endpoint endpoint, CancellationToken cancellationToken = default)
+        {
+            await Task.Yield();
+            SavedEndpoints[endpoint.Id.Name] = endpoint;
+        }
+
+        public async Task RecordEndpointThroughput(string endpointName, ThroughputSource throughputSource,
+            IList<EndpointDailyThroughput> throughput, CancellationToken cancellationToken = default)
+        {
+            await Task.Yield();
+            RecordedThroughput[endpointName] = throughput.Sum(t => t.MessageCount);
+        }
+
+        public Task<IEnumerable<Endpoint>> GetAllEndpoints(bool includePlatformEndpoints, CancellationToken cancellationToken = default) =>
+            throw new NotSupportedException();
+
+        public Task<IEnumerable<(EndpointIdentifier Id, Endpoint Endpoint)>> GetEndpoints(IList<EndpointIdentifier> endpointIds, CancellationToken cancellationToken = default) =>
+            throw new NotSupportedException();
+
+        public Task<IDictionary<string, IEnumerable<ThroughputData>>> GetEndpointThroughputByQueueName(IList<string> queueNames, CancellationToken cancellationToken = default) =>
+            throw new NotSupportedException();
+
+        public Task UpdateUserIndicatorOnEndpoints(List<UpdateUserIndicator> userIndicatorUpdates, CancellationToken cancellationToken = default) =>
+            throw new NotSupportedException();
+
+        public Task<bool> IsThereThroughputForLastXDays(int days, CancellationToken cancellationToken = default) =>
+            throw new NotSupportedException();
+
+        public Task<bool> IsThereThroughputForLastXDaysForSource(int days, ThroughputSource throughputSource, bool includeToday, CancellationToken cancellationToken = default) =>
+            throw new NotSupportedException();
+
+        public Task<BrokerMetadata> GetBrokerMetadata(CancellationToken cancellationToken = default) =>
+            throw new NotSupportedException();
+
+        public Task SaveBrokerMetadata(BrokerMetadata brokerMetadata, CancellationToken cancellationToken = default) =>
+            throw new NotSupportedException();
+
+        public Task<AuditServiceMetadata> GetAuditServiceMetadata(CancellationToken cancellationToken = default) =>
+            throw new NotSupportedException();
+
+        public Task SaveAuditServiceMetadata(AuditServiceMetadata auditServiceMetadata, CancellationToken cancellationToken = default) =>
+            throw new NotSupportedException();
+
+        public Task<List<string>> GetReportMasks(CancellationToken cancellationToken = default) =>
+            throw new NotSupportedException();
+
+        public Task SaveReportMasks(List<string> reportMasks, CancellationToken cancellationToken = default) =>
+            throw new NotSupportedException();
+
+        public Task<LicensedEndpointDetails> GetLicensedEndpointDetails(CancellationToken cancellationToken = default) =>
+            throw new NotSupportedException();
+
+        public Task SaveLicensedEndpointDetails(LicensedEndpointDetails result, CancellationToken cancellationToken = default) =>
+            throw new NotSupportedException();
+    }
+
     class BrokerThroughputQuery_WithSanitization : IBrokerThroughputQuery
     {
         public Dictionary<string, string> Data => throw new NotImplementedException();

src/Particular.LicensingComponent/MonitoringThroughput/MonitoringService.cs

@@ -18,11 +18,11 @@ public async Task RecordMonitoringThroughput(byte[] throughputMessage, Cancellat
             message = await JsonSerializer.DeserializeAsync<RecordEndpointThroughputData>(stream, cancellationToken: cancellationToken);
         }
 
-        if (message != null && message.EndpointThroughputData != null)
+        if (message is not null && message.EndpointThroughputData is { } endpointThroughputData)
         {
-            Debug.WriteLine($"Throughput data from {message.StartDateTime:yyyy-MM-dd HH:mm} to {message.EndDateTime:yyyy-MM-dd HH:mm} for {message.EndpointThroughputData?.Length} endpoint(s)");
+            Debug.WriteLine($"Throughput data from {message.StartDateTime:yyyy-MM-dd HH:mm} to {message.EndDateTime:yyyy-MM-dd HH:mm} for {endpointThroughputData.Length} endpoint(s)");
 
-            message.EndpointThroughputData?.ToList().ForEach(async e =>
+            foreach (var e in endpointThroughputData)
             {
                 var endpoint = await dataStore.GetEndpoint(e.Name, ThroughputSource.Monitoring, cancellationToken);
                 if (endpoint == null)
@@ -41,7 +41,7 @@ public async Task RecordMonitoringThroughput(byte[] throughputMessage, Cancellat
 
                     await dataStore.RecordEndpointThroughput(e.Name, ThroughputSource.Monitoring, [endpointThroughput], cancellationToken);
                 }
-            });
+            }
         }
     }
 

src/ServiceControl.AcceptanceTesting/.editorconfig

@@ -2,11 +2,3 @@
 
 # Justification: Usage is from test projects
 dotnet_diagnostic.CA2007.severity = none
-
-# These types sit on NServiceBus.AcceptanceTesting's extension points, which drive them without a
-# CancellationToken: IEndpointSetupTemplate, IComponentBehavior/ComponentRunner, and the scenario
-# Done/When callbacks whose delegate shapes the framework fixes. A token added here could only ever
-# be CancellationToken.None. Everything else in this project takes and forwards one.
-[{EndpointTemplates/*.cs,InfrastructureConfig/*.cs,ScenarioWithEndpointBehaviorExtensions.cs,Sequence.cs,DispatchRawMessages.cs}]
-dotnet_diagnostic.PS0013.severity = none
-dotnet_diagnostic.PS0018.severity = none

src/ServiceControl.AcceptanceTesting/DispatchRawMessages.cs

@@ -9,6 +9,8 @@
     using NServiceBus.Features;
     using NServiceBus.Transport;
 
+#pragma warning disable PS0013
+#pragma warning disable PS0018
     public abstract class DispatchRawMessages<TContext> : Feature
         where TContext : ScenarioContext
     {
@@ -79,4 +81,6 @@ protected override Task OnStop(IMessageSession session, CancellationToken cancel
             ScenarioContext scenarioContext;
         }
     }
+#pragma warning restore PS0018
+#pragma warning restore PS0013
 }

src/ServiceControl.AcceptanceTesting/EndpointTemplates/DefaultServerBase.cs

@@ -16,7 +16,11 @@ protected DefaultServerBase() : this(new ConfigureEndpointLearningTransport())
         {
         }
 
+#pragma warning disable PS0013
+#pragma warning disable PS0018
         public virtual async Task<EndpointConfiguration> GetConfiguration(RunDescriptor runDescriptor, EndpointCustomizationConfiguration endpointCustomizations, Func<EndpointConfiguration, Task> configurationBuilderCustomization)
+#pragma warning restore PS0018
+#pragma warning restore PS0013
         {
             var endpointConfiguration = new EndpointConfiguration(endpointCustomizations.EndpointName);
 

src/ServiceControl.AcceptanceTesting/EndpointTemplates/IConfigureEndpointTestExecution.cs

@@ -8,6 +8,7 @@
     /// Provide a mechanism in acceptance tests for transports and persistences
     /// to configure an endpoint for a test and then clean up afterwards.
     /// </summary>
+#pragma warning disable PS0018
     public interface IConfigureEndpointTestExecution
     {
         /// <summary>
@@ -32,4 +33,5 @@ public interface IConfigureEndpointTestExecution
         /// <returns>An async Task.</returns>
         Task Cleanup();
     }
+#pragma warning restore PS0018
 }

src/ServiceControl.AcceptanceTesting/InfrastructureConfig/ConfigureEndpointInMemoryPersistence.cs

@@ -5,6 +5,7 @@
     using NServiceBus;
     using NServiceBus.AcceptanceTesting.Support;
 
+#pragma warning disable PS0018
     public class ConfigureEndpointInMemoryPersistence : IConfigureEndpointTestExecution
     {
         public Task Configure(string endpointName, EndpointConfiguration configuration, RunSettings settings, PublisherMetadata publisherMetadata)
@@ -17,4 +18,5 @@ public Task Cleanup() =>
             // Nothing required for in-memory persistence
             Task.CompletedTask;
     }
+#pragma warning restore PS0018
 }