Toy server: support the maintenance-notification opt-in, and sending events

Deliberately on RedisServer rather than a bespoke test subclass. The opt-in is an
ordinary command, so any test should be able to use it, and a server that never
sends a notification is the normal case - every OSS, Valkey and Garnet build
behaves that way, and so will our own docker topology. The interesting behaviour
is all client-side, so the fake should not be a special place.

CLIENT MAINT_NOTIFICATIONS <ON|OFF> [parameter value ...] per the contract: a
bare ON is valid and means "server defaults", moving-endpoint-type is the only
parameter defined so far, and its five values are validated. Unknown parameters
are refused rather than ignored - the client is asking the server to do something
specific, and silently not doing it is worse than saying no. State is per
connection, including a count, since re-arming after a reconnect is a requirement
and a count is what distinguishes that from having opted in once.

MaintenanceNotifications selects how the server answers: accept, reject as an
unknown subcommand (what a server that never heard of it does), or reject as
disabled (what one with the feature flag off does). A client has to survive all
three, so a test has to be able to ask for all three.

Sending covers MOVING with or without an address, the shard-scoped and
slot-scoped families, explicit or generated sequence ids - the contract never
defines those, so repeating one deliberately is part of what the fake owes us -
and a raw-push hook for malformed frames. Notifications go only to connections
that opted in; a raw push is not gated, which is the contrast the tests assert.

Two bugs the tests caught: the count returned was clients *visited* rather than
sent to, because ForAllClients' Action overload returns one per client
regardless; and an assertion comparing against ClientCount was racy, since under
RESP2 the subscription connection can register between the send and the read.

Author: mgravell

tests/StackExchange.Redis.Tests/MaintenanceOptInServerTests.cs

@@ -0,0 +1,159 @@
+using System;
+using System.Linq;
+using System.Threading.Tasks;
+using Xunit;
+using static StackExchange.Redis.Server.RedisServer;
+
+namespace StackExchange.Redis.Tests;
+
+/// <summary>
+/// The server half of the maintenance-notification contract, exercised directly. The client does not opt in
+/// yet, so these drive the command as a caller would - which is also how any other test will be able to opt
+/// in once the client does, since this is ordinary server functionality rather than a special test server.
+/// </summary>
+public class MaintenanceOptInServerTests(ITestOutputHelper log)
+{
+    private static InProcessTestServer CreateServer(ITestOutputHelper log) => new(log);
+
+    private static async Task<RedisResult> OptInAsync(IConnectionMultiplexer conn, InProcessTestServer server, params object[] args)
+        => await conn.GetServer(server.DefaultEndPoint).ExecuteAsync("client", args.Prepend("maint_notifications").ToArray());
+
+    [Fact]
+    public async Task BareOnIsAcceptedAndRecorded()
+    {
+        // "CLIENT MAINT_NOTIFICATIONS ON" with no parameters is explicitly valid, and means "server defaults"
+        using var server = CreateServer(log);
+        await using var conn = await server.ConnectAsync(defaultOnly: true);
+
+        Assert.Equal("OK", (string?)await OptInAsync(conn, server, "on"));
+
+        var client = Assert.Single(OptedIn(server));
+        Assert.Null(client.MovingEndpointType); // server defaults, not a value we invented
+        Assert.Equal(1, client.MaintenanceNotificationOptInCount);
+    }
+
+    [Theory]
+    [InlineData("internal-ip")]
+    [InlineData("internal-fqdn")]
+    [InlineData("external-ip")]
+    [InlineData("external-fqdn")]
+    [InlineData("none")]
+    public async Task EveryDefinedEndpointTypeIsAccepted(string endpointType)
+    {
+        using var server = CreateServer(log);
+        await using var conn = await server.ConnectAsync(defaultOnly: true);
+
+        Assert.Equal("OK", (string?)await OptInAsync(conn, server, "on", "moving-endpoint-type", endpointType));
+        Assert.Equal(endpointType, Assert.Single(OptedIn(server)).MovingEndpointType);
+    }
+
+    [Fact]
+    public async Task OffClearsTheOptIn()
+    {
+        using var server = CreateServer(log);
+        await using var conn = await server.ConnectAsync(defaultOnly: true);
+
+        await OptInAsync(conn, server, "on", "moving-endpoint-type", "external-fqdn");
+        Assert.Single(OptedIn(server));
+
+        Assert.Equal("OK", (string?)await OptInAsync(conn, server, "off"));
+        Assert.Empty(OptedIn(server));
+    }
+
+    [Theory]
+    [InlineData("sideways")]                              // not on/off
+    [InlineData("on", "moving-endpoint-type", "sideways")] // undefined endpoint type
+    [InlineData("on", "not-a-parameter", "value")]         // unknown parameter
+    [InlineData("on", "moving-endpoint-type")]             // parameter with no value
+    public async Task MalformedOptInIsRejected(params string[] args)
+    {
+        using var server = CreateServer(log);
+        await using var conn = await server.ConnectAsync(defaultOnly: true);
+
+        var ex = await Assert.ThrowsAsync<RedisServerException>(
+            async () => await OptInAsync(conn, server, args.Cast<object>().ToArray()));
+        log.WriteLine(ex.Message);
+        Assert.Empty(OptedIn(server));
+    }
+
+    [Theory]
+    [InlineData(MaintenanceNotificationSupport.UnknownSubcommand)]
+    [InlineData(MaintenanceNotificationSupport.Disabled)]
+    public async Task UnsupportingServerRejectsTheOptIn(MaintenanceNotificationSupport support)
+    {
+        // the two ways a real server refuses: it has never heard of the subcommand (OSS, Valkey, Garnet), or
+        // it knows it and has the feature flag off. A client has to survive both
+        using var server = CreateServer(log);
+        server.MaintenanceNotifications = support;
+        await using var conn = await server.ConnectAsync(defaultOnly: true);
+
+        var ex = await Assert.ThrowsAsync<RedisServerException>(
+            async () => await OptInAsync(conn, server, "on"));
+        log.WriteLine($"{support}: {ex.Message}");
+        Assert.Empty(OptedIn(server));
+    }
+
+    [Fact]
+    public async Task NotificationsGoOnlyToConnectionsThatOptedIn()
+    {
+        // a real server sends to the connections that asked; sending to everything would let a client pass a
+        // test it should fail, by receiving notifications it never subscribed to
+        using var server = CreateServer(log);
+        await using var optedIn = await server.ConnectAsync(defaultOnly: true);
+        await using var notOptedIn = await server.ConnectAsync(defaultOnly: true);
+
+        await OptInAsync(optedIn, server, "on", "moving-endpoint-type", "external-fqdn");
+        var subscribed = OptedIn(server).Count();
+        log.WriteLine($"{subscribed} of {server.ClientCount} connections opted in");
+
+        var sent = server.SendShardNotification(null, MaintenanceNotificationKind.Migrating, timeSeconds: 5);
+        Assert.Equal(subscribed, sent);
+        Assert.NotEqual(server.ClientCount, sent);
+    }
+
+    [Fact]
+    public async Task SequenceIdsAdvanceAndCanBeRepeated()
+    {
+        // the contract never defines these, so a client's use of them is its own invention - which means being
+        // able to repeat one deliberately is part of what the fake owes us
+        using var server = CreateServer(log);
+        await using var conn = await server.ConnectAsync(defaultOnly: true);
+        await OptInAsync(conn, server, "on");
+
+        var first = server.NextMaintenanceSequenceId;
+        Assert.Equal(1, server.SendShardNotification(null, MaintenanceNotificationKind.Migrating, 5));
+        Assert.Equal(first + 1, server.NextMaintenanceSequenceId);
+
+        // and an explicit id does not advance the counter, so a replay stays a replay
+        Assert.Equal(1, server.SendShardNotification(null, MaintenanceNotificationKind.Migrating, 5, sequenceId: first));
+        Assert.Equal(first + 1, server.NextMaintenanceSequenceId);
+    }
+
+    [Fact]
+    public async Task RawPushIsNotGatedByOptIn()
+    {
+        // the malformed-payload hook, and the contrast is the point: with nobody opted in, a notification
+        // reaches no one while a raw push still lands. Asserted as gated-versus-not rather than against
+        // ClientCount, which is read at a different instant - under RESP2 the subscription connection can
+        // register in between, so comparing totals is a race rather than a property
+        using var server = CreateServer(log);
+        await using var conn = await server.ConnectAsync(defaultOnly: true);
+
+        var gated = server.SendShardNotification(null, MaintenanceNotificationKind.Migrating, timeSeconds: 5);
+        var ungated = server.SendRawPush(null, "NOT_A_REAL_KIND", "nonsense");
+        log.WriteLine($"notification reached {gated}, raw push reached {ungated}");
+
+        Assert.Equal(0, gated);
+        Assert.True(ungated > 0, "a raw push should reach connections that never opted in");
+    }
+
+    private static System.Collections.Generic.IEnumerable<Server.RedisClient> OptedIn(InProcessTestServer server)
+    {
+        var found = new System.Collections.Generic.List<Server.RedisClient>();
+        server.ForAllClients(c =>
+        {
+            if (c.MaintenanceNotifications) found.Add(c);
+        });
+        return found;
+    }
+}

toys/StackExchange.Redis.Server/RedisClient.cs

@@ -99,6 +99,22 @@ public bool TryReadRequest(ReadOnlySequence<byte> data, out long consumed)
         }
 
         public RedisServer.Node Node => node;
+
+        /// <summary>
+        /// Whether this connection has opted in to maintenance notifications, and with which endpoint type -
+        /// per connection, because that is how the real opt-in is scoped, so a test can assert that every
+        /// connection opted in rather than merely that one did.
+        /// </summary>
+        public bool MaintenanceNotifications { get; internal set; }
+
+        /// <summary>The <c>moving-endpoint-type</c> this connection asked for, or null for server defaults.</summary>
+        public string MovingEndpointType { get; internal set; }
+
+        /// <summary>
+        /// How many times this connection has sent the opt-in. Re-arming on reconnect is a requirement, and a
+        /// count is what distinguishes "opted in once" from "opted in again after reconnecting".
+        /// </summary>
+        public int MaintenanceNotificationOptInCount { get; internal set; }
         public int SkipReplies { get; set; }
         public void SkipAllReplies() => SkipReplies = -1;
         internal bool ShouldSkipResponse()

toys/StackExchange.Redis.Server/RedisServer.Maintenance.cs

@@ -0,0 +1,160 @@
+using System;
+using System.Net;
+using RESPite;
+using RESPite.Messages;
+
+namespace StackExchange.Redis.Server
+{
+    /// <summary>
+    /// Sending maintenance notifications. Deliberately on the server itself rather than on a bespoke test
+    /// subclass: the opt-in is a normal command any test may use, and injecting a notification is a normal
+    /// thing any test may want to do. A server that never sends one is the ordinary case - that is what every
+    /// OSS build does - so the interesting behaviour is on the client side either way.
+    /// </summary>
+    public partial class RedisServer
+    {
+        /// <summary>
+        /// The notification types defined by the maintenance-notification contract. <c>SMOVING</c> and
+        /// <c>SFAILING_OVER</c> were proposed upstream but never landed, and are deliberately absent.
+        /// </summary>
+        public enum MaintenanceNotificationKind
+        {
+            /// <summary>This endpoint is being replaced; the payload names its successor.</summary>
+            Moving,
+
+            /// <summary>A shard is migrating away from this node.</summary>
+            Migrating,
+
+            /// <summary>The migration has completed.</summary>
+            Migrated,
+
+            /// <summary>This node is failing over.</summary>
+            FailingOver,
+
+            /// <summary>The failover has completed.</summary>
+            FailedOver,
+
+            /// <summary>Slots are migrating (OSS cluster family).</summary>
+            SlotMigrating,
+
+            /// <summary>Slots have migrated (OSS cluster family).</summary>
+            SlotMigrated,
+        }
+
+        private static string GetName(MaintenanceNotificationKind kind) => kind switch
+        {
+            MaintenanceNotificationKind.Moving => "MOVING",
+            MaintenanceNotificationKind.Migrating => "MIGRATING",
+            MaintenanceNotificationKind.Migrated => "MIGRATED",
+            MaintenanceNotificationKind.FailingOver => "FAILING_OVER",
+            MaintenanceNotificationKind.FailedOver => "FAILED_OVER",
+            MaintenanceNotificationKind.SlotMigrating => "SMIGRATING",
+            MaintenanceNotificationKind.SlotMigrated => "SMIGRATED",
+            _ => throw new ArgumentOutOfRangeException(nameof(kind)),
+        };
+
+        private int _maintenanceSequence;
+
+        /// <summary>
+        /// The sequence id given to the next notification, unless one is supplied explicitly. The contract does
+        /// not define these, so a client's use of them is its own invention - which is worth being able to
+        /// exercise, including by repeating one.
+        /// </summary>
+        public int NextMaintenanceSequenceId => _maintenanceSequence + 1;
+
+        /// <summary>
+        /// Sends <c>MOVING</c> to one client, or to every client that opted in when <paramref name="client"/>
+        /// is null. A null <paramref name="newEndpoint"/> is the documented no-address form, which a client
+        /// must handle whether or not it asked for <c>none</c>.
+        /// </summary>
+        /// <returns>The number of clients the notification was sent to.</returns>
+        public int SendMoving(RedisClient client, int timeSeconds, EndPoint newEndpoint, int? sequenceId = null)
+            => Send(client, MaintenanceNotificationKind.Moving, timeSeconds, sequenceId, newEndpoint, null);
+
+        /// <summary>
+        /// Sends one of the shard-scoped notifications. <paramref name="timeSeconds"/> is a remaining-time
+        /// delta and may legitimately be zero or negative for a connection that arrived mid-window.
+        /// </summary>
+        /// <returns>The number of clients the notification was sent to.</returns>
+        public int SendShardNotification(RedisClient client, MaintenanceNotificationKind kind, int timeSeconds, string shardIds = null, int? sequenceId = null)
+            => Send(client, kind, timeSeconds, sequenceId, null, shardIds);
+
+        /// <summary>
+        /// Sends a slot-scoped notification (<c>SMIGRATING</c> / <c>SMIGRATED</c>) carrying a slot list in the
+        /// contract's comma-and-range form, e.g. <c>"123,456,789-1000"</c>.
+        /// </summary>
+        /// <returns>The number of clients the notification was sent to.</returns>
+        public int SendSlotNotification(RedisClient client, MaintenanceNotificationKind kind, string slots, int? sequenceId = null)
+            => Send(client, kind, null, sequenceId, null, slots);
+
+        /// <summary>
+        /// Sends an arbitrary push frame to a client, for the cases a well-formed notification cannot express:
+        /// an unknown type, a malformed payload, extra trailing elements.
+        /// </summary>
+        /// <returns>The number of clients the frame was sent to.</returns>
+        public int SendRawPush(RedisClient client, params string[] parts)
+        {
+            var frame = TypedRedisValue.Rent(parts.Length, out var span, RespPrefix.Push);
+            for (int i = 0; i < parts.Length; i++)
+            {
+                span[i] = TypedRedisValue.BulkString(parts[i]);
+            }
+            return Dispatch(client, frame, requireOptIn: false);
+        }
+
+        private int Send(
+            RedisClient client,
+            MaintenanceNotificationKind kind,
+            int? timeSeconds,
+            int? sequenceId,
+            EndPoint newEndpoint,
+            string extra)
+        {
+            // [type, seqID, ...] - the sequence id is an integer, which is precisely why these frames could
+            // not be treated as pub/sub: element 1 is not a channel name
+            int count = 2 + (timeSeconds.HasValue ? 1 : 0) + (newEndpoint is not null || kind == MaintenanceNotificationKind.Moving ? 1 : 0) + (extra is not null ? 1 : 0);
+            var frame = TypedRedisValue.Rent(count, out var span, RespPrefix.Push);
+
+            int index = 0;
+            span[index++] = TypedRedisValue.SimpleString(GetName(kind));
+            span[index++] = TypedRedisValue.Integer(sequenceId ?? System.Threading.Interlocked.Increment(ref _maintenanceSequence));
+            if (timeSeconds.HasValue) span[index++] = TypedRedisValue.Integer(timeSeconds.GetValueOrDefault());
+            if (kind == MaintenanceNotificationKind.Moving)
+            {
+                // null rather than absent when there is no address: the client must cope with both
+                span[index++] = newEndpoint is null
+                    ? TypedRedisValue.BulkString(RedisValue.Null)
+                    : TypedRedisValue.BulkString(Format.ToString(newEndpoint));
+            }
+            else if (newEndpoint is not null)
+            {
+                span[index++] = TypedRedisValue.BulkString(Format.ToString(newEndpoint));
+            }
+            if (extra is not null) span[index] = TypedRedisValue.BulkString(extra);
+
+            return Dispatch(client, frame, requireOptIn: true);
+        }
+
+        private int Dispatch(RedisClient client, in TypedRedisValue frame, bool requireOptIn)
+        {
+            if (client is not null)
+            {
+                client.AddOutbound(frame);
+                return 1;
+            }
+
+            // a real server sends only to connections that asked, so sending to all means all *opted-in*.
+            // Counting the sends rather than the clients visited: the Action overload of ForAllClients returns
+            // one per client regardless, which would report every connection as a recipient
+            var copy = frame;
+            return ForAllClients(
+                requireOptIn,
+                (target, gated) =>
+                {
+                    if (gated && !target.MaintenanceNotifications) return 0;
+                    target.AddOutbound(copy);
+                    return 1;
+                });
+        }
+    }
+}