diff --git a/src/MicroPlumberd.Tests.App/SnapshotDomain/TallyAggregate.cs b/src/MicroPlumberd.Tests.App/SnapshotDomain/TallyAggregate.cs new file mode 100644 index 0000000..34a2432 --- /dev/null +++ b/src/MicroPlumberd.Tests.App/SnapshotDomain/TallyAggregate.cs @@ -0,0 +1,30 @@ +namespace MicroPlumberd.Tests.App.SnapshotDomain; + +/// +/// A snapshotting aggregate whose state SHOWS a re-fold (issue #28): every TallyAdded increments Count and appends its Mark, +/// so an event applied twice reads as "abb" / Count 3, never as the same last-write-wins value. +/// SnapshotEvery is large: the rows write their snapshots themselves. +/// +[Aggregate(SnapshotEvery = 1000)] +public partial class TallyAggregate(Guid id) : AggregateBase(id) +{ + public new TallyState State => base.State; + public record TallyState { public int Count { get; set; } public string Seen { get; set; } = ""; } + + private static TallyState Given(TallyState state, TallyAdded ev) => state with { Count = state.Count + 1, Seen = state.Seen + ev.Mark }; + + /// A type "added in a later release": the build that wrote a pre-#26 snapshot did not map it; this one does. + private static TallyState Given(TallyState state, TallyNoted ev) => state with { Count = state.Count + 1, Seen = state.Seen + ev.Mark }; + + public static TallyAggregate Start(Guid id, string mark) + { + var r = Empty(id); + r.AppendPendingChange(new TallyAdded { Mark = mark }); + return r; + } + + public void Add(string mark) => AppendPendingChange(new TallyAdded { Mark = mark }); +} + +public record TallyAdded { public string? Mark { get; set; } } +public record TallyNoted { public string? Mark { get; set; } } diff --git a/src/MicroPlumberd.Tests/Integration/SnapshotRevisionTests.cs b/src/MicroPlumberd.Tests/Integration/SnapshotRevisionTests.cs new file mode 100644 index 0000000..6b4231a --- /dev/null +++ b/src/MicroPlumberd.Tests/Integration/SnapshotRevisionTests.cs @@ -0,0 +1,230 @@ +using System.Text.Json.Nodes; +using FluentAssertions; +using KurrentDB.Client; +using MicroPlumberd.Api; +using MicroPlumberd.Services; +using MicroPlumberd.Testing; +using MicroPlumberd.Tests.App.SnapshotDomain; +using MicroPlumberd.Tests.Utils; +using Microsoft.Extensions.DependencyInjection; + +namespace MicroPlumberd.Tests.Integration; + +/// +/// Issue #28. A snapshot written by a build before #26 recorded a COUNT of the events that build could map as its version. +/// Over a stream holding events it did not know, that version is LOW; loading from it read on at version + 1 and RE-FOLDED +/// events already in the snapshot's state. These rows write such a snapshot the way that build wrote it — the same event, +/// the same metadata, without the marker this fix adds — over a real stream, and load through Plumber.Get. +/// TallyAggregate's state shows a re-fold: "abb" / Count 3, never a last-write-wins value that hides it. +/// +[TestCategory("Integration")] +public class SnapshotRevisionTests : IClassFixture +{ + private readonly IPlumberApi _plumber; + private readonly PlumberEngine _engine; + private readonly EventStoreServer _es; + + public SnapshotRevisionTests(EventStoreServer es) + { + // Through DI, as an application gets it: a snapshot policy (ISnapshotPolicy<> → AttributeSnaphotPolicy<>, registered by + // AddPlumberd) is what makes Get read a snapshot at all — Plumber.Create alone has none. + var sp = new ServiceCollection().AddPlumberd(es.GetEventStoreSettings()).BuildServiceProvider(); + _plumber = sp.GetRequiredService(); + _engine = sp.GetRequiredService(); + _es = es; + } + + private string Stream(Guid id) => _plumber.Config.Conventions.GetStreamIdConvention(null!, typeof(TallyAggregate), id); + private string SnapshotStream(Guid id) => _plumber.Config.Conventions.GetStreamIdSnapshotConvention(null!, typeof(TallyAggregate.TallyState), id); + + private async Task Append(Guid id, params object[] events) => + await _plumber.AppendEvents(Stream(id), StreamState.Any, events); + + private async Task AppendUnknown(Guid id) + { + await using var client = new KurrentDBClient(_es.GetEventStoreSettings()); + await client.AppendToStreamAsync(Stream(id), StreamState.Any, + [new EventData(Uuid.NewUuid(), "EventFromANewerRelease", "{}"u8.ToArray(), "{}"u8.ToArray())]); + } + + /// + /// A snapshot as a build before #28 wrote it: written through the real AppendSnapshot, then re-appended as the stream's + /// latest snapshot with the SnapshotVersionIsRevision marker removed — the same type, data and metadata otherwise. + /// + private async Task LegacySnapshot(Guid id, TallyAggregate.TallyState state, long version) + { + await _plumber.AppendSnapshot(state, id, version); + await using var client = new KurrentDBClient(_es.GetEventStoreSettings()); + var written = (await client.ReadStreamAsync(Direction.Backwards, SnapshotStream(id), StreamPosition.End, maxCount: 1).ToListAsync())[0].Event; + var metadata = JsonNode.Parse(written.Metadata.Span)!.AsObject(); + metadata.Remove("SnapshotVersionIsRevision").Should().BeTrue("the real AppendSnapshot writes the marker"); + await client.AppendToStreamAsync(SnapshotStream(id), StreamState.Any, + [new EventData(Uuid.NewUuid(), written.EventType, written.Data, System.Text.Encoding.UTF8.GetBytes(metadata.ToJsonString()))]); + (await _plumber.GetSnapshot(id, typeof(TallyAggregate.TallyState)))!.VersionIsRevision.Should().BeFalse("precondition: a legacy snapshot"); + } + + /// + /// THE ISSUE. The old build saw [a, (unknown), b] as two events and snapshotted {ab, 2} at version 1 — the revision is 2. + /// Reading on from 2 re-folds b: "abb". Rebuilt from the stream: "ab", the unknown skipped, version 2. + /// + [Fact] + public async Task ALegacySnapshotOverAnUnknownEvent_DoesNotReFold() + { + await _es.StartInDocker(); + var id = Guid.NewGuid(); + await Append(id, new TallyAdded { Mark = "a" }); + await AppendUnknown(id); + await Append(id, new TallyAdded { Mark = "b" }); + await LegacySnapshot(id, new TallyAggregate.TallyState { Count = 2, Seen = "ab" }, version: 1); + + var tally = await _plumber.Get(id); + + tally.State.Seen.Should().Be("ab"); + tally.State.Count.Should().Be(2); + tally.Version.Should().Be(2); + } + + /// + /// The UPGRADE case — and the plausible wrong fix's blind spot. The old build did not map TallyNoted (added later); it + /// snapshotted [a, x, b] as {ab, 2} at version 1. THIS build maps TallyNoted, so "count this build's mapped events to + /// find where the snapshot ends" lands after x — and re-folds b. Rebuilt: "axb", 3 — the state this build folds. + /// + [Fact] + public async Task ALegacySnapshotOverATypeThisBuildNowMaps_IsRebuilt_NotPositionedByThisBuildsCount() + { + await _es.StartInDocker(); + var id = Guid.NewGuid(); + await Append(id, new TallyAdded { Mark = "a" }, new TallyNoted { Mark = "x" }, new TallyAdded { Mark = "b" }); + await LegacySnapshot(id, new TallyAggregate.TallyState { Count = 2, Seen = "ab" }, version: 1); + + var tally = await _plumber.Get(id); + + tally.State.Seen.Should().Be("axb"); + tally.State.Count.Should().Be(3); + tally.Version.Should().Be(2); + } + + /// + /// A legacy snapshot that is RIGHT — the event after its version was written after it — is still USED (not every legacy + /// snapshot is rebuilt: that would cost every library user a full replay). The snapshot's state is a sentinel {S, 100}: + /// used → "Sb" / 101; rebuilt → "ab" / 2. + /// + [Fact] + public async Task ALegacySnapshotWhoseNextEventCameAfterIt_IsUsed() + { + await _es.StartInDocker(); + var id = Guid.NewGuid(); + await Append(id, new TallyAdded { Mark = "a" }); + await LegacySnapshot(id, new TallyAggregate.TallyState { Count = 100, Seen = "S" }, version: 0); + await Append(id, new TallyAdded { Mark = "b" }); + + var tally = await _plumber.Get(id); + + tally.State.Seen.Should().Be("Sb"); + tally.State.Count.Should().Be(101); + tally.Version.Should().Be(1); + } + + /// + /// A snapshot this build writes carries the marker: its version IS the revision, so it is used even where the order check + /// alone would doubt it — here the event after it was written BEFORE it (another writer between a save and its snapshot). + /// Sentinel state: used → "Sb" / 101 (over [a, (unknown), b], version 1 is the revision of the unknown event). + /// + [Fact] + public async Task ASnapshotThisBuildWrote_IsUsed_EvenWhenTheNextEventPrecedesIt() + { + await _es.StartInDocker(); + var id = Guid.NewGuid(); + await Append(id, new TallyAdded { Mark = "a" }); + await AppendUnknown(id); + await Append(id, new TallyAdded { Mark = "b" }); + await _plumber.AppendSnapshot(new TallyAggregate.TallyState { Count = 100, Seen = "S" }, id, 1); + (await _plumber.GetSnapshot(id, typeof(TallyAggregate.TallyState)))!.VersionIsRevision.Should().BeTrue(); + + var tally = await _plumber.Get(id); + + tally.State.Seen.Should().Be("Sb"); + tally.State.Count.Should().Be(101); + tally.Version.Should().Be(2); + } + + /// + /// Compatibility the other way: an OLDER build (before this marker) reading a snapshot THIS build wrote is unaffected — the + /// extra metadata key is ignored. Read here exactly as the pre-marker reader reads (master 875ab10b, PlumberEngine.GetSnapshot + /// 1305-1315): the latest event of the snapshot stream, its type the snapshot event name, its data through the serializer, + /// Created and SnapshotVersion from the metadata — and nothing else. + /// + [Fact] + public async Task AMarkedSnapshot_ReadsThroughThePreMarkerReader_Unchanged() + { + await _es.StartInDocker(); + var id = Guid.NewGuid(); + var state = new TallyAggregate.TallyState { Count = 7, Seen = "abcdefg" }; + await _plumber.AppendSnapshot(state, id, 6); + + await using var client = new KurrentDBClient(_es.GetEventStoreSettings()); + var e = (await client.ReadStreamAsync(Direction.Backwards, SnapshotStream(id), StreamPosition.End, maxCount: 1).ToListAsync())[0].Event; + + e.EventType.Should().Be(_plumber.Config.Conventions.SnapshotEventNameConvention(typeof(TallyAggregate.TallyState))); + var value = _plumber.Config.SerializerFactory(typeof(TallyAggregate.TallyState)).Deserialize(null!, e.Data.Span, typeof(TallyAggregate.TallyState)); + value.Should().BeEquivalentTo(state); + using var metadata = System.Text.Json.JsonDocument.Parse(e.Metadata); + DateTimeOffset.Parse(metadata.RootElement.GetProperty("Created").GetString()!).Should().BeCloseTo(DateTimeOffset.Now, TimeSpan.FromMinutes(5)); + metadata.RootElement.GetProperty("SnapshotVersion").GetInt64().Should().Be(6); + metadata.RootElement.GetProperty("SnapshotVersionIsRevision").GetBoolean().Should().BeTrue("the marker is an ADDED key the old reader skips"); + } + + /// + /// rev-reach a3f8d0c32 (m1): "written after" is decided by the $all COMMIT POSITION, never by Created (the server's wall + /// clock). A real store cannot plant a clock step, so the decision is isolated in PlumberEngine.WrittenAfter and fed two + /// records whose Created order CONTRADICTS their positions — what a clock stepped back between the two writes produces. + /// The event at version + 1 committed BEFORE the snapshot (position 100 < 200) but stamped LATER (Created +1 h): + /// it is NOT "written after" → the short legacy snapshot is not used. + /// + [Fact] + public void WrittenAfter_IsDecidedByCommitPosition_NotByCreated() + { + var t0 = DateTime.UtcNow; + var next = Record(position: 100, created: t0.AddHours(1)); // committed first, clock says later + var snapshot = Record(position: 200, created: t0); // committed second, clock says earlier + next.Created.Should().BeAfter(snapshot.Created, "precondition: the records' clocks contradict their commit order"); + + PlumberEngine.WrittenAfter(next, snapshot).Should().BeFalse("the event at v+1 committed BEFORE the snapshot, whatever its clock says"); + PlumberEngine.WrittenAfter(snapshot, next).Should().BeTrue(); + } + + /// + /// The seam row proves the comparison; this proves its INPUT (lead + rev-reach 2883377b9). Through the real read path + /// Get's snapshot branch uses (SnapshotOrderRecords), both records carry a server-populated $all Position: non-zero, not + /// Position.Start, and in commit order — the event at v+1 appended first has the smaller one. Were a server to send 0, + /// WrittenAfter would call every legacy snapshot "may be low": all rebuilt, safely but silently. + /// + [Fact] + public async Task TheRecordsTheDecisionReads_CarryServerPositions_InCommitOrder() + { + await _es.StartInDocker(); + var id = Guid.NewGuid(); + await Append(id, new TallyAdded { Mark = "a" }, new TallyAdded { Mark = "b" }); // v+1 = revision 1, appended first + await LegacySnapshot(id, new TallyAggregate.TallyState { Count = 1, Seen = "a" }, version: 0); + + var (next, snapshot) = await _engine.SnapshotOrderRecords(null!, typeof(TallyAggregate), id, typeof(TallyAggregate.TallyState), 0, default); + + next.Should().NotBeNull(); snapshot.Should().NotBeNull(); + next!.EventNumber.ToInt64().Should().Be(1, "the event at version + 1"); + next.Position.CommitPosition.Should().BeGreaterThan(0UL); + snapshot!.Position.CommitPosition.Should().BeGreaterThan(0UL); + next.Position.Should().NotBe(Position.Start); + snapshot.Position.Should().NotBe(Position.Start); + next.Position.Should().BeLessThan(snapshot.Position, "the event at v+1 was committed before the snapshot"); + } + + private static EventRecord Record(ulong position, DateTime created) => + new("Tally-x", Uuid.NewUuid(), StreamPosition.FromInt64(0), new Position(position, position), + new Dictionary + { + ["type"] = "TallyAdded", + ["created"] = new DateTimeOffset(created).ToUnixTimeMilliseconds().ToString() + "0000", + ["content-type"] = "application/json" + }, + ReadOnlyMemory.Empty, ReadOnlyMemory.Empty); +} diff --git a/src/MicroPlumberd/Abstractions/IPlumber.cs b/src/MicroPlumberd/Abstractions/IPlumber.cs index d0d135b..b6c8ac9 100644 --- a/src/MicroPlumberd/Abstractions/IPlumber.cs +++ b/src/MicroPlumberd/Abstractions/IPlumber.cs @@ -53,6 +53,13 @@ public abstract record Snapshot /// Gets the version of the snapshot. /// public long Version { get; internal set; } + + /// + /// Whether is known to be the STREAM REVISION the snapshot covers (written by a build with #26/#28). + /// A snapshot without it was written by an earlier build, which counted only the events it could map: over a stream + /// holding events it did not know, its version is LOW (issue #28). + /// + public bool VersionIsRevision { get; internal set; } } /// diff --git a/src/MicroPlumberd/Conventions.cs b/src/MicroPlumberd/Conventions.cs index b948f9c..f759d5c 100644 --- a/src/MicroPlumberd/Conventions.cs +++ b/src/MicroPlumberd/Conventions.cs @@ -646,6 +646,13 @@ public static class MetadataExtensions return null; } + /// + /// Whether the snapshot's SnapshotVersion is a STREAM REVISION — written by a build where it is (issue #28). Absent on + /// snapshots of earlier builds, whose version counted only the events they could map. + /// + public static bool SnapshotVersionIsRevision(this Metadata m) => + m.Data.TryGetProperty("SnapshotVersionIsRevision", out var v) && v.ValueKind == System.Text.Json.JsonValueKind.True; + /// /// Extracts the correlation ID from metadata. /// diff --git a/src/MicroPlumberd/PlumberEngine.cs b/src/MicroPlumberd/PlumberEngine.cs index d0781b3..0444cbb 100644 --- a/src/MicroPlumberd/PlumberEngine.cs +++ b/src/MicroPlumberd/PlumberEngine.cs @@ -848,6 +848,9 @@ public async Task Get(OperationContext context, object id, Cance if (GetPolicy() != null && aggregate is IStatefull i) { var snapshot = await GetSnapshot(context,id, i.SnapshotType, token); + if (snapshot != null && !snapshot.VersionIsRevision && + await LegacySnapshotMayBeLow(context, typeof(TOwner), id, i.SnapshotType, snapshot.Version, token)) + snapshot = null; // rebuilt from the stream: always right, where reading on from snapshot.Version + 1 would re-fold if (snapshot != null) { i.Initialize(snapshot.Value, new StateInfo(snapshot.Version, snapshot.Created)); @@ -867,6 +870,54 @@ public async Task Get(OperationContext context, object id, Cance return aggregate; } + /// + /// Issue #28. A snapshot written before #26 recorded a COUNT of the events its build could map, not the stream revision: + /// over a stream holding events that build did not know, the version is LOW, and reading on from version + 1 re-folds + /// events already in the snapshot's state. Such a snapshot carries no marker, and its true revision cannot be recovered + /// from the number alone — the build that wrote it mapped a set of types this build may not share (it may know them now). + /// What CAN be told: the snapshot was appended after the events it covers. If the event at version + 1 was written + /// BEFORE the snapshot, the snapshot may cover it → may be low → do not use it (the caller rebuilds from the stream, which + /// is always right). Otherwise version + 1 was written after the snapshot: the version is the revision, use it. + /// (A writer that appended between another's save and its snapshot also reads as "may be low": a needless rebuild, never + /// a wrong state.) The order is the $all COMMIT POSITION of both records (), never a timestamp. + /// + private async Task LegacySnapshotMayBeLow(OperationContext context, Type owner, object id, Type snapshotType, long version, CancellationToken token) + { + var (next, snapshot) = await SnapshotOrderRecords(context, owner, id, snapshotType, version, token); + if (next == null) return false; // nothing after the snapshot's version: nothing to re-fold + if (snapshot == null) return true; + return !WrittenAfter(next, snapshot); + } + + /// + /// The two records the legacy-snapshot decision compares, read as reads them: the aggregate + /// stream's event at + 1, and the snapshot stream's latest event. Their $all Position comes + /// from the SERVER on a stream read (KurrentDB.Client builds it from the server's commit/prepare position); a row pins + /// that it is populated and in commit order — a zero would make every legacy snapshot "may be low" (rebuilt), silently. + /// + internal async Task<(EventRecord? Next, EventRecord? Snapshot)> SnapshotOrderRecords(OperationContext context, Type owner, object id, + Type snapshotType, long version, CancellationToken token) + { + var nextRead = Client.ReadStreamAsync(Direction.Forwards, Conventions.GetStreamIdConvention(context, owner, id), + StreamPosition.FromInt64(version + 1), maxCount: 1, resolveLinkTos: false, cancellationToken: token); + if (await nextRead.ReadState == ReadState.StreamNotFound) return (null, null); + var next = (await nextRead.FirstOrDefaultAsync(token)).Event; + if (next == null) return (null, null); + + var snapRead = Client.ReadStreamAsync(Direction.Backwards, Conventions.GetStreamIdSnapshotConvention(context, snapshotType, id), + StreamPosition.End, maxCount: 1, resolveLinkTos: false, cancellationToken: token); + if (await snapRead.ReadState == ReadState.StreamNotFound) return (next, null); + return (next, (await snapRead.FirstOrDefaultAsync(token)).Event); + } + + /// + /// Whether was committed after — by the $all COMMIT POSITION, which is + /// monotonic across streams. NOT Created (rev-reach a3f8d0c32): that is the server's wall clock, and a clock step + /// BACK between the write of the event at version + 1 and the snapshot's write would make a SHORT legacy snapshot look + /// as if its next event came later — the snapshot used, the event re-folded: the very state #28 fixes. + /// + internal static bool WrittenAfter(EventRecord later, EventRecord earlier) => later.Position > earlier.Position; + /// The revision of the last event a read went past, mapped or not. private sealed class LastRevisionRead { @@ -974,7 +1025,9 @@ await Client.AppendToStreamAsync(streamId, version == -1 ? StreamState.NoStream /// The write result indicating success or failure. public async Task AppendSnapshot(OperationContext context, object snapshot, object id, long version, StreamState? state = null, CancellationToken token = default) { - var m = Conventions.GetMetadata(context, null, snapshot, new { SnapshotVersion = version }); + // SnapshotVersionIsRevision (issue #28): the version is the aggregate's — since #26 the STREAM REVISION. A snapshot + // without the marker comes from an earlier build, whose version may be low; Get checks such a snapshot before using it. + var m = Conventions.GetMetadata(context, null, snapshot, new { SnapshotVersion = version, SnapshotVersionIsRevision = true }); var stateType = snapshot.GetType(); var streamId = Conventions.GetStreamIdSnapshotConvention(context, stateType, id); var evId = Conventions.GetEventIdConvention(context, null, snapshot); @@ -1312,6 +1365,7 @@ public async Task SaveNew(OperationContext context, T aggregate s.Created = m.Created().Value; s.Value = evt; s.Version = m.SnapshotVersion() ?? 0; + s.VersionIsRevision = m.SnapshotVersionIsRevision(); return s; }