From 106414709e2358d3cd53e490a92c04378c18c929 Mon Sep 17 00:00:00 2001 From: John Simons Date: Mon, 17 Aug 2026 16:51:36 +1000 Subject: [PATCH] Properly await throughput recording in MonitoringService Using List.ForEach with an async lambda resulted in async void calls that were not awaited, meaning the service could return before all endpoint data was persisted. Replacing this with a standard foreach loop ensures every operation is correctly awaited. --- .../MonitoringService_Tests.cs | 106 ++++++++++++++++++ .../MonitoringThroughput/MonitoringService.cs | 8 +- 2 files changed, 110 insertions(+), 4 deletions(-) diff --git a/src/Particular.LicensingComponent.UnitTests/MonitoringService_Tests.cs b/src/Particular.LicensingComponent.UnitTests/MonitoringService_Tests.cs index aa13b9852f..55f14310ac 100644 --- a/src/Particular.LicensingComponent.UnitTests/MonitoringService_Tests.cs +++ b/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 SavedEndpoints { get; } = new(); + + public ConcurrentDictionary RecordedThroughput { get; } = new(); + + public async Task 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 throughput, CancellationToken cancellationToken = default) + { + await Task.Yield(); + RecordedThroughput[endpointName] = throughput.Sum(t => t.MessageCount); + } + + public Task> GetAllEndpoints(bool includePlatformEndpoints, CancellationToken cancellationToken = default) => + throw new NotSupportedException(); + + public Task> GetEndpoints(IList endpointIds, CancellationToken cancellationToken = default) => + throw new NotSupportedException(); + + public Task>> GetEndpointThroughputByQueueName(IList queueNames, CancellationToken cancellationToken = default) => + throw new NotSupportedException(); + + public Task UpdateUserIndicatorOnEndpoints(List userIndicatorUpdates, CancellationToken cancellationToken = default) => + throw new NotSupportedException(); + + public Task IsThereThroughputForLastXDays(int days, CancellationToken cancellationToken = default) => + throw new NotSupportedException(); + + public Task IsThereThroughputForLastXDaysForSource(int days, ThroughputSource throughputSource, bool includeToday, CancellationToken cancellationToken = default) => + throw new NotSupportedException(); + + public Task GetBrokerMetadata(CancellationToken cancellationToken = default) => + throw new NotSupportedException(); + + public Task SaveBrokerMetadata(BrokerMetadata brokerMetadata, CancellationToken cancellationToken = default) => + throw new NotSupportedException(); + + public Task GetAuditServiceMetadata(CancellationToken cancellationToken = default) => + throw new NotSupportedException(); + + public Task SaveAuditServiceMetadata(AuditServiceMetadata auditServiceMetadata, CancellationToken cancellationToken = default) => + throw new NotSupportedException(); + + public Task> GetReportMasks(CancellationToken cancellationToken = default) => + throw new NotSupportedException(); + + public Task SaveReportMasks(List reportMasks, CancellationToken cancellationToken = default) => + throw new NotSupportedException(); + + public Task GetLicensedEndpointDetails(CancellationToken cancellationToken = default) => + throw new NotSupportedException(); + + public Task SaveLicensedEndpointDetails(LicensedEndpointDetails result, CancellationToken cancellationToken = default) => + throw new NotSupportedException(); + } + class BrokerThroughputQuery_WithSanitization : IBrokerThroughputQuery { public Dictionary Data => throw new NotImplementedException(); diff --git a/src/Particular.LicensingComponent/MonitoringThroughput/MonitoringService.cs b/src/Particular.LicensingComponent/MonitoringThroughput/MonitoringService.cs index b0dd964a06..d27ec786f3 100644 --- a/src/Particular.LicensingComponent/MonitoringThroughput/MonitoringService.cs +++ b/src/Particular.LicensingComponent/MonitoringThroughput/MonitoringService.cs @@ -18,11 +18,11 @@ public async Task RecordMonitoringThroughput(byte[] throughputMessage, Cancellat message = await JsonSerializer.DeserializeAsync(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); } - }); + } } }