Keep the saga ETag when an audit instance lacks the saga

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.

src/ServiceControl.Audit.Persistence.EFCore/Implementation/AuditDataStore.cs

@@ -126,9 +126,11 @@ public Task<QueryResult<SagaHistory>> 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<SagaHistory>.Empty();
+                return QueryResult<SagaHistory>.Empty(new QueryStatsInfo(version, 0));
             }
 
             var history = new SagaHistory
@@ -139,8 +141,6 @@ public Task<QueryResult<SagaHistory>> QuerySagaHistoryById(Guid input, Cancellat
                 Changes = [.. snapshots.Select(ToStateChange)]
             };
 
-            var version = DataVersion.OverRows([("changes", snapshots.Count)], snapshots, snapshot => [snapshot.CreatedOn, snapshot.Id]);
-
             return new QueryResult<SagaHistory>(history, new QueryStatsInfo(version, snapshots.Count));
         }, cancellationToken);
 
@@ -164,6 +164,7 @@ Task<QueryResult<IList<MessagesView>>> Page(
 
             IList<MessagesView> 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<IList<MessagesView>>(results, new QueryStatsInfo(version, total));

src/ServiceControl.Audit.Persistence.RavenDB/RavenAuditDataStore.cs

@@ -26,7 +26,7 @@ public async Task<QueryResult<SagaHistory>> QuerySagaHistoryById(Guid input, Can
                     .Statistics(out var stats)
                     .SingleOrDefaultAsync(x => x.SagaId == input, token: cancellationToken);
 
-            return sagaHistory == null ? QueryResult<SagaHistory>.Empty() : new QueryResult<SagaHistory>(sagaHistory, stats.ToQueryStatsInfo());
+            return new QueryResult<SagaHistory>(sagaHistory, stats.ToQueryStatsInfo());
         }
 
         public Task<QueryResult<IList<MessagesView>>> GetMessages(bool includeSystemMessages, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange, CancellationToken cancellationToken = default) =>

src/ServiceControl.Audit.Persistence.Tests.RavenDB/RetentionTests.cs

@@ -46,6 +46,25 @@ Task<QueryResult<IList<MessagesView>>> 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<QueryResult<IList<MessagesView>>> QueryMessages(CancellationToken cancellationToken) =>
+                MessagesViewStore.QueryMessages("MyMessageId", new PagingInfo(), new SortInfo("Id", "asc"), cancellationToken: cancellationToken);
+        }
+
         [Test]
         public async Task SagaSnapshotRetention()
         {

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<QueryResult<IList<MessagesView>>> 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";
+    }
+}

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();
+        }
     }
 }

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)
         {

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);
+    }
+}

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<TOut> Empty() => new QueryResult<TOut>(null, QueryStatsInfo.Zero);
+        public static QueryResult<TOut> Empty(QueryStatsInfo queryStatsInfo) => new QueryResult<TOut>(null, queryStatsInfo);
     }
 }