From 9231d6d7f3a98826e34574901e947661b5e28b08 Mon Sep 17 00:00:00 2001 From: Rafal Maciag Date: Sun, 18 Oct 2026 16:49:55 +0000 Subject: [PATCH 1/3] =?UTF-8?q?fix:=20a=20snapshot=20written=20before=20#2?= =?UTF-8?q?6=20no=20longer=20re-folds=20events=20=E2=80=94=20new=20snapsho?= =?UTF-8?q?ts=20mark=20their=20version=20as=20the=20stream=20revision;=20a?= =?UTF-8?q?n=20unmarked=20one=20is=20used=20only=20when=20the=20event=20af?= =?UTF-8?q?ter=20it=20was=20written=20after=20it,=20else=20the=20aggregate?= =?UTF-8?q?=20is=20rebuilt=20(#28)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes #28 Co-Authored-By: Claude Opus 5.5 --- .../SnapshotDomain/TallyAggregate.cs | 30 +++ .../Integration/SnapshotRevisionTests.cs | 174 ++++++++++++++++++ src/MicroPlumberd/Abstractions/IPlumber.cs | 7 + src/MicroPlumberd/Conventions.cs | 7 + src/MicroPlumberd/PlumberEngine.cs | 34 +++- 5 files changed, 251 insertions(+), 1 deletion(-) create mode 100644 src/MicroPlumberd.Tests.App/SnapshotDomain/TallyAggregate.cs create mode 100644 src/MicroPlumberd.Tests/Integration/SnapshotRevisionTests.cs 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..8546ed8 --- /dev/null +++ b/src/MicroPlumberd.Tests/Integration/SnapshotRevisionTests.cs @@ -0,0 +1,174 @@ +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 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. + _plumber = new ServiceCollection().AddPlumberd(es.GetEventStoreSettings()).BuildServiceProvider() + .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"); + } +} 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..478f5ed 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,32 @@ 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 store's own commit timestamps on both streams. + /// + private async Task LegacySnapshotMayBeLow(OperationContext context, Type owner, object id, Type snapshotType, long version, CancellationToken token) + { + var next = Client.ReadStreamAsync(Direction.Forwards, Conventions.GetStreamIdConvention(context, owner, id), + StreamPosition.FromInt64(version + 1), maxCount: 1, resolveLinkTos: false, cancellationToken: token); + if (await next.ReadState == ReadState.StreamNotFound) return false; + var after = await next.FirstOrDefaultAsync(token); + if (after.Event == null) return false; // nothing after the snapshot's version: nothing to re-fold + + var snap = Client.ReadStreamAsync(Direction.Backwards, Conventions.GetStreamIdSnapshotConvention(context, snapshotType, id), + StreamPosition.End, maxCount: 1, resolveLinkTos: false, cancellationToken: token); + if (await snap.ReadState == ReadState.StreamNotFound) return true; + var written = await snap.FirstOrDefaultAsync(token); + return written.Event == null || after.Event.Created <= written.Event.Created; + } + /// The revision of the last event a read went past, mapped or not. private sealed class LastRevisionRead { @@ -974,7 +1003,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 +1343,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; } From 7a78bbab812b7f7f310e89ed36c1067ae3a22d9b Mon Sep 17 00:00:00 2001 From: Rafal Maciag Date: Sun, 18 Oct 2026 17:43:22 +0000 Subject: [PATCH 2/3] =?UTF-8?q?fix(#28):=20judge=20'written=20after'=20by?= =?UTF-8?q?=20the=20$all=20commit=20POSITION,=20not=20the=20server's=20wal?= =?UTF-8?q?l-clock=20Created=20(rev-reach=20a3f8d0c32=20m1)=20=E2=80=94=20?= =?UTF-8?q?a=20clock=20step=20back=20could=20make=20a=20short=20legacy=20s?= =?UTF-8?q?napshot=20look=20valid=20and=20re-fold?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 5.5 --- .../Integration/SnapshotRevisionTests.cs | 29 +++++++++++++++++++ src/MicroPlumberd/PlumberEngine.cs | 12 ++++++-- 2 files changed, 39 insertions(+), 2 deletions(-) diff --git a/src/MicroPlumberd.Tests/Integration/SnapshotRevisionTests.cs b/src/MicroPlumberd.Tests/Integration/SnapshotRevisionTests.cs index 8546ed8..1730b12 100644 --- a/src/MicroPlumberd.Tests/Integration/SnapshotRevisionTests.cs +++ b/src/MicroPlumberd.Tests/Integration/SnapshotRevisionTests.cs @@ -171,4 +171,33 @@ public async Task AMarkedSnapshot_ReadsThroughThePreMarkerReader_Unchanged() 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(); + } + + 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/PlumberEngine.cs b/src/MicroPlumberd/PlumberEngine.cs index 478f5ed..29a18f7 100644 --- a/src/MicroPlumberd/PlumberEngine.cs +++ b/src/MicroPlumberd/PlumberEngine.cs @@ -879,7 +879,7 @@ await LegacySnapshotMayBeLow(context, typeof(TOwner), id, i.SnapshotType, snapsh /// 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 store's own commit timestamps on both streams. + /// 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) { @@ -893,9 +893,17 @@ private async Task LegacySnapshotMayBeLow(OperationContext context, Type o StreamPosition.End, maxCount: 1, resolveLinkTos: false, cancellationToken: token); if (await snap.ReadState == ReadState.StreamNotFound) return true; var written = await snap.FirstOrDefaultAsync(token); - return written.Event == null || after.Event.Created <= written.Event.Created; + return written.Event == null || !WrittenAfter(after.Event, written.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 { From 328c0dc74434f51c2c5525015e924f3eabb19490 Mon Sep 17 00:00:00 2001 From: Rafal Maciag Date: Sun, 18 Oct 2026 17:52:52 +0000 Subject: [PATCH 3/3] =?UTF-8?q?test(#28):=20pin=20the=20INPUT=20of=20the?= =?UTF-8?q?=20position=20decision=20=E2=80=94=20through=20the=20real=20rea?= =?UTF-8?q?d=20path=20(SnapshotOrderRecords),=20both=20records=20carry=20s?= =?UTF-8?q?erver=20$all=20positions,=20non-zero,=20in=20commit=20order=20(?= =?UTF-8?q?lead=20+=20rev-reach=202883377b9)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 5.5 --- .../Integration/SnapshotRevisionTests.cs | 31 +++++++++++++++++-- src/MicroPlumberd/PlumberEngine.cs | 30 +++++++++++++----- 2 files changed, 51 insertions(+), 10 deletions(-) diff --git a/src/MicroPlumberd.Tests/Integration/SnapshotRevisionTests.cs b/src/MicroPlumberd.Tests/Integration/SnapshotRevisionTests.cs index 1730b12..6b4231a 100644 --- a/src/MicroPlumberd.Tests/Integration/SnapshotRevisionTests.cs +++ b/src/MicroPlumberd.Tests/Integration/SnapshotRevisionTests.cs @@ -21,14 +21,16 @@ namespace MicroPlumberd.Tests.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. - _plumber = new ServiceCollection().AddPlumberd(es.GetEventStoreSettings()).BuildServiceProvider() - .GetRequiredService(); + var sp = new ServiceCollection().AddPlumberd(es.GetEventStoreSettings()).BuildServiceProvider(); + _plumber = sp.GetRequiredService(); + _engine = sp.GetRequiredService(); _es = es; } @@ -191,6 +193,31 @@ public void WrittenAfter_IsDecidedByCommitPosition_NotByCreated() 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 diff --git a/src/MicroPlumberd/PlumberEngine.cs b/src/MicroPlumberd/PlumberEngine.cs index 29a18f7..0444cbb 100644 --- a/src/MicroPlumberd/PlumberEngine.cs +++ b/src/MicroPlumberd/PlumberEngine.cs @@ -883,17 +883,31 @@ await LegacySnapshotMayBeLow(context, typeof(TOwner), id, i.SnapshotType, snapsh /// private async Task LegacySnapshotMayBeLow(OperationContext context, Type owner, object id, Type snapshotType, long version, CancellationToken token) { - var next = Client.ReadStreamAsync(Direction.Forwards, Conventions.GetStreamIdConvention(context, owner, id), + 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 next.ReadState == ReadState.StreamNotFound) return false; - var after = await next.FirstOrDefaultAsync(token); - if (after.Event == null) return false; // nothing after the snapshot's version: nothing to re-fold + if (await nextRead.ReadState == ReadState.StreamNotFound) return (null, null); + var next = (await nextRead.FirstOrDefaultAsync(token)).Event; + if (next == null) return (null, null); - var snap = Client.ReadStreamAsync(Direction.Backwards, Conventions.GetStreamIdSnapshotConvention(context, snapshotType, id), + var snapRead = Client.ReadStreamAsync(Direction.Backwards, Conventions.GetStreamIdSnapshotConvention(context, snapshotType, id), StreamPosition.End, maxCount: 1, resolveLinkTos: false, cancellationToken: token); - if (await snap.ReadState == ReadState.StreamNotFound) return true; - var written = await snap.FirstOrDefaultAsync(token); - return written.Event == null || !WrittenAfter(after.Event, written.Event); + if (await snapRead.ReadState == ReadState.StreamNotFound) return (next, null); + return (next, (await snapRead.FirstOrDefaultAsync(token)).Event); } ///