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.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/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.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.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/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.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 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();