From e899fa069cb82fed71cf4be9a5fd257b4020ea5e Mon Sep 17 00:00:00 2001 From: Ramon Smits Date: Fri, 4 Sep 2026 17:00:11 +0200 Subject: [PATCH 1/2] =?UTF-8?q?=F0=9F=90=9B=20Terminate=20message=20querie?= =?UTF-8?q?s=20that=20exceed=20a=20configurable=20time=20limit?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Customer debug packages showed audit message queries against MessagesViewIndexWithFullTextSearch running for up to ~65 minutes. RavenDB's own Databases.QueryTimeoutInSec (default 300) does not stop them because the server renews the operation deadline while a query keeps making progress, so large scans/sorts run for hours and their sorted results spill into the server's Temp folder, filling the disk. The message view and search queries (all messages, per endpoint, search, conversation) now run under a linked CancellationTokenSource that cancels after a configurable time limit, default 1 minute, maximum 1 hour. Cancelling the client request aborts the call to the database server, which does terminate the query server-side. On expiry a TimeoutException names the setting to adjust; invalid values fall back to the default. One QueryTimeLimit helper in ServiceControl.Infrastructure serves every persister, so they share one setting with one behavior, a hard wall-clock deadline per data store call. Its rule is that whatever exception surfaces after the deadline fired is the timeout: RavenDB and Npgsql raise OperationCanceledException, Microsoft.Data.SqlClient raises SqlException("Operation cancelled by user"), which a plain OperationCanceledException catch never sees. A test against the real SQL Server and PostgreSQL containers slows the command down server-side to cover that. - ServiceControl.Audit/QueryTimeoutInSeconds bounds the audit RavenAuditDataStore message view queries. The licensing audit counts and the saga history lookup are not under the limit. - ServiceControl/QueryTimeoutInSeconds bounds the primary instance IMessagesViewDataStore queries on RavenDB, SQL Server and PostgreSQL, which use the same unbounded sorted index query shape. On SQL Server and PostgreSQL these queries also use the limit as their per-command timeout, so Database/CommandTimeout cannot undercut it. Co-authored-by: Mauro Servienti --- .../DatabaseConfiguration.cs | 5 +- .../RavenAuditDataStore.cs | 161 ++++++++------- .../RavenPersistenceConfiguration.cs | 24 ++- .../ConfigurationValidationTests.cs | 51 +++++ .../QueryTimeoutTests.cs | 46 +++++ .../SharedEmbeddedServer.cs | 2 +- .../QueryTimeLimitTests.cs | 135 +++++++++++++ .../QueryTimeLimit.cs | 78 ++++++++ .../EFPersistenceConfigurationBase.cs | 1 + .../Implementation/DataStoreBase.cs | 20 ++ .../Implementation/MessagesViewDataStore.cs | 10 +- .../ErrorMessagesDataStore.cs | 187 +++++++++--------- .../RavenPersistenceConfiguration.cs | 3 +- .../RavenPersisterSettings.cs | 2 +- .../PersistenceTestsContext.cs | 12 ++ .../QueryTimeoutTests.cs | 55 ++++++ .../PersistenceTestsContext.cs | 12 ++ .../EFCore/MessagesViewQueryTimeoutTests.cs | 89 +++++++++ .../EFCore/QueryTimeoutConfigurationTests.cs | 89 +++++++++ .../PersistenceSettings.cs | 13 +- 20 files changed, 817 insertions(+), 178 deletions(-) create mode 100644 src/ServiceControl.Audit.Persistence.Tests.RavenDB/QueryTimeoutTests.cs create mode 100644 src/ServiceControl.Infrastructure.Tests/QueryTimeLimitTests.cs create mode 100644 src/ServiceControl.Infrastructure/QueryTimeLimit.cs create mode 100644 src/ServiceControl.Persistence.Tests.RavenDB/QueryTimeoutTests.cs create mode 100644 src/ServiceControl.Persistence.Tests/EFCore/MessagesViewQueryTimeoutTests.cs create mode 100644 src/ServiceControl.Persistence.Tests/EFCore/QueryTimeoutConfigurationTests.cs diff --git a/src/ServiceControl.Audit.Persistence.RavenDB/DatabaseConfiguration.cs b/src/ServiceControl.Audit.Persistence.RavenDB/DatabaseConfiguration.cs index 2c0e0cd5d9..b9a6550383 100644 --- a/src/ServiceControl.Audit.Persistence.RavenDB/DatabaseConfiguration.cs +++ b/src/ServiceControl.Audit.Persistence.RavenDB/DatabaseConfiguration.cs @@ -11,7 +11,8 @@ public class DatabaseConfiguration( int dataSpaceRemainingThreshold, int minimumStorageLeftRequiredForIngestion, ServerConfiguration serverConfiguration, - TimeSpan bulkInsertCommitTimeout) + TimeSpan bulkInsertCommitTimeout, + TimeSpan queryTimeout) { public string Name { get; } = name; @@ -30,5 +31,7 @@ public class DatabaseConfiguration( public int MinimumStorageLeftRequiredForIngestion { get; internal set; } = minimumStorageLeftRequiredForIngestion; //Setting for ATT only public TimeSpan BulkInsertCommitTimeout { get; } = bulkInsertCommitTimeout; + + public TimeSpan QueryTimeout { get; } = queryTimeout; } } diff --git a/src/ServiceControl.Audit.Persistence.RavenDB/RavenAuditDataStore.cs b/src/ServiceControl.Audit.Persistence.RavenDB/RavenAuditDataStore.cs index 56ab14f906..18e2aa7706 100644 --- a/src/ServiceControl.Audit.Persistence.RavenDB/RavenAuditDataStore.cs +++ b/src/ServiceControl.Audit.Persistence.RavenDB/RavenAuditDataStore.cs @@ -1,4 +1,4 @@ -namespace ServiceControl.Audit.Persistence.RavenDB +namespace ServiceControl.Audit.Persistence.RavenDB { using System; using System.Collections.Generic; @@ -11,6 +11,7 @@ using Raven.Client.Documents; using ServiceControl.Audit.Auditing; using ServiceControl.Audit.Infrastructure; + using ServiceControl.Infrastructure; using ServiceControl.SagaAudit; using Transformers; @@ -28,81 +29,86 @@ public async Task> QuerySagaHistoryById(Guid input, Can return sagaHistory == null ? QueryResult.Empty() : new QueryResult(sagaHistory, stats.ToQueryStatsInfo()); } - public async Task>> GetMessages(bool includeSystemMessages, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange, CancellationToken cancellationToken = default) - { - using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); - var results = await session.Query(GetIndexName(isFullTextSearchEnabled)) - .Statistics(out var stats) - .FilterBySentTimeRange(timeSentRange) - .IncludeSystemMessagesWhere(includeSystemMessages) - .Sort(sortInfo) - .Paging(pagingInfo) - .ToMessagesView() - .ToListAsync(token: cancellationToken); - - return new QueryResult>(results, stats.ToQueryStatsInfo()); - } - - public async Task>> QueryMessages(string searchParam, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange, CancellationToken cancellationToken = default) - { - using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); - var results = await session.Query(GetIndexName(isFullTextSearchEnabled)) - .Statistics(out var stats) - .Search(x => x.Query, searchParam) - .FilterBySentTimeRange(timeSentRange) - .Sort(sortInfo) - .Paging(pagingInfo) - .ToMessagesView() - .ToListAsync(token: cancellationToken); - - return new QueryResult>(results, stats.ToQueryStatsInfo()); - } - - public async Task>> QueryMessagesByReceivingEndpointAndKeyword(string endpoint, string keyword, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange, CancellationToken cancellationToken = default) - { - using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); - var results = await session.Query(GetIndexName(isFullTextSearchEnabled)) - .Statistics(out var stats) - .Search(x => x.Query, keyword) - .Where(m => m.ReceivingEndpointName == endpoint) - .FilterBySentTimeRange(timeSentRange) - .Sort(sortInfo) - .Paging(pagingInfo) - .ToMessagesView() - .ToListAsync(token: cancellationToken); - - return new QueryResult>(results, stats.ToQueryStatsInfo()); - } - - public async Task>> QueryMessagesByReceivingEndpoint(bool includeSystemMessages, string endpointName, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange, CancellationToken cancellationToken = default) - { - using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); - var results = await session.Query(GetIndexName(isFullTextSearchEnabled)) - .Statistics(out var stats) - .IncludeSystemMessagesWhere(includeSystemMessages) - .Where(m => m.ReceivingEndpointName == endpointName) - .FilterBySentTimeRange(timeSentRange) - .Sort(sortInfo) - .Paging(pagingInfo) - .ToMessagesView() - .ToListAsync(token: cancellationToken); - - return new QueryResult>(results, stats.ToQueryStatsInfo()); - } + public Task>> GetMessages(bool includeSystemMessages, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange, CancellationToken cancellationToken = default) => + WithQueryTimeout(async token => + { + using var session = await sessionProvider.OpenSession(cancellationToken: token); + var results = await session.Query(GetIndexName(isFullTextSearchEnabled)) + .Statistics(out var stats) + .FilterBySentTimeRange(timeSentRange) + .IncludeSystemMessagesWhere(includeSystemMessages) + .Sort(sortInfo) + .Paging(pagingInfo) + .ToMessagesView() + .ToListAsync(token: token); + + return new QueryResult>(results, stats.ToQueryStatsInfo()); + }, cancellationToken); + + public Task>> QueryMessages(string searchParam, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange, CancellationToken cancellationToken = default) => + WithQueryTimeout(async token => + { + using var session = await sessionProvider.OpenSession(cancellationToken: token); + var results = await session.Query(GetIndexName(isFullTextSearchEnabled)) + .Statistics(out var stats) + .Search(x => x.Query, searchParam) + .FilterBySentTimeRange(timeSentRange) + .Sort(sortInfo) + .Paging(pagingInfo) + .ToMessagesView() + .ToListAsync(token: token); + + return new QueryResult>(results, stats.ToQueryStatsInfo()); + }, cancellationToken); + + public Task>> QueryMessagesByReceivingEndpointAndKeyword(string endpoint, string keyword, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange, CancellationToken cancellationToken = default) => + WithQueryTimeout(async token => + { + using var session = await sessionProvider.OpenSession(cancellationToken: token); + var results = await session.Query(GetIndexName(isFullTextSearchEnabled)) + .Statistics(out var stats) + .Search(x => x.Query, keyword) + .Where(m => m.ReceivingEndpointName == endpoint) + .FilterBySentTimeRange(timeSentRange) + .Sort(sortInfo) + .Paging(pagingInfo) + .ToMessagesView() + .ToListAsync(token: token); + + return new QueryResult>(results, stats.ToQueryStatsInfo()); + }, cancellationToken); + + public Task>> QueryMessagesByReceivingEndpoint(bool includeSystemMessages, string endpointName, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange, CancellationToken cancellationToken = default) => + WithQueryTimeout(async token => + { + using var session = await sessionProvider.OpenSession(cancellationToken: token); + var results = await session.Query(GetIndexName(isFullTextSearchEnabled)) + .Statistics(out var stats) + .IncludeSystemMessagesWhere(includeSystemMessages) + .Where(m => m.ReceivingEndpointName == endpointName) + .FilterBySentTimeRange(timeSentRange) + .Sort(sortInfo) + .Paging(pagingInfo) + .ToMessagesView() + .ToListAsync(token: token); + + return new QueryResult>(results, stats.ToQueryStatsInfo()); + }, cancellationToken); + + public Task>> QueryMessagesByConversationId(string conversationId, PagingInfo pagingInfo, SortInfo sortInfo, CancellationToken cancellationToken = default) => + WithQueryTimeout(async token => + { + using var session = await sessionProvider.OpenSession(cancellationToken: token); + var results = await session.Query(GetIndexName(isFullTextSearchEnabled)) + .Statistics(out var stats) + .Where(m => m.ConversationId == conversationId) + .Sort(sortInfo) + .Paging(pagingInfo) + .ToMessagesView() + .ToListAsync(token: token); - public async Task>> QueryMessagesByConversationId(string conversationId, PagingInfo pagingInfo, SortInfo sortInfo, CancellationToken cancellationToken = default) - { - using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); - var results = await session.Query(GetIndexName(isFullTextSearchEnabled)) - .Statistics(out var stats) - .Where(m => m.ConversationId == conversationId) - .Sort(sortInfo) - .Paging(pagingInfo) - .ToMessagesView() - .ToListAsync(token: cancellationToken); - - return new QueryResult>(results, stats.ToQueryStatsInfo()); - } + return new QueryResult>(results, stats.ToQueryStatsInfo()); + }, cancellationToken); public async Task GetMessageBody(string messageId, CancellationToken cancellationToken = default) { @@ -184,8 +190,11 @@ public async Task>> QueryAuditCounts(string endpoi return new QueryResult>(results, QueryStatsInfo.Zero); } + Task WithQueryTimeout(Func> query, CancellationToken cancellationToken) => + QueryTimeLimit.Run(query, databaseConfiguration.QueryTimeout, RavenPersistenceConfiguration.QueryTimeoutSettingName, cancellationToken); + static string GetIndexName(bool isFullTextSearchEnabled) => isFullTextSearchEnabled ? "MessagesViewIndexWithFullTextSearch" : "MessagesViewIndex"; bool isFullTextSearchEnabled = databaseConfiguration.EnableFullTextSearch; } -} \ No newline at end of file +} diff --git a/src/ServiceControl.Audit.Persistence.RavenDB/RavenPersistenceConfiguration.cs b/src/ServiceControl.Audit.Persistence.RavenDB/RavenPersistenceConfiguration.cs index eced9e35b2..09a0850ad8 100644 --- a/src/ServiceControl.Audit.Persistence.RavenDB/RavenPersistenceConfiguration.cs +++ b/src/ServiceControl.Audit.Persistence.RavenDB/RavenPersistenceConfiguration.cs @@ -1,4 +1,4 @@ -namespace ServiceControl.Audit.Persistence.RavenDB +namespace ServiceControl.Audit.Persistence.RavenDB { using System; using System.Collections.Generic; @@ -23,6 +23,8 @@ public class RavenPersistenceConfiguration : IPersistenceConfiguration public const string MinimumStorageLeftRequiredForIngestionKey = "MinimumStorageLeftRequiredForIngestion"; public const string BulkInsertCommitTimeoutInSecondsKey = "BulkInsertCommitTimeoutInSeconds"; public const string DataSpaceRemainingThresholdKey = "DataSpaceRemainingThreshold"; + public const string QueryTimeoutInSecondsKey = QueryTimeLimit.SettingName; + public const string QueryTimeoutSettingName = "ServiceControl.Audit/" + QueryTimeLimit.SettingName; public IEnumerable ConfigurationKeys => new[]{ DatabaseNameKey, @@ -37,7 +39,8 @@ public class RavenPersistenceConfiguration : IPersistenceConfiguration RavenDbLogLevelKey, DataSpaceRemainingThresholdKey, MinimumStorageLeftRequiredForIngestionKey, - BulkInsertCommitTimeoutInSecondsKey + BulkInsertCommitTimeoutInSecondsKey, + QueryTimeoutInSecondsKey }; public string Name => "RavenDB"; @@ -121,6 +124,8 @@ internal static DatabaseConfiguration GetDatabaseConfiguration(PersistenceSettin var bulkInsertTimeout = TimeSpan.FromSeconds(GetBulkInsertCommitTimeout(settings)); + var queryTimeout = GetQueryTimeout(settings); + return new DatabaseConfiguration( databaseName, expirationProcessTimerInSeconds, @@ -130,7 +135,8 @@ internal static DatabaseConfiguration GetDatabaseConfiguration(PersistenceSettin dataSpaceRemainingThreshold, minimumStorageLeftRequiredForIngestion, serverConfiguration, - bulkInsertTimeout); + bulkInsertTimeout, + queryTimeout); } static int GetExpirationProcessTimerInSeconds(PersistenceSettings settings) @@ -185,6 +191,18 @@ static int GetBulkInsertCommitTimeout(PersistenceSettings settings) return bulkInsertCommitTimeoutInSeconds; } + static TimeSpan GetQueryTimeout(PersistenceSettings settings) + { + var queryTimeoutInSeconds = QueryTimeLimit.DefaultSeconds; + + if (settings.PersisterSpecificSettings.TryGetValue(QueryTimeoutInSecondsKey, out var queryTimeoutString)) + { + queryTimeoutInSeconds = int.Parse(queryTimeoutString); + } + + return QueryTimeLimit.Validate(queryTimeoutInSeconds, QueryTimeoutSettingName, Logger); + } + static string GetLogPath(PersistenceSettings settings) { if (!settings.PersisterSpecificSettings.TryGetValue(LogPathKey, out var logPath)) diff --git a/src/ServiceControl.Audit.Persistence.Tests.RavenDB/ConfigurationValidationTests.cs b/src/ServiceControl.Audit.Persistence.Tests.RavenDB/ConfigurationValidationTests.cs index aaed138170..a450381824 100644 --- a/src/ServiceControl.Audit.Persistence.Tests.RavenDB/ConfigurationValidationTests.cs +++ b/src/ServiceControl.Audit.Persistence.Tests.RavenDB/ConfigurationValidationTests.cs @@ -104,6 +104,57 @@ public void Should_throw_if_both_path_or_connection_string_is_configured() Assert.Throws(() => RavenPersistenceConfiguration.GetDatabaseConfiguration(settings)); } + [Test] + public void Should_default_query_timeout_to_one_minute() + { + var settings = BuildSettings(); + + settings.PersisterSpecificSettings[RavenPersistenceConfiguration.ConnectionStringKey] = "connection string"; + + var configuration = RavenPersistenceConfiguration.GetDatabaseConfiguration(settings); + + Assert.That(configuration.QueryTimeout, Is.EqualTo(TimeSpan.FromMinutes(1))); + } + + [Test] + public void Should_apply_query_timeout_setting() + { + var settings = BuildSettings(); + + settings.PersisterSpecificSettings[RavenPersistenceConfiguration.ConnectionStringKey] = "connection string"; + settings.PersisterSpecificSettings[RavenPersistenceConfiguration.QueryTimeoutInSecondsKey] = "120"; + + var configuration = RavenPersistenceConfiguration.GetDatabaseConfiguration(settings); + + Assert.That(configuration.QueryTimeout, Is.EqualTo(TimeSpan.FromSeconds(120))); + } + + [Test] + public void Should_fall_back_to_default_query_timeout_when_value_is_not_positive() + { + var settings = BuildSettings(); + + settings.PersisterSpecificSettings[RavenPersistenceConfiguration.ConnectionStringKey] = "connection string"; + settings.PersisterSpecificSettings[RavenPersistenceConfiguration.QueryTimeoutInSecondsKey] = "0"; + + var configuration = RavenPersistenceConfiguration.GetDatabaseConfiguration(settings); + + Assert.That(configuration.QueryTimeout, Is.EqualTo(TimeSpan.FromMinutes(1))); + } + + [Test] + public void Should_fall_back_to_default_query_timeout_when_value_exceeds_maximum() + { + var settings = BuildSettings(); + + settings.PersisterSpecificSettings[RavenPersistenceConfiguration.ConnectionStringKey] = "connection string"; + settings.PersisterSpecificSettings[RavenPersistenceConfiguration.QueryTimeoutInSecondsKey] = "3700"; + + var configuration = RavenPersistenceConfiguration.GetDatabaseConfiguration(settings); + + Assert.That(configuration.QueryTimeout, Is.EqualTo(TimeSpan.FromMinutes(1))); + } + PersistenceSettings BuildSettings() { return new PersistenceSettings(TimeSpan.FromMinutes(2), true, 100000); diff --git a/src/ServiceControl.Audit.Persistence.Tests.RavenDB/QueryTimeoutTests.cs b/src/ServiceControl.Audit.Persistence.Tests.RavenDB/QueryTimeoutTests.cs new file mode 100644 index 0000000000..d5ccadde48 --- /dev/null +++ b/src/ServiceControl.Audit.Persistence.Tests.RavenDB/QueryTimeoutTests.cs @@ -0,0 +1,46 @@ +namespace ServiceControl.UnitTests; + +using System; +using System.Threading; +using System.Threading.Tasks; +using NUnit.Framework; +using Raven.Client.Documents.Session; +using ServiceControl.Audit.Infrastructure; +using ServiceControl.Audit.Persistence.RavenDB; + +class QueryTimeoutTests +{ + [Test] + public void Should_cancel_query_that_exceeds_the_allowed_query_time() + { + var dataStore = new RavenAuditDataStore(new NeverCompletingSessionProvider(), BuildConfiguration(queryTimeout: TimeSpan.FromMilliseconds(50))); + + var exception = Assert.ThrowsAsync(() => dataStore.QueryMessages("search", new PagingInfo(), new SortInfo("time_sent", "desc"), new DateTimeRange((DateTime?)null, null))); + + Assert.That(exception.InnerException, Is.InstanceOf()); + } + + [Test] + public void Should_propagate_caller_cancellation_instead_of_timing_out() + { + var dataStore = new RavenAuditDataStore(new NeverCompletingSessionProvider(), BuildConfiguration(queryTimeout: TimeSpan.FromSeconds(30))); + + using var callerTokenSource = new CancellationTokenSource(TimeSpan.FromMilliseconds(50)); + + var exception = Assert.CatchAsync(() => dataStore.QueryMessages("search", new PagingInfo(), new SortInfo("time_sent", "desc"), new DateTimeRange((DateTime?)null, null), callerTokenSource.Token)); + + Assert.That(exception, Is.InstanceOf()); + } + + static DatabaseConfiguration BuildConfiguration(TimeSpan queryTimeout) => + new("audit", 60, true, TimeSpan.FromMinutes(5), 120000, 5, 5, new ServerConfiguration("http://localhost:33334"), TimeSpan.FromSeconds(60), queryTimeout); + + class NeverCompletingSessionProvider : IRavenSessionProvider + { + public async ValueTask OpenSession(SessionOptions options = default, CancellationToken cancellationToken = default) + { + await Task.Delay(Timeout.Infinite, cancellationToken); + return null; + } + } +} diff --git a/src/ServiceControl.Audit.Persistence.Tests.RavenDB/SharedEmbeddedServer.cs b/src/ServiceControl.Audit.Persistence.Tests.RavenDB/SharedEmbeddedServer.cs index 0cf229eb72..c3bf3fb38e 100644 --- a/src/ServiceControl.Audit.Persistence.Tests.RavenDB/SharedEmbeddedServer.cs +++ b/src/ServiceControl.Audit.Persistence.Tests.RavenDB/SharedEmbeddedServer.cs @@ -37,7 +37,7 @@ public static async Task GetInstance(CancellationToken cancell var logsMode = "Operations"; var serverUrl = $"http://localhost:{PortUtility.GetAssignedOrAvailablePort(33334)}"; - var databaseConfiguration = new DatabaseConfiguration("audit", 60, true, TimeSpan.FromMinutes(5), 120000, 5, 5, new ServerConfiguration(dbPath, serverUrl, logPath, logsMode), TimeSpan.FromSeconds(60)); + var databaseConfiguration = new DatabaseConfiguration("audit", 60, true, TimeSpan.FromMinutes(5), 120000, 5, 5, new ServerConfiguration(dbPath, serverUrl, logPath, logsMode), TimeSpan.FromSeconds(60), TimeSpan.FromSeconds(30)); var serverConfig = databaseConfiguration.ServerConfiguration; // TODO: See if more refactoring can be done in configuration classes diff --git a/src/ServiceControl.Infrastructure.Tests/QueryTimeLimitTests.cs b/src/ServiceControl.Infrastructure.Tests/QueryTimeLimitTests.cs new file mode 100644 index 0000000000..4f9260d4e7 --- /dev/null +++ b/src/ServiceControl.Infrastructure.Tests/QueryTimeLimitTests.cs @@ -0,0 +1,135 @@ +namespace ServiceControl.Infrastructure.Tests; + +using System; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; +using NUnit.Framework; +using ServiceControl.Infrastructure; + +[TestFixture] +public class QueryTimeLimitTests +{ + const string SettingName = "ServiceControl.Test/QueryTimeoutInSeconds"; + + [Test] + public async Task A_query_within_the_limit_returns_its_result() + { + var result = await QueryTimeLimit.Run(_ => Task.FromResult(42), TimeSpan.FromSeconds(30), SettingName, CancellationToken.None); + + Assert.That(result, Is.EqualTo(42)); + } + + [Test] + public void A_query_over_the_limit_is_cancelled_and_reported_as_a_timeout_naming_the_setting() + { + var exception = Assert.ThrowsAsync(() => + QueryTimeLimit.Run(Hang, TimeSpan.FromMilliseconds(50), SettingName, CancellationToken.None)); + + Assert.That(exception.Message, Does.Contain(SettingName)); + Assert.That(exception.InnerException, Is.InstanceOf()); + } + + [Test] + public void An_exception_the_provider_raises_for_the_cancellation_is_still_reported_as_a_timeout() + { + // Microsoft.Data.SqlClient reports a cancelled command as SqlException("Operation cancelled by user"), + // not as OperationCanceledException. Once the deadline has fired, whatever surfaces is the timeout. + var exception = Assert.ThrowsAsync(() => + QueryTimeLimit.Run(HangThenFailLikeSqlClient, TimeSpan.FromMilliseconds(50), SettingName, CancellationToken.None)); + + Assert.That(exception.InnerException, Is.InstanceOf()); + Assert.That(exception.InnerException.Message, Is.EqualTo("Operation cancelled by user.")); + } + + [Test] + public void A_failure_before_the_deadline_is_not_a_timeout() + { + Assert.ThrowsAsync(() => + QueryTimeLimit.Run(_ => throw new InvalidOperationException("boom"), TimeSpan.FromSeconds(30), SettingName, CancellationToken.None)); + } + + [Test] + public void Caller_cancellation_is_propagated_as_cancellation_not_as_a_timeout() + { + using var caller = new CancellationTokenSource(TimeSpan.FromMilliseconds(50)); + + var exception = Assert.CatchAsync(() => QueryTimeLimit.Run(Hang, TimeSpan.FromSeconds(30), SettingName, caller.Token)); + + Assert.That(exception, Is.InstanceOf()); + Assert.That(exception, Is.Not.InstanceOf()); + } + + [Test] + public void Caller_cancellation_that_the_provider_reports_as_its_own_error_is_still_cancellation() + { + using var caller = new CancellationTokenSource(TimeSpan.FromMilliseconds(50)); + + var exception = Assert.CatchAsync(() => QueryTimeLimit.Run(HangThenFailLikeSqlClient, TimeSpan.FromSeconds(30), SettingName, caller.Token)); + + Assert.That(exception, Is.InstanceOf()); + Assert.That(exception.InnerException, Is.InstanceOf()); + } + + [TestCase(60, 60)] + [TestCase(300, 300)] + [TestCase(3600, 3600)] + public void The_configured_seconds_are_used_when_inside_the_allowed_range(int configured, int expectedSeconds) + { + var limit = QueryTimeLimit.Validate(configured, SettingName, NullLogger.Instance); + + Assert.That(limit, Is.EqualTo(TimeSpan.FromSeconds(expectedSeconds))); + } + + [TestCase(0)] + [TestCase(-5)] + [TestCase(3601)] + public void Values_outside_the_allowed_range_fall_back_to_the_default(int configured) + { + var logger = new RecordingLogger(); + + var limit = QueryTimeLimit.Validate(configured, SettingName, logger); + + Assert.That(limit, Is.EqualTo(TimeSpan.FromSeconds(QueryTimeLimit.DefaultSeconds))); + Assert.That(logger.Errors, Has.Count.EqualTo(1)); + Assert.That(logger.Errors[0], Does.Contain(SettingName)); + } + + static async Task Hang(CancellationToken cancellationToken) + { + await Task.Delay(Timeout.Infinite, cancellationToken); + return 0; + } + + static async Task HangThenFailLikeSqlClient(CancellationToken cancellationToken) + { + try + { + await Task.Delay(Timeout.Infinite, cancellationToken); + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + throw new InvalidOperationException("Operation cancelled by user."); + } + + return 0; + } + + class RecordingLogger : ILogger + { + public System.Collections.Generic.List Errors { get; } = []; + + public IDisposable BeginScope(TState state) where TState : notnull => null; + + public bool IsEnabled(LogLevel logLevel) => true; + + public void Log(LogLevel logLevel, EventId eventId, TState state, Exception exception, Func formatter) + { + if (logLevel == LogLevel.Error) + { + Errors.Add(formatter(state, exception)); + } + } + } +} diff --git a/src/ServiceControl.Infrastructure/QueryTimeLimit.cs b/src/ServiceControl.Infrastructure/QueryTimeLimit.cs new file mode 100644 index 0000000000..c800e7df9a --- /dev/null +++ b/src/ServiceControl.Infrastructure/QueryTimeLimit.cs @@ -0,0 +1,78 @@ +namespace ServiceControl.Infrastructure; + +using System; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Extensions.Logging; +using ServiceControl.Configuration; + +/// +/// Bounds a data store query to one wall-clock deadline. Cancelling the client-side call aborts the request to the +/// database server, which terminates the query server-side; a server-side query timeout alone is not enough, as +/// RavenDB renews its deadline while a query keeps making progress and a single query can then run for hours. +/// +public static class QueryTimeLimit +{ + public const string SettingName = "QueryTimeoutInSeconds"; + public const int DefaultSeconds = 60; + public const int MaxSeconds = 3600; + + public static readonly TimeSpan Default = TimeSpan.FromSeconds(DefaultSeconds); + + /// The query, which must observe the token it is handed. + /// The wall-clock limit for the whole query. + /// The fully qualified setting the timeout message names, e.g. "ServiceControl/QueryTimeoutInSeconds". + /// The caller's token. Its cancellation surfaces as cancellation, not as a timeout. + public static async Task Run(Func> query, TimeSpan limit, string settingName, CancellationToken cancellationToken = default) + { + using var deadline = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + deadline.CancelAfter(limit); + + try + { + return await query(deadline.Token).ConfigureAwait(false); + } + // Once the deadline has fired, whatever the provider raises is the cancellation: RavenDB and Npgsql raise + // OperationCanceledException, Microsoft.Data.SqlClient raises SqlException("Operation cancelled by user"). +#pragma warning disable PS0019 // Catching Exception is the point: the filter attributes it to the deadline, not to the exception type + catch (Exception e) when (deadline.Token.IsCancellationRequested) +#pragma warning restore PS0019 + { + if (cancellationToken.IsCancellationRequested) + { + // The caller cancelled (e.g. the HTTP request was aborted), not the query time limit + if (e is OperationCanceledException) + { + throw; + } + + throw new OperationCanceledException("The query was cancelled by the caller.", e, cancellationToken); + } + + throw Timeout(e, limit, settingName); + } + } + + static TimeoutException Timeout(Exception cause, TimeSpan limit, string settingName) => + new($"The query did not complete within the allowed query time of {limit.TotalSeconds:0} seconds and was cancelled. The '{settingName}' setting can be used to change the allowed database query time.", cause); + + public static TimeSpan Read(SettingsRootNamespace settingsRootNamespace, ILogger logger) => + Validate(SettingsReader.Read(settingsRootNamespace, SettingName, DefaultSeconds), $"{settingsRootNamespace}/{SettingName}", logger); + + public static TimeSpan Validate(int seconds, string settingName, ILogger logger) + { + if (seconds <= 0) + { + logger.LogError("{SettingName} must be greater than zero. Defaulting to {QueryTimeoutInSecondsDefault}", settingName, DefaultSeconds); + return Default; + } + + if (seconds > MaxSeconds) + { + logger.LogError("{SettingName} cannot be larger than {MaxQueryTimeoutInSeconds}. Defaulting to {QueryTimeoutInSecondsDefault}", settingName, MaxSeconds, DefaultSeconds); + return Default; + } + + return TimeSpan.FromSeconds(seconds); + } +} diff --git a/src/ServiceControl.Persistence.EFCore/Abstractions/EFPersistenceConfigurationBase.cs b/src/ServiceControl.Persistence.EFCore/Abstractions/EFPersistenceConfigurationBase.cs index 62529efbf2..0d8d5f8a2f 100644 --- a/src/ServiceControl.Persistence.EFCore/Abstractions/EFPersistenceConfigurationBase.cs +++ b/src/ServiceControl.Persistence.EFCore/Abstractions/EFPersistenceConfigurationBase.cs @@ -40,6 +40,7 @@ public PersistenceSettings CreateSettings(SettingsRootNamespace settingsRootName settings.EventsRetentionPeriod = SettingsReader.Read(settingsRootNamespace, EventsRetentionPeriodKey, EFPersisterSettings.DefaultEventsRetentionPeriod); settings.SubscriptionCacheDuration = SettingsReader.Read(settingsRootNamespace, SubscriptionCacheDurationKey, EFPersisterSettings.DefaultSubscriptionCacheDuration); settings.ExternalIntegrationsDispatchingBatchSize = ReadExternalIntegrationsDispatchingBatchSize(settingsRootNamespace); + settings.QueryTimeout = QueryTimeLimit.Read(settingsRootNamespace, LoggerUtil.CreateStaticLogger()); return settings; } diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/DataStoreBase.cs b/src/ServiceControl.Persistence.EFCore/Implementation/DataStoreBase.cs index 5dc89f5fe0..7bcaec7f2e 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/DataStoreBase.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/DataStoreBase.cs @@ -3,8 +3,11 @@ namespace ServiceControl.Persistence.EFCore.Implementation; using System; using System.Runtime.CompilerServices; using System.Threading.Tasks; +using Abstractions; using DbContexts; +using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.DependencyInjection; +using ServiceControl.Infrastructure; /// /// Base class for data stores that provides helper methods to simplify scope and DbContext management @@ -23,6 +26,23 @@ protected async Task ExecuteWithDbContext(Func + /// Executes a read-only query with a scoped DbContext under the configured query time limit. Cancelling the + /// query sends a cancel signal to the database server, terminating the query server-side. The limit bounds the + /// whole data store call, which can span multiple SQL commands, to one wall-clock deadline; the provider's own + /// per-command timeout (Database/CommandTimeout) is raised to the same value so it cannot undercut it. + /// Only for queries: a write aborted by a deadline would be harmful. + /// + protected async Task ExecuteQueryWithDbContext(Func> query, CancellationToken cancellationToken = default) + { + await using var scope = scopeFactory.CreateAsyncScope(); + var settings = scope.ServiceProvider.GetRequiredService(); + var dbContext = scope.ServiceProvider.GetRequiredService(); + dbContext.Database.SetCommandTimeout(settings.QueryTimeout); + + return await QueryTimeLimit.Run(token => query(dbContext, token), settings.QueryTimeout, PersistenceSettings.QueryTimeoutSettingName, cancellationToken); + } + /// /// Executes an operation with a scoped DbContext, without returning a result /// diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/MessagesViewDataStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/MessagesViewDataStore.cs index 2b2d69d2db..c285095cc3 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/MessagesViewDataStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/MessagesViewDataStore.cs @@ -10,14 +10,14 @@ namespace ServiceControl.Persistence.EFCore.Implementation; public class MessagesViewDataStore(IServiceScopeFactory scopeFactory, IFullTextSearchDialect fullTextSearch) : DataStoreBase(scopeFactory), IMessagesViewDataStore { public Task>> GetAllMessages(PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages, DateTimeRange? timeSentRange = null, CancellationToken cancellationToken = default) => - ExecuteWithDbContext((dbContext, token) => dbContext.FailedMessages + ExecuteQueryWithDbContext((dbContext, token) => dbContext.FailedMessages .AsNoTracking() .IncludeSystemMessagesWhere(includeSystemMessages) .FilterBySentTimeRange(timeSentRange) .ToPagedMessagesResult(pagingInfo, sortInfo, token), cancellationToken); public Task>> GetAllMessagesForEndpoint(string endpointName, PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages, DateTimeRange? timeSentRange = null, CancellationToken cancellationToken = default) => - ExecuteWithDbContext((dbContext, token) => dbContext.FailedMessages + ExecuteQueryWithDbContext((dbContext, token) => dbContext.FailedMessages .AsNoTracking() .Where(message => message.ReceivingEndpointName == endpointName) .IncludeSystemMessagesWhere(includeSystemMessages) @@ -26,18 +26,18 @@ public Task>> GetAllMessagesForEndpoint(string e // includeSystemMessages is unused here: a conversation is incomplete without the system messages that took part in it. public Task>> GetAllMessagesByConversation(string conversationId, PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages, CancellationToken cancellationToken = default) => - ExecuteWithDbContext((dbContext, token) => dbContext.FailedMessages + ExecuteQueryWithDbContext((dbContext, token) => dbContext.FailedMessages .AsNoTracking() .Where(message => message.ConversationId == conversationId) .ToPagedMessagesResult(pagingInfo, sortInfo, token), cancellationToken); public Task>> GetAllMessagesForSearch(string searchTerms, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange? timeSentRange = null, CancellationToken cancellationToken = default) => - ExecuteWithDbContext((dbContext, token) => Search(dbContext.FailedMessages.AsNoTracking(), searchTerms) + ExecuteQueryWithDbContext((dbContext, token) => Search(dbContext.FailedMessages.AsNoTracking(), searchTerms) .FilterBySentTimeRange(timeSentRange) .ToPagedMessagesResult(pagingInfo, sortInfo, token), cancellationToken); public Task>> SearchEndpointMessages(string endpointName, string searchKeyword, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange? timeSentRange = null, CancellationToken cancellationToken = default) => - ExecuteWithDbContext((dbContext, token) => Search(dbContext.FailedMessages.AsNoTracking(), searchKeyword) + ExecuteQueryWithDbContext((dbContext, token) => Search(dbContext.FailedMessages.AsNoTracking(), searchKeyword) .Where(message => message.ReceivingEndpointName == endpointName) .FilterBySentTimeRange(timeSentRange) .ToPagedMessagesResult(pagingInfo, sortInfo, token), cancellationToken); diff --git a/src/ServiceControl.Persistence.RavenDB/ErrorMessagesDataStore.cs b/src/ServiceControl.Persistence.RavenDB/ErrorMessagesDataStore.cs index a381261bd6..52bb6f3b5e 100644 --- a/src/ServiceControl.Persistence.RavenDB/ErrorMessagesDataStore.cs +++ b/src/ServiceControl.Persistence.RavenDB/ErrorMessagesDataStore.cs @@ -1,4 +1,4 @@ -namespace ServiceControl.Persistence.RavenDB +namespace ServiceControl.Persistence.RavenDB { using System; using System.Collections.Generic; @@ -23,6 +23,7 @@ using ServiceControl.MessageFailures.Api; using ServiceControl.Operations; using ServiceControl.Operations.BodyStorage; + using ServiceControl.Infrastructure; using ServiceControl.Persistence.Infrastructure; using ServiceControl.Recoverability; @@ -30,127 +31,135 @@ class ErrorMessagesDataStore( IRavenSessionProvider sessionProvider, IRavenDocumentStoreProvider documentStoreProvider, ExpirationManager expirationManager, + RavenPersisterSettings settings, ILogger logger) : IMessagesViewDataStore, IFailedMessageQueryDataStore, IFailedMessageLifecycleDataStore, IFailedMessageRetryDataStore { - public async Task>> GetAllMessages( + public Task>> GetAllMessages( PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages, DateTimeRange timeSentRange, CancellationToken cancellationToken = default - ) - { - using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); - var query = session.Query() - .IncludeSystemMessagesWhere(includeSystemMessages) - .FilterBySentTimeRange(timeSentRange) - .Statistics(out var stats) - .Sort(sortInfo) - .Paging(pagingInfo) - .OfType() - .TransformToMessageView(); - - var results = await query.ToListAsync(cancellationToken); - - return new QueryResult>(results, stats.ToQueryStatsInfo()); - } - - public async Task>> GetAllMessagesForEndpoint( + ) => + WithQueryTimeout(async token => + { + using var session = await sessionProvider.OpenSession(cancellationToken: token); + var query = session.Query() + .IncludeSystemMessagesWhere(includeSystemMessages) + .FilterBySentTimeRange(timeSentRange) + .Statistics(out var stats) + .Sort(sortInfo) + .Paging(pagingInfo) + .OfType() + .TransformToMessageView(); + + var results = await query.ToListAsync(token); + + return new QueryResult>(results, stats.ToQueryStatsInfo()); + }, cancellationToken); + + public Task>> GetAllMessagesForEndpoint( string endpointName, PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages, DateTimeRange timeSentRange, CancellationToken cancellationToken = default - ) - { - using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); - var query = session.Query() - .IncludeSystemMessagesWhere(includeSystemMessages) - .FilterBySentTimeRange(timeSentRange) - .Where(m => m.ReceivingEndpointName == endpointName) - .Statistics(out var stats) - .Sort(sortInfo) - .Paging(pagingInfo) - .OfType() - .TransformToMessageView(); - - var results = await query.ToListAsync(cancellationToken); - - - return new QueryResult>(results, stats.ToQueryStatsInfo()); - } - - public async Task>> SearchEndpointMessages( + ) => + WithQueryTimeout(async token => + { + using var session = await sessionProvider.OpenSession(cancellationToken: token); + var query = session.Query() + .IncludeSystemMessagesWhere(includeSystemMessages) + .FilterBySentTimeRange(timeSentRange) + .Where(m => m.ReceivingEndpointName == endpointName) + .Statistics(out var stats) + .Sort(sortInfo) + .Paging(pagingInfo) + .OfType() + .TransformToMessageView(); + + var results = await query.ToListAsync(token); + + return new QueryResult>(results, stats.ToQueryStatsInfo()); + }, cancellationToken); + + public Task>> SearchEndpointMessages( string endpointName, string searchKeyword, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange, CancellationToken cancellationToken = default - ) - { - using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); - var query = session.Query() - .Statistics(out var stats) - .Search(x => x.Query, searchKeyword) - .Where(m => m.ReceivingEndpointName == endpointName) - .FilterBySentTimeRange(timeSentRange) - .Sort(sortInfo) - .Paging(pagingInfo) - .OfType() - .TransformToMessageView(); - - var results = await query.ToListAsync(cancellationToken); - - return new QueryResult>(results, stats.ToQueryStatsInfo()); - } - - public async Task>> GetAllMessagesByConversation( + ) => + WithQueryTimeout(async token => + { + using var session = await sessionProvider.OpenSession(cancellationToken: token); + var query = session.Query() + .Statistics(out var stats) + .Search(x => x.Query, searchKeyword) + .Where(m => m.ReceivingEndpointName == endpointName) + .FilterBySentTimeRange(timeSentRange) + .Sort(sortInfo) + .Paging(pagingInfo) + .OfType() + .TransformToMessageView(); + + var results = await query.ToListAsync(token); + + return new QueryResult>(results, stats.ToQueryStatsInfo()); + }, cancellationToken); + + public Task>> GetAllMessagesByConversation( string conversationId, PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages, CancellationToken cancellationToken = default - ) - { - using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); - var query = session.Query() - .Statistics(out var stats) - .Where(m => m.ConversationId == conversationId) - .Sort(sortInfo) - .Paging(pagingInfo) - .OfType() - .TransformToMessageView(); + ) => + WithQueryTimeout(async token => + { + using var session = await sessionProvider.OpenSession(cancellationToken: token); + var query = session.Query() + .Statistics(out var stats) + .Where(m => m.ConversationId == conversationId) + .Sort(sortInfo) + .Paging(pagingInfo) + .OfType() + .TransformToMessageView(); - var results = await query.ToListAsync(cancellationToken); + var results = await query.ToListAsync(token); - return new QueryResult>(results, stats.ToQueryStatsInfo()); - } + return new QueryResult>(results, stats.ToQueryStatsInfo()); + }, cancellationToken); - public async Task>> GetAllMessagesForSearch( + public Task>> GetAllMessagesForSearch( string searchTerms, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange, CancellationToken cancellationToken = default - ) - { - using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); - var query = session.Query() - .Statistics(out var stats) - .Search(x => x.Query, searchTerms) - .FilterBySentTimeRange(timeSentRange) - .Sort(sortInfo) - .Paging(pagingInfo) - .OfType() - .TransformToMessageView(); - - var results = await query.ToListAsync(cancellationToken); - - return new QueryResult>(results, stats.ToQueryStatsInfo()); - } + ) => + WithQueryTimeout(async token => + { + using var session = await sessionProvider.OpenSession(cancellationToken: token); + var query = session.Query() + .Statistics(out var stats) + .Search(x => x.Query, searchTerms) + .FilterBySentTimeRange(timeSentRange) + .Sort(sortInfo) + .Paging(pagingInfo) + .OfType() + .TransformToMessageView(); + + var results = await query.ToListAsync(token); + + return new QueryResult>(results, stats.ToQueryStatsInfo()); + }, cancellationToken); + + Task WithQueryTimeout(Func> query, CancellationToken cancellationToken) => + QueryTimeLimit.Run(query, settings.QueryTimeout, PersistenceSettings.QueryTimeoutSettingName, cancellationToken); public async Task MarkAsArchived(string failedMessageId, CancellationToken cancellationToken = default) { diff --git a/src/ServiceControl.Persistence.RavenDB/RavenPersistenceConfiguration.cs b/src/ServiceControl.Persistence.RavenDB/RavenPersistenceConfiguration.cs index 10a68fec56..fe40bb3421 100644 --- a/src/ServiceControl.Persistence.RavenDB/RavenPersistenceConfiguration.cs +++ b/src/ServiceControl.Persistence.RavenDB/RavenPersistenceConfiguration.cs @@ -1,4 +1,4 @@ -namespace ServiceControl.Persistence.RavenDB +namespace ServiceControl.Persistence.RavenDB { using System; using System.IO; @@ -39,6 +39,7 @@ public PersistenceSettings CreateSettings(SettingsRootNamespace settingsRootName EventsRetentionPeriod = SettingsReader.Read(settingsRootNamespace, EventsRetentionPeriodKey, TimeSpan.FromDays(14)), AuditRetentionPeriod = SettingsReader.Read(settingsRootNamespace, AuditRetentionPeriodKey, TimeSpan.Zero), ExternalIntegrationsDispatchingBatchSize = ReadExternalIntegrationsDispatchingBatchSize(settingsRootNamespace), + QueryTimeout = QueryTimeLimit.Read(settingsRootNamespace, LoggerUtil.CreateStaticLogger()), MaintenanceMode = SettingsReader.Read(settingsRootNamespace, MaintenanceModeKey, false), LogPath = SettingsReader.Read(settingsRootNamespace, RavenBootstrapper.LogsPathKey, DefaultLogLocation()), LogsMode = logsMode, diff --git a/src/ServiceControl.Persistence.RavenDB/RavenPersisterSettings.cs b/src/ServiceControl.Persistence.RavenDB/RavenPersisterSettings.cs index 81958f5bb4..0f59fd9554 100644 --- a/src/ServiceControl.Persistence.RavenDB/RavenPersisterSettings.cs +++ b/src/ServiceControl.Persistence.RavenDB/RavenPersisterSettings.cs @@ -1,4 +1,4 @@ -using System; +using System; using Particular.LicensingComponent.Contracts; using ServiceControl.Persistence; using ServiceControl.Persistence.RavenDB.CustomChecks; diff --git a/src/ServiceControl.Persistence.Tests.PostgreSql/PersistenceTestsContext.cs b/src/ServiceControl.Persistence.Tests.PostgreSql/PersistenceTestsContext.cs index d05de5de27..82b256d0f3 100644 --- a/src/ServiceControl.Persistence.Tests.PostgreSql/PersistenceTestsContext.cs +++ b/src/ServiceControl.Persistence.Tests.PostgreSql/PersistenceTestsContext.cs @@ -9,6 +9,7 @@ namespace ServiceControl.Persistence.Tests; using EFCore.PostgreSql; using MessageFailures; using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Diagnostics; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; using Npgsql; @@ -82,6 +83,17 @@ public async Task CompleteDatabaseOperation() public PersistenceSettings PersistenceSettings { get; set; } + /// + /// Adds an EF Core interceptor to the persistence's own DbContext registration. + /// + public void InterceptDatabaseCommands(IServiceCollection services, IInterceptor interceptor) => + services.ConfigureDbContext(options => options.AddInterceptors(interceptor)); + + /// + /// SQL that makes the server sleep, to prepend to a query that has to still be running when a deadline fires. + /// + public string SqlToDelayFor(TimeSpan delay) => $"SELECT pg_sleep({delay.TotalSeconds:0});"; + public string GenerateFailedMessageRecordId(string messageId) => messageId; public Task InsertFailedMessages(params FailedMessage[] messages) => InsertFailedMessagesDirect(host.Services, messages); diff --git a/src/ServiceControl.Persistence.Tests.RavenDB/QueryTimeoutTests.cs b/src/ServiceControl.Persistence.Tests.RavenDB/QueryTimeoutTests.cs new file mode 100644 index 0000000000..1070051f39 --- /dev/null +++ b/src/ServiceControl.Persistence.Tests.RavenDB/QueryTimeoutTests.cs @@ -0,0 +1,55 @@ +namespace ServiceControl.Persistence.Tests; + +using System; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Extensions.Logging.Abstractions; +using NUnit.Framework; +using Raven.Client.Documents.Session; +using ServiceControl.Persistence.Infrastructure; +using ServiceControl.Persistence.RavenDB; + +class QueryTimeoutTests +{ + [Test] + public void Should_cancel_query_that_exceeds_the_allowed_query_time() + { + var dataStore = BuildDataStore(queryTimeout: TimeSpan.FromMilliseconds(50)); + + var exception = Assert.ThrowsAsync(() => dataStore.GetAllMessagesForSearch("search", new PagingInfo(), new SortInfo("time_sent", "desc"), timeSentRange: null)); + + Assert.That(exception.InnerException, Is.InstanceOf()); + } + + [Test] + public void Should_propagate_caller_cancellation_instead_of_timing_out() + { + var dataStore = BuildDataStore(queryTimeout: TimeSpan.FromSeconds(30)); + + using var callerTokenSource = new CancellationTokenSource(TimeSpan.FromMilliseconds(50)); + + var exception = Assert.CatchAsync(() => dataStore.GetAllMessagesForSearch("search", new PagingInfo(), new SortInfo("time_sent", "desc"), timeSentRange: null, cancellationToken: callerTokenSource.Token)); + + Assert.That(exception, Is.InstanceOf()); + } + + static ErrorMessagesDataStore BuildDataStore(TimeSpan queryTimeout) + { + var settings = new RavenPersisterSettings + { + ErrorRetentionPeriod = TimeSpan.FromDays(10), + QueryTimeout = queryTimeout + }; + + return new ErrorMessagesDataStore(new NeverCompletingSessionProvider(), null, new ExpirationManager(settings), settings, NullLogger.Instance); + } + + class NeverCompletingSessionProvider : IRavenSessionProvider + { + public async ValueTask OpenSession(SessionOptions options = default, CancellationToken cancellationToken = default) + { + await Task.Delay(Timeout.Infinite, cancellationToken); + return null; + } + } +} diff --git a/src/ServiceControl.Persistence.Tests.SqlServer/PersistenceTestsContext.cs b/src/ServiceControl.Persistence.Tests.SqlServer/PersistenceTestsContext.cs index 9077dd38dd..76dfa275a2 100644 --- a/src/ServiceControl.Persistence.Tests.SqlServer/PersistenceTestsContext.cs +++ b/src/ServiceControl.Persistence.Tests.SqlServer/PersistenceTestsContext.cs @@ -9,6 +9,7 @@ namespace ServiceControl.Persistence.Tests; using MessageFailures; using Microsoft.Data.SqlClient; using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Diagnostics; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; using ServiceControl.Persistence.EFCore.Abstractions; @@ -87,6 +88,17 @@ public async Task CompleteDatabaseOperation() public PersistenceSettings PersistenceSettings { get; set; } + /// + /// Adds an EF Core interceptor to the persistence's own DbContext registration. + /// + public void InterceptDatabaseCommands(IServiceCollection services, IInterceptor interceptor) => + services.ConfigureDbContext(options => options.AddInterceptors(interceptor)); + + /// + /// SQL that makes the server sleep, to prepend to a query that has to still be running when a deadline fires. + /// + public string SqlToDelayFor(TimeSpan delay) => $"WAITFOR DELAY '{delay:hh\\:mm\\:ss}';"; + public string GenerateFailedMessageRecordId(string messageId) => messageId; public Task InsertFailedMessages(params FailedMessage[] messages) => InsertFailedMessagesDirect(host.Services, messages); diff --git a/src/ServiceControl.Persistence.Tests/EFCore/MessagesViewQueryTimeoutTests.cs b/src/ServiceControl.Persistence.Tests/EFCore/MessagesViewQueryTimeoutTests.cs new file mode 100644 index 0000000000..69a4924208 --- /dev/null +++ b/src/ServiceControl.Persistence.Tests/EFCore/MessagesViewQueryTimeoutTests.cs @@ -0,0 +1,89 @@ +namespace ServiceControl.Persistence.Tests; + +using System; +using System.Data.Common; +using System.Diagnostics; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.EntityFrameworkCore.Diagnostics; +using Microsoft.Extensions.DependencyInjection; +using NUnit.Framework; +using ServiceControl.CompositeViews.Messages; +using ServiceControl.Persistence.EFCore.Abstractions; +using ServiceControl.Persistence.Infrastructure; + +// Runs against the real provider: the way a cancelled command surfaces differs per ADO.NET provider +// (Microsoft.Data.SqlClient raises SqlException, Npgsql raises OperationCanceledException), which a stubbed +// session cannot show. +[TestFixture] +class MessagesViewQueryTimeoutTests : PersistenceTestBase +{ + readonly SlowCommands slowCommands = new(); + + public MessagesViewQueryTimeoutTests() => + RegisterServices = services => PersistenceTestsContext.InterceptDatabaseCommands(services, slowCommands); + + [Test] + public void A_search_over_the_query_time_limit_is_cancelled_and_reported_as_a_timeout_naming_the_setting() + { + Settings.QueryTimeout = TimeSpan.FromSeconds(1); + slowCommands.DelaySql = PersistenceTestsContext.SqlToDelayFor(TimeSpan.FromSeconds(20)); + + var stopwatch = Stopwatch.StartNew(); + var exception = Assert.ThrowsAsync(() => Search()); + stopwatch.Stop(); + + Assert.That(stopwatch.Elapsed, Is.LessThan(TimeSpan.FromSeconds(10)), "the query must be cancelled, not waited out"); + Assert.That(exception.Message, Does.Contain(PersistenceSettings.QueryTimeoutSettingName)); + Assert.That(slowCommands.CommandsSeen, Is.GreaterThan(0), "the interceptor did not see the command, so the query was not slowed down"); + } + + [Test] + public async Task A_search_within_the_query_time_limit_returns_its_result() + { + Settings.QueryTimeout = TimeSpan.FromSeconds(30); + + var result = await Search(); + + Assert.That(result.Results, Is.Empty); + } + + [Test] + public async Task The_per_command_timeout_of_a_search_is_the_query_time_limit() + { + // The provider's own command timeout (Database/CommandTimeout, default 30s) must not undercut the query + // time limit, or raising QueryTimeoutInSeconds would have no effect and the error would not name it. + Settings.QueryTimeout = TimeSpan.FromSeconds(123); + + await Search(); + + Assert.That(slowCommands.CommandTimeoutSeen, Is.EqualTo(123)); + } + + EFPersisterSettings Settings => (EFPersisterSettings)PersistenceSettings; + + Task>> Search() => + ServiceProvider.GetRequiredService() + .GetAllMessagesForSearch("anything", new PagingInfo(), new SortInfo("time_sent", "desc")); + + // Prepends a server-side sleep to every query so the command is still running when the deadline fires. + class SlowCommands : DbCommandInterceptor + { + public string DelaySql { get; set; } + public int CommandsSeen { get; private set; } + public int? CommandTimeoutSeen { get; private set; } + + public override ValueTask> ReaderExecutingAsync(DbCommand command, CommandEventData eventData, InterceptionResult result, CancellationToken cancellationToken = default) + { + CommandsSeen++; + CommandTimeoutSeen = command.CommandTimeout; + + if (DelaySql != null) + { + command.CommandText = DelaySql + Environment.NewLine + command.CommandText; + } + + return base.ReaderExecutingAsync(command, eventData, result, cancellationToken); + } + } +} diff --git a/src/ServiceControl.Persistence.Tests/EFCore/QueryTimeoutConfigurationTests.cs b/src/ServiceControl.Persistence.Tests/EFCore/QueryTimeoutConfigurationTests.cs new file mode 100644 index 0000000000..e9f6697274 --- /dev/null +++ b/src/ServiceControl.Persistence.Tests/EFCore/QueryTimeoutConfigurationTests.cs @@ -0,0 +1,89 @@ +namespace ServiceControl.Persistence.Tests; + +using System; +using NUnit.Framework; +using ServiceControl.Configuration; +using ServiceControl.Persistence.EFCore.Abstractions; + +// Covers the settings-reader validation for the query time limit applied to message view queries. +[TestFixture] +[NonParallelizable] +class QueryTimeoutConfigurationTests +{ + static readonly SettingsRootNamespace TestNamespace = new("ServiceControl"); + + const string QueryTimeoutVariable = "SERVICECONTROL_QUERYTIMEOUTINSECONDS"; + + static readonly string[] Keys = + [ + QueryTimeoutVariable, + "SERVICECONTROL_DATABASE_CONNECTIONSTRING", + "SERVICECONTROL_ERRORRETENTIONPERIOD", + "SERVICECONTROL_MESSAGEBODY_STORAGETYPE", + "SERVICECONTROL_MESSAGEBODY_FILESYSTEM_STORAGEPATH" + ]; + + [SetUp] + public void SetUp() + { + ClearKeys(); + Environment.SetEnvironmentVariable("SERVICECONTROL_DATABASE_CONNECTIONSTRING", "Server=nowhere"); + Environment.SetEnvironmentVariable("SERVICECONTROL_ERRORRETENTIONPERIOD", "10.00:00:00"); + Environment.SetEnvironmentVariable("SERVICECONTROL_MESSAGEBODY_STORAGETYPE", "FileSystem"); + Environment.SetEnvironmentVariable("SERVICECONTROL_MESSAGEBODY_FILESYSTEM_STORAGEPATH", "/var/bodies"); + } + + [TearDown] + public void TearDown() => ClearKeys(); + + static void ClearKeys() + { + foreach (var key in Keys) + { + Environment.SetEnvironmentVariable(key, null); + } + } + + [Test] + public void Defaults_to_one_minute() + { + var settings = CreateSettings(); + + Assert.That(settings.QueryTimeout, Is.EqualTo(TimeSpan.FromMinutes(1))); + } + + [Test] + public void Reads_the_configured_value() + { + Environment.SetEnvironmentVariable(QueryTimeoutVariable, "120"); + + var settings = CreateSettings(); + + Assert.That(settings.QueryTimeout, Is.EqualTo(TimeSpan.FromSeconds(120))); + } + + [TestCase("0")] + [TestCase("-5")] + [TestCase("3700")] + public void Falls_back_to_the_default_for_values_outside_the_allowed_range(string value) + { + Environment.SetEnvironmentVariable(QueryTimeoutVariable, value); + + var settings = CreateSettings(); + + Assert.That(settings.QueryTimeout, Is.EqualTo(TimeSpan.FromMinutes(1))); + } + + static EFPersisterSettings CreateSettings() => + (EFPersisterSettings)new TestPersistenceConfiguration().CreateSettings(TestNamespace); + + sealed class TestPersistenceConfiguration : EFPersistenceConfigurationBase + { + public override IPersistence Create(PersistenceSettings settings) => throw new NotSupportedException(); + + protected override EFPersisterSettings CreateSettings(string connectionString, BodyStorageSettings bodyStorage) => + new TestPersisterSettings { ConnectionString = connectionString, BodyStorage = bodyStorage }; + } + + sealed class TestPersisterSettings : EFPersisterSettings; +} diff --git a/src/ServiceControl.Persistence/PersistenceSettings.cs b/src/ServiceControl.Persistence/PersistenceSettings.cs index a11b359c22..fb7c3b5ff7 100644 --- a/src/ServiceControl.Persistence/PersistenceSettings.cs +++ b/src/ServiceControl.Persistence/PersistenceSettings.cs @@ -1,6 +1,7 @@ -namespace ServiceControl.Persistence +namespace ServiceControl.Persistence { using System; + using ServiceControl.Infrastructure; /// /// Base settings that apply across all Persisters @@ -20,5 +21,15 @@ public abstract class PersistenceSettings public bool EnableFullTextSearchOnBodies { get; set; } = true; public TimeSpan? OverrideCustomCheckRepeatTime { get; set; } + + /// + /// Wall-clock limit for the message view queries, see . + /// + public TimeSpan QueryTimeout { get; set; } = QueryTimeLimit.Default; + + /// + /// The setting is read from, as named in the timeout error. + /// + public const string QueryTimeoutSettingName = "ServiceControl/" + QueryTimeLimit.SettingName; } } \ No newline at end of file From ec9cf12d323c6c8be46cb33943ce9b6d526f6e20 Mon Sep 17 00:00:00 2001 From: Ramon Smits Date: Fri, 4 Sep 2026 17:00:11 +0200 Subject: [PATCH 2/2] =?UTF-8?q?=E2=9C=A8=20Answer=20a=20timed-out=20query?= =?UTF-8?q?=20with=20504=20and=20keep=20the=20composite's=20partial=20data?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A data store's TimeoutException now becomes a 504 Gateway Timeout with a problem body naming the setting, on both the primary and the audit host, so a caller can tell a timeout from a crash and from an empty result. The primary's scatter-gather no longer hides what went wrong. A local timeout is absorbed the way a remote failure already was, a remote 504 or error is a missing instance rather than an instance with no data, and the response names every missing instance in the X-Particular-Incomplete-Results header (instanceId:timeout|unavailable| error) and carries no ETag. Only when no instance that was asked answered and one of them timed out does the composite itself fail with the timeout. The signal is a header rather than a body field because the composite endpoints return a bare array; an envelope would change the response schema and need a new API for every existing client. The in-process audit counts used for licensing throughput fail instead of recording a partial sum as the day's throughput. The remote instance HttpClient timeout follows ServiceControl/QueryTimeoutInSeconds plus a 30 second margin instead of the 100 second HttpClient default, so raising the limit on both instances does not make the primary give up on the audit instance first. --- .../WebApplicationExtensions.cs | 2 + .../QueryTimeout/QueryTimeoutResponse.cs | 44 ++++ .../QueryTimeoutResponseTests.cs | 57 +++++ .../Infrastructure/QueryResult.cs | 31 +++ .../ScatterGather/IncompleteResultsTests.cs | 232 ++++++++++++++++++ .../RemoteInstanceHttpClientTests.cs | 31 +++ .../Messages/GetMessages2Controller.cs | 1 + .../GetMessagesByConversationController.cs | 4 +- .../Messages/GetMessagesController.cs | 14 +- .../Messages/ScatterGatherApi.cs | 87 ++++++- .../Messages/ScatterGatherRemoteOnly.cs | 2 + .../Infrastructure/Api/AuditCountApi.cs | 26 +- .../Infrastructure/WebApi/Cors.cs | 4 +- .../WebApi/HttpResponseExtensions.cs | 36 +++ ...moteInstanceServiceCollectionExtensions.cs | 10 +- .../SagaAudit/SagasController.cs | 2 +- .../WebApplicationExtensions.cs | 2 + 17 files changed, 559 insertions(+), 26 deletions(-) create mode 100644 src/ServiceControl.Hosting/QueryTimeout/QueryTimeoutResponse.cs create mode 100644 src/ServiceControl.Infrastructure.Tests/QueryTimeoutResponseTests.cs create mode 100644 src/ServiceControl.UnitTests/ScatterGather/IncompleteResultsTests.cs create mode 100644 src/ServiceControl.UnitTests/ScatterGather/RemoteInstanceHttpClientTests.cs diff --git a/src/ServiceControl.Audit/WebApplicationExtensions.cs b/src/ServiceControl.Audit/WebApplicationExtensions.cs index 1a3ec118ff..1f7adb9f99 100644 --- a/src/ServiceControl.Audit/WebApplicationExtensions.cs +++ b/src/ServiceControl.Audit/WebApplicationExtensions.cs @@ -4,6 +4,7 @@ namespace ServiceControl.Audit; using Microsoft.AspNetCore.Builder; using ServiceControl.Hosting.ForwardedHeaders; using ServiceControl.Hosting.Https; +using ServiceControl.Hosting.QueryTimeout; using ServiceControl.Hosting.RequestId; using ServiceControl.Infrastructure; @@ -12,6 +13,7 @@ public static class WebApplicationExtensions public static void UseServiceControlAudit(this WebApplication app, ForwardedHeadersSettings forwardedHeadersSettings, HttpsSettings httpsSettings) { app.UseRequestIdHeader(); + app.UseQueryTimeoutResponse(); app.UseServiceControlForwardedHeaders(forwardedHeadersSettings); app.UseServiceControlHttps(httpsSettings); app.UseResponseCompression(); diff --git a/src/ServiceControl.Hosting/QueryTimeout/QueryTimeoutResponse.cs b/src/ServiceControl.Hosting/QueryTimeout/QueryTimeoutResponse.cs new file mode 100644 index 0000000000..5e28116fd9 --- /dev/null +++ b/src/ServiceControl.Hosting/QueryTimeout/QueryTimeoutResponse.cs @@ -0,0 +1,44 @@ +#nullable enable +namespace ServiceControl.Hosting.QueryTimeout; + +using System; +using System.Threading.Tasks; +using Microsoft.AspNetCore.Builder; +using Microsoft.AspNetCore.Http; +using Microsoft.AspNetCore.Mvc; +using Microsoft.Extensions.Logging; +using ServiceControl.Infrastructure; + +/// +/// Answers a request whose database query ran out of its allowed query time with 504 Gateway Timeout and a +/// problem body that names the setting, so a caller can tell a timeout from a crash and from an empty result. +/// The data stores raise for it, see . +/// +public static class QueryTimeoutResponse +{ + static readonly ILogger logger = LoggerUtil.CreateStaticLogger(typeof(QueryTimeoutResponse)); + + public static void UseQueryTimeoutResponse(this WebApplication app) => app.Use(Wrap); + + public static RequestDelegate Wrap(RequestDelegate next) => async context => + { + try + { + await next(context).ConfigureAwait(false); + } + catch (TimeoutException e) when (!context.Response.HasStarted) + { + logger.LogWarning(e, "The query behind {Method} {Path} did not complete within its allowed query time", context.Request.Method, context.Request.Path); + + context.Response.StatusCode = StatusCodes.Status504GatewayTimeout; + // Deliberately no "type" URI (RFC 9457 then treats it as about:blank): the body stays purely + // descriptive rather than linking to documentation, which only search or permalink URLs may do. + await context.Response.WriteAsJsonAsync(new ProblemDetails + { + Status = StatusCodes.Status504GatewayTimeout, + Title = "The query did not complete within the allowed query time", + Detail = e.Message + }, options: null, contentType: "application/problem+json", context.RequestAborted).ConfigureAwait(false); + } + }; +} diff --git a/src/ServiceControl.Infrastructure.Tests/QueryTimeoutResponseTests.cs b/src/ServiceControl.Infrastructure.Tests/QueryTimeoutResponseTests.cs new file mode 100644 index 0000000000..16edafacec --- /dev/null +++ b/src/ServiceControl.Infrastructure.Tests/QueryTimeoutResponseTests.cs @@ -0,0 +1,57 @@ +namespace ServiceControl.Infrastructure.Tests; + +using System; +using System.IO; +using System.Text.Json; +using System.Threading.Tasks; +using Microsoft.AspNetCore.Http; +using Microsoft.AspNetCore.Mvc; +using NUnit.Framework; +using ServiceControl.Hosting.QueryTimeout; + +[TestFixture] +public class QueryTimeoutResponseTests +{ + [Test] + public async Task A_query_timeout_becomes_a_504_problem_that_names_the_setting() + { + var context = new DefaultHttpContext(); + context.Response.Body = new MemoryStream(); + + var pipeline = QueryTimeoutResponse.Wrap(_ => throw new TimeoutException("The query did not complete within 60 seconds. The 'ServiceControl/QueryTimeoutInSeconds' setting can be used to change it.")); + + await pipeline(context); + + Assert.That(context.Response.StatusCode, Is.EqualTo(StatusCodes.Status504GatewayTimeout)); + Assert.That(context.Response.ContentType, Does.StartWith("application/problem+json")); + + context.Response.Body.Position = 0; + var problem = await JsonSerializer.DeserializeAsync(context.Response.Body); + Assert.That(problem.Status, Is.EqualTo(504)); + Assert.That(problem.Detail, Does.Contain("ServiceControl/QueryTimeoutInSeconds")); + } + + [Test] + public async Task A_response_that_completes_is_passed_through() + { + var context = new DefaultHttpContext(); + + var pipeline = QueryTimeoutResponse.Wrap(httpContext => + { + httpContext.Response.StatusCode = StatusCodes.Status204NoContent; + return Task.CompletedTask; + }); + + await pipeline(context); + + Assert.That(context.Response.StatusCode, Is.EqualTo(StatusCodes.Status204NoContent)); + } + + [Test] + public void Other_failures_are_not_its_business() + { + var pipeline = QueryTimeoutResponse.Wrap(_ => throw new InvalidOperationException("boom")); + + Assert.ThrowsAsync(() => pipeline(new DefaultHttpContext())); + } +} diff --git a/src/ServiceControl.Persistence/Infrastructure/QueryResult.cs b/src/ServiceControl.Persistence/Infrastructure/QueryResult.cs index 519217f1c5..34a0fac3d4 100644 --- a/src/ServiceControl.Persistence/Infrastructure/QueryResult.cs +++ b/src/ServiceControl.Persistence/Infrastructure/QueryResult.cs @@ -1,7 +1,26 @@ namespace ServiceControl.Persistence.Infrastructure { + using System.Collections.Generic; using System.Threading.Tasks; + /// + /// Why an instance contributed nothing to a query. + /// + public enum QueryFailure + { + /// The query ran out of its allowed query time. + TimedOut, + /// The instance could not be reached. + Unavailable, + /// The instance answered with an error. + Failed + } + + /// + /// An instance whose data a composite result is missing. + /// + public sealed record IncompleteInstance(string InstanceId, QueryFailure Reason); + public class QueryResult(TOut? results, QueryStatsInfo queryStatsInfo) where TOut : class { @@ -16,8 +35,20 @@ public class QueryResult(TOut? results, QueryStatsInfo queryStatsInfo) public QueryStatsInfo QueryStats { get; } = queryStatsInfo; + /// + /// Why this instance contributed nothing. Null when it answered, also when it answered with no data. + /// + public QueryFailure? Failure { get; init; } + + /// + /// The instances a composite result is missing. Empty when every instance answered. + /// + public IReadOnlyList IncompleteInstances { get; init; } = []; + public static QueryResult Empty() => new(null, QueryStatsInfo.Zero); + public static QueryResult Failed(QueryFailure reason) => new(null, QueryStatsInfo.Zero) { Failure = reason }; + public static implicit operator Task>(QueryResult instance) => Task.FromResult(instance); } } \ No newline at end of file diff --git a/src/ServiceControl.UnitTests/ScatterGather/IncompleteResultsTests.cs b/src/ServiceControl.UnitTests/ScatterGather/IncompleteResultsTests.cs new file mode 100644 index 0000000000..84bfe00c10 --- /dev/null +++ b/src/ServiceControl.UnitTests/ScatterGather/IncompleteResultsTests.cs @@ -0,0 +1,232 @@ +#pragma warning disable PS0003 // Make the CancellationToken parameter optional — HttpMessageHandler.SendAsync override signature is fixed + +namespace ServiceControl.UnitTests.ScatterGather; + +using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.Linq; +using System.Net; +using System.Net.Http; +using System.Threading; +using System.Threading.Tasks; +using CompositeViews.MessageCounting; +using CompositeViews.Messages; +using Microsoft.AspNetCore.Http; +using Microsoft.Extensions.Logging.Abstractions; +using NUnit.Framework; +using ServiceBus.Management.Infrastructure.Settings; +using ServiceControl.Api.Contracts; +using ServiceControl.Infrastructure.Api; +using ServiceControl.Infrastructure.WebApi; +using ServiceControl.Persistence.Infrastructure; + +/// +/// A timed-out or failed instance is not an instance with no data: the composite keeps what the others +/// returned and says which instances are missing, and only gives up when nothing answered. +/// +[TestFixture] +class IncompleteResultsTests +{ + const string RemoteAddress = "http://audit-1/api"; + const string OtherRemoteAddress = "http://audit-2/api"; + + [Test] + public async Task A_local_query_timeout_keeps_the_remote_data_and_reports_the_local_instance() + { + var settings = Settings(RemoteAddress); + var factory = new FakeHttpClientFactory(); + factory.Register(settings.RemoteInstances[0], Healthy("remote-msg")); + + var api = new TestApi(settings, factory, _ => throw new TimeoutException("query time limit")); + + var result = await api.Execute(Context(), "/api/messages"); + + Assert.That(result.Results.Select(m => m.MessageId), Is.EqualTo(["remote-msg"])); + Assert.That(result.IncompleteInstances, Is.EqualTo([new IncompleteInstance(settings.InstanceId, QueryFailure.TimedOut)])); + } + + [Test] + public async Task A_remote_query_timeout_keeps_the_local_data_and_reports_the_remote_instance() + { + var settings = Settings(RemoteAddress); + var factory = new FakeHttpClientFactory(); + factory.Register(settings.RemoteInstances[0], Status(HttpStatusCode.GatewayTimeout)); + + var api = new TestApi(settings, factory, Local("local-msg")); + + var result = await api.Execute(Context(), "/api/messages"); + + Assert.That(result.Results.Select(m => m.MessageId), Is.EqualTo(["local-msg"])); + Assert.That(result.IncompleteInstances, Is.EqualTo([new IncompleteInstance(settings.RemoteInstances[0].InstanceId, QueryFailure.TimedOut)])); + Assert.That(settings.RemoteInstances[0].TemporarilyUnavailable, Is.False, "a remote whose query timed out is up; it must not be skipped on the next query"); + } + + [Test] + public async Task A_remote_error_is_reported_as_a_failure_not_as_no_data() + { + var settings = Settings(RemoteAddress); + var factory = new FakeHttpClientFactory(); + factory.Register(settings.RemoteInstances[0], Status(HttpStatusCode.InternalServerError)); + + var api = new TestApi(settings, factory, Local("local-msg")); + + var result = await api.Execute(Context(), "/api/messages"); + + Assert.That(result.IncompleteInstances, Is.EqualTo([new IncompleteInstance(settings.RemoteInstances[0].InstanceId, QueryFailure.Failed)])); + } + + [Test] + public async Task A_temporarily_unavailable_remote_is_reported_as_missing() + { + var settings = Settings(RemoteAddress); + settings.RemoteInstances[0].TemporarilyUnavailable = true; + + var api = new TestApi(settings, new FakeHttpClientFactory(), Local("local-msg")); + + var result = await api.Execute(Context(), "/api/messages"); + + Assert.That(result.Results.Select(m => m.MessageId), Is.EqualTo(["local-msg"])); + Assert.That(result.IncompleteInstances, Is.EqualTo([new IncompleteInstance(settings.RemoteInstances[0].InstanceId, QueryFailure.Unavailable)])); + } + + [Test] + public async Task An_incomplete_result_carries_no_version() + { + var settings = Settings(RemoteAddress); + var factory = new FakeHttpClientFactory(); + factory.Register(settings.RemoteInstances[0], Status(HttpStatusCode.GatewayTimeout)); + + var result = await new TestApi(settings, factory, Local("local-msg")).Execute(Context(), "/api/messages"); + + Assert.That(result.QueryStats.Version.HasValue, Is.False, "a client must not cache an incomplete page as if it were the whole answer"); + } + + [Test] + public async Task A_complete_result_reports_nothing_missing() + { + var settings = Settings(RemoteAddress); + var factory = new FakeHttpClientFactory(); + factory.Register(settings.RemoteInstances[0], Healthy("remote-msg")); + + var result = await new TestApi(settings, factory, Local("local-msg")).Execute(Context(), "/api/messages"); + + Assert.That(result.IncompleteInstances, Is.Empty); + Assert.That(result.Results, Has.Count.EqualTo(2)); + } + + [Test] + public void When_no_instance_answered_and_one_timed_out_the_query_is_a_timeout() + { + var settings = Settings(RemoteAddress); + var factory = new FakeHttpClientFactory(); + factory.Register(settings.RemoteInstances[0], Status(HttpStatusCode.GatewayTimeout)); + + var api = new TestApi(settings, factory, _ => throw new TimeoutException("query time limit")); + + var exception = Assert.ThrowsAsync(() => api.Execute(Context(), "/api/messages")); + + Assert.That(exception.Message, Does.Contain(settings.RemoteInstances[0].InstanceId)); + } + + [Test] + public void A_remote_only_query_whose_only_remote_timed_out_is_a_timeout() + { + var settings = Settings(RemoteAddress); + var factory = new FakeHttpClientFactory(); + factory.Register(settings.RemoteInstances[0], Status(HttpStatusCode.GatewayTimeout)); + + var api = new GetAuditCountsForEndpointApi(settings, factory, new HttpContextAccessor(), NullLogger.Instance); + + Assert.ThrowsAsync(() => api.Execute(new AuditCountsForEndpointContext(new PagingInfo(), "Sales"), "/api/endpoints/Sales/audit-count")); + } + + [Test] + public void Audit_counts_missing_an_instance_are_not_recorded_as_the_endpoint_throughput() + { + var settings = Settings(RemoteAddress, OtherRemoteAddress); + var factory = new FakeHttpClientFactory(); + factory.Register(settings.RemoteInstances[0], Json>([new AuditCount { UtcDate = DateTime.UtcNow.Date, Count = 5 }])); + factory.Register(settings.RemoteInstances[1], Status(HttpStatusCode.GatewayTimeout)); + + var auditCountApi = new AuditCountApi(new GetAuditCountsForEndpointApi(settings, factory, new HttpContextAccessor(), NullLogger.Instance)); + + var exception = Assert.CatchAsync(() => auditCountApi.GetEndpointAuditCounts("Sales")); + + Assert.That(exception.Message, Does.Contain(settings.RemoteInstances[1].InstanceId), "a partial sum recorded as the day's throughput would under-count the license usage for good"); + } + + [Test] + public void The_response_names_the_missing_instances_in_a_header() + { + var context = new DefaultHttpContext(); + var result = new QueryResult>([], QueryStatsInfo.Zero) + { + IncompleteInstances = [new IncompleteInstance("audit-1", QueryFailure.TimedOut), new IncompleteInstance("audit-2", QueryFailure.Unavailable)] + }; + + context.Response.WithScatterGatherResult(result, new PagingInfo()); + + Assert.That(context.Response.Headers[HttpResponseExtensions.IncompleteResultsHeader].ToString(), Is.EqualTo("audit-1:timeout, audit-2:unavailable")); + } + + [Test] + public void A_complete_response_has_no_incomplete_results_header() + { + var context = new DefaultHttpContext(); + + context.Response.WithScatterGatherResult(new QueryResult>([], QueryStatsInfo.Zero), new PagingInfo()); + + Assert.That(context.Response.Headers.ContainsKey(HttpResponseExtensions.IncompleteResultsHeader), Is.False); + } + + static Settings Settings(params string[] remotes) => new() + { + RemoteInstances = remotes.Select(address => new RemoteInstanceSetting(address)).ToArray() + }; + + static ScatterGatherApiMessageViewContext Context() => new(new PagingInfo(), new SortInfo("time_sent", "desc")); + + static Func>>> Local(string messageId) => + _ => Task.FromResult(new QueryResult>([new MessagesView { MessageId = messageId }], new QueryStatsInfo(DataVersion.FromToken("local-etag"), 1))); + + static Func> Healthy(string messageId) => + Json>([new MessagesView { MessageId = messageId }]); + + static Func> Json(T body) => + (_, _) => + { + var response = new HttpResponseMessage(HttpStatusCode.OK) { Content = new StringContent(System.Text.Json.JsonSerializer.Serialize(body, SerializerOptions.Default)) }; + response.Headers.TryAddWithoutValidation("Total-Count", "1"); + response.Headers.TryAddWithoutValidation("ETag", "\"remote-etag\""); + return Task.FromResult(response); + }; + + static Func> Status(HttpStatusCode statusCode) => + (_, _) => Task.FromResult(new HttpResponseMessage(statusCode)); + + class TestApi(Settings settings, IHttpClientFactory factory, Func>>> local) + : ScatterGatherApiMessageView(new object(), settings, factory, new HttpContextAccessor(), NullLogger.Instance) + { + protected override Task>> LocalQuery(ScatterGatherApiMessageViewContext input, CancellationToken cancellationToken = default) => local(cancellationToken); + } + + class FakeHttpClientFactory : IHttpClientFactory + { + readonly ConcurrentDictionary handlers = new(); + + public void Register(RemoteInstanceSetting remote, Func> responder) => + handlers[remote.InstanceId] = (new StubHandler(responder), remote.BaseAddress); + + public HttpClient CreateClient(string name) + { + var (handler, baseAddress) = handlers[name]; + return new HttpClient(handler, disposeHandler: false) { BaseAddress = new Uri(baseAddress) }; + } + } + + class StubHandler(Func> responder) : HttpMessageHandler + { + protected override Task SendAsync(HttpRequestMessage request, CancellationToken cancellationToken) => responder(request, cancellationToken); + } +} diff --git a/src/ServiceControl.UnitTests/ScatterGather/RemoteInstanceHttpClientTests.cs b/src/ServiceControl.UnitTests/ScatterGather/RemoteInstanceHttpClientTests.cs new file mode 100644 index 0000000000..aab6f84e60 --- /dev/null +++ b/src/ServiceControl.UnitTests/ScatterGather/RemoteInstanceHttpClientTests.cs @@ -0,0 +1,31 @@ +namespace ServiceControl.UnitTests.ScatterGather; + +using System; +using System.Net.Http; +using Microsoft.Extensions.DependencyInjection; +using NUnit.Framework; +using ServiceBus.Management.Infrastructure.Settings; +using ServiceControl.Infrastructure.WebApi; +using ServiceControl.Persistence; + +[TestFixture] +class RemoteInstanceHttpClientTests +{ + [Test] + public void The_remote_client_waits_out_the_query_time_limit_plus_a_margin_for_the_answer() + { + // A remote's query is allowed the same query time as ours, and its 504 still has to travel back. + var settings = new Settings { RemoteInstances = [new RemoteInstanceSetting("http://audit/api")] }; + + var services = new ServiceCollection(); + services.AddSingleton(new TestPersistenceSettings { QueryTimeout = TimeSpan.FromMinutes(5) }); + services.AddRemoteInstancesHttpClients(settings); + using var provider = services.BuildServiceProvider(); + + var client = provider.GetRequiredService().CreateClient(settings.RemoteInstances[0].InstanceId); + + Assert.That(client.Timeout, Is.EqualTo(TimeSpan.FromMinutes(5) + RemoteInstanceServiceCollectionExtensions.ResponseMargin)); + } + + class TestPersistenceSettings : PersistenceSettings; +} diff --git a/src/ServiceControl/CompositeViews/Messages/GetMessages2Controller.cs b/src/ServiceControl/CompositeViews/Messages/GetMessages2Controller.cs index 6c1af26c31..6c6faac067 100644 --- a/src/ServiceControl/CompositeViews/Messages/GetMessages2Controller.cs +++ b/src/ServiceControl/CompositeViews/Messages/GetMessages2Controller.cs @@ -67,6 +67,7 @@ public async Task> Messages( Response.WithTotalCount(result.QueryStats.TotalCount); Response.WithEtag(result.QueryStats.Version); + Response.WithIncompleteResults(result.IncompleteInstances); return result.Results; } diff --git a/src/ServiceControl/CompositeViews/Messages/GetMessagesByConversationController.cs b/src/ServiceControl/CompositeViews/Messages/GetMessagesByConversationController.cs index 5b21ae3246..10559fb6a2 100644 --- a/src/ServiceControl/CompositeViews/Messages/GetMessagesByConversationController.cs +++ b/src/ServiceControl/CompositeViews/Messages/GetMessagesByConversationController.cs @@ -1,4 +1,4 @@ -namespace ServiceControl.CompositeViews.Messages +namespace ServiceControl.CompositeViews.Messages { using System.Collections.Generic; using System.Threading; @@ -27,7 +27,7 @@ public async Task> Messages([FromQuery] PagingInfo pagingInf new MessagesByConversationContext(pagingInfo, sortInfo, includeSystemMessages, conversationId), Request.GetEncodedPathAndQuery(), cancellationToken); - Response.WithQueryStatsAndPagingInfo(result.QueryStats, pagingInfo); + Response.WithScatterGatherResult(result, pagingInfo); return result.Results; } } diff --git a/src/ServiceControl/CompositeViews/Messages/GetMessagesController.cs b/src/ServiceControl/CompositeViews/Messages/GetMessagesController.cs index 8469628992..7cc05c8b50 100644 --- a/src/ServiceControl/CompositeViews/Messages/GetMessagesController.cs +++ b/src/ServiceControl/CompositeViews/Messages/GetMessagesController.cs @@ -48,7 +48,7 @@ public async Task> Messages([FromQuery] PagingInfo pagingInf new ScatterGatherApiMessageViewWithSystemMessagesContext(pagingInfo, sortInfo, includeSystemMessages), Request.GetEncodedPathAndQuery(), cancellationToken); - Response.WithQueryStatsAndPagingInfo(result.QueryStats, pagingInfo); + Response.WithScatterGatherResult(result, pagingInfo); return result.Results; } @@ -64,7 +64,7 @@ public async Task> MessagesForEndpoint([FromQuery] PagingInf new AllMessagesForEndpointContext(pagingInfo, sortInfo, includeSystemMessages, endpoint), Request.GetEncodedPathAndQuery(), cancellationToken); - Response.WithQueryStatsAndPagingInfo(result.QueryStats, pagingInfo); + Response.WithScatterGatherResult(result, pagingInfo); return result.Results; } @@ -77,7 +77,7 @@ public async Task> GetEndpointAuditCounts([FromQuery] PagingIn QueryResult> result = await auditCountsForEndpointApi.Execute( new AuditCountsForEndpointContext(pagingInfo, endpoint), Request.GetEncodedPathAndQuery(), cancellationToken); - Response.WithQueryStatsAndPagingInfo(result.QueryStats, pagingInfo); + Response.WithScatterGatherResult(result, pagingInfo); return result.Results; } @@ -130,7 +130,7 @@ public async Task> Search([FromQuery] PagingInfo pagingInfo, QueryResult> result = await api.Execute(new SearchApiContext(pagingInfo, sortInfo, q), Request.GetEncodedPathAndQuery(), cancellationToken); - Response.WithQueryStatsAndPagingInfo(result.QueryStats, pagingInfo); + Response.WithScatterGatherResult(result, pagingInfo); return result.Results; } @@ -144,7 +144,7 @@ public async Task> SearchByKeyWord([FromQuery] PagingInfo pa new SearchApiContext(pagingInfo, sortInfo, keyword?.Replace("/", @"\")), Request.GetEncodedPathAndQuery(), cancellationToken); - Response.WithQueryStatsAndPagingInfo(result.QueryStats, pagingInfo); + Response.WithScatterGatherResult(result, pagingInfo); return result.Results; } @@ -157,7 +157,7 @@ public async Task> Search([FromQuery] PagingInfo pagingInfo, QueryResult> result = await endpointApi.Execute( new SearchEndpointContext(pagingInfo, sortInfo, Keyword: q, Endpoint: endpoint), Request.GetEncodedPathAndQuery(), cancellationToken); - Response.WithQueryStatsAndPagingInfo(result.QueryStats, pagingInfo); + Response.WithScatterGatherResult(result, pagingInfo); return result.Results; } @@ -170,7 +170,7 @@ public async Task> SearchByKeyword([FromQuery] PagingInfo pa QueryResult> result = await endpointApi.Execute( new SearchEndpointContext(pagingInfo, sortInfo, Keyword: keyword, Endpoint: endpoint), Request.GetEncodedPathAndQuery(), cancellationToken); - Response.WithQueryStatsAndPagingInfo(result.QueryStats, pagingInfo); + Response.WithScatterGatherResult(result, pagingInfo); return result.Results; } } diff --git a/src/ServiceControl/CompositeViews/Messages/ScatterGatherApi.cs b/src/ServiceControl/CompositeViews/Messages/ScatterGatherApi.cs index f207315409..1af5732b9f 100644 --- a/src/ServiceControl/CompositeViews/Messages/ScatterGatherApi.cs +++ b/src/ServiceControl/CompositeViews/Messages/ScatterGatherApi.cs @@ -26,6 +26,15 @@ internal static DataVersion ReadEtag(HttpResponseHeaders headers) => headers.TryGetValues("ETag", out var values) ? DataVersion.FromClient(values.FirstOrDefault()) : DataVersion.None; + + internal static string Describe(IEnumerable incomplete) => + string.Join(", ", incomplete.Select(instance => $"{instance.InstanceId} {instance.Reason switch + { + QueryFailure.TimedOut => "timed out", + QueryFailure.Unavailable => "is unavailable", + QueryFailure.Failed => "failed", + _ => "failed" + }}")); } public record ScatterGatherContext(PagingInfo PagingInfo); @@ -58,26 +67,65 @@ public async Task> Execute(TIn input, string pathAndQuery, Can { LocalCall(input, instanceId, cancellationToken) }; + var unavailable = new List>(); foreach (var remote in remotes) { if (remote.TemporarilyUnavailable) { + unavailable.Add(new QueryResult(null, QueryStatsInfo.Zero) { Failure = QueryFailure.Unavailable, InstanceId = remote.InstanceId }); continue; } tasks.Add(RemoteCall(HttpClientFactory.CreateClient(remote.InstanceId), pathAndQuery, remote, authorizationHeader, cancellationToken)); } - var results = await Task.WhenAll(tasks); + // The local result stays first: ProcessResults gives it precedence when de-duplicating. + var results = (await Task.WhenAll(tasks)).Concat(unavailable).ToArray(); var response = AggregateResults(input, results); + ThrowWhenNothingAnswered(results, response.IncompleteInstances); + return response; } + /// + /// A missing instance is reported, not hidden; but when no instance that was asked answered and at least + /// one of them ran out of its query time, there is nothing to report but the timeout. + /// + void ThrowWhenNothingAnswered(QueryResult[] results, IReadOnlyList incomplete) + { + var anyAnswered = results.Any(result => result.Failure is null && (LocalInstanceParticipates || !result.IsLocalInstance)); + + if (anyAnswered || incomplete.All(instance => instance.Reason != QueryFailure.TimedOut)) + { + return; + } + + throw new TimeoutException($"No instance completed the query within its allowed query time. {Describe(incomplete)}"); + } + + /// + /// Whether this instance's own data store is a source for the query. An API that only forwards to the + /// remotes answers "nothing" locally without that meaning anything about the data. + /// + protected virtual bool LocalInstanceParticipates => true; + async Task> LocalCall(TIn input, string instanceId, CancellationToken cancellationToken) { - var result = await LocalQuery(input, cancellationToken); + QueryResult result; + + try + { + result = await LocalQuery(input, cancellationToken); + } + catch (TimeoutException e) + { + // The same treatment a remote gets: this instance's data is missing, the others' is not. + logger.LogWarning(e, "The local query did not complete within its allowed query time"); + result = QueryResult.Failed(QueryFailure.TimedOut); + } + result.InstanceId = instanceId; result.IsLocalInstance = true; return result; @@ -92,7 +140,13 @@ internal QueryResult AggregateResults(TIn input, QueryResult[] resul return new QueryResult( combinedResults, AggregateStats(input, results, combinedResults) - ); + ) + { + IncompleteInstances = results + .Where(result => result.Failure is not null) + .Select(result => new IncompleteInstance(result.InstanceId, result.Failure.Value)) + .ToArray() + }; } protected abstract TOut ProcessResults(TIn input, QueryResult[] results); @@ -158,7 +212,22 @@ async Task> FetchAndParse(HttpClient httpClient, string pathAn { logger.LogWarning("Authentication failed when querying remote instance at {RemoteInstanceBaseAddress}. Ensure authentication is correctly configured.", remoteInstanceSetting.BaseAddress); - return QueryResult.Empty(); + return QueryResult.Failed(QueryFailure.Failed); + } + + if (rawResponse.StatusCode == HttpStatusCode.GatewayTimeout) + { + // The remote's own query ran out of its allowed query time; its problem body names the remote's setting. + logger.LogWarning("The query on remote instance at {RemoteInstanceBaseAddress} did not complete within its allowed query time: {Detail}", + remoteInstanceSetting.BaseAddress, await rawResponse.Content.ReadAsStringAsync(cancellationToken)); + return QueryResult.Failed(QueryFailure.TimedOut); + } + + if (!rawResponse.IsSuccessStatusCode) + { + logger.LogWarning("Remote instance at {RemoteInstanceBaseAddress} answered {StatusCode} {ReasonPhrase}", + remoteInstanceSetting.BaseAddress, (int)rawResponse.StatusCode, rawResponse.ReasonPhrase); + return QueryResult.Failed(QueryFailure.Failed); } return await ParseResult(rawResponse, cancellationToken); @@ -170,23 +239,23 @@ async Task> FetchAndParse(HttpClient httpClient, string pathAn httpRequestException, "An HttpRequestException occurred when querying remote instance at {RemoteInstanceBaseAddress}. The instance will be temporarily disabled", remoteInstanceSetting.BaseAddress); - return QueryResult.Empty(); + return QueryResult.Failed(QueryFailure.Unavailable); } catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) { // The caller gave up on the whole scatter-gather, so this is not a per-remote timeout - // to be absorbed into an empty result. + // to be absorbed into a failed result. throw; } - catch (OperationCanceledException) // Intentional, used to gracefully handle timeout + catch (OperationCanceledException) // Intentional, used to gracefully handle the HttpClient timeout { logger.LogWarning("Failed to query remote instance at {RemoteInstanceBaseAddress} due to a timeout", remoteInstanceSetting.BaseAddress); - return QueryResult.Empty(); + return QueryResult.Failed(QueryFailure.TimedOut); } catch (Exception exception) { logger.LogWarning(exception, "Failed to query remote instance at {RemoteInstanceBaseAddress}", remoteInstanceSetting.BaseAddress); - return QueryResult.Empty(); + return QueryResult.Failed(QueryFailure.Failed); } } diff --git a/src/ServiceControl/CompositeViews/Messages/ScatterGatherRemoteOnly.cs b/src/ServiceControl/CompositeViews/Messages/ScatterGatherRemoteOnly.cs index f867a39694..85794e5421 100644 --- a/src/ServiceControl/CompositeViews/Messages/ScatterGatherRemoteOnly.cs +++ b/src/ServiceControl/CompositeViews/Messages/ScatterGatherRemoteOnly.cs @@ -16,6 +16,8 @@ public abstract class ScatterGatherRemoteOnly(Settings settings, IHtt { protected sealed override Task> LocalQuery(TIn input, CancellationToken cancellationToken = default) => QueryResult.Empty(); + protected sealed override bool LocalInstanceParticipates => false; + protected sealed override QueryStatsInfo AggregateStats(TIn input, IEnumerable> results, TOut processedResults) => AggregateStatsFromRemotesOnly(results); } diff --git a/src/ServiceControl/Infrastructure/Api/AuditCountApi.cs b/src/ServiceControl/Infrastructure/Api/AuditCountApi.cs index 8258688842..df6cc8fc16 100644 --- a/src/ServiceControl/Infrastructure/Api/AuditCountApi.cs +++ b/src/ServiceControl/Infrastructure/Api/AuditCountApi.cs @@ -1,16 +1,34 @@ -namespace ServiceControl.Infrastructure.Api; +namespace ServiceControl.Infrastructure.Api; +using System; using System.Collections.Generic; +using System.Linq; using System.Threading; using System.Threading.Tasks; using CompositeViews.MessageCounting; +using CompositeViews.Messages; using Persistence.Infrastructure; using ServiceControl.Api; using ServiceControl.Api.Contracts; class AuditCountApi(GetAuditCountsForEndpointApi auditCountsForEndpointApi) : IAuditCountApi { - public async Task> GetEndpointAuditCounts(string endpoint, CancellationToken cancellationToken = default) => - (await auditCountsForEndpointApi.Execute(new AuditCountsForEndpointContext(new PagingInfo(), endpoint), - $"/api/endpoints/{endpoint}/audit-count", cancellationToken)).Results; + public async Task> GetEndpointAuditCounts(string endpoint, CancellationToken cancellationToken = default) + { + var result = await auditCountsForEndpointApi.Execute(new AuditCountsForEndpointContext(new PagingInfo(), endpoint), + $"/api/endpoints/{endpoint}/audit-count", cancellationToken); + + // A sum that is missing an instance is not that endpoint's throughput. Recorded as such, the day would + // be under-counted for good; failing leaves it to be collected on the next run. + if (result.IncompleteInstances.Count > 0) + { + var message = $"The audit counts for endpoint '{endpoint}' are incomplete: {ScatterGatherApiBase.Describe(result.IncompleteInstances)}"; + + throw result.IncompleteInstances.Any(instance => instance.Reason == QueryFailure.TimedOut) + ? new TimeoutException(message) + : new InvalidOperationException(message); + } + + return result.Results; + } } \ No newline at end of file diff --git a/src/ServiceControl/Infrastructure/WebApi/Cors.cs b/src/ServiceControl/Infrastructure/WebApi/Cors.cs index 4321e8eafc..9be6758713 100644 --- a/src/ServiceControl/Infrastructure/WebApi/Cors.cs +++ b/src/ServiceControl/Infrastructure/WebApi/Cors.cs @@ -1,4 +1,4 @@ -namespace ServiceControl.Infrastructure.WebApi +namespace ServiceControl.Infrastructure.WebApi { using Microsoft.AspNetCore.Cors.Infrastructure; using ServiceControl.Hosting.RequestId; @@ -26,7 +26,7 @@ public static CorsPolicy GetDefaultPolicy(CorsSettings settings) } // Expose custom headers that clients need to read from responses - builder.WithExposedHeaders(["ETag", "Last-Modified", "Link", "Total-Count", "X-Particular-Version", "Content-Disposition", RequestIdHeader.HeaderName]); + builder.WithExposedHeaders(["ETag", "Last-Modified", "Link", "Total-Count", "X-Particular-Version", "Content-Disposition", RequestIdHeader.HeaderName, HttpResponseExtensions.IncompleteResultsHeader]); // Allow standard headers required for API requests builder.WithHeaders(["Origin", "X-Requested-With", "Content-Type", "Accept", "Authorization"]); // Allow all HTTP methods used by the ServiceControl API diff --git a/src/ServiceControl/Infrastructure/WebApi/HttpResponseExtensions.cs b/src/ServiceControl/Infrastructure/WebApi/HttpResponseExtensions.cs index 1fcb30247f..cf44cf3f0d 100644 --- a/src/ServiceControl/Infrastructure/WebApi/HttpResponseExtensions.cs +++ b/src/ServiceControl/Infrastructure/WebApi/HttpResponseExtensions.cs @@ -11,6 +11,42 @@ namespace ServiceControl.Infrastructure.WebApi static class HttpResponseExtensions { + /// + /// Names the instances whose data the response is missing, as "instanceId:reason" entries, so a client + /// can tell an incomplete page from a complete one. Absent when every instance answered. + /// + /// + /// A header rather than a JSON field on purpose: the composite endpoints return a bare array (or a single + /// object), so the only place for this in the body would be an envelope around it. That changes the + /// response schema and would need a new API for every client that reads these endpoints today. + /// The header keeps the API backward compatible, the way Total-Count, ETag and Link already do for the + /// rest of the list metadata; a client that does not know the header simply ignores it. + /// + public const string IncompleteResultsHeader = "X-Particular-Incomplete-Results"; + + public static void WithScatterGatherResult(this HttpResponse response, QueryResult result, PagingInfo pagingInfo) + where T : class + { + response.WithQueryStatsAndPagingInfo(result.QueryStats, pagingInfo); + response.WithIncompleteResults(result.IncompleteInstances); + } + + public static void WithIncompleteResults(this HttpResponse response, IReadOnlyList incomplete) + { + if (incomplete.Count == 0) + { + return; + } + + response.WithHeader(IncompleteResultsHeader, string.Join(", ", incomplete.Select(instance => $"{instance.InstanceId}:{instance.Reason switch + { + QueryFailure.TimedOut => "timeout", + QueryFailure.Unavailable => "unavailable", + QueryFailure.Failed => "error", + _ => "error" + }}"))); + } + public static void WithTotalCount(this HttpResponse response, long totalCount) => response.WithHeader("Total-Count", totalCount.ToString(CultureInfo.InvariantCulture)); public static void WithEtag(this HttpResponse response, DataVersion version) diff --git a/src/ServiceControl/Infrastructure/WebApi/RemoteInstanceServiceCollectionExtensions.cs b/src/ServiceControl/Infrastructure/WebApi/RemoteInstanceServiceCollectionExtensions.cs index 40dfd8645f..09eda3b16d 100644 --- a/src/ServiceControl/Infrastructure/WebApi/RemoteInstanceServiceCollectionExtensions.cs +++ b/src/ServiceControl/Infrastructure/WebApi/RemoteInstanceServiceCollectionExtensions.cs @@ -7,6 +7,7 @@ namespace ServiceControl.Infrastructure.WebApi; using System.Net.Http.Headers; using Microsoft.Extensions.DependencyInjection; using ServiceBus.Management.Infrastructure.Settings; +using ServiceControl.Persistence; using Yarp.ReverseProxy.Forwarder; static class RemoteInstanceServiceCollectionExtensions @@ -28,15 +29,22 @@ public static void AddHttpForwarding(this IServiceCollection services) })); } + /// + /// How much longer than the query time limit a remote gets to answer: its own query is allowed that limit + /// (the audit instance has the same setting), and its answer, a 504 included, still has to travel back. + /// + public static readonly TimeSpan ResponseMargin = TimeSpan.FromSeconds(30); + public static void AddRemoteInstancesHttpClients(this IServiceCollection services, Settings settings) { foreach (var remoteInstance in settings.RemoteInstances) { - var remoteClientBuilder = services.AddHttpClient(remoteInstance.InstanceId, client => + var remoteClientBuilder = services.AddHttpClient(remoteInstance.InstanceId, (serviceProvider, client) => { client.DefaultRequestHeaders.Accept.Add(new MediaTypeWithQualityHeaderValue("application/json")); // Application settings might contain remote URLs with /api. We strip that away to be a real base address. client.BaseAddress = new Uri(remoteInstance.BaseAddress); + client.Timeout = serviceProvider.GetRequiredService().QueryTimeout + ResponseMargin; }); remoteClientBuilder.UseSocketsHttpHandler((handler, _) => diff --git a/src/ServiceControl/SagaAudit/SagasController.cs b/src/ServiceControl/SagaAudit/SagasController.cs index f0ff9f28b6..ed6c0aea4f 100644 --- a/src/ServiceControl/SagaAudit/SagasController.cs +++ b/src/ServiceControl/SagaAudit/SagasController.cs @@ -22,7 +22,7 @@ public async Task Sagas([FromQuery] PagingInfo pagingInfo, Guid id, QueryResult result = await getSagaByIdApi.Execute(new SagaByIdContext(pagingInfo, id), Request.GetEncodedPathAndQuery(), cancellationToken); - Response.WithQueryStatsAndPagingInfo(result.QueryStats, pagingInfo); + Response.WithScatterGatherResult(result, pagingInfo); return result.Results; } } diff --git a/src/ServiceControl/WebApplicationExtensions.cs b/src/ServiceControl/WebApplicationExtensions.cs index 888759811f..ec2f69f03d 100644 --- a/src/ServiceControl/WebApplicationExtensions.cs +++ b/src/ServiceControl/WebApplicationExtensions.cs @@ -4,6 +4,7 @@ namespace ServiceControl; using Microsoft.AspNetCore.Builder; using ServiceControl.Hosting.ForwardedHeaders; using ServiceControl.Hosting.Https; +using ServiceControl.Hosting.QueryTimeout; using ServiceControl.Hosting.RequestId; using ServiceControl.Infrastructure; using ServiceControl.Infrastructure.Health; @@ -13,6 +14,7 @@ public static class WebApplicationExtensions public static void UseServiceControl(this WebApplication app, ForwardedHeadersSettings forwardedHeadersSettings, HttpsSettings httpsSettings) { app.UseRequestIdHeader(); + app.UseQueryTimeoutResponse(); app.UseServiceControlForwardedHeaders(forwardedHeadersSettings); app.UseServiceControlHttps(httpsSettings); app.UseResponseCompression();