Resolve critical failure in high-integrity-mode with -MOVED response (#3049)

* failing test for #3048

* Critical fix for high-integrity-mode re -MOVED

* release notes

Author: mgravell

docs/ReleaseNotes.md

@@ -8,7 +8,7 @@ Current package versions:
 
 ## Unreleased
 
-- none
+- Critical fix for high-integrity mode in cluster scenarios ([#3049 by @mgravell](https://github.com/StackExchange/StackExchange.Redis/issues/3049))
 
 ## 2.12.8
 

src/StackExchange.Redis/PhysicalConnection.cs

@@ -1914,19 +1914,25 @@ private void MatchResult(in RawResult result)
 
             Trace("Response to: " + msg);
             _readStatus = ReadStatus.ComputeResult;
-            if (msg.ComputeResult(this, result))
+
+            // need to capture HIT promptly, as -MOVED could cause a resend with a new high-integrity token
+            // (a lazy approach would be to not rotate, but: we'd rather avoid that; the -MOVED case is rare)
+            var highIntegrityToken = msg.HighIntegrityToken;
+
+            bool computed = msg.ComputeResult(this, result);
+            if (computed)
             {
                 _readStatus = msg.ResultBoxIsAsync ? ReadStatus.CompletePendingMessageAsync : ReadStatus.CompletePendingMessageSync;
-                if (!msg.IsHighIntegrity)
+                if (highIntegrityToken == 0)
                 {
                     // can't complete yet if needs checksum
                     msg.Complete();
                 }
             }
-            if (msg.IsHighIntegrity)
+            if (highIntegrityToken != 0)
             {
-                // stash this for the next non-OOB response
-                Volatile.Write(ref _awaitingToken, msg);
+                // stash this for the next non-OOB response, retaining the old HIT iff we had a -MOVED etc
+                Volatile.Write(ref _awaitingToken, computed ? msg : new DummyHighIntegrityMessage(msg, highIntegrityToken));
             }
 
             _readStatus = ReadStatus.MatchResultComplete;
@@ -2455,4 +2461,17 @@ internal bool HasPendingCallerFacingItems()
             }
         }
     }
+
+    internal sealed class DummyHighIntegrityMessage : Message
+    {
+        // note: we don't create this message very often - only when a HIT gets a -MOVED or similar
+        public DummyHighIntegrityMessage(Message msg, uint highIntegrityToken) : base(msg.Db, msg.Flags, msg.Command)
+        {
+            WithHighIntegrity(highIntegrityToken);
+        }
+
+        public override int ArgCount => 0;
+        protected override void WriteImpl(PhysicalConnection physical)
+            => throw new NotSupportedException("This message cannot be written; it is a place-holder for high-integrity scenarios.");
+    }
 }

tests/StackExchange.Redis.Tests/HighIntegrityMovedUnitTests.cs

@@ -0,0 +1,40 @@
+using System.Threading.Tasks;
+using Xunit;
+
+namespace StackExchange.Redis.Tests;
+
+public class HighIntegrityMovedUnitTests(ITestOutputHelper log)
+{
+    [Theory]
+    [InlineData(false)]
+    [InlineData(true)]
+    public async Task HighIntegritySurvivesMovedResponse(bool highIntegrity)
+    {
+        using var server = new InProcessTestServer(log) { ServerType = ServerType.Cluster };
+        var secondary = server.AddEmptyNode();
+
+        var config = server.GetClientConfig();
+        config.HighIntegrity = highIntegrity;
+        using var client = await ConnectionMultiplexer.ConnectAsync(config);
+
+        RedisKey a = "a", b = "b"; // known to be in different slots
+        Assert.NotEqual(ServerSelectionStrategy.GetClusterSlot(a), ServerSelectionStrategy.GetClusterSlot(b));
+
+        var db = client.GetDatabase();
+        var x = db.StringIncrementAsync(a);
+        var y = db.StringIncrementAsync(b);
+        await x;
+        await y;
+        Assert.Equal(1, await db.StringGetAsync(a));
+        Assert.Equal(1, await db.StringGetAsync(b));
+
+        // now force a -MOVED response
+        server.Migrate(a, secondary);
+        x = db.StringIncrementAsync(a);
+        y = db.StringIncrementAsync(b);
+        await x;
+        await y;
+        Assert.Equal(2, await db.StringGetAsync(a));
+        Assert.Equal(2, await db.StringGetAsync(b));
+    }
+}