From eab024c4cfbd3770cd80889432e64cb2b2855745 Mon Sep 17 00:00:00 2001 From: Phil Bastian Date: Thu, 30 Jul 2026 09:30:30 +0800 Subject: [PATCH 1/6] 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 2/6] 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 3/6] 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 4/6] 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 5/6] 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 6/6] 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 {