Use a dedicated pub/sub connection under RESP3 by default

docs/Configuration.md

@@ -100,6 +100,7 @@ The `ConfigurationOptions` object has a wide range of properties, all of which a
 | tunnel={string}        | `Tunnel`               | `null`                       | Tunnel for connections (use `http:{proxy url}` for "connect"-based proxy server)                          |
 | setlib={bool}          | `SetClientLibrary`     | `true`                       | Whether to attempt to use `CLIENT SETINFO` to set the library name/version on the connection              |
 | protocol={string}      | `Protocol`             | `null`                       | Redis protocol to use; see section below                                                                  |
+| sharedSubscriptionConnection={bool} | `SharedSubscriptionConnection` | `false` | Under RESP3, share the interactive connection with pub/sub rather than using a dedicated one; see section below |
 | highIntegrity={bool}   | `HighIntegrity`        | `false`                      | High integrity (incurs overhead) sequence checking on every command; see section below                    |
 | defaults={string}      | `Defaults`             | `null`                       | Selects a named defaults provider; see section below                                                      |
 | maintNotifications={string} | `MaintenanceNotifications` | `disabled`           | Whether to ask servers for maintenance notifications; see section below                                    |
@@ -266,7 +267,7 @@ Both options can be customized or disabled (set to `""`), via the `.Configuratio
 
 These settings are also used by the `IServer.MakeMaster()` method, which can set the tie-breaker in the database and broadcast the configuration change message. The configuration message can also be used separately to primary/replica changes simply to request all nodes to refresh their configurations, via the `ConnectionMultiplexer.PublishReconfigure` method.
 
-The configuration channel is subscribed to on every server connection, under both RESP2 and RESP3. Like any other pub/sub channel, it has the `ChannelPrefix` (if one is configured) applied - both when subscribing to it and when the library publishes to it - so a client with `channelPrefix=app1-` listens on `app1-__Booksleeve_MasterChanged`. Two consequences follow:
+The configuration channel is subscribed to on every server, under both RESP2 and RESP3 - on the subscription connection, or on the interactive connection when it is shared with pub/sub (`SharedSubscriptionConnection`); this is enough to make the server treat that connection as a pub/sub client. Like any other pub/sub channel, it has the `ChannelPrefix` (if one is configured) applied - both when subscribing to it and when the library publishes to it - so a client with `channelPrefix=app1-` listens on `app1-__Booksleeve_MasterChanged`. Two consequences follow:
 
 - If you announce a change yourself rather than through the library (for example `PUBLISH app1-__Booksleeve_MasterChanged "*"` from `redis-cli`), publish to the *prefixed* name, once for each distinct prefix in use; clients with a different prefix will not hear it. A publish to the unprefixed name is only heard by clients that have no prefix.
 - If you use ACLs that restrict channels, the user needs access to the *prefixed* channel (for example `&app1-__Booksleeve_MasterChanged`), in addition to any channels the application itself uses.
@@ -394,9 +395,13 @@ saying so.
 
 ## Redis protocol
 
-RESP3 is a newer protocol (available on v6 servers and above) which allows (among other changes) pub/sub messages to be communicated on the *same* connection - which can be very
-desirable in servers with a large number of clients; under RESP2, pub/sub requires a separate connection to the server. The protocol handshake needs to happen very early in the
-connection, so the library only attempts RESP3 when it has reason to expect it to work.
+RESP3 is a newer protocol (available on v6 servers and above) which allows (among other changes) pub/sub messages to be communicated on the *same* connection - which can be
+desirable in servers with a large number of clients; under RESP2, pub/sub requires a separate connection to the server. The library still uses a separate pub/sub connection under
+RESP3 unless `SharedSubscriptionConnection` is enabled: the server applies its (much tighter) `pubsub` output-buffer limits to any connection with an active subscription, which
+can get a shared connection - and every command in flight on it - closed. Before enabling it, check that the server's `pubsub` limits (`CONFIG GET client-output-buffer-limit`)
+are suitable for your largest replies, not just your pub/sub messages; see [RESP3](Resp3#pubsub-connections) for details.
+
+The protocol handshake needs to happen very early in the connection, so the library only attempts RESP3 when it has reason to expect it to work.
 
 The library determines whether to use RESP3 by:
 - The `HELLO` command has been disabled: RESP2 is used

docs/Resp3.md

@@ -3,7 +3,8 @@
 RESP2 and RESP3 are evolutions of the Redis protocol, with RESP3 existing from Redis server version 6 onwards (v7.2+ for Redis Enterprise). The main differences are:
 
 1. RESP3 can carry out-of-band / "push" messages on a single connection, where-as RESP2 requires a separate connection for out-of-band (pub/sub) messages
-    - this single connection can be of huge benefit in high-usage servers, as it halves the number of connections required
+    - SE.Redis still uses a separate pub/sub connection by default under RESP3, because of how servers limit subscribed connections; sharing a single connection
+      is opt-in - see [Pub/sub connections](#pubsub-connections) below
 2. RESP3 supports *additional* out-of-band messages that cannot be expressed in RESP2, which allows advanced features such as "smart client handoffs" (also called
    "hitless upgrades"; a family of server maintenance notifications)
     - these features allow for greater stability in complex deployments, and are implemented in SE.Redis: see
@@ -17,6 +18,35 @@ migration work that may be required. In particular, some commands *return differ
 (essentially an interleaved flat array). SE.Redis has been updated to handle these cases transparently, but if you are using `Execute[Async]` or `ScriptEvaluate[Async]` (or if
 you are using an additional library that issues ad-hoc commands or scripts on your behalf) you may need to update your processing code to compensate for this. This is discussed more below.
 
+# Pub/sub connections
+
+RESP3 *allows* pub/sub to share the connection used for ordinary commands, which halves the number of connections. SE.Redis does **not** do this by default: as under RESP2,
+it uses a dedicated subscription connection to each server. This is because the server treats any connection with an active subscription as a *pub/sub client*, and applies the
+`pubsub` class of [`client-output-buffer-limit`](https://redis.io/docs/latest/develop/reference/clients/#output-buffer-limits) to it. The default is:
+
+```
+client-output-buffer-limit pubsub 32mb 8mb 60
+```
+
+meaning the server closes the connection if its pending output ever exceeds 32MiB, or stays above 8MiB for 60 seconds. Those limits are designed for subscribers (where a slow
+reader would otherwise grow the server's memory without bound), but they are easily reached by ordinary traffic: a single large value, a large `MGET`/`HGETALL`/`LRANGE`,
+or a burst of pipelined replies. Ordinary (`normal`) clients have no such limit by default. On a shared connection, the server closing it fails *every* in-flight command, not
+just pub/sub - and since SE.Redis subscribes to the [configuration channel](Configuration#tiebreakers-and-configuration-change-announcements) automatically, a shared
+connection is a pub/sub client even if your application never subscribes to anything.
+
+To share the connection anyway, set `SharedSubscriptionConnection` (`sharedSubscriptionConnection=true` in the configuration string). This has no effect under RESP2.
+**Before enabling it, check that the server's `pubsub` limits suit your workload**: they must accommodate the largest reply (and the largest burst of pipelined or batched
+replies) that any command on the connection can produce, not just your pub/sub messages. You can inspect them with:
+
+```
+CONFIG GET client-output-buffer-limit
+```
+
+and (where permitted) change just the `pubsub` class with, for example, `CONFIG SET client-output-buffer-limit "pubsub 256mb 64mb 60"` (or `0 0 0` to remove the limit),
+remembering to persist it in the server configuration. Many managed services do not allow `CONFIG`, and may apply different limits; if you cannot confirm the limits,
+leave sharing disabled. A disconnect caused by these limits is logged by the server as `scheduled to be closed ASAP for overcoming of output buffer limits`, and the
+client sees it as an unexpected socket close.
+
 # Enabling RESP3
 
 RESP2 and RESP3 are both supported options (if the server does not support RESP3, RESP2 will always be used). To make full use of the benefits of RESP3,

docs/SyncOverAsync.md

@@ -123,8 +123,8 @@ service while the real fix is made. It is not a licence to keep the blocking cal
 
 Two caveats worth knowing before you enable it:
 
-- it costs a reader and a writer thread **for each node you connect to** (not RESP2 pub/sub connections, which
-  stay on the thread-pool; RESP3 does not use separate pub/sub connections), so think about it before enabling
+- it costs a reader and a writer thread **for each node you connect to** (not pub/sub connections, which stay on
+  the thread-pool - unless RESP3 is sharing the interactive connection with pub/sub via `SharedSubscriptionConnection`), so think about it before enabling
   it against a very wide cluster, where that scales with the number of shards;
 - it is deliberately opt-in, and set process-wide at startup rather than per-connection.
 

src/StackExchange.Redis/Configuration/DefaultOptionsProvider.cs

@@ -184,6 +184,12 @@ public static DefaultOptionsProvider GetProvider(EndPoint endpoint)
         /// </remarks>
         public virtual bool HighIntegrity => false;
 
+        /// <summary>
+        /// A Boolean value that specifies whether, under RESP3, pub/sub should share the interactive connection rather than
+        /// using a dedicated subscription connection; see <see cref="ConfigurationOptions.SharedSubscriptionConnection"/>.
+        /// </summary>
+        public virtual bool SharedSubscriptionConnection => false;
+
         /// <summary>
         /// The number of times to repeat the initial connect cycle if no servers respond promptly.
         /// </summary>

src/StackExchange.Redis/ConfigurationOptions.cs

@@ -185,6 +185,7 @@ internal const string
                 MaintenanceRelaxedWindowMax = "maintRelaxedWindowMax",
                 MaintenancePostEventRelaxedDuration = "maintPostEventRelaxed",
                 HighIntegrity = "highIntegrity",
+                SharedSubscriptionConnection = "sharedSubscriptionConnection",
                 TcpKeepAlive = "tcpKeepAlive";
 
             private static readonly Dictionary<string, string> normalizedOptions = new[]
@@ -229,6 +230,7 @@ internal const string
                 MaintenanceRelaxedWindowMax,
                 MaintenancePostEventRelaxedDuration,
                 HighIntegrity,
+                SharedSubscriptionConnection,
                 TcpKeepAlive,
             }.ToDictionary(x => x, StringComparer.OrdinalIgnoreCase);
 
@@ -290,6 +292,8 @@ private enum OptionFlags : ulong
             MaintenanceRelaxedWindowMaxHasValue = 1UL << 38,
             MaintenancePostEventRelaxedDurationHasValue = 1UL << 39,
             TopologyRefreshSecondsHasValue = 1UL << 41,
+            SharedSubscriptionConnectionHasValue = 1UL << 42,
+            SharedSubscriptionConnectionValue = 1UL << 43,
         }
 
         private OptionFlags optionFlags;
@@ -508,6 +512,23 @@ public bool HighIntegrity
             set => SetBooleanWithValue(OptionFlags.HighIntegrityHasValue, OptionFlags.HighIntegrityValue, value);
         }
 
+        /// <summary>
+        /// A Boolean value that specifies whether, under RESP3, pub/sub should share the interactive connection rather than
+        /// using a dedicated subscription connection. Has no effect under RESP2, which always requires a separate connection.
+        /// </summary>
+        /// <remarks>
+        /// Sharing halves the number of connections, but the server classifies any connection with an active subscription
+        /// (including the library's own configuration channel) as a pub/sub client, which by default applies much stricter
+        /// output-buffer limits (<c>client-output-buffer-limit pubsub 32mb 8mb 60</c>). A large reply, or a burst of pipelined
+        /// replies, can then cause the server to close the connection - taking all in-flight commands with it. Only enable this
+        /// if replies are known to be small, or the server's pub/sub output-buffer limits have been relaxed.
+        /// </remarks>
+        public bool SharedSubscriptionConnection
+        {
+            get => HasValue(OptionFlags.SharedSubscriptionConnectionHasValue) ? IsSet(OptionFlags.SharedSubscriptionConnectionValue) : Defaults.SharedSubscriptionConnection;
+            set => SetBooleanWithValue(OptionFlags.SharedSubscriptionConnectionHasValue, OptionFlags.SharedSubscriptionConnectionValue, value);
+        }
+
         /// <summary>
         /// Create a certificate validation check that checks against the supplied issuer even when not known by the machine.
         /// </summary>
@@ -1260,6 +1281,7 @@ public string ToString(bool includePassword)
             Append(sb, OptionKeys.DefaultDatabase, OptionFlags.DefaultDatabaseHasValue, in defaultDatabase);
             Append(sb, OptionKeys.SetClientLibrary, OptionFlags.SetClientLibraryHasValue, OptionFlags.SetClientLibraryValue);
             Append(sb, OptionKeys.HighIntegrity, OptionFlags.HighIntegrityHasValue, OptionFlags.HighIntegrityValue);
+            Append(sb, OptionKeys.SharedSubscriptionConnection, OptionFlags.SharedSubscriptionConnectionHasValue, OptionFlags.SharedSubscriptionConnectionValue);
             if (HasValue(OptionFlags.ProtocolHasValue)) Append(sb, OptionKeys.Protocol, FormatProtocol(_protocol));
             // only when the caller set it *and* it can be named: an inferred provider must not be baked into
             // the string, or re-parsing would pin a choice that was only ever a guess from the endpoints
@@ -1514,6 +1536,9 @@ private ConfigurationOptions DoParse(string configuration, bool ignoreUnknown)
                         case OptionKeys.HighIntegrity:
                             HighIntegrity = OptionKeys.ParseBoolean(key, value);
                             break;
+                        case OptionKeys.SharedSubscriptionConnection:
+                            SharedSubscriptionConnection = OptionKeys.ParseBoolean(key, value);
+                            break;
                         case OptionKeys.TcpKeepAlive:
                             TcpKeepAlive = OptionKeys.ParseBoolean(key, value);
                             break;

src/StackExchange.Redis/ConnectionMultiplexer.FeatureFlags.cs

@@ -80,8 +80,9 @@ public static bool GetFeatureFlag(string flag)
     /// </summary>
     /// <remarks>
     /// For tests and diagnostics: the <see cref="FeatureFlags.DedicatedThreads"/> flag is a request, and this
-    /// is what actually happened. Note that under RESP3 there is no separate subscription connection, so
-    /// asking about <see cref="ConnectionType.Subscription"/> answers about the shared one.
+    /// is what actually happened. Note that under RESP3 with <see cref="ConfigurationOptions.SharedSubscriptionConnection"/>
+    /// there is no separate subscription connection, so asking about <see cref="ConnectionType.Subscription"/> answers
+    /// about the shared one.
     /// </remarks>
     bool? IInternalConnectionMultiplexer.IsSyncReader(EndPoint endpoint, ConnectionType connectionType)
         => GetPhysical(endpoint, connectionType)?.IsSyncReader;

src/StackExchange.Redis/ConnectionMultiplexer.cs

@@ -1742,7 +1742,8 @@ private void ActivateAllServers(ILogger? log)
         /// anything we are about to depend on being connected.
         /// </para>
         /// <para>
-        /// Both legs, where the subscription connection is a separate one: under RESP2 the waiters registered
+        /// Both legs, where the subscription connection is a separate one (always under RESP2, and under RESP3
+        /// unless <see cref="ConfigurationOptions.SharedSubscriptionConnection"/> is enabled): under RESP2 the waiters registered
         /// by <see cref="ServerEndPoint.OnConnectedAsync"/> are only completed once that second connection is
         /// up, so activating the interactive bridge alone leaves such a wait hanging.
         /// </para>
@@ -1751,8 +1752,8 @@ internal static void ActivateServer(ServerEndPoint server, ILogger? log)
         {
             // bool hasSubscriptions = GetSubscriptionsCount() != 0;
             server.Activate(ConnectionType.Interactive, log);
-            // if (hasSubscriptions && server.SupportsSubscriptions && !server.KnowOrAssumeResp3())
-            if (server.SupportsSubscriptions && !server.KnowOrAssumeResp3())
+            // if (hasSubscriptions && server.SupportsSubscriptions && !server.SharesSubscriptionConnection())
+            if (server.SupportsSubscriptions && !server.SharesSubscriptionConnection())
             {
                 // Intentionally not logging the sub connection
                 server.Activate(ConnectionType.Subscription, null);

src/StackExchange.Redis/PhysicalBridge.cs

@@ -824,8 +824,8 @@ private WriteResult WriteMessageInsideLock(PhysicalConnection physical, Message
                 && Protocol is RedisProtocol.Resp2
                 && ServerEndPoint.TryRerouteToSubscriptionBridge(message, this))
             {
-                // this was queued while we expected RESP3 - where subscriptions share the interactive
-                // connection - but the handshake resolved to RESP2, which needs them on their own connection.
+                // this was queued while we expected RESP3 - where subscriptions can share the interactive
+                // connection (SharedSubscriptionConnection) - but the handshake resolved to RESP2, which needs them on their own connection.
                 // Writing it here would put *this* connection into subscriber mode, which rejects every
                 // ordinary command on it from then on; see #3154
                 Trace($"Rerouting {message.CommandAndKey} to the subscription connection (RESP2)");

src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt

@@ -1,4 +1,7 @@
 #nullable enable
+StackExchange.Redis.ConfigurationOptions.SharedSubscriptionConnection.get -> bool
+StackExchange.Redis.ConfigurationOptions.SharedSubscriptionConnection.set -> void
+virtual StackExchange.Redis.Configuration.DefaultOptionsProvider.SharedSubscriptionConnection.get -> bool
 StackExchange.Redis.ConfigurationOptions.ConnectionAttemptCompleted -> System.EventHandler<StackExchange.Redis.ConnectionAttemptCompletedEventArgs!>?
 StackExchange.Redis.ConnectionAttemptCompletedEventArgs
 StackExchange.Redis.ConnectionAttemptCompletedEventArgs.ClientCertificateIssuer.get -> string?