From 9eda6433ec7d688ed80c86641558f2599459f3e4 Mon Sep 17 00:00:00 2001 From: Warwick Schroeder Date: Fri, 23 Oct 2026 13:31:21 +0800 Subject: [PATCH] Implement data versioning and enhance saga history tracking with new tests --- docs/data-versioning-design.md | 2 + .../Implementation/AuditDataStore.cs | 7 +- .../RavenAuditDataStore.cs | 2 +- .../RetentionTests.cs | 19 ++++ .../AuditTests.Versions.cs | 93 +++++++++++++++++++ .../EFCore/EFRetentionTests.cs | 46 +++++++++ .../SagaHistoryTests.cs | 40 ++++++++ .../VersionAssert.cs | 31 +++++++ .../Infrastructure/QueryResult.cs | 2 +- .../ScatterGather/SagaScatterGatherTests.cs | 90 ++++++++++++++++++ .../Messages/ScatterGatherApi.cs | 6 ++ 11 files changed, 333 insertions(+), 5 deletions(-) create mode 100644 src/ServiceControl.Audit.Persistence.Tests/AuditTests.Versions.cs create mode 100644 src/ServiceControl.Audit.Persistence.Tests/VersionAssert.cs create mode 100644 src/ServiceControl.UnitTests/ScatterGather/SagaScatterGatherTests.cs diff --git a/docs/data-versioning-design.md b/docs/data-versioning-design.md index 44935db1c8..0b7b310bdd 100644 --- a/docs/data-versioning-design.md +++ b/docs/data-versioning-design.md @@ -35,6 +35,8 @@ Every term's value, and every field inside a row, is **length prefixed**. Withou `OverRows` names rows by position, so a caller whose query has no `ORDER BY` has to sort them first or the validator churns. +The audit instance's EF Core persisters give each row only its `CreatedOn` and `Id` as fields. That covers everything the row shows only because audit rows are inserted and deleted, never updated. A change that updates an audit row in place has to version over the fields the response renders instead. + ## Absence `DataVersion.None` is `default`, and it means there is no version to offer. Two parties that both know nothing have not established that nothing changed, so **absence must never answer `304`**: an empty validator that matched itself would serve a cached page for every request forever. diff --git a/src/ServiceControl.Audit.Persistence.EFCore/Implementation/AuditDataStore.cs b/src/ServiceControl.Audit.Persistence.EFCore/Implementation/AuditDataStore.cs index 4bd9e8d840..89daf8cd42 100644 --- a/src/ServiceControl.Audit.Persistence.EFCore/Implementation/AuditDataStore.cs +++ b/src/ServiceControl.Audit.Persistence.EFCore/Implementation/AuditDataStore.cs @@ -126,9 +126,11 @@ public Task> QuerySagaHistoryById(Guid input, Cancellat .Take(SagaHistoryCap) .ToListAsync(token); + var version = DataVersion.OverRows([("changes", snapshots.Count)], snapshots, snapshot => [snapshot.CreatedOn, snapshot.Id]); + if (snapshots.Count == 0) { - return QueryResult.Empty(); + return QueryResult.Empty(new QueryStatsInfo(version, 0)); } var history = new SagaHistory @@ -139,8 +141,6 @@ public Task> QuerySagaHistoryById(Guid input, Cancellat Changes = [.. snapshots.Select(ToStateChange)] }; - var version = DataVersion.OverRows([("changes", snapshots.Count)], snapshots, snapshot => [snapshot.CreatedOn, snapshot.Id]); - return new QueryResult(history, new QueryStatsInfo(version, snapshots.Count)); }, cancellationToken); @@ -164,6 +164,7 @@ Task>> Page( IList results = [.. rows.Select(row => row.ToMessagesView())]; + // Audit rows are only ever inserted or deleted, so a row's identity stands for every field the page shows. var version = DataVersion.OverRows([("messages", total)], rows, row => [row.CreatedOn, row.Id]); return new QueryResult>(results, new QueryStatsInfo(version, total)); diff --git a/src/ServiceControl.Audit.Persistence.RavenDB/RavenAuditDataStore.cs b/src/ServiceControl.Audit.Persistence.RavenDB/RavenAuditDataStore.cs index a1c27b4927..29a663e80b 100644 --- a/src/ServiceControl.Audit.Persistence.RavenDB/RavenAuditDataStore.cs +++ b/src/ServiceControl.Audit.Persistence.RavenDB/RavenAuditDataStore.cs @@ -26,7 +26,7 @@ public async Task> QuerySagaHistoryById(Guid input, Can .Statistics(out var stats) .SingleOrDefaultAsync(x => x.SagaId == input, token: cancellationToken); - return sagaHistory == null ? QueryResult.Empty() : new QueryResult(sagaHistory, stats.ToQueryStatsInfo()); + return new QueryResult(sagaHistory, stats.ToQueryStatsInfo()); } public Task>> GetMessages(bool includeSystemMessages, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange, CancellationToken cancellationToken = default) => diff --git a/src/ServiceControl.Audit.Persistence.Tests.RavenDB/RetentionTests.cs b/src/ServiceControl.Audit.Persistence.Tests.RavenDB/RetentionTests.cs index 5ae566bea8..7bdc6f00a3 100644 --- a/src/ServiceControl.Audit.Persistence.Tests.RavenDB/RetentionTests.cs +++ b/src/ServiceControl.Audit.Persistence.Tests.RavenDB/RetentionTests.cs @@ -46,6 +46,25 @@ Task>> QueryMessages(CancellationToken cancellat MessagesViewStore.QueryMessages("MyMessageId", new PagingInfo(), new SortInfo("Id", "asc"), cancellationToken: cancellationToken); } + [Test] + public async Task A_list_version_moves_when_expiration_removes_a_message() + { + await WaitForIndexesToBeBuilt(); + + await IngestProcessedMessagesAudits(MakeMessage("MyMessageId")); + + var beforeExpiration = await QueryMessages(TestContext.CurrentContext.CancellationToken); + Assert.That(beforeExpiration.Results, Has.Count.EqualTo(1), "the message expired before the first read"); + + var afterExpiration = await WaitUntil(QueryMessages, result => result.Results.Count == 0); + + Assert.That(afterExpiration.Results, Is.Empty, "expiration did not remove the message"); + VersionAssert.Moved(beforeExpiration.QueryStats.Version, afterExpiration.QueryStats.Version, "a client holding the old version would keep showing the removed message"); + + Task>> QueryMessages(CancellationToken cancellationToken) => + MessagesViewStore.QueryMessages("MyMessageId", new PagingInfo(), new SortInfo("Id", "asc"), cancellationToken: cancellationToken); + } + [Test] public async Task SagaSnapshotRetention() { diff --git a/src/ServiceControl.Audit.Persistence.Tests/AuditTests.Versions.cs b/src/ServiceControl.Audit.Persistence.Tests/AuditTests.Versions.cs new file mode 100644 index 0000000000..b3f74eae10 --- /dev/null +++ b/src/ServiceControl.Audit.Persistence.Tests/AuditTests.Versions.cs @@ -0,0 +1,93 @@ +namespace ServiceControl.Audit.Persistence.Tests +{ + using System; + using System.Collections.Generic; + using System.Linq; + using System.Threading.Tasks; + using Auditing; + using Auditing.MessagesView; + using Monitoring; + using NUnit.Framework; + using ServiceControl.Audit.Infrastructure; + + partial class AuditTests + { + [Test] + public async Task A_list_version_moves_when_a_matching_message_arrives([Values] ListQuery query) + { + var conversationId = Guid.NewGuid().ToString(); + await IngestProcessedMessagesAudits(MakeReceivedMessage(conversationId)); + + var issued = await List(query, conversationId); + var repeated = await List(query, conversationId); + + await IngestProcessedMessagesAudits(MakeReceivedMessage(conversationId)); + + var afterArrival = await List(query, conversationId); + + Assert.That(afterArrival.Results, Has.Count.EqualTo(2), $"the {query} query did not see the new message"); + VersionAssert.Matches(issued.QueryStats.Version, repeated.QueryStats.Version, "nothing was written between the two reads"); + VersionAssert.Moved(repeated.QueryStats.Version, afterArrival.QueryStats.Version, "a client holding the old version would never see the new message"); + } + + [Test] + public async Task A_list_version_moves_when_the_total_moves_under_an_unchanged_page() + { + var onePage = new PagingInfo(page: 1, pageSize: 1); + var byMessageId = new SortInfo("message_id", "asc"); + + await IngestProcessedMessagesAudits(MakeMessage("b")); + var before = await MessagesViewStore.GetMessages(false, onePage, byMessageId, cancellationToken: TestContext.CurrentContext.CancellationToken); + + await IngestProcessedMessagesAudits(MakeMessage("c")); + var after = await MessagesViewStore.GetMessages(false, onePage, byMessageId, cancellationToken: TestContext.CurrentContext.CancellationToken); + + using (Assert.EnterMultipleScope()) + { + Assert.That(after.Results.Select(message => message.MessageId), Is.EqualTo(new[] { "b" }), "the page itself was meant to stay the same"); + Assert.That(after.QueryStats.TotalCount, Is.EqualTo(2)); + } + + VersionAssert.Moved(before.QueryStats.Version, after.QueryStats.Version, "Total-Count went from one to two, so a client holding the old version would keep showing one"); + } + + public enum ListQuery + { + All, + Search, + Endpoint, + EndpointAndKeyword, + Conversation + } + + ProcessedMessage MakeReceivedMessage(string conversationId) + { + var message = MakeMessage(conversationId: conversationId); + message.MessageMetadata["ReceivingEndpoint"] = new EndpointDetails { Name = ReceivingEndpoint }; + return message; + } + + async Task>> List(ListQuery query, string conversationId) + { + var paging = new PagingInfo(); + var sort = new SortInfo("time_sent", "desc"); + var cancellationToken = TestContext.CurrentContext.CancellationToken; + + var result = query switch + { + ListQuery.All => await MessagesViewStore.GetMessages(false, paging, sort, cancellationToken: cancellationToken), + ListQuery.Search => await MessagesViewStore.QueryMessages("MyMessageType", paging, sort, cancellationToken: cancellationToken), + ListQuery.Endpoint => await MessagesViewStore.QueryMessagesByReceivingEndpoint(false, ReceivingEndpoint, paging, sort, cancellationToken: cancellationToken), + ListQuery.EndpointAndKeyword => await MessagesViewStore.QueryMessagesByReceivingEndpointAndKeyword(ReceivingEndpoint, "MyMessageType", paging, sort, cancellationToken: cancellationToken), + ListQuery.Conversation => await MessagesViewStore.QueryMessagesByConversationId(conversationId, paging, sort, cancellationToken), + _ => throw new ArgumentOutOfRangeException(nameof(query), query, null) + }; + + Assert.That(result.Results, Is.Not.Empty, $"the {query} query found nothing, so its version says nothing about the message"); + + return result; + } + + const string ReceivingEndpoint = "SomeEndpoint"; + } +} diff --git a/src/ServiceControl.Audit.Persistence.Tests/EFCore/EFRetentionTests.cs b/src/ServiceControl.Audit.Persistence.Tests/EFCore/EFRetentionTests.cs index d06a996705..3ba9131298 100644 --- a/src/ServiceControl.Audit.Persistence.Tests/EFCore/EFRetentionTests.cs +++ b/src/ServiceControl.Audit.Persistence.Tests/EFCore/EFRetentionTests.cs @@ -8,6 +8,7 @@ namespace ServiceControl.Audit.Persistence.Tests using NUnit.Framework; using ServiceControl.Audit.Infrastructure; using ServiceControl.Audit.Persistence.EFCore.Infrastructure; + using ServiceControl.SagaAudit; class EFRetentionTests : EFPersistenceTestFixture { @@ -36,6 +37,43 @@ public async Task Removes_rows_older_than_the_retention_period() Assert.That(remaining.Results.Select(message => message.MessageId), Is.EqualTo(new[] { live.Headers[Headers.MessageId] })); } + [Test] + public async Task A_list_version_moves_when_a_sweep_and_an_arrival_leave_the_total_unchanged() + { + await Ingest(MakeMessage()); + var before = await MessagesViewStore.GetMessages(true, new PagingInfo(), new SortInfo("time_sent", "desc")); + + configuration.TimeProvider.Advance(TimeSpan.FromDays(3)); + await configuration.CreateRetention().SweepNow(); + var live = MakeMessage(); + await Ingest(live); + var after = await MessagesViewStore.GetMessages(true, new PagingInfo(), new SortInfo("time_sent", "desc")); + + using (Assert.EnterMultipleScope()) + { + Assert.That(after.Results.Select(message => message.MessageId), Is.EqualTo(new[] { live.Headers[Headers.MessageId] }), "the sweep did not swap the expired message for the live one"); + Assert.That(after.QueryStats.TotalCount, Is.EqualTo(before.QueryStats.TotalCount), "the total was meant to hold, as it does at the count cap"); + } + + VersionAssert.Moved(before.QueryStats.Version, after.QueryStats.Version, "the total held, so only the rows can move the version"); + } + + [Test] + public async Task A_saga_version_moves_when_a_sweep_and_an_arrival_leave_the_change_count_unchanged() + { + var sagaId = Guid.NewGuid(); + await IngestSagaSnapshot(new SagaSnapshot { SagaId = sagaId, Status = SagaStateChangeStatus.New }); + var before = await SagaHistoryStore.QuerySagaHistoryById(sagaId); + + configuration.TimeProvider.Advance(TimeSpan.FromDays(3)); + await configuration.CreateRetention().SweepNow(); + await IngestSagaSnapshot(new SagaSnapshot { SagaId = sagaId, Status = SagaStateChangeStatus.Updated }); + var after = await SagaHistoryStore.QuerySagaHistoryById(sagaId); + + Assert.That(after.Results.Changes.Select(change => change.Status), Is.EqualTo(new[] { SagaStateChangeStatus.Updated }), "the sweep did not swap the expired change for the new one"); + VersionAssert.Moved(before.QueryStats.Version, after.QueryStats.Version, "the change count held, so only the rows can move the version"); + } + [Test] public async Task Keeps_rows_until_the_retention_period_has_passed() { @@ -84,5 +122,13 @@ public async Task The_retention_lock_is_released_with_its_handle() Assert.That(afterRelease, Is.Not.Null); } } + + async Task IngestSagaSnapshot(SagaSnapshot snapshot) + { + await using var unitOfWork = await StartAuditUnitOfWork(1); + await unitOfWork.RecordSagaSnapshot(snapshot); + await unitOfWork.Complete(); + await configuration.CompleteDBOperation(); + } } } diff --git a/src/ServiceControl.Audit.Persistence.Tests/SagaHistoryTests.cs b/src/ServiceControl.Audit.Persistence.Tests/SagaHistoryTests.cs index 6063d280b0..cc7cb1fe17 100644 --- a/src/ServiceControl.Audit.Persistence.Tests/SagaHistoryTests.cs +++ b/src/ServiceControl.Audit.Persistence.Tests/SagaHistoryTests.cs @@ -58,6 +58,46 @@ await IngestSagaAudits( Assert.That(queryResult.Results.Changes, Has.Count.EqualTo(2)); } + [Test] + public async Task A_missing_saga_still_reports_a_version() + { + var queryResult = await SagaHistoryStore.QuerySagaHistoryById(Guid.NewGuid(), TestContext.CurrentContext.CancellationToken); + + Assert.That(queryResult.QueryStats.Version.HasValue, Is.True, + "the query ran and found nothing, and without a version the primary's merged saga history loses its ETag"); + } + + [Test] + public async Task A_saga_version_moves_when_its_first_change_arrives() + { + var sagaId = Guid.NewGuid(); + var missing = await SagaHistoryStore.QuerySagaHistoryById(sagaId, TestContext.CurrentContext.CancellationToken); + + await IngestSagaAudits(new SagaSnapshot { SagaId = sagaId, Status = SagaStateChangeStatus.New }); + + var found = await SagaHistoryStore.QuerySagaHistoryById(sagaId, TestContext.CurrentContext.CancellationToken); + + VersionAssert.Moved(missing.QueryStats.Version, found.QueryStats.Version, "a client holding the not-found version would never see the saga"); + } + + [Test] + public async Task A_saga_version_moves_when_a_change_arrives() + { + var sagaId = Guid.NewGuid(); + await IngestSagaAudits(new SagaSnapshot { SagaId = sagaId, Status = SagaStateChangeStatus.New }); + + var issued = await SagaHistoryStore.QuerySagaHistoryById(sagaId, TestContext.CurrentContext.CancellationToken); + var repeated = await SagaHistoryStore.QuerySagaHistoryById(sagaId, TestContext.CurrentContext.CancellationToken); + + await IngestSagaAudits(new SagaSnapshot { SagaId = sagaId, Status = SagaStateChangeStatus.Updated }); + + var afterChange = await SagaHistoryStore.QuerySagaHistoryById(sagaId, TestContext.CurrentContext.CancellationToken); + + Assert.That(afterChange.Results.Changes, Has.Count.EqualTo(2), "the query did not see the new change"); + VersionAssert.Matches(issued.QueryStats.Version, repeated.QueryStats.Version, "nothing was written between the two reads"); + VersionAssert.Moved(repeated.QueryStats.Version, afterChange.QueryStats.Version, "a client holding the old version would never see the new change"); + } + async Task IngestSagaAudits(params SagaSnapshot[] snapshots) { diff --git a/src/ServiceControl.Audit.Persistence.Tests/VersionAssert.cs b/src/ServiceControl.Audit.Persistence.Tests/VersionAssert.cs new file mode 100644 index 0000000000..784ee65b9e --- /dev/null +++ b/src/ServiceControl.Audit.Persistence.Tests/VersionAssert.cs @@ -0,0 +1,31 @@ +namespace ServiceControl.Audit.Persistence.Tests +{ + using NUnit.Framework; + using ServiceControl.Infrastructure; + + static class VersionAssert + { + public static void Moved(DataVersion before, DataVersion after, string because) + { + using (Assert.EnterMultipleScope()) + { + Assert.That(before.HasValue, Is.True, "there was no version to move"); + Assert.That(after.HasValue, Is.True, "the version went missing rather than moving"); + Assert.That(after.Matches(before), Is.False, because); + } + } + + public static void Matches(DataVersion first, DataVersion second, string because) + { + using (Assert.EnterMultipleScope()) + { + Assert.That(first.HasValue, Is.True, "there was no version to hold"); + Assert.That(second.HasValue, Is.True, "the version went missing rather than holding"); + Assert.That(second.Matches(first), Is.True, because); + } + } + + public static bool Matches(this DataVersion one, DataVersion other) => + one.HasValue && other.HasValue && one.Equals(other); + } +} diff --git a/src/ServiceControl.Audit.Persistence/Infrastructure/QueryResult.cs b/src/ServiceControl.Audit.Persistence/Infrastructure/QueryResult.cs index 4303ad6071..6d76fa00d3 100644 --- a/src/ServiceControl.Audit.Persistence/Infrastructure/QueryResult.cs +++ b/src/ServiceControl.Audit.Persistence/Infrastructure/QueryResult.cs @@ -25,6 +25,6 @@ public QueryResult(TOut results, QueryStatsInfo queryStatsInfo) : base(results, public TOut Results { get; } - public static QueryResult Empty() => new QueryResult(null, QueryStatsInfo.Zero); + public static QueryResult Empty(QueryStatsInfo queryStatsInfo) => new QueryResult(null, queryStatsInfo); } } \ No newline at end of file diff --git a/src/ServiceControl.UnitTests/ScatterGather/SagaScatterGatherTests.cs b/src/ServiceControl.UnitTests/ScatterGather/SagaScatterGatherTests.cs new file mode 100644 index 0000000000..d02d24abab --- /dev/null +++ b/src/ServiceControl.UnitTests/ScatterGather/SagaScatterGatherTests.cs @@ -0,0 +1,90 @@ +namespace ServiceControl.UnitTests.ScatterGather; + +using System; +using System.Net; +using System.Net.Http; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.AspNetCore.Http; +using Microsoft.Extensions.Logging.Abstractions; +using NUnit.Framework; +using ServiceBus.Management.Infrastructure.Settings; +using ServiceControl.Infrastructure.WebApi; +using ServiceControl.Persistence.Infrastructure; +using ServiceControl.SagaAudit; + +/// +/// A saga usually lives on one audit instance, so every other instance answers that it has none. +/// +[TestFixture] +class SagaScatterGatherTests +{ + [Test] + public async Task An_instance_without_the_saga_answers_rather_than_fails() + { + var sagaId = Guid.NewGuid(); + var settings = Settings("http://audit-1/api", "http://audit-2/api"); + var factory = new FakeHttpClientFactory(); + factory.Register(settings.RemoteInstances[0], Found(sagaId, "\"has-it\"")); + factory.Register(settings.RemoteInstances[1], NotFound("\"has-none\"")); + + var api = new GetSagaByIdApi(settings, factory, new HttpContextAccessor(), NullLogger.Instance); + + var result = await api.Execute(new SagaByIdContext(new PagingInfo(), sagaId), $"/api/sagas/{sagaId}"); + + using (Assert.EnterMultipleScope()) + { + Assert.That(result.Results?.SagaId, Is.EqualTo(sagaId)); + Assert.That(result.IncompleteInstances, Is.Empty, "an instance that has no such saga still answered"); + Assert.That(result.QueryStats.Version.HasValue, Is.True, "every instance sent a version, so the composite has one to send"); + } + } + + [Test] + public async Task An_older_instance_without_the_saga_answers_but_leaves_no_version() + { + var sagaId = Guid.NewGuid(); + var settings = Settings("http://audit-1/api", "http://audit-2/api"); + var factory = new FakeHttpClientFactory(); + factory.Register(settings.RemoteInstances[0], Found(sagaId, "\"has-it\"")); + factory.Register(settings.RemoteInstances[1], NotFound(etag: null)); + + var api = new GetSagaByIdApi(settings, factory, new HttpContextAccessor(), NullLogger.Instance); + + var result = await api.Execute(new SagaByIdContext(new PagingInfo(), sagaId), $"/api/sagas/{sagaId}"); + + using (Assert.EnterMultipleScope()) + { + Assert.That(result.Results?.SagaId, Is.EqualTo(sagaId)); + Assert.That(result.IncompleteInstances, Is.Empty, "an instance from before the fix still answered"); + Assert.That(result.QueryStats.Version.HasValue, Is.False, "one instance offered no version, so the composite cannot promise anything"); + } + } + + static Settings Settings(params string[] remotes) => new() + { + RemoteInstances = Array.ConvertAll(remotes, address => new RemoteInstanceSetting(address)) + }; + + static Func> Found(Guid sagaId, string etag) => + (_, _) => + { + var history = new SagaHistory { Id = sagaId, SagaId = sagaId, SagaType = "MySaga", Changes = [] }; + var response = new HttpResponseMessage(HttpStatusCode.OK) { Content = new StringContent(System.Text.Json.JsonSerializer.Serialize(history, SerializerOptions.Default)) }; + response.Headers.TryAddWithoutValidation("ETag", etag); + return Task.FromResult(response); + }; + + // What an audit instance sends for a missing saga: ASP.NET Core turns the null result into a 204. + static Func> NotFound(string etag) => + (_, _) => + { + var response = new HttpResponseMessage(HttpStatusCode.NoContent); + if (etag != null) + { + response.Headers.TryAddWithoutValidation("ETag", etag); + } + + return Task.FromResult(response); + }; +} diff --git a/src/ServiceControl/CompositeViews/Messages/ScatterGatherApi.cs b/src/ServiceControl/CompositeViews/Messages/ScatterGatherApi.cs index da91191eb2..4146c41b66 100644 --- a/src/ServiceControl/CompositeViews/Messages/ScatterGatherApi.cs +++ b/src/ServiceControl/CompositeViews/Messages/ScatterGatherApi.cs @@ -209,6 +209,12 @@ async Task> FetchAndParse(HttpClient httpClient, string pathAn return QueryResult.Empty(); } + // A remote that found nothing returns null, which ASP.NET Core sends as a 204 with no body to parse. + if (rawResponse.StatusCode == HttpStatusCode.NoContent) + { + return new QueryResult(null, new QueryStatsInfo(ReadEtag(rawResponse.Headers), 0)); + } + if (rawResponse.StatusCode is HttpStatusCode.Unauthorized or HttpStatusCode.Forbidden) { logger.LogWarning("Authentication failed when querying remote instance at {RemoteInstanceBaseAddress}. Ensure authentication is correctly configured.",