From eab024c4cfbd3770cd80889432e64cb2b2855745 Mon Sep 17 00:00:00 2001 From: Phil Bastian Date: Thu, 30 Jul 2026 09:30:30 +0800 Subject: [PATCH 01/13] remove any EndpointDocuments that haven't seen any throughput data for the reporting period --- .../InMemoryLicensingDataStore.cs | 9 +++++++++ .../ILicensingDataStore.cs | 2 ++ ...ughputCollector_ThroughputSummary_Tests.cs | 19 +++++++++++++++++-- .../ThroughputCollector.cs | 9 ++++++++- .../ThroughputDataExtensions.cs | 2 +- .../Implementation/LicensingDataStore.cs | 1 + .../Throughput/LicensingDataStore.cs | 17 ++++++++++++++++- 7 files changed, 54 insertions(+), 5 deletions(-) diff --git a/src/Particular.LicensingComponent.Persistence.InMemory/InMemoryLicensingDataStore.cs b/src/Particular.LicensingComponent.Persistence.InMemory/InMemoryLicensingDataStore.cs index 46639e6305..cd5e67c868 100644 --- a/src/Particular.LicensingComponent.Persistence.InMemory/InMemoryLicensingDataStore.cs +++ b/src/Particular.LicensingComponent.Persistence.InMemory/InMemoryLicensingDataStore.cs @@ -179,6 +179,15 @@ public Task SaveLicensedEndpointDetails(LicensedEndpointDetails result, Cancella return Task.CompletedTask; } + public Task RemoveEndpoints(EndpointIdentifier[] endpointIds, CancellationToken cancellationToken) + { + foreach (var id in endpointIds) + { + endpoints.Remove(id); + } + return Task.CompletedTask; + } + class EndpointCollection : KeyedCollection { protected override EndpointIdentifier GetKeyForItem(Endpoint item) => item.Id; diff --git a/src/Particular.LicensingComponent.Persistence/ILicensingDataStore.cs b/src/Particular.LicensingComponent.Persistence/ILicensingDataStore.cs index 3a73a50891..d2121fcde5 100644 --- a/src/Particular.LicensingComponent.Persistence/ILicensingDataStore.cs +++ b/src/Particular.LicensingComponent.Persistence/ILicensingDataStore.cs @@ -15,6 +15,8 @@ public interface ILicensingDataStore Task SaveEndpoint(Endpoint endpoint, CancellationToken cancellationToken); + Task RemoveEndpoints(EndpointIdentifier[] endpointIds, CancellationToken cancellationToken); + Task>> GetEndpointThroughputByQueueName(IList queueNames, CancellationToken cancellationToken); Task RecordEndpointThroughput(string endpointName, ThroughputSource throughputSource, DateOnly date, long messageCount, CancellationToken cancellationToken) => diff --git a/src/Particular.LicensingComponent.UnitTests/ThroughputCollector/ThroughputCollector_ThroughputSummary_Tests.cs b/src/Particular.LicensingComponent.UnitTests/ThroughputCollector/ThroughputCollector_ThroughputSummary_Tests.cs index b3007fe41f..53aa86ae08 100644 --- a/src/Particular.LicensingComponent.UnitTests/ThroughputCollector/ThroughputCollector_ThroughputSummary_Tests.cs +++ b/src/Particular.LicensingComponent.UnitTests/ThroughputCollector/ThroughputCollector_ThroughputSummary_Tests.cs @@ -177,10 +177,11 @@ await DataStore.CreateBuilder() } [Test] - public async Task Should_return_correct_max_daily_throughput_in_summary_when_endpoint_has_no_throughput() + public async Task Should_return_correct_max_daily_throughput_in_summary_when_endpoint_has_zero_throughput() { // Arrange - await DataStore.CreateBuilder().AddEndpoint().Build(); + await DataStore.CreateBuilder().AddEndpoint().WithThroughput(new ThroughputData([ + new EndpointDailyThroughput(new DateOnly(2025, 1, 10), 0)])).Build(); // Act var summary = await ThroughputCollector.GetThroughputSummary(default); @@ -191,6 +192,20 @@ public async Task Should_return_correct_max_daily_throughput_in_summary_when_end Assert.That(summary[0].MaxDailyThroughput, Is.EqualTo(0), $"Incorrect MaxDailyThroughput recorded for {summary[0].Name}"); } + [Test] + public async Task Should_not_return_endpoint_in_summary_when_endpoint_has_no_throughput() + { + // Arrange + await DataStore.CreateBuilder().AddEndpoint().Build(); + + // Act + var summary = await ThroughputCollector.GetThroughputSummary(default); + + // Assert + Assert.That(summary, Is.Not.Null); + Assert.That(summary, Is.Empty, "Invalid number of endpoints in throughput summary"); + } + [Test] public async Task Should_return_correct_max_daily_throughput_in_summary_when_data_from_multiple_sources_and_name_is_different() { diff --git a/src/Particular.LicensingComponent/ThroughputCollector.cs b/src/Particular.LicensingComponent/ThroughputCollector.cs index 32f70aaf40..04edae5b59 100644 --- a/src/Particular.LicensingComponent/ThroughputCollector.cs +++ b/src/Particular.LicensingComponent/ThroughputCollector.cs @@ -224,7 +224,14 @@ async IAsyncEnumerable GetDistinctEndpointData([EnumeratorCancella var userIndicator = UserIndicator(endpointGroupPerQueue) ?? null; - yield return new EndpointData(endpointName, throughputData, userIndicator, EndpointScope(endpointGroupPerQueue), EndpointIndicators(endpointGroupPerQueue), IsKnownEndpoint(endpointGroupPerQueue)); + if (throughputData.Any(x => x.Any())) + { + yield return new EndpointData(endpointName, throughputData, userIndicator, EndpointScope(endpointGroupPerQueue), EndpointIndicators(endpointGroupPerQueue), IsKnownEndpoint(endpointGroupPerQueue)); + } + else + { + await dataStore.RemoveEndpoints([.. endpointGroupPerQueue.Select(endpoint => endpoint.Id)], cancellationToken); + } } } diff --git a/src/Particular.LicensingComponent/ThroughputDataExtensions.cs b/src/Particular.LicensingComponent/ThroughputDataExtensions.cs index ce19589692..18a11ed2f6 100644 --- a/src/Particular.LicensingComponent/ThroughputDataExtensions.cs +++ b/src/Particular.LicensingComponent/ThroughputDataExtensions.cs @@ -51,5 +51,5 @@ public static long AverageMonthlyThroughput(this List throughput } public static bool HasDataFromSource(this IDictionary> throughputPerQueue, ThroughputSource source) => - throughputPerQueue.Any(queueName => queueName.Value.Any(data => data.ThroughputSource == source && data.Count > 0)); + throughputPerQueue.Any(queueThroughput => queueThroughput.Value.Any(data => data.ThroughputSource == source && data.Count > 0)); } \ No newline at end of file diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/LicensingDataStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/LicensingDataStore.cs index acf8fe9102..f57591a034 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/LicensingDataStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/LicensingDataStore.cs @@ -14,6 +14,7 @@ class LicensingDataStore : ILicensingDataStore public Task IsThereThroughputForLastXDays(int days, CancellationToken cancellationToken) => throw new NotImplementedException(); public Task IsThereThroughputForLastXDaysForSource(int days, ThroughputSource throughputSource, bool includeToday, CancellationToken cancellationToken) => throw new NotImplementedException(); public Task RecordEndpointThroughput(string endpointName, ThroughputSource throughputSource, IList throughput, CancellationToken cancellationToken) => throw new NotImplementedException(); + public Task RemoveEndpoints(EndpointIdentifier[] endpointIds, CancellationToken cancellationToken) => throw new NotImplementedException(); public Task SaveAuditServiceMetadata(AuditServiceMetadata auditServiceMetadata, CancellationToken cancellationToken) => throw new NotImplementedException(); public Task SaveBrokerMetadata(BrokerMetadata brokerMetadata, CancellationToken cancellationToken) => throw new NotImplementedException(); public Task SaveEndpoint(Particular.LicensingComponent.Contracts.Endpoint endpoint, CancellationToken cancellationToken) => throw new NotImplementedException(); diff --git a/src/ServiceControl.Persistence.RavenDB/Throughput/LicensingDataStore.cs b/src/ServiceControl.Persistence.RavenDB/Throughput/LicensingDataStore.cs index 26b666cdcb..40faf054e6 100644 --- a/src/ServiceControl.Persistence.RavenDB/Throughput/LicensingDataStore.cs +++ b/src/ServiceControl.Persistence.RavenDB/Throughput/LicensingDataStore.cs @@ -25,6 +25,8 @@ class LicensingDataStore( const string ReportMasksDocumentId = "ReportMasks"; const string LicencedEndpointDetailsDocumentId = "LicensedEndpointDetails"; + const int ThroughputPeriodMonths = 14; + static readonly AuditServiceMetadata DefaultAuditServiceMetadata = new([], []); static readonly BrokerMetadata DefaultBrokerMetadata = new(null, []); static readonly ReportConfigurationDocument DefaultReportConfiguration = new(); @@ -121,6 +123,19 @@ public async Task SaveEndpoint(Endpoint endpoint, CancellationToken cancellation await session.SaveChangesAsync(cancellationToken); } + public async Task RemoveEndpoints(EndpointIdentifier[] endpointIds, CancellationToken cancellationToken) + { + var documentIds = endpointIds.Select(id => id.GenerateDocumentId()); + + var store = await storeProvider.GetDocumentStore(cancellationToken); + using IAsyncDocumentSession session = store.OpenAsyncSession(databaseConfiguration.Name); + + foreach (var documentId in documentIds) { + session.Delete(documentId); + } + await session.SaveChangesAsync(cancellationToken); + } + public async Task>> GetEndpointThroughputByQueueName(IList queueNames, CancellationToken cancellationToken) { var results = queueNames.ToDictionary(queueName => queueName, _ => new List() as IEnumerable); @@ -128,7 +143,7 @@ public async Task>> GetEndpointT var store = await storeProvider.GetDocumentStore(cancellationToken); using IAsyncDocumentSession session = store.OpenAsyncSession(databaseConfiguration.Name); - var from = DateTime.UtcNow.AddMonths(-14); + var from = DateTime.UtcNow.AddMonths(-ThroughputPeriodMonths); var query = session.Query() .Where(document => document.SanitizedName.In(queueNames)) .Include(builder => builder.IncludeTimeSeries(ThroughputTimeSeriesName, from)); From a6de4b9ff94e13174c68b8718bf181fcd5e14f68 Mon Sep 17 00:00:00 2001 From: Phil Bastian Date: Fri, 31 Jul 2026 15:21:54 +0800 Subject: [PATCH 02/13] fix warnings --- .../Throughput/LicensingDataStore.cs | 3 ++- src/ServiceControl/Licensing/LicenseController.cs | 2 -- 2 files changed, 2 insertions(+), 3 deletions(-) diff --git a/src/ServiceControl.Persistence.RavenDB/Throughput/LicensingDataStore.cs b/src/ServiceControl.Persistence.RavenDB/Throughput/LicensingDataStore.cs index 40faf054e6..dc0b269a85 100644 --- a/src/ServiceControl.Persistence.RavenDB/Throughput/LicensingDataStore.cs +++ b/src/ServiceControl.Persistence.RavenDB/Throughput/LicensingDataStore.cs @@ -130,7 +130,8 @@ public async Task RemoveEndpoints(EndpointIdentifier[] endpointIds, Cancellation var store = await storeProvider.GetDocumentStore(cancellationToken); using IAsyncDocumentSession session = store.OpenAsyncSession(databaseConfiguration.Name); - foreach (var documentId in documentIds) { + foreach (var documentId in documentIds) + { session.Delete(documentId); } await session.SaveChangesAsync(cancellationToken); diff --git a/src/ServiceControl/Licensing/LicenseController.cs b/src/ServiceControl/Licensing/LicenseController.cs index 51efd95cc3..cd9580773c 100644 --- a/src/ServiceControl/Licensing/LicenseController.cs +++ b/src/ServiceControl/Licensing/LicenseController.cs @@ -1,10 +1,8 @@ #nullable enable namespace ServiceControl.Licensing { - using System; using System.IO; using System.IO.Compression; - using System.Text; using System.Text.Json; using System.Threading; using System.Threading.Tasks; From e7b46ceedb1543415b7ca332b0ca3e7f67a0b89e Mon Sep 17 00:00:00 2001 From: Phil Bastian Date: Fri, 31 Jul 2026 15:42:14 +0800 Subject: [PATCH 03/13] update tests --- ...hroughputCollector_Report_Throughput_Tests.cs | 16 +++++++++++++++- 1 file changed, 15 insertions(+), 1 deletion(-) diff --git a/src/Particular.LicensingComponent.UnitTests/ThroughputCollector/ThroughputCollector_Report_Throughput_Tests.cs b/src/Particular.LicensingComponent.UnitTests/ThroughputCollector/ThroughputCollector_Report_Throughput_Tests.cs index 6603abcad0..c991f87e6c 100644 --- a/src/Particular.LicensingComponent.UnitTests/ThroughputCollector/ThroughputCollector_Report_Throughput_Tests.cs +++ b/src/Particular.LicensingComponent.UnitTests/ThroughputCollector/ThroughputCollector_Report_Throughput_Tests.cs @@ -194,7 +194,7 @@ await DataStore.CreateBuilder() public async Task Should_return_correct_throughput_in_report_when_endpoint_has_no_throughput() { // Arrange - await DataStore.CreateBuilder().AddEndpoint().Build(); + await DataStore.CreateBuilder().AddEndpoint().WithThroughput(ThroughputSource.Broker, data: [0]).Build(); // Act var report = await ThroughputCollector.GenerateThroughputReport("", null, default); @@ -211,6 +211,20 @@ public async Task Should_return_correct_throughput_in_report_when_endpoint_has_n } } + [Test] + public async Task Should_not_return_endpoint_in_report_when_endpoint_has_no_throughput() + { + // Arrange + await DataStore.CreateBuilder().AddEndpoint().Build(); + + // Act + var report = await ThroughputCollector.GenerateThroughputReport("", null, default); + + // Assert + Assert.That(report, Is.Not.Null); + Assert.That(report.ReportData.Queues.Count, Is.Zero, "Invalid number of endpoints in throughput report"); + } + [Test] public async Task Should_return_correct_throughput_in_report_when_data_from_multiple_sources_and_name_is_different() { From 5d383e14a9c46d9e0a7ffc43073ccdae6e9aa368 Mon Sep 17 00:00:00 2001 From: Phil Bastian Date: Wed, 5 Aug 2026 08:54:07 +0800 Subject: [PATCH 04/13] ensure send-only endpoints populate with zero throughput from audit so that they don't get removed when throughput is calculated --- .../AuditThroughputCollectorHostedService.cs | 6 +++++- .../Indexes/MessagesViewIndex.cs | 6 ++++-- .../RavenAuditDataStore.cs | 11 +++++++++++ 3 files changed, 20 insertions(+), 3 deletions(-) diff --git a/src/Particular.LicensingComponent/AuditThroughput/AuditThroughputCollectorHostedService.cs b/src/Particular.LicensingComponent/AuditThroughput/AuditThroughputCollectorHostedService.cs index 91c90dbb95..3a22f73065 100644 --- a/src/Particular.LicensingComponent/AuditThroughput/AuditThroughputCollectorHostedService.cs +++ b/src/Particular.LicensingComponent/AuditThroughput/AuditThroughputCollectorHostedService.cs @@ -74,11 +74,15 @@ async Task GatherThroughput(CancellationToken cancellationToken) var auditCounts = (await auditQuery.GetAuditCountForEndpoint(knownEndpointsLookup[endpointId].UrlName, cancellationToken)).ToList(); - if (endpoint == null) + if (endpoint == null && auditCounts.Count > 0) { endpoint = ConvertToEndpoint(knownEndpointsLookup[endpointId]); await dataStore.SaveEndpoint(endpoint, cancellationToken); } + else if (endpoint is null) + { + return; + } var missingAuditThroughput = auditCounts .Where(auditCount => auditCount.UtcDate > endpoint.LastCollectedDate && diff --git a/src/ServiceControl.Audit.Persistence.RavenDB/Indexes/MessagesViewIndex.cs b/src/ServiceControl.Audit.Persistence.RavenDB/Indexes/MessagesViewIndex.cs index 6e348f08c4..fa9a3bc75f 100644 --- a/src/ServiceControl.Audit.Persistence.RavenDB/Indexes/MessagesViewIndex.cs +++ b/src/ServiceControl.Audit.Persistence.RavenDB/Indexes/MessagesViewIndex.cs @@ -21,13 +21,14 @@ from message in messages TimeSent = (DateTime)message.MessageMetadata["TimeSent"], ProcessedAt = message.ProcessedAt, ReceivingEndpointName = ((EndpointDetails)message.MessageMetadata["ReceivingEndpoint"]).Name, + SendingEndpointName = ((EndpointDetails)message.MessageMetadata["SendingEndpoint"]).Name, CriticalTime = (TimeSpan?)message.MessageMetadata["CriticalTime"], ProcessingTime = (TimeSpan?)message.MessageMetadata["ProcessingTime"], DeliveryTime = (TimeSpan?)message.MessageMetadata["DeliveryTime"], Query = message.MessageMetadata.Select(_ => _.Value.ToString()).Union(new[] { - string.Join(" ", message.Headers.Select(x => x.Value)) - }).ToArray(), + string.Join(" ", message.Headers.Select(x => x.Value)) + }).ToArray(), ConversationId = (string)message.MessageMetadata["ConversationId"] }; @@ -48,6 +49,7 @@ public class SortAndFilterOptions public MessageStatus Status { get; set; } public DateTime ProcessedAt { get; set; } public string ReceivingEndpointName { get; set; } + public string SendingEndpointName { get; set; } public TimeSpan? CriticalTime { get; set; } public TimeSpan? ProcessingTime { get; set; } public TimeSpan? DeliveryTime { get; set; } diff --git a/src/ServiceControl.Audit.Persistence.RavenDB/RavenAuditDataStore.cs b/src/ServiceControl.Audit.Persistence.RavenDB/RavenAuditDataStore.cs index 97f90af67c..88d9b56ba6 100644 --- a/src/ServiceControl.Audit.Persistence.RavenDB/RavenAuditDataStore.cs +++ b/src/ServiceControl.Audit.Persistence.RavenDB/RavenAuditDataStore.cs @@ -138,6 +138,9 @@ public async Task>> QueryAuditCounts(string endpoi .OrderBy(m => m.ProcessedAt) .FirstOrDefaultAsync(token: cancellationToken); + var hasSent = await session.Query(indexName) + .AnyAsync(m => m.SendingEndpointName == endpointName, token: cancellationToken); + if (oldestMsg != null) { var endDate = DateTime.UtcNow.Date.AddDays(1); @@ -166,6 +169,14 @@ public async Task>> QueryAuditCounts(string endpoi } } } + else if (hasSent) + { + results.Add(new AuditCount + { + UtcDate = DateTime.UtcNow.Date, + Count = 0 + }); + } return new QueryResult>(results, QueryStatsInfo.Zero); } From 3c3a5b75ba04cfc6228f632e8d3a63340902e6ee Mon Sep 17 00:00:00 2001 From: Phil Bastian Date: Thu, 6 Aug 2026 11:10:58 +0800 Subject: [PATCH 05/13] apply new index field to fulltextindex too --- .../Indexes/MessagesViewIndexWithFullTextSearch.cs | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/src/ServiceControl.Audit.Persistence.RavenDB/Indexes/MessagesViewIndexWithFullTextSearch.cs b/src/ServiceControl.Audit.Persistence.RavenDB/Indexes/MessagesViewIndexWithFullTextSearch.cs index 9eb433d6e0..1dca30993f 100644 --- a/src/ServiceControl.Audit.Persistence.RavenDB/Indexes/MessagesViewIndexWithFullTextSearch.cs +++ b/src/ServiceControl.Audit.Persistence.RavenDB/Indexes/MessagesViewIndexWithFullTextSearch.cs @@ -21,14 +21,15 @@ from message in messages TimeSent = (DateTime)message.MessageMetadata["TimeSent"], ProcessedAt = message.ProcessedAt, ReceivingEndpointName = ((EndpointDetails)message.MessageMetadata["ReceivingEndpoint"]).Name, + SendingEndpointName = ((EndpointDetails)message.MessageMetadata["SendingEndpoint"]).Name, CriticalTime = (TimeSpan?)message.MessageMetadata["CriticalTime"], ProcessingTime = (TimeSpan?)message.MessageMetadata["ProcessingTime"], DeliveryTime = (TimeSpan?)message.MessageMetadata["DeliveryTime"], Query = message.MessageMetadata.Select(_ => _.Value.ToString()).Union(new[] { - string.Join(" ", message.Headers.Select(x => x.Value)), - LoadAttachment(message, "body").GetContentAsString() - }).ToArray(), + string.Join(" ", message.Headers.Select(x => x.Value)), + LoadAttachment(message, "body").GetContentAsString() + }).ToArray(), ConversationId = (string)message.MessageMetadata["ConversationId"] }; From 8a5b5f94feec2dbc5d34b9f54c404e29a7a23fc8 Mon Sep 17 00:00:00 2001 From: Phil Bastian Date: Thu, 6 Aug 2026 12:29:48 +0800 Subject: [PATCH 06/13] add tests for sendonly audit endpoint behaviour --- ...tThroughputCollectorHostedService_Tests.cs | 88 +++++++++++++++++++ .../AuditThroughputCollectorHostedService.cs | 2 +- .../AuditCountingTests.cs | 37 +++++++- 3 files changed, 124 insertions(+), 3 deletions(-) diff --git a/src/Particular.LicensingComponent.UnitTests/AuditThroughputCollectorHostedService_Tests.cs b/src/Particular.LicensingComponent.UnitTests/AuditThroughputCollectorHostedService_Tests.cs index 88a22d2cfd..4fa629dfdc 100644 --- a/src/Particular.LicensingComponent.UnitTests/AuditThroughputCollectorHostedService_Tests.cs +++ b/src/Particular.LicensingComponent.UnitTests/AuditThroughputCollectorHostedService_Tests.cs @@ -203,6 +203,48 @@ await Task.Run(async () => } } + [Test] + public async Task Should_only_create_new_endpoint_when_audit_counts_exist() + { + // Arrange + using var tokenSource = new CancellationTokenSource(TimeSpan.FromSeconds(3)); + var token = tokenSource.Token; + var fakeTimeProvider = new FakeTimeProvider(); + + var date = DateOnly.FromDateTime(DateTime.UtcNow.AddDays(-1)); + var auditQuery = new AuditQuery_WithTwoEndpointsAndSelectiveCounts( + endpointWithoutCounts: "EndpointNoData", + endpointWithCounts: "EndpointWithData", + throughputDate: date, + throughputCount: 5); + + using var auditThroughputCollectorHostedService = new AuditThroughputCollectorHostedService( + NullLogger.Instance, configuration.ThroughputSettings, DataStore, + auditQuery, fakeTimeProvider) + { DelayStart = TimeSpan.Zero }; + + // Act + await auditThroughputCollectorHostedService.StartAsync(token); + await Task.Run(async () => + { + do + { + await Task.Delay(TimeSpan.FromMilliseconds(50)); + } while (!token.IsCancellationRequested); + }); + await auditThroughputCollectorHostedService.StopAsync(token); + + var endpointWithoutCounts = await DataStore.GetEndpoint("EndpointNoData", ThroughputSource.Audit, default); + var endpointWithCounts = await DataStore.GetEndpoint("EndpointWithData", ThroughputSource.Audit, default); + + // Assert + using (Assert.EnterMultipleScope()) + { + Assert.That(endpointWithoutCounts, Is.Null, "Endpoint with empty auditCounts should not be created"); + Assert.That(endpointWithCounts, Is.Not.Null, "Endpoint with auditCounts should be created"); + } + } + class AuditQuery_NoAuditRemotes : IAuditQuery { public SemanticVersion MinAuditCountsVersion => new(4, 29, 0); @@ -335,4 +377,50 @@ public string SanitizeEndpointName(string endpointName) public string SanitizedEndpointNameCleanser(string endpointName) => endpointName; } + + class AuditQuery_WithTwoEndpointsAndSelectiveCounts : IAuditQuery + { + public AuditQuery_WithTwoEndpointsAndSelectiveCounts( + string endpointWithoutCounts, + string endpointWithCounts, + DateOnly throughputDate, + long throughputCount) + { + this.endpointWithoutCounts = endpointWithoutCounts; + this.endpointWithCounts = endpointWithCounts; + this.throughputDate = throughputDate; + this.throughputCount = throughputCount; + } + + public SemanticVersion MinAuditCountsVersion => new(4, 29, 0); + public Func ValidRemoteInstances => _ => true; + + public Task> GetKnownEndpoints(CancellationToken cancellationToken) => + Task.FromResult>( + [ + new ServiceControlEndpoint { Name = endpointWithoutCounts, HeartbeatsEnabled = true }, + new ServiceControlEndpoint { Name = endpointWithCounts, HeartbeatsEnabled = true } + ]); + + public Task> GetAuditCountForEndpoint(string endpointUrlName, CancellationToken cancellationToken) + { + if (endpointUrlName == endpointWithCounts) + { + return Task.FromResult>([new AuditCount { UtcDate = throughputDate, Count = throughputCount }]); + } + + return Task.FromResult>([]); + } + + public Task> GetAuditRemotes(CancellationToken cancellationToken) => + Task.FromResult>([]); + + public Task TestAuditConnection(CancellationToken cancellationToken) => + Task.FromResult(new ConnectionSettingsTestResult { ConnectionSuccessful = true, ConnectionErrorMessages = [] }); + + readonly string endpointWithoutCounts; + readonly string endpointWithCounts; + readonly DateOnly throughputDate; + readonly long throughputCount; + } } \ No newline at end of file diff --git a/src/Particular.LicensingComponent/AuditThroughput/AuditThroughputCollectorHostedService.cs b/src/Particular.LicensingComponent/AuditThroughput/AuditThroughputCollectorHostedService.cs index 3a22f73065..dbb8b46e30 100644 --- a/src/Particular.LicensingComponent/AuditThroughput/AuditThroughputCollectorHostedService.cs +++ b/src/Particular.LicensingComponent/AuditThroughput/AuditThroughputCollectorHostedService.cs @@ -81,7 +81,7 @@ async Task GatherThroughput(CancellationToken cancellationToken) } else if (endpoint is null) { - return; + continue; } var missingAuditThroughput = auditCounts diff --git a/src/ServiceControl.Audit.Persistence.Tests/AuditCountingTests.cs b/src/ServiceControl.Audit.Persistence.Tests/AuditCountingTests.cs index e067bd33b9..3344510c40 100644 --- a/src/ServiceControl.Audit.Persistence.Tests/AuditCountingTests.cs +++ b/src/ServiceControl.Audit.Persistence.Tests/AuditCountingTests.cs @@ -67,7 +67,36 @@ string ScrubDates(string input) }, ScrubDates); } - ProcessedMessage MakeMessage(string processingEndpoint, DateTime processedAt, bool systemMessage) + [Test] + public async Task Should_return_zero_throughput_entry_when_SendOnly() + { + // Arrange + var today = DateTime.UtcNow.Date; + const string sendOnlyEndpoint = "SendOnlyEndpoint"; + + var messages = new[] + { + // Endpoint sent a message, but did not receive any + MakeMessage("SomeOtherEndpoint", sendOnlyEndpoint, today, false) + }; + + await IngestProcessedMessagesAudits(messages); + + // Act + var result = (await DataStore.QueryAuditCounts(sendOnlyEndpoint, TestContext.CurrentContext.CancellationToken)).Results; + + // Assert + Assert.That(result, Is.Not.Empty, "Expected non-empty result for endpoint that only sent messages"); + Assert.That(result, Has.Count.EqualTo(1), "Expected single audit count for send-only endpoint"); + using (Assert.EnterMultipleScope()) + { + Assert.That(result[0].UtcDate, Is.EqualTo(today), "Expected today's date placeholder"); + Assert.That(result[0].Count, Is.Zero, "Expected zero throughput count for send-only endpoint"); + } + } + + static ProcessedMessage MakeMessage(string processingEndpoint, DateTime processedAt, bool systemMessage) => MakeMessage(processingEndpoint, null, processedAt, systemMessage); + static ProcessedMessage MakeMessage(string processingEndpoint, string sendingEndpoint, DateTime processedAt, bool systemMessage) { var messageId = Guid.NewGuid().ToString(); var messageType = "MyMessageType"; @@ -85,8 +114,12 @@ ProcessedMessage MakeMessage(string processingEndpoint, DateTime processedAt, bo { "MessageType", messageType }, { "IsRetried", false }, { "ConversationId", messageId }, - { "ReceivingEndpoint", new EndpointDetails { Name = processingEndpoint } } + { "ReceivingEndpoint", new EndpointDetails { Name = processingEndpoint } }, }; + if (!string.IsNullOrEmpty(sendingEndpoint)) + { + metadata.Add("SendingEndpoint", new EndpointDetails { Name = sendingEndpoint }); + } var headers = new Dictionary { From f3e941ecea491123266039ab06b996a0c7c66f42 Mon Sep 17 00:00:00 2001 From: Phil Bastian Date: Mon, 17 Aug 2026 08:10:31 +0800 Subject: [PATCH 07/13] fix in-memory test run --- .../InMemoryAuditDataStore.cs | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/src/ServiceControl.Audit.Persistence.InMemory/InMemoryAuditDataStore.cs b/src/ServiceControl.Audit.Persistence.InMemory/InMemoryAuditDataStore.cs index f06b0681b7..6b6c397cb2 100644 --- a/src/ServiceControl.Audit.Persistence.InMemory/InMemoryAuditDataStore.cs +++ b/src/ServiceControl.Audit.Persistence.InMemory/InMemoryAuditDataStore.cs @@ -5,11 +5,13 @@ using System.Linq; using System.Threading; using System.Threading.Tasks; + using OpenTelemetry.Trace; using ServiceControl.Audit.Auditing; using ServiceControl.Audit.Auditing.BodyStorage; using ServiceControl.Audit.Auditing.MessagesView; using ServiceControl.Audit.Infrastructure; using ServiceControl.SagaAudit; + using static System.Runtime.InteropServices.JavaScript.JSType; class InMemoryAuditDataStore : IAuditDataStore { @@ -173,6 +175,7 @@ Task GetMessageBodyFromMetadata(string messageId) public Task>> QueryAuditCounts(string endpointName, CancellationToken cancellationToken) { + var hasSent = messageViews.Any(m => m.SendingEndpoint?.Name == endpointName); var results = messageViews .Where(m => m.ReceivingEndpoint.Name == endpointName && !m.IsSystemMessage) .GroupBy(m => m.ProcessedAt.ToUniversalTime().Date) @@ -184,7 +187,13 @@ public Task>> QueryAuditCounts(string endpointName .OrderBy(r => r.UtcDate) .ToList(); - return Task.FromResult(new QueryResult>(results, QueryStatsInfo.Zero)); + return results.Count == 0 && hasSent + ? Task.FromResult(new QueryResult>([new AuditCount + { + UtcDate = messageViews.First().ProcessedAt.ToUniversalTime().Date, + Count = 0 + }], QueryStatsInfo.Zero)) + : Task.FromResult(new QueryResult>(results, QueryStatsInfo.Zero)); } public Task SaveProcessedMessage(ProcessedMessage processedMessage) From 4336db07b9c9269106d34fce93403826077dd98c Mon Sep 17 00:00:00 2001 From: Phil Bastian Date: Mon, 17 Aug 2026 09:11:26 +0800 Subject: [PATCH 08/13] make build happy --- .../InMemoryLicensingDataStore.cs | 2 +- .../ILicensingDataStore.cs | 2 +- .../Implementation/LicensingDataStore.cs | 2 +- .../Throughput/LicensingDataStore.cs | 2 +- 4 files changed, 4 insertions(+), 4 deletions(-) diff --git a/src/Particular.LicensingComponent.Persistence.InMemory/InMemoryLicensingDataStore.cs b/src/Particular.LicensingComponent.Persistence.InMemory/InMemoryLicensingDataStore.cs index a8963a44db..5d63fe73ec 100644 --- a/src/Particular.LicensingComponent.Persistence.InMemory/InMemoryLicensingDataStore.cs +++ b/src/Particular.LicensingComponent.Persistence.InMemory/InMemoryLicensingDataStore.cs @@ -179,7 +179,7 @@ public Task SaveLicensedEndpointDetails(LicensedEndpointDetails result, Cancella return Task.CompletedTask; } - public Task RemoveEndpoints(EndpointIdentifier[] endpointIds, CancellationToken cancellationToken) + public Task RemoveEndpoints(EndpointIdentifier[] endpointIds, CancellationToken cancellationToken = default) { foreach (var id in endpointIds) { diff --git a/src/Particular.LicensingComponent.Persistence/ILicensingDataStore.cs b/src/Particular.LicensingComponent.Persistence/ILicensingDataStore.cs index df5cbf7d8c..567ae4089c 100644 --- a/src/Particular.LicensingComponent.Persistence/ILicensingDataStore.cs +++ b/src/Particular.LicensingComponent.Persistence/ILicensingDataStore.cs @@ -15,7 +15,7 @@ public interface ILicensingDataStore Task SaveEndpoint(Endpoint endpoint, CancellationToken cancellationToken = default); - Task RemoveEndpoints(EndpointIdentifier[] endpointIds, CancellationToken cancellationToken); + Task RemoveEndpoints(EndpointIdentifier[] endpointIds, CancellationToken cancellationToken = default); Task>> GetEndpointThroughputByQueueName(IList queueNames, CancellationToken cancellationToken = default); diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/LicensingDataStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/LicensingDataStore.cs index fde4bbc9a0..694f828209 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/LicensingDataStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/LicensingDataStore.cs @@ -246,7 +246,7 @@ static async Task TryRecordEndpointThroughput(ServiceControlDbContext cont }); } - public Task RemoveEndpoints(EndpointIdentifier[] endpointIds, CancellationToken cancellationToken) => throw new NotImplementedException(); + public Task RemoveEndpoints(EndpointIdentifier[] endpointIds, CancellationToken cancellationToken = default) => throw new NotImplementedException(); public Task UpdateUserIndicatorOnEndpoints(List userIndicatorUpdates, CancellationToken cancellationToken = default) => ExecuteWithDbContext(async (context, token) => diff --git a/src/ServiceControl.Persistence.RavenDB/Throughput/LicensingDataStore.cs b/src/ServiceControl.Persistence.RavenDB/Throughput/LicensingDataStore.cs index b949e048d8..ef4f23402d 100644 --- a/src/ServiceControl.Persistence.RavenDB/Throughput/LicensingDataStore.cs +++ b/src/ServiceControl.Persistence.RavenDB/Throughput/LicensingDataStore.cs @@ -121,7 +121,7 @@ public async Task SaveEndpoint(Endpoint endpoint, CancellationToken cancellation await session.SaveChangesAsync(cancellationToken); } - public async Task RemoveEndpoints(EndpointIdentifier[] endpointIds, CancellationToken cancellationToken) + public async Task RemoveEndpoints(EndpointIdentifier[] endpointIds, CancellationToken cancellationToken = default) { var documentIds = endpointIds.Select(id => id.GenerateDocumentId()); From 1b826068707f8a0b2feba694d5cec43bcd73a605 Mon Sep 17 00:00:00 2001 From: Phil Bastian Date: Mon, 17 Aug 2026 09:20:09 +0800 Subject: [PATCH 09/13] make build happy --- .../AuditThroughputCollectorHostedService_Tests.cs | 4 ++-- .../ThroughputCollector_Report_Throughput_Tests.cs | 2 +- .../ThroughputCollector_ThroughputSummary_Tests.cs | 2 +- 3 files changed, 4 insertions(+), 4 deletions(-) diff --git a/src/Particular.LicensingComponent.UnitTests/AuditThroughputCollectorHostedService_Tests.cs b/src/Particular.LicensingComponent.UnitTests/AuditThroughputCollectorHostedService_Tests.cs index e827e5328f..05cd7a6c21 100644 --- a/src/Particular.LicensingComponent.UnitTests/AuditThroughputCollectorHostedService_Tests.cs +++ b/src/Particular.LicensingComponent.UnitTests/AuditThroughputCollectorHostedService_Tests.cs @@ -234,8 +234,8 @@ await Task.Run(async () => }); await auditThroughputCollectorHostedService.StopAsync(token); - var endpointWithoutCounts = await DataStore.GetEndpoint("EndpointNoData", ThroughputSource.Audit, default); - var endpointWithCounts = await DataStore.GetEndpoint("EndpointWithData", ThroughputSource.Audit, default); + var endpointWithoutCounts = await DataStore.GetEndpoint("EndpointNoData", ThroughputSource.Audit, CancellationToken.None); + var endpointWithCounts = await DataStore.GetEndpoint("EndpointWithData", ThroughputSource.Audit, CancellationToken.None); // Assert using (Assert.EnterMultipleScope()) diff --git a/src/Particular.LicensingComponent.UnitTests/ThroughputCollector/ThroughputCollector_Report_Throughput_Tests.cs b/src/Particular.LicensingComponent.UnitTests/ThroughputCollector/ThroughputCollector_Report_Throughput_Tests.cs index 1f69dfb125..5ae45f22a0 100644 --- a/src/Particular.LicensingComponent.UnitTests/ThroughputCollector/ThroughputCollector_Report_Throughput_Tests.cs +++ b/src/Particular.LicensingComponent.UnitTests/ThroughputCollector/ThroughputCollector_Report_Throughput_Tests.cs @@ -219,7 +219,7 @@ public async Task Should_not_return_endpoint_in_report_when_endpoint_has_no_thro await DataStore.CreateBuilder().AddEndpoint().Build(); // Act - var report = await ThroughputCollector.GenerateThroughputReport("", null, default); + var report = await ThroughputCollector.GenerateThroughputReport("", null, CancellationToken.None); // Assert Assert.That(report, Is.Not.Null); diff --git a/src/Particular.LicensingComponent.UnitTests/ThroughputCollector/ThroughputCollector_ThroughputSummary_Tests.cs b/src/Particular.LicensingComponent.UnitTests/ThroughputCollector/ThroughputCollector_ThroughputSummary_Tests.cs index 0cb311706b..ddb3725bec 100644 --- a/src/Particular.LicensingComponent.UnitTests/ThroughputCollector/ThroughputCollector_ThroughputSummary_Tests.cs +++ b/src/Particular.LicensingComponent.UnitTests/ThroughputCollector/ThroughputCollector_ThroughputSummary_Tests.cs @@ -184,7 +184,7 @@ await DataStore.CreateBuilder().AddEndpoint().WithThroughput(new ThroughputData( new EndpointDailyThroughput(new DateOnly(2025, 1, 10), 0)])).Build(); // Act - var summary = await ThroughputCollector.GetThroughputSummary(default); + var summary = await ThroughputCollector.GetThroughputSummary(CancellationToken.None); // Assert Assert.That(summary, Is.Not.Null); From 1658b95721ad0136c12e80550cd440efae669645 Mon Sep 17 00:00:00 2001 From: Phil Bastian Date: Mon, 17 Aug 2026 09:21:20 +0800 Subject: [PATCH 10/13] make build happy --- .../AuditThroughputCollectorHostedService_Tests.cs | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/src/Particular.LicensingComponent.UnitTests/AuditThroughputCollectorHostedService_Tests.cs b/src/Particular.LicensingComponent.UnitTests/AuditThroughputCollectorHostedService_Tests.cs index 05cd7a6c21..e548393052 100644 --- a/src/Particular.LicensingComponent.UnitTests/AuditThroughputCollectorHostedService_Tests.cs +++ b/src/Particular.LicensingComponent.UnitTests/AuditThroughputCollectorHostedService_Tests.cs @@ -395,14 +395,14 @@ public AuditQuery_WithTwoEndpointsAndSelectiveCounts( public SemanticVersion MinAuditCountsVersion => new(4, 29, 0); public Func ValidRemoteInstances => _ => true; - public Task> GetKnownEndpoints(CancellationToken cancellationToken) => + public Task> GetKnownEndpoints(CancellationToken cancellationToken = default) => Task.FromResult>( [ new ServiceControlEndpoint { Name = endpointWithoutCounts, HeartbeatsEnabled = true }, new ServiceControlEndpoint { Name = endpointWithCounts, HeartbeatsEnabled = true } ]); - public Task> GetAuditCountForEndpoint(string endpointUrlName, CancellationToken cancellationToken) + public Task> GetAuditCountForEndpoint(string endpointUrlName, CancellationToken cancellationToken = default) { if (endpointUrlName == endpointWithCounts) { @@ -412,10 +412,10 @@ public Task> GetAuditCountForEndpoint(string endpointUrl return Task.FromResult>([]); } - public Task> GetAuditRemotes(CancellationToken cancellationToken) => + public Task> GetAuditRemotes(CancellationToken cancellationToken = default) => Task.FromResult>([]); - public Task TestAuditConnection(CancellationToken cancellationToken) => + public Task TestAuditConnection(CancellationToken cancellationToken = default) => Task.FromResult(new ConnectionSettingsTestResult { ConnectionSuccessful = true, ConnectionErrorMessages = [] }); readonly string endpointWithoutCounts; From 4eca3b2c210c967f1d2496a932f7e895c1a0f18a Mon Sep 17 00:00:00 2001 From: Phil Bastian Date: Mon, 17 Aug 2026 09:26:46 +0800 Subject: [PATCH 11/13] remove unneeded includes --- .../InMemoryAuditDataStore.cs | 2 -- 1 file changed, 2 deletions(-) diff --git a/src/ServiceControl.Audit.Persistence.InMemory/InMemoryAuditDataStore.cs b/src/ServiceControl.Audit.Persistence.InMemory/InMemoryAuditDataStore.cs index 92ebc6bd9a..1ea4212771 100644 --- a/src/ServiceControl.Audit.Persistence.InMemory/InMemoryAuditDataStore.cs +++ b/src/ServiceControl.Audit.Persistence.InMemory/InMemoryAuditDataStore.cs @@ -5,13 +5,11 @@ using System.Linq; using System.Threading; using System.Threading.Tasks; - using OpenTelemetry.Trace; using ServiceControl.Audit.Auditing; using ServiceControl.Audit.Auditing.BodyStorage; using ServiceControl.Audit.Auditing.MessagesView; using ServiceControl.Audit.Infrastructure; using ServiceControl.SagaAudit; - using static System.Runtime.InteropServices.JavaScript.JSType; class InMemoryAuditDataStore : IAuditDataStore { From 4bb78affde25b70523eb808dfe808e59895bd6d5 Mon Sep 17 00:00:00 2001 From: Phil Bastian Date: Tue, 18 Aug 2026 11:52:08 +0800 Subject: [PATCH 12/13] refactors from code review --- .../AuditThroughputCollectorHostedService.cs | 11 ++++++----- .../RavenAuditDataStore.cs | 19 +++++++++++-------- 2 files changed, 17 insertions(+), 13 deletions(-) diff --git a/src/Particular.LicensingComponent/AuditThroughput/AuditThroughputCollectorHostedService.cs b/src/Particular.LicensingComponent/AuditThroughput/AuditThroughputCollectorHostedService.cs index 16281cd2a8..5b345182dd 100644 --- a/src/Particular.LicensingComponent/AuditThroughput/AuditThroughputCollectorHostedService.cs +++ b/src/Particular.LicensingComponent/AuditThroughput/AuditThroughputCollectorHostedService.cs @@ -78,15 +78,16 @@ async Task GatherThroughput(CancellationToken cancellationToken) var auditCounts = (await auditQuery.GetAuditCountForEndpoint(knownEndpointsLookup[endpointId].UrlName, cancellationToken)).ToList(); - if (endpoint == null && auditCounts.Count > 0) + if (endpoint is null) { + if (auditCounts.Count <= 0) + { + continue; + } + endpoint = ConvertToEndpoint(knownEndpointsLookup[endpointId]); await dataStore.SaveEndpoint(endpoint, cancellationToken); } - else if (endpoint is null) - { - continue; - } var missingAuditThroughput = auditCounts .Where(auditCount => auditCount.UtcDate > endpoint.LastCollectedDate && diff --git a/src/ServiceControl.Audit.Persistence.RavenDB/RavenAuditDataStore.cs b/src/ServiceControl.Audit.Persistence.RavenDB/RavenAuditDataStore.cs index bb83359372..6c39ad103c 100644 --- a/src/ServiceControl.Audit.Persistence.RavenDB/RavenAuditDataStore.cs +++ b/src/ServiceControl.Audit.Persistence.RavenDB/RavenAuditDataStore.cs @@ -138,9 +138,6 @@ public async Task>> QueryAuditCounts(string endpoi .OrderBy(m => m.ProcessedAt) .FirstOrDefaultAsync(token: cancellationToken); - var hasSent = await session.Query(indexName) - .AnyAsync(m => m.SendingEndpointName == endpointName, token: cancellationToken); - if (oldestMsg != null) { var endDate = DateTime.UtcNow.Date.AddDays(1); @@ -169,13 +166,19 @@ public async Task>> QueryAuditCounts(string endpoi } } } - else if (hasSent) + else { - results.Add(new AuditCount + var hasSent = await session.Query(indexName) + .AnyAsync(m => m.SendingEndpointName == endpointName, token: cancellationToken); + + if (hasSent) { - UtcDate = DateTime.UtcNow.Date, - Count = 0 - }); + results.Add(new AuditCount + { + UtcDate = DateTime.UtcNow.Date, + Count = 0 + }); + } } return new QueryResult>(results, QueryStatsInfo.Zero); From 4e79ff50767503c477da4adeb6435bb41c481834 Mon Sep 17 00:00:00 2001 From: Phil Bastian Date: Tue, 18 Aug 2026 12:42:40 +0800 Subject: [PATCH 13/13] fix interface implementation --- .../MonitoringService_Tests.cs | 1 + 1 file changed, 1 insertion(+) diff --git a/src/Particular.LicensingComponent.UnitTests/MonitoringService_Tests.cs b/src/Particular.LicensingComponent.UnitTests/MonitoringService_Tests.cs index 55f14310ac..438cb729dd 100644 --- a/src/Particular.LicensingComponent.UnitTests/MonitoringService_Tests.cs +++ b/src/Particular.LicensingComponent.UnitTests/MonitoringService_Tests.cs @@ -263,6 +263,7 @@ public Task GetLicensedEndpointDetails(CancellationToke public Task SaveLicensedEndpointDetails(LicensedEndpointDetails result, CancellationToken cancellationToken = default) => throw new NotSupportedException(); + public Task RemoveEndpoints(EndpointIdentifier[] endpointIds, CancellationToken cancellationToken = default) => throw new NotImplementedException(); } class BrokerThroughputQuery_WithSanitization : IBrokerThroughputQuery