feat(worker): versioned OnWriteComplete completion cache

2026-08-09 12:21:51 -04:00
parent aec95b78c9
commit fc23a65cca
2 changed files with 297 additions and 0 deletions
@@ -0,0 +1,145 @@
+using System;
+using Google.Protobuf.Collections;
+using ZB.MOM.WW.MxGateway.Contracts.Proto;
+using ZB.MOM.WW.MxGateway.Worker.MxAccess;
+
+namespace ZB.MOM.WW.MxGateway.Worker.Tests.MxAccess;
+
+/// <summary>
+///     Unit tests for <see cref="MxAccessWriteCompletionCache"/>. The cache is
+///     consumed by the write command executor's bounded pump-wait so a
+///     WriteSecured/WriteSecured2 reply can carry the correlated
+///     OnWriteComplete outcome; its version-baseline contract is exercised in
+///     isolation here before the STA / COM plumbing gets layered on top.
+/// </summary>
+public sealed class MxAccessWriteCompletionCacheTests
+{
+    /// <summary>Verifies that Record bumps the version per key and keys stay isolated.</summary>
+    [Fact]
+    public void Record_IncrementsVersionPerKey()
+    {
+        MxAccessWriteCompletionCache cache = new();
+
+        Assert.Equal(0UL, cache.CurrentVersion(7, 21));
+
+        cache.Record(7, 21, BuildStatuses(detail: 100));
+        Assert.Equal(1UL, cache.CurrentVersion(7, 21));
+
+        cache.Record(7, 21, BuildStatuses(detail: 200));
+        Assert.Equal(2UL, cache.CurrentVersion(7, 21));
+
+        cache.Record(7, 22, BuildStatuses(detail: 300));
+        Assert.Equal(1UL, cache.CurrentVersion(7, 22));
+        Assert.Equal(2UL, cache.CurrentVersion(7, 21));
+    }
+
+    /// <summary>Verifies that a completion newer than the baseline is returned with its status rows.</summary>
+    [Fact]
+    public void TryWaitForCompletion_WhenCompletionNewerThanBaseline_ReturnsStatuses()
+    {
+        MxAccessWriteCompletionCache cache = new();
+        cache.Record(7, 21, BuildStatuses(detail: 4321));
+
+        bool found = cache.TryWaitForCompletion(
+            7,
+            21,
+            sinceVersion: 0UL,
+            deadlineUtc: DateTime.UtcNow.AddSeconds(5),
+            pumpStep: static () => { },
+            out RepeatedField<MxStatusProxy> statuses);
+
+        Assert.True(found);
+        MxStatusProxy row = Assert.Single(statuses);
+        Assert.Equal(4321, row.Detail);
+        Assert.Equal(MxStatusCategory.Ok, row.Category);
+    }
+
+    /// <summary>
+    ///     Verifies that a completion recorded before the baseline was captured is
+    ///     never misattributed to the waiting write: only a strictly newer version
+    ///     satisfies the wait, so a stale row times the wait out.
+    /// </summary>
+    [Fact]
+    public void TryWaitForCompletion_WhenOnlyStaleCompletion_TimesOut()
+    {
+        MxAccessWriteCompletionCache cache = new();
+        cache.Record(7, 21, BuildStatuses(detail: 4321));
+        ulong baseline = cache.CurrentVersion(7, 21);
+
+        bool found = cache.TryWaitForCompletion(
+            7,
+            21,
+            sinceVersion: baseline,
+            deadlineUtc: DateTime.UtcNow.AddMilliseconds(50),
+            pumpStep: static () => { },
+            out RepeatedField<MxStatusProxy> statuses);
+
+        Assert.False(found);
+        Assert.Empty(statuses);
+    }
+
+    /// <summary>
+    ///     Verifies the pump loop is what lets a completion land: the completion is
+    ///     recorded from inside a later pump step (standing in for the STA
+    ///     dispatching the OnWriteComplete message) and the wait then succeeds.
+    /// </summary>
+    [Fact]
+    public void TryWaitForCompletion_InvokesPumpStepEachIteration()
+    {
+        MxAccessWriteCompletionCache cache = new();
+        int pumpCalls = 0;
+
+        bool found = cache.TryWaitForCompletion(
+            7,
+            21,
+            sinceVersion: 0UL,
+            deadlineUtc: DateTime.UtcNow.AddSeconds(5),
+            pumpStep: () =>
+            {
+                pumpCalls++;
+                if (pumpCalls == 2)
+                {
+                    cache.Record(7, 21, BuildStatuses(detail: 55));
+                }
+            },
+            out RepeatedField<MxStatusProxy> statuses);
+
+        Assert.True(found);
+        Assert.True(pumpCalls >= 2);
+        Assert.Equal(55, Assert.Single(statuses).Detail);
+    }
+
+    /// <summary>Verifies that Record stores an independent clone of the caller's rows.</summary>
+    [Fact]
+    public void Record_ClonesStatuses()
+    {
+        MxAccessWriteCompletionCache cache = new();
+        RepeatedField<MxStatusProxy> callerRows = BuildStatuses(detail: 77);
+
+        cache.Record(7, 21, callerRows);
+        callerRows[0].Detail = 999;
+        callerRows.Add(new MxStatusProxy());
+
+        Assert.True(cache.TryWaitForCompletion(
+            7,
+            21,
+            sinceVersion: 0UL,
+            deadlineUtc: DateTime.UtcNow.AddSeconds(5),
+            pumpStep: static () => { },
+            out RepeatedField<MxStatusProxy> statuses));
+        Assert.Equal(77, Assert.Single(statuses).Detail);
+    }
+
+    private static RepeatedField<MxStatusProxy> BuildStatuses(int detail)
+    {
+        return new RepeatedField<MxStatusProxy>
+        {
+            new MxStatusProxy
+            {
+                Success = 1,
+                Category = MxStatusCategory.Ok,
+                Detail = detail,
+            },
+        };
+    }
+}