Refactor and improve data versioning / etags

docs/data-versioning-design.md

@@ -0,0 +1,51 @@
+# Data versioning design
+
+## What it is
+
+A **data version** is the short opaque label a query result carries so that a client asking for it again can be told "nothing has changed" instead of being sent the whole answer. On the wire it is an HTTP entity-tag: the response carries `ETag`, the client sends it back as `If-None-Match`, and a matching request is answered `304 Not Modified` with no body.
+
+One value type carries it end to end: `DataVersion` in `src/ServiceControl.Persistence/Infrastructure/DataVersion.cs`. Every persister produces one, `QueryStatsInfo.Version` carries it out of the persistence layer, and the Web API turns it into the header. It is a `readonly struct`, so `default` is a legitimate value and no variable of the type can be null.
+
+This is the primary (error) instance only. The audit instance still carries a `string ETag` on its own `QueryStatsInfo` and has not been converted.
+
+## The one rule
+
+**If a field the response renders can change without the version changing, a client caches that page for ever and nothing reveals it.** No log line, no exception, no failing test.
+
+The promise is scoped to **one URL**, because a client only ever sends a validator back to the URL that issued it. So what must never happen is one URL answering `304` when its own body would have differed. Two different URLs sharing a value is harmless: an HTTP cache is keyed on the whole URL.
+
+That scoping is what makes a backend's own token usable. RavenDB's result etag stands for the state of the index behind the query, so it moves on any write the query could see, but it says nothing about which page was asked for: every `/api/errors` URL shares one value, whatever the page, sort or filter. The EF Core persisters compose over the rows they returned, so theirs differ per page. Both satisfy the rule.
+
+## Making one
+
+| Factory                                 | Use it for                                                                                      |
+| --------------------------------------- | ----------------------------------------------------------------------------------------------- |
+| `FromToken(string)` / `FromToken(long)` | a token the backend already produces, such as a RavenDB index etag or document change vector    |
+| `Compose(terms)`                        | named terms over aggregates, where an aggregate provably moves with the fields it stands in for |
+| `OverRows(summary, rows, fields)`       | a list the response renders row by row: summary terms for the whole set, plus one term per row  |
+| `Combine(instances)`                    | one version for a result gathered from several instances                                        |
+| `FromClient(header)`                    | a validator a caller sent back, in any shape an old or current instance might emit              |
+
+`Compose` hashes its description through `DeterministicGuid.MakeId`, so the emitted tag is a GUID rather than the underlying values.
+
+Term names and every field inside a row are **length prefixed**. Without that, free user text carrying a delimiter could make two different results digest identically: a failure group titled `x.y` with an empty `Type` would collide with one titled `x` whose `Type` is `y`. `Format` accepts strings, `bool`, `DateTime` and `DateTimeOffset` (both by ticks) and anything `IFormattable` under the invariant culture, and **throws** on anything else, because a type whose `ToString` is not a documented function of its content would pin the version silently.
+
+`OverRows` names rows by position, so a caller whose query has no `ORDER BY` has to sort them first or the validator churns.
+
+## Absence
+
+`DataVersion.None` is `default`, and it **matches nothing, not even itself**. Two parties that both know nothing have not established that nothing changed, so an empty-string validator matching itself would answer `304` for every request.
+
+Absence propagates in the safe direction. `WithEtag` writes no header for `None`, so no header means no `If-None-Match`, which means the full body. `Combine` returns `None` as soon as any instance reports none, rather than quietly claiming to cover an instance it could not see.
+
+## Reaching the client
+
+`WithEtag` emits **every** tag weak, as `W/"…"`. Nothing here can promise the response bytes: response compression rewrites them without touching the tag, and no endpoint enables range processing, which is the one thing an exact validator would buy. RFC 9110 requires `If-None-Match` to use the weak comparison anyway, so the marking costs nothing.
+
+`NotModifiedStatusHttpHandler` turns a matching request into a `304`. It compares with `EntityTagHeaderValue.Compare(useStrongComparison: false)`, because `Equals` on that type compares strength as well as the tag and its own documentation says not to use it for this. `*` matches whenever a representation exists.
+
+## Across instances
+
+Scatter-gather endpoints merge one version per instance through `Combine`. It is keyed on instance id and sorted ordinally, so the composite is independent of the order instances answered in but still moves if two instances swap which validator they report.
+
+An API whose own instance holds none of the data drops its own empty result before aggregating, via `AggregateStatsFromRemotesOnly`. Left in, its version-less placeholder would take the whole composite to `None` and the endpoint would emit no tag at all.

docs/eventlog-design.md

@@ -40,7 +40,7 @@ Timestamps are when the thing happened, so an item can land in the middle of the
 `IEventLogDataStore` has two methods, and its XML docs are the binding contract:
 
 - `Add(EventLogItem)` persists one item. **Identity is the store's to assign** and surfaces on `EventLogItemView.Id`. That makes `Id` opaque: a stable key within one store, not something to parse.
-- `GetEventLogItems(PagingInfo, knownVersion)` returns a `QueryResult` carrying the page, the total count independent of paging, and an `ETag`. Two obligations: the `ETag` is surfaced **verbatim**, so whatever the client echoes back arrives here unchanged and can be compared, and it **must change when retention removes items**, not only when one is added, since nothing else tells a polling client its cached page has gone stale.
+- `GetEventLogItems(PagingInfo)` returns a `QueryResult` carrying the page, the total count independent of paging, and an `ETag`. Two obligations: the `ETag` is surfaced **verbatim**, so whatever the client echoes back arrives here unchanged and can be compared, and it **must change when retention removes items**, not only when one is added, since nothing else tells a polling client its cached page has gone stale.
 
 ## Retention
 

src/ServiceControl.AcceptanceTests.PostgreSql/ServiceControl.AcceptanceTests.PostgreSql.csproj

@@ -51,9 +51,6 @@
     <Compile Remove="..\ServiceControl.AcceptanceTests\Recoverability\ExternalIntegration\When_a_reedit_solves_a_failed_msg.cs" />
     <Compile Remove="..\ServiceControl.AcceptanceTests\Recoverability\ExternalIntegration\When_encountered_an_error.cs" />
     <Compile Remove="..\ServiceControl.AcceptanceTests\Recoverability\When_edited_message_fails_to_process.cs" />
-
-    <!-- The EF custom-check query does not provide an ETag. Addressed by a separate PR. -->
-    <Compile Remove="..\ServiceControl.AcceptanceTests\WebApi\When_a_request_is_repeated_with_its_etag.cs" />
   </ItemGroup>
 
 </Project>

src/ServiceControl.AcceptanceTests.RavenDB/Recoverability/MessageFailures/FailedErrorsController.cs

@@ -5,6 +5,7 @@
     using Infrastructure.WebApi;
     using Microsoft.AspNetCore.Mvc;
     using Operations;
+    using Persistence.Infrastructure;
     using Persistence.RavenDB;
     using Raven.Client.Documents;
 
@@ -28,7 +29,7 @@ public async Task<FailedErrorsCountReponse> GetFailedErrorsCount(CancellationTok
 
             var count = await query.CountAsync(cancellationToken);
 
-            Response.WithEtag(stats.ResultEtag.ToString());
+            Response.WithEtag(DataVersion.FromToken(stats.ResultEtag.ToString()));
 
             return new FailedErrorsCountReponse { Count = count };
         }

src/ServiceControl.AcceptanceTests.RavenDB/Recoverability/MessageFailures/FailedMessageRetriesController.cs

@@ -4,6 +4,7 @@
     using System.Threading.Tasks;
     using Infrastructure.WebApi;
     using Microsoft.AspNetCore.Mvc;
+    using Persistence.Infrastructure;
     using Persistence.RavenDB;
     using Raven.Client.Documents;
     using ServiceControl.Recoverability;
@@ -24,7 +25,7 @@ public async Task<FailedMessageRetriesCountReponse> GetFailedMessageRetriesCount
             using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken);
             await session.Query<FailedMessageRetry>().Statistics(out var stats).ToListAsync(cancellationToken);
 
-            Response.WithEtag(stats.ResultEtag.ToString());
+            Response.WithEtag(DataVersion.FromToken(stats.ResultEtag.ToString()));
 
             return new FailedMessageRetriesCountReponse { Count = stats.TotalResults };
         }

src/ServiceControl.AcceptanceTests.SqlServer/ServiceControl.AcceptanceTests.SqlServer.csproj

@@ -51,9 +51,6 @@
     <Compile Remove="..\ServiceControl.AcceptanceTests\Recoverability\ExternalIntegration\When_a_reedit_solves_a_failed_msg.cs" />
     <Compile Remove="..\ServiceControl.AcceptanceTests\Recoverability\ExternalIntegration\When_encountered_an_error.cs" />
     <Compile Remove="..\ServiceControl.AcceptanceTests\Recoverability\When_edited_message_fails_to_process.cs" />
-
-    <!-- The EF custom-check query does not provide an ETag. Addressed by a separate PR. -->
-    <Compile Remove="..\ServiceControl.AcceptanceTests\WebApi\When_a_request_is_repeated_with_its_etag.cs" />
   </ItemGroup>
 
 </Project>

src/ServiceControl.AcceptanceTests/WebApi/When_a_request_is_repeated_with_its_etag.cs

@@ -2,34 +2,26 @@ namespace ServiceControl.AcceptanceTests.WebApi
 {
     using System.Net;
     using System.Net.Http;
+    using System.Net.Http.Json;
     using System.Threading.Tasks;
     using AcceptanceTesting;
     using NServiceBus.AcceptanceTesting;
     using NUnit.Framework;
-    using Recoverability.MessageRedirects;
 
     class When_a_request_is_repeated_with_its_etag : AcceptanceTest
     {
-        [TestCase("/api/customchecks", "GET", false)]
-        [TestCase("/api/redirects", "GET", true)]
-        [TestCase("/api/redirect", "HEAD", true)]
-        public async Task Should_answer_not_modified(string url, string method, bool seedARedirect)
+        [TestCase("/api/customchecks", "GET")]
+        [TestCase("/api/redirects", "GET")]
+        [TestCase("/api/redirect", "HEAD")]
+        [TestCase("/api/errors/queues/addresses", "GET")]
+        public async Task Should_answer_not_modified(string url, string method)
         {
             Answer issued = null;
             Answer repeated = null;
 
             await Define<Context>()
                 .Done(async ctx =>
                 {
-                    if (seedARedirect)
-                    {
-                        await this.Post("/api/redirects", new RedirectRequest
-                        {
-                            fromphysicaladdress = "endpointA@machine1",
-                            tophysicaladdress = "endpointB@machine2"
-                        }, status => status is not HttpStatusCode.Created);
-                    }
-
                     // Internal custom checks re-report on a timer, so the validator can move between
                     // the two requests.
                     for (var attempt = 0; attempt < 5; attempt++)
@@ -61,6 +53,42 @@ await Define<Context>()
             Assert.That(repeated.TotalCount, Is.Not.Null.And.EqualTo(issued.TotalCount), $"{method} {url} did not carry its Total-Count through to the 304");
         }
 
+        [Test]
+        public async Task Should_answer_with_a_new_etag_once_the_data_moves()
+        {
+            Answer before = null;
+            Answer after = null;
+
+            await Define<Context>()
+                .Done(async ctx =>
+                {
+                    before = await Ask("GET", "/api/redirects", ifNoneMatch: null);
+
+                    if (before.Etag == null)
+                    {
+                        return false;
+                    }
+
+                    using var created = await HttpClient.PostAsJsonAsync("/api/redirects", new
+                    {
+                        FromPhysicalAddress = "SomeEndpoint@MACHINE",
+                        ToPhysicalAddress = "OtherEndpoint@MACHINE"
+                    });
+
+                    created.EnsureSuccessStatusCode();
+
+                    after = await Ask("GET", "/api/redirects", before.Etag);
+
+                    return true;
+                })
+                .Run();
+
+            Assert.That(after.Status, Is.EqualTo(HttpStatusCode.OK),
+                "a redirect was added, so the client's validator is stale and it has to be sent the new list");
+            Assert.That(after.Etag, Is.Not.Null.And.Not.EqualTo(before.Etag),
+                "the body changed, so the validator has to move with it or the next poll caches the stale list forever");
+        }
+
         async Task<Answer> Ask(string method, string url, string ifNoneMatch)
         {
             using var response = await Send(method, url, ifNoneMatch);

src/ServiceControl.Audit.Persistence.RavenDB/Extensions/RavenQueryStatisticsExtensions.cs

@@ -1,13 +1,18 @@
 namespace ServiceControl.Audit.Persistence.RavenDB.Extensions
 {
+    using System.Globalization;
     using Auditing.MessagesView;
     using Raven.Client.Documents.Session;
 
     static class RavenQueryStatisticsExtensions
     {
-        public static QueryStatsInfo ToQueryStatsInfo(this QueryStatistics stats)
-        {
-            return new QueryStatsInfo($"{stats.ResultEtag}", stats.TotalResults);
-        }
+        /// <summary>
+        /// RavenDB's result etag hashes the state of the index behind the query: every collection's last
+        /// document and tombstone etag, how far the index has processed, and its definition. It therefore
+        /// moves on any write the query could see, which is all a validator has to do, because a client only
+        /// ever sends one back on a request for the same URL.
+        /// </summary>
+        public static QueryStatsInfo ToQueryStatsInfo(this QueryStatistics stats) =>
+            new(stats.ResultEtag?.ToString(CultureInfo.InvariantCulture) ?? string.Empty, stats.TotalResults);
     }
-}
+}

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

@@ -25,7 +25,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, new QueryStatsInfo($"{stats.ResultEtag}", stats.TotalResults));
+            return sagaHistory == null ? QueryResult<SagaHistory>.Empty() : new QueryResult<SagaHistory>(sagaHistory, stats.ToQueryStatsInfo());
         }
 
         public async Task<QueryResult<IList<MessagesView>>> GetMessages(bool includeSystemMessages, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange, CancellationToken cancellationToken = default)

src/ServiceControl.Audit/Auditing/MessagesView/GetMessages2Controller.cs

@@ -54,6 +54,7 @@ public async Task<IList<MessagesView>> GetAllMessages(
         }
 
         Response.WithTotalCount(result.QueryStats.TotalCount);
+        Response.WithEtag(result.QueryStats.ETag);
 
         return result.Results;
     }

src/ServiceControl.Persistence.EFCore/Implementation/BodyStorage/BodyStorage.cs

@@ -8,6 +8,7 @@ namespace ServiceControl.Persistence.EFCore.Implementation.BodyStorage;
 using ServiceControl.Persistence.EFCore.DbContexts;
 using ServiceControl.Persistence.EFCore.Entities;
 using ServiceControl.Persistence.EFCore.Infrastructure;
+using ServiceControl.Persistence.Infrastructure;
 
 /// <summary>
 /// Resolves a message body from wherever it was stored.
@@ -28,12 +29,15 @@ public async Task<MessageBodyResult> TryFetch(string bodyId, CancellationToken c
             return MessageBodyResult.NotFound();
         }
 
-        // Bodies are immutable per message, so the id is a stable ETag.
-        var uniqueMessageId = row.UniqueMessageId.ToString();
+        // Ingestion updates the existing row rather than adding one, so the message id is unchanged
+        // and cannot serve as a version alone. LastModified is written on every upsert.
+        var version = DataVersion.Compose(
+            ("uniqueMessageId", row.UniqueMessageId),
+            ("lastModified", row.LastModified));
 
         if (row.BodyStoredExternally)
         {
-            var external = await storagePersistence.ReadBody(uniqueMessageId, cancellationToken);
+            var external = await storagePersistence.ReadBody(row.UniqueMessageId.ToString(), cancellationToken);
 
             if (external == null)
             {
@@ -46,7 +50,7 @@ public async Task<MessageBodyResult> TryFetch(string bodyId, CancellationToken c
                 return MessageBodyResult.Empty();
             }
 
-            return MessageBodyResult.Available(new MessageBodyStreamContent(external.Stream, external.ContentType, external.BodySize, uniqueMessageId));
+            return MessageBodyResult.Available(new MessageBodyStreamContent(external.Stream, external.ContentType, external.BodySize, version));
         }
 
         if (row.BodyText != null)
@@ -62,7 +66,7 @@ public async Task<MessageBodyResult> TryFetch(string bodyId, CancellationToken c
                 new MemoryStream(bytes, writable: false),
                 row.BodyContentType ?? "text/plain",
                 bytes.Length,
-                uniqueMessageId));
+                version));
         }
 
         if (row.BodySize == 0)
@@ -98,7 +102,8 @@ public async Task<MessageBodyResult> TryFetch(string bodyId, CancellationToken c
                 BodyText = message.BodyText,
                 BodyStoredExternally = message.BodyStoredExternally,
                 BodySize = message.BodySize,
-                BodyContentType = message.BodyContentType
+                BodyContentType = message.BodyContentType,
+                LastModified = message.LastModified
             })
             .FirstOrDefaultAsync(cancellationToken);
 
@@ -109,5 +114,6 @@ sealed class BodyRow
         public bool BodyStoredExternally { get; init; }
         public int BodySize { get; init; }
         public string? BodyContentType { get; init; }
+        public DateTime LastModified { get; init; }
     }
 }

src/ServiceControl.Persistence.EFCore/Implementation/CustomCheckDataStore.cs

@@ -4,6 +4,8 @@ namespace ServiceControl.Persistence.EFCore.Implementation;
 using Microsoft.EntityFrameworkCore;
 using Microsoft.Extensions.DependencyInjection;
 using ServiceControl.Contracts.CustomChecks;
+using ServiceControl.Operations;
+using ServiceControl.Persistence.EFCore.Infrastructure;
 using ServiceControl.Persistence.Infrastructure;
 
 public class CustomCheckDataStore(IServiceScopeFactory scopeFactory) : DataStoreBase(scopeFactory), ICustomChecksDataStore
@@ -59,21 +61,31 @@ public Task<QueryResult<IList<CustomCheck>>> GetStats(PagingInfo paging, string?
             _ => query
         };
 
-        var page = await query
+        var checks = await query
             .OrderBy(c => c.ReportedAt)
+            .ThenBy(c => c.Id)
             .Skip(paging.Offset)
             .Take(paging.PageSize)
+            .Select(c => new CustomCheck
+            {
+                Id = c.Id.ToString(),
+                CustomCheckId = c.CustomCheckId,
+                Category = c.Category,
+                Status = c.Status,
+                ReportedAt = c.ReportedAt,
+                FailureReason = c.FailureReason,
+                OriginatingEndpoint = new EndpointDetails
+                {
+                    Name = c.OriginatingEndpointName,
+                    Host = c.OriginatingEndpointHost,
+                    HostId = c.OriginatingEndpointHostId
+                }
+            })
             .ToListAsync(token);
 
-        return new QueryResult<IList<CustomCheck>>(page.Select(c => new CustomCheck
-        {
-            Id = c.Id.ToString(),
-            CustomCheckId = c.CustomCheckId,
-            Category = c.Category,
-            Status = c.Status,
-            ReportedAt = c.ReportedAt,
-            FailureReason = c.FailureReason
-        }).ToList(), new QueryStatsInfo("", page.Count, false));
+        var totalCount = await query.CountAsync(token);
+
+        return new QueryResult<IList<CustomCheck>>(checks, checks.ToQueryStatsInfo(totalCount));
     }, cancellationToken);
 
     public Task DeleteCustomCheck(Guid id, CancellationToken cancellationToken = default) => ExecuteWithDbContext(async (context, token) => await context.CustomChecks.AsNoTracking().Where(cc => cc.Id == id).ExecuteDeleteAsync(token), cancellationToken);