diff --git a/docs/Configuration.md b/docs/Configuration.md
index d92f5153b..72bc9b2b9 100644
--- a/docs/Configuration.md
+++ b/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
diff --git a/docs/Resp3.md b/docs/Resp3.md
index 9f851d61e..2d3d05111 100644
--- a/docs/Resp3.md
+++ b/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,
diff --git a/docs/SyncOverAsync.md b/docs/SyncOverAsync.md
index a1b2dd7a2..fb0b359d7 100644
--- a/docs/SyncOverAsync.md
+++ b/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.
diff --git a/src/StackExchange.Redis/Configuration/DefaultOptionsProvider.cs b/src/StackExchange.Redis/Configuration/DefaultOptionsProvider.cs
index 15b227f50..00d4de888 100644
--- a/src/StackExchange.Redis/Configuration/DefaultOptionsProvider.cs
+++ b/src/StackExchange.Redis/Configuration/DefaultOptionsProvider.cs
@@ -184,6 +184,12 @@ public static DefaultOptionsProvider GetProvider(EndPoint endpoint)
///
public virtual bool HighIntegrity => false;
+ ///
+ /// A Boolean value that specifies whether, under RESP3, pub/sub should share the interactive connection rather than
+ /// using a dedicated subscription connection; see .
+ ///
+ public virtual bool SharedSubscriptionConnection => false;
+
///
/// The number of times to repeat the initial connect cycle if no servers respond promptly.
///
diff --git a/src/StackExchange.Redis/ConfigurationOptions.cs b/src/StackExchange.Redis/ConfigurationOptions.cs
index 4ea52bc04..d0eae12df 100644
--- a/src/StackExchange.Redis/ConfigurationOptions.cs
+++ b/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 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);
}
+ ///
+ /// 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.
+ ///
+ ///
+ /// 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 (client-output-buffer-limit pubsub 32mb 8mb 60). 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.
+ ///
+ public bool SharedSubscriptionConnection
+ {
+ get => HasValue(OptionFlags.SharedSubscriptionConnectionHasValue) ? IsSet(OptionFlags.SharedSubscriptionConnectionValue) : Defaults.SharedSubscriptionConnection;
+ set => SetBooleanWithValue(OptionFlags.SharedSubscriptionConnectionHasValue, OptionFlags.SharedSubscriptionConnectionValue, value);
+ }
+
///
/// Create a certificate validation check that checks against the supplied issuer even when not known by the machine.
///
@@ -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;
diff --git a/src/StackExchange.Redis/ConnectionMultiplexer.FeatureFlags.cs b/src/StackExchange.Redis/ConnectionMultiplexer.FeatureFlags.cs
index 73180546f..b22d709f3 100644
--- a/src/StackExchange.Redis/ConnectionMultiplexer.FeatureFlags.cs
+++ b/src/StackExchange.Redis/ConnectionMultiplexer.FeatureFlags.cs
@@ -80,8 +80,9 @@ public static bool GetFeatureFlag(string flag)
///
///
/// For tests and diagnostics: the flag is a request, and this
- /// is what actually happened. Note that under RESP3 there is no separate subscription connection, so
- /// asking about answers about the shared one.
+ /// is what actually happened. Note that under RESP3 with
+ /// there is no separate subscription connection, so asking about answers
+ /// about the shared one.
///
bool? IInternalConnectionMultiplexer.IsSyncReader(EndPoint endpoint, ConnectionType connectionType)
=> GetPhysical(endpoint, connectionType)?.IsSyncReader;
diff --git a/src/StackExchange.Redis/ConnectionMultiplexer.cs b/src/StackExchange.Redis/ConnectionMultiplexer.cs
index e24c5d360..b2d38cdc1 100644
--- a/src/StackExchange.Redis/ConnectionMultiplexer.cs
+++ b/src/StackExchange.Redis/ConnectionMultiplexer.cs
@@ -1742,7 +1742,8 @@ private void ActivateAllServers(ILogger? log)
/// anything we are about to depend on being connected.
///
///
- /// 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 is enabled): under RESP2 the waiters registered
/// by are only completed once that second connection is
/// up, so activating the interactive bridge alone leaves such a wait hanging.
///
@@ -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);
diff --git a/src/StackExchange.Redis/PhysicalBridge.cs b/src/StackExchange.Redis/PhysicalBridge.cs
index 83d04c889..2ead01179 100644
--- a/src/StackExchange.Redis/PhysicalBridge.cs
+++ b/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)");
diff --git a/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt b/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt
index 2bfa4e24f..6cc0a9b34 100644
--- a/src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt
+++ b/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.ClientCertificateIssuer.get -> string?
diff --git a/src/StackExchange.Redis/ServerEndPoint.cs b/src/StackExchange.Redis/ServerEndPoint.cs
index 109fd5129..eb9d9fa02 100644
--- a/src/StackExchange.Redis/ServerEndPoint.cs
+++ b/src/StackExchange.Redis/ServerEndPoint.cs
@@ -142,7 +142,16 @@ public int Databases
public bool IsConnecting => interactive?.IsConnecting == true;
public bool IsConnected => interactive?.IsConnected == true;
- public bool IsSubscriberConnected => KnowOrAssumeResp3() ? IsConnected : subscription?.IsConnected == true;
+ public bool IsSubscriberConnected => SharesSubscriptionConnection() ? IsConnected : subscription?.IsConnected == true;
+
+ ///
+ /// Whether pub/sub shares the interactive connection: only under RESP3, and only when opted into via
+ /// . Otherwise subscriptions get a dedicated
+ /// connection, as they always do under RESP2 - because the server classifies any connection with a live
+ /// subscription as a pub/sub client, applying the (much tighter) pub/sub output-buffer limits to it; see #3263.
+ ///
+ public bool SharesSubscriptionConnection() => Multiplexer.RawConfig.SharedSubscriptionConnection && KnowOrAssumeResp3();
+
public bool KnowOrAssumeResp3()
{
var protocol = interactive?.Protocol;
@@ -214,7 +223,7 @@ internal Exception? LastException
}
internal State InteractiveConnectionState => interactive?.ConnectionState ?? State.Disconnected;
- internal State SubscriptionConnectionState => KnowOrAssumeResp3() ? InteractiveConnectionState : subscription?.ConnectionState ?? State.Disconnected;
+ internal State SubscriptionConnectionState => SharesSubscriptionConnection() ? InteractiveConnectionState : subscription?.ConnectionState ?? State.Disconnected;
public long OperationCount => (interactive?.OperationCount ?? 0) + (subscription?.OperationCount ?? 0);
@@ -415,10 +424,10 @@ public void Dispose()
switch (type)
{
case ConnectionType.Interactive:
- case ConnectionType.Subscription when KnowOrAssumeResp3():
- return interactive ?? (create ? interactive = CreateBridge(ConnectionType.Interactive, log) : null);
+ case ConnectionType.Subscription when SharesSubscriptionConnection():
+ return interactive ?? (create ? GetOrCreateBridge(ref interactive, ConnectionType.Interactive, log) : null);
case ConnectionType.Subscription:
- return subscription ?? (create ? subscription = CreateBridge(ConnectionType.Subscription, log) : null);
+ return subscription ?? (create ? GetOrCreateBridge(ref subscription, ConnectionType.Subscription, log) : null);
default:
return null;
}
@@ -443,9 +452,9 @@ public void Dispose()
break;
}
- return (message.IsForSubscriptionBridge && !KnowOrAssumeResp3())
- ? subscription ??= CreateBridge(ConnectionType.Subscription, null)
- : interactive ??= CreateBridge(ConnectionType.Interactive, null);
+ return (message.IsForSubscriptionBridge && !SharesSubscriptionConnection())
+ ? subscription ?? GetOrCreateBridge(ref subscription, ConnectionType.Subscription, null)
+ : interactive ?? GetOrCreateBridge(ref interactive, ConnectionType.Interactive, null);
}
///
@@ -458,7 +467,7 @@ internal bool TryRerouteToSubscriptionBridge(Message message, PhysicalBridge fro
if (isDisposed) return false;
// deliberately not via GetBridge: that consults the same expectation that got us here
- var target = subscription ??= CreateBridge(ConnectionType.Subscription, null);
+ var target = subscription ?? GetOrCreateBridge(ref subscription, ConnectionType.Subscription, null);
if (target is null || ReferenceEquals(target, from)) return false;
target.AcceptRerouted(message);
@@ -476,13 +485,13 @@ internal bool TryRerouteToSubscriptionBridge(Message message, PhysicalBridge fro
case RedisCommand.PUNSUBSCRIBE:
case RedisCommand.SSUBSCRIBE:
case RedisCommand.SUNSUBSCRIBE:
- if (!KnowOrAssumeResp3())
+ if (!SharesSubscriptionConnection())
{
- return subscription ?? (create ? subscription = CreateBridge(ConnectionType.Subscription, null) : null);
+ return subscription ?? (create ? GetOrCreateBridge(ref subscription, ConnectionType.Subscription, null) : null);
}
break;
}
- return interactive ?? (create ? interactive = CreateBridge(ConnectionType.Interactive, null) : null);
+ return interactive ?? (create ? GetOrCreateBridge(ref interactive, ConnectionType.Interactive, null) : null);
}
public RedisFeatures GetFeatures() => new RedisFeatures(version);
@@ -753,7 +762,8 @@ internal Task Close(ConnectionType connectionType)
try
{
var tmp = GetBridge(connectionType, create: false);
- if (tmp == null || !tmp.IsConnected || !Multiplexer.CommandMap.IsAvailable(RedisCommand.QUIT))
+ if (tmp == null || !tmp.IsConnected || !Multiplexer.CommandMap.IsAvailable(RedisCommand.QUIT)
+ || (connectionType == ConnectionType.Subscription && ReferenceEquals(tmp, interactive))) // shared: QUIT once, via Interactive
{
return Task.CompletedTask;
}
@@ -947,7 +957,7 @@ internal void OnDisconnected(PhysicalBridge bridge)
if (bridge == interactive)
{
CompletePendingConnectionMonitors("Disconnected");
- if (Protocol is RedisProtocol.Resp3)
+ if (SharesSubscriptionConnection())
{
Multiplexer.UpdateSubscriptions();
}
@@ -1046,9 +1056,10 @@ internal void OnRepeatedConnectFailure(int consecutiveFailures)
private static int NudgeFromZeroTicks(int ticks) => ticks == 0 ? 1 : ticks;
///
- /// Subscribes to the configuration-change broadcast on a RESP3 connection. With RESP3 there is no
- /// separate subscription connection, and so no subscription handshake - which is where RESP2 subscribes
- /// to it. Left at that, the channel would silently have no subscriber, and the manual
+ /// Subscribes to the configuration-change broadcast on a RESP3 interactive connection that is shared with
+ /// pub/sub (). Then there is no separate
+ /// subscription connection, and so no subscription handshake - which is where a dedicated subscription
+ /// connection subscribes to it. Left at that, the channel would silently have no subscriber, and the manual
/// PUBLISH that clients have long used to announce a topology change would reach nobody.
///
///
@@ -1082,28 +1093,32 @@ internal void OnFullyEstablished(PhysicalConnection connection, string source)
// is *this specific* connection using RESP3? (without reference to config preferences)
bool isResp3 = connection?.Protocol is >= RedisProtocol.Resp3;
+ // and if so, is pub/sub sharing it? (if not, subscriptions live on a dedicated connection)
+ bool shared = isResp3 && bridge == interactive && Multiplexer.RawConfig.SharedSubscriptionConnection;
if (connection is not null && bridge == interactive)
{
ReconcileMaintenanceNotifications(connection);
}
- if (bridge == subscription || isResp3)
+ if (bridge == subscription || shared)
{
// Note: this MUST be fire and forget, because we might be in the middle of a Sync processing
// TracerProcessor which is executing this line inside a SetResultCore().
// Since we're issuing commands inside a SetResult path in a message, we'd create a deadlock by waiting.
Multiplexer.EnsureSubscriptions(CommandFlags.FireAndForget);
- if (isResp3 && bridge == interactive)
+ if (shared)
{
SubscribeToConfigurationChannel(bridge);
}
}
- else if (SupportsSubscriptions && Multiplexer.RawConfig.Protocol > RedisProtocol.Resp2)
+ else if (bridge == interactive && !isResp3 && SupportsSubscriptions
+ && Multiplexer.RawConfig.SharedSubscriptionConnection && Multiplexer.RawConfig.Protocol > RedisProtocol.Resp2)
{
- // interactive, and we wanted RESP3+, but we didn't get it; spin up pub/sub
+ // interactive, and we wanted to share it with pub/sub under RESP3+, but we didn't get it; spin up
+ // pub/sub (when not sharing, ActivateServer has already done so, as under RESP2)
Activate(ConnectionType.Subscription, null);
}
- if (IsConnected && (IsSubscriberConnected || !SupportsSubscriptions || isResp3))
+ if (IsConnected && (IsSubscriberConnected || !SupportsSubscriptions || shared))
{
// Only connect on the second leg - we can accomplish this by checking both
// Or the first leg, if we're only making 1 connection because subscriptions aren't supported
@@ -1327,6 +1342,27 @@ internal ValueTask WriteDirectOrQueueFireAndForgetAsync(PhysicalConnection? c
return default;
}
+ ///
+ /// Lazily creates (and starts connecting) the bridge in , safely against a concurrent
+ /// caller doing the same - e.g. ActivateServer on the caller's thread racing OnFullyEstablished
+ /// on an IO thread. A bridge connects as soon as it is created, so a plain ??= that loses the race
+ /// leaks a second, unreferenced connection to the server.
+ ///
+ private PhysicalBridge? GetOrCreateBridge(ref PhysicalBridge? field, ConnectionType type, ILogger? log)
+ {
+ var existing = Volatile.Read(ref field);
+ if (existing is not null) return existing;
+
+ var created = CreateBridge(type, log);
+ if (created is null) return null;
+
+ existing = Interlocked.CompareExchange(ref field, created, null);
+ if (existing is null) return created;
+
+ created.Dispose(); // lost the race: use the winner, and close the connection we started
+ return existing;
+ }
+
private PhysicalBridge? CreateBridge(ConnectionType type, ILogger? log)
{
if (Multiplexer.IsDisposed) return null;
diff --git a/tests/StackExchange.Redis.Tests/ClientKillTests.cs b/tests/StackExchange.Redis.Tests/ClientKillTests.cs
index 8ac4f2c39..cf94ae8e4 100644
--- a/tests/StackExchange.Redis.Tests/ClientKillTests.cs
+++ b/tests/StackExchange.Redis.Tests/ClientKillTests.cs
@@ -11,22 +11,30 @@ namespace StackExchange.Redis.Tests;
public class ClientKillTests(ITestOutputHelper output) : TestBase(output)
{
- [Fact]
- public async Task ClientKill()
+ [Theory]
+ [InlineData(false)]
+ [InlineData(true)]
+ public async Task ClientKill(bool sharedSubscriptionConnection)
{
SetExpectedAmbientFailureCount(-1);
- await using var otherConnection = Create(allowAdmin: true, shared: false, backlogPolicy: BacklogPolicy.FailFast, require: RedisFeatures.v7_4_0_rc1);
+ await using var otherConnection = Create(
+ allowAdmin: true,
+ shared: false,
+ backlogPolicy: BacklogPolicy.FailFast,
+ require: RedisFeatures.v7_4_0_rc1,
+ configuration: sharedSubscriptionConnection ? GetConfiguration() + ",sharedSubscriptionConnection=true" : null);
var id = otherConnection.GetDatabase().Execute(RedisCommand.CLIENT.ToString(), RedisLiterals.ID);
await using var conn = Create(allowAdmin: true, shared: false, backlogPolicy: BacklogPolicy.FailFast);
var server = conn.GetServer(conn.GetEndPoints()[0]);
- // RESP3 has no separate subscription connection, so the interactive connection carries the
- // configuration-channel subscription - and the server counts a subscribed client as pubsub, not normal.
+ // when RESP3 shares the interactive connection with pub/sub, it carries the configuration-channel
+ // subscription - and the server counts a subscribed client as pubsub, not normal (which is exactly
+ // why sharing is opt-in: pubsub clients get much tighter output-buffer limits)
var protocol = TestContext.Current.GetProtocol();
var client = server.ClientList().Single(x => x.Id == id.AsInt64());
Assert.Equal(protocol, client.Protocol);
- var expectedType = protocol == RedisProtocol.Resp3 ? ClientType.PubSub : ClientType.Normal;
+ var expectedType = protocol == RedisProtocol.Resp3 && sharedSubscriptionConnection ? ClientType.PubSub : ClientType.Normal;
Assert.Equal(expectedType, client.ClientType);
long result = server.ClientKill(id.AsInt64(), expectedType, null, true);
diff --git a/tests/StackExchange.Redis.Tests/ClusterTests.cs b/tests/StackExchange.Redis.Tests/ClusterTests.cs
index b10ab0d00..94513858e 100644
--- a/tests/StackExchange.Redis.Tests/ClusterTests.cs
+++ b/tests/StackExchange.Redis.Tests/ClusterTests.cs
@@ -39,12 +39,17 @@ public async Task ExportConfiguration()
Assert.True(File.Exists("cluster.zip"));
}
- [Fact]
- public async Task ConnectUsesSingleSocket()
+ [Theory]
+ [InlineData(false)]
+ [InlineData(true)]
+ public async Task ConnectUsesSingleSocket(bool sharedSubscriptionConnection)
{
for (int i = 0; i < 5; i++)
{
- await using var conn = Create(failMessage: i + ": ", log: Writer);
+ await using var conn = Create(
+ failMessage: i + ": ",
+ log: Writer,
+ configuration: sharedSubscriptionConnection ? GetConfiguration() + ",sharedSubscriptionConnection=true" : null);
foreach (var ep in conn.GetEndPoints())
{
@@ -58,7 +63,7 @@ public async Task ConnectUsesSingleSocket()
var srv = conn.GetServer(ep);
var counters = srv.GetCounters();
Assert.Equal(1, counters.Interactive.SocketCount);
- Assert.Equal(TestContext.Current.IsResp3() ? 0 : 1, counters.Subscription.SocketCount);
+ Assert.Equal(sharedSubscriptionConnection && TestContext.Current.IsResp3() ? 0 : 1, counters.Subscription.SocketCount);
}
}
}
diff --git a/tests/StackExchange.Redis.Tests/ConfigTests.cs b/tests/StackExchange.Redis.Tests/ConfigTests.cs
index 6b487dbda..bab313224 100644
--- a/tests/StackExchange.Redis.Tests/ConfigTests.cs
+++ b/tests/StackExchange.Redis.Tests/ConfigTests.cs
@@ -467,7 +467,7 @@ public async Task ConnectWithSubscribeDisabled()
Assert.True(conn.IsConnected);
var servers = conn.GetServerSnapshot();
Assert.True(servers[0].IsConnected);
- if (!TestContext.Current.IsResp3())
+ if (!servers[0].SharesSubscriptionConnection())
{
Assert.False(servers[0].IsSubscriberConnected);
}
@@ -703,10 +703,13 @@ public void Apply()
Assert.Equal(result, options);
}
- [Fact]
- public async Task BeforeSocketConnect()
+ [Theory]
+ [InlineData(false)]
+ [InlineData(true)]
+ public async Task BeforeSocketConnect(bool sharedSubscriptionConnection)
{
var options = Parse(TestConfig.Current.PrimaryServerAndPort);
+ options.SharedSubscriptionConnection = sharedSubscriptionConnection;
int count = 0;
options.BeforeSocketConnect = (endpoint, connType, socket) =>
{
@@ -717,7 +720,7 @@ public async Task BeforeSocketConnect()
};
await using var conn = ConnectionMultiplexer.Connect(options);
Assert.True(conn.IsConnected);
- Assert.Equal(options.TryResp3() ? 1 : 2, count);
+ Assert.Equal(sharedSubscriptionConnection && options.TryResp3() ? 1 : 2, count);
var endpoint = conn.GetServerSnapshot()[0];
var interactivePhysical = endpoint.GetBridge(ConnectionType.Interactive)?.TryConnect(null);
@@ -898,6 +901,26 @@ public void CheckHighIntegrity(bool? assigned, bool expected, string cs)
Assert.Equal(expected, parsed.HighIntegrity);
}
+ [Theory]
+ [InlineData(null, false, "dummy")]
+ [InlineData(false, false, "dummy,sharedSubscriptionConnection=False")]
+ [InlineData(true, true, "dummy,sharedSubscriptionConnection=True")]
+ public void CheckSharedSubscriptionConnection(bool? assigned, bool expected, string cs)
+ {
+ var options = Parse("dummy");
+ if (assigned.HasValue) options.SharedSubscriptionConnection = assigned.Value;
+
+ Assert.Equal(expected, options.SharedSubscriptionConnection);
+ Assert.Equal(cs, RemoveTestDefaults(options.ToString()));
+
+ var clone = options.Clone();
+ Assert.Equal(expected, clone.SharedSubscriptionConnection);
+ Assert.Equal(cs, RemoveTestDefaults(clone.ToString()));
+
+ var parsed = Parse(cs);
+ Assert.Equal(expected, parsed.SharedSubscriptionConnection);
+ }
+
[Theory]
[InlineData("maintRelaxedTimeout=20", 20, 60, 40)]
[InlineData("maintRelaxedWindowMax=45", 10, 45, 20)]
diff --git a/tests/StackExchange.Redis.Tests/ConfigurationChannelUnitTests.cs b/tests/StackExchange.Redis.Tests/ConfigurationChannelUnitTests.cs
index 7045d6cc6..6de5b9f38 100644
--- a/tests/StackExchange.Redis.Tests/ConfigurationChannelUnitTests.cs
+++ b/tests/StackExchange.Redis.Tests/ConfigurationChannelUnitTests.cs
@@ -1,4 +1,4 @@
-using System;
+using System;
using System.Threading.Tasks;
using Xunit;
@@ -7,28 +7,32 @@ namespace StackExchange.Redis.Tests;
///
/// The configuration-change channel is how a client is told, by hand, that the topology moved: a PUBLISH to
/// it makes every subscribed client refresh. Under RESP2 the dedicated subscription connection subscribes to it as
-/// part of its handshake; RESP3 has no such connection, and the subscription was never made. See #3254.
+/// part of its handshake; a RESP3 interactive connection shared with pub/sub has no such connection, and the
+/// subscription was never made. See #3254.
///
[RunPerProtocol]
public class ConfigurationChannelUnitTests(ITestOutputHelper log)
{
private const string Channel = "__Booksleeve_MasterChanged";
- private static ConfigurationOptions Configure(InProcessTestServer server, bool prefix)
+ private static ConfigurationOptions Configure(InProcessTestServer server, bool prefix, bool shared)
{
var config = server.GetClientConfig();
config.ConfigurationChannel = Channel; // the test server turns this off by default
+ config.SharedSubscriptionConnection = shared;
if (prefix) config.ChannelPrefix = RedisChannel.Literal("testuser-");
return config;
}
[Theory]
- [InlineData(false)]
- [InlineData(true)]
- public async Task PublishingToTheConfigurationChannelIsHeard(bool prefix)
+ [InlineData(false, false)]
+ [InlineData(true, false)]
+ [InlineData(false, true)]
+ [InlineData(true, true)]
+ public async Task PublishingToTheConfigurationChannelIsHeard(bool prefix, bool shared)
{
using var server = new InProcessTestServer(log);
- var config = Configure(server, prefix);
+ var config = Configure(server, prefix, shared);
await using var conn = await ConnectionMultiplexer.ConnectAsync(config);
var heard = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
@@ -48,14 +52,16 @@ public async Task PublishingToTheConfigurationChannelIsHeard(bool prefix)
}
[Theory]
- [InlineData(false)]
- [InlineData(true)]
- public async Task TheLibrarysOwnBroadcastIsHeard(bool prefix)
+ [InlineData(false, false)]
+ [InlineData(true, false)]
+ [InlineData(false, true)]
+ [InlineData(true, true)]
+ public async Task TheLibrarysOwnBroadcastIsHeard(bool prefix, bool shared)
{
// the channel is subscribed with the channel prefix applied, so it must be fired with it too - or a
// client that changes a server's role announces it to a channel that nobody is listening on
using var server = new InProcessTestServer(log);
- await using var conn = await ConnectionMultiplexer.ConnectAsync(Configure(server, prefix));
+ await using var conn = await ConnectionMultiplexer.ConnectAsync(Configure(server, prefix, shared));
var heard = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
conn.ConfigurationChangedBroadcast += (_, e) => heard.TrySetResult(e);
diff --git a/tests/StackExchange.Redis.Tests/ConnectingFailDetectionTests.cs b/tests/StackExchange.Redis.Tests/ConnectingFailDetectionTests.cs
index 10ad1afa0..747a7c101 100644
--- a/tests/StackExchange.Redis.Tests/ConnectingFailDetectionTests.cs
+++ b/tests/StackExchange.Redis.Tests/ConnectingFailDetectionTests.cs
@@ -128,8 +128,8 @@ public async Task Issue922_ReconnectRaised()
var server = conn.GetServer(TestConfig.Current.PrimaryServerAndPort);
var protocol = server.Protocol;
- // RESP2 has interactive+subscriber connections; RESP3 uses one connection for both.
- var expectedCount = protocol is RedisProtocol.Resp3 ? 1 : 2;
+ // interactive+subscriber connections, unless RESP3 is sharing one connection for both (opt-in)
+ var expectedCount = ((IInternalConnectionMultiplexer)conn).GetServerEndPoint(server.EndPoint).SharesSubscriptionConnection() ? 1 : 2;
Log($"Using {protocol.GetString()}; expecting {expectedCount} reconnect event(s)");
Assert.SkipUnless(server.CanSimulateConnectionFailure(), "Skipping because server cannot simulate connection failure");
diff --git a/tests/StackExchange.Redis.Tests/DedicatedThreadsTests.cs b/tests/StackExchange.Redis.Tests/DedicatedThreadsTests.cs
index b982e8289..63e8755b6 100644
--- a/tests/StackExchange.Redis.Tests/DedicatedThreadsTests.cs
+++ b/tests/StackExchange.Redis.Tests/DedicatedThreadsTests.cs
@@ -1,4 +1,4 @@
-using System;
+using System;
using System.Net;
using System.Threading.Tasks;
using Xunit;
@@ -27,18 +27,19 @@ public class DedicatedThreadsTests(ITestOutputHelper output) : TestBase(output)
private const string Flag = "DedicatedThreads";
/// Connect with the flag in a known state, and put it back afterwards.
- private async Task WithFlagAsync(bool enabled, Func assert)
+ private async Task WithFlagAsync(bool enabled, Func assert, bool sharedSubscriptionConnection = false)
{
var wasSet = ConnectionMultiplexer.GetFeatureFlag(Flag);
ConnectionMultiplexer.SetFeatureFlag(Flag, enabled);
try
{
- await using var conn = Create(shared: false);
+ await using var conn = Create(
+ shared: false,
+ configuration: sharedSubscriptionConnection ? GetConfiguration() + ",sharedSubscriptionConnection=true" : null);
var endpoint = conn.GetEndPoints()[0];
// the flag is consumed while the connection is being established, so make sure one exists before
- // asking anything about it - and touch pub/sub too, since under RESP2 that is a second connection
- // that is not created until it is needed
+ // asking anything about it - and touch pub/sub too, since that can be a second connection
var db = conn.GetDatabase();
await db.PingAsync();
await conn.GetSubscriber().PingAsync();
@@ -72,15 +73,16 @@ public Task WithTheFlag_TheInteractiveConnectionOwnsItsThreads() => WithFlagAsyn
/// quoted per node rather than per connection.
///
///
- /// Only checkable under RESP2. Under RESP3 there *is* no separate subscription connection - the bridge
- /// lookup returns the interactive one - so the question does not arise, and asserting "false" there would
- /// be asserting against the shared connection we just required to be true.
+ /// Only checkable with a dedicated subscription connection. When RESP3 shares the interactive connection
+ /// with pub/sub (opt-in) the bridge lookup returns the interactive one, so the question does not arise, and
+ /// asserting "false" there would be asserting against the shared connection we just required to be true.
///
- [Fact]
- public Task WithTheFlag_PubSubStaysOnTheThreadPool() => WithFlagAsync(true, (conn, endpoint) =>
+ [Theory]
+ [InlineData(false)]
+ [InlineData(true)]
+ public Task WithTheFlag_PubSubStaysOnTheThreadPool(bool sharedSubscriptionConnection) => WithFlagAsync(true, (conn, endpoint) =>
{
- var protocol = conn.GetServerEndPoint(endpoint).Protocol ?? RedisProtocol.Resp2;
- if (protocol >= RedisProtocol.Resp3)
+ if (conn.GetServerEndPoint(endpoint).SharesSubscriptionConnection())
{
// one connection carries both, so the subscription lookup is the interactive connection
Assert.True(conn.IsSyncReader(endpoint, ConnectionType.Subscription));
@@ -97,5 +99,5 @@ public Task WithTheFlag_PubSubStaysOnTheThreadPool() => WithFlagAsync(true, (con
}
return Task.CompletedTask;
- });
+ }, sharedSubscriptionConnection);
}
diff --git a/tests/StackExchange.Redis.Tests/DefaultOptionsTests.cs b/tests/StackExchange.Redis.Tests/DefaultOptionsTests.cs
index 1f508f262..38b3fd3b5 100644
--- a/tests/StackExchange.Redis.Tests/DefaultOptionsTests.cs
+++ b/tests/StackExchange.Redis.Tests/DefaultOptionsTests.cs
@@ -218,15 +218,22 @@ public void MaintenanceNotificationDefaultPerProvider(string hostName, Maintenan
Assert.Equal(expected, options.MaintenanceNotifications);
}
+ ///
+ /// A dedicated subscription connection unless pub/sub is explicitly allowed to share the RESP3 interactive one;
+ /// either way the interactive connection must not be left subscribed (and so classed as pub/sub by the server).
+ ///
[Theory]
- [InlineData(RedisProtocol.Resp2)]
- [InlineData(RedisProtocol.Resp3)]
- public async Task AzureManagedRedisConnectsWithoutSubscriptionConnection(RedisProtocol protocol)
+ [InlineData(RedisProtocol.Resp2, false)]
+ [InlineData(RedisProtocol.Resp3, false)]
+ [InlineData(RedisProtocol.Resp2, true)]
+ [InlineData(RedisProtocol.Resp3, true)]
+ public async Task AzureManagedRedisConnectionCount(RedisProtocol protocol, bool sharedSubscriptionConnection)
{
using var serverObj = new InProcessTestServer(Output, new DnsEndPoint("contoso.redis.azure.net", 10000), useSsl: true);
var config = serverObj.GetClientConfig();
config.ClientName = Guid.NewGuid().ToString().Replace("-", "");
config.Protocol = protocol;
+ config.SharedSubscriptionConnection = sharedSubscriptionConnection;
await using var conn = await ConnectionMultiplexer.ConnectAsync(config, Writer);
@@ -244,11 +251,11 @@ public async Task AzureManagedRedisConnectsWithoutSubscriptionConnection(RedisPr
Assert.Equal(0, self.ShardedSubscriptionCount);
Assert.Equal(protocol, self.Protocol);
- var expectedCount = protocol is RedisProtocol.Resp3 ? 1 : 2;
+ var expectedCount = protocol is RedisProtocol.Resp3 && sharedSubscriptionConnection ? 1 : 2;
Assert.Equal(expectedCount, serverObj.ClientCount);
Assert.Equal(expectedCount, namedClients.Length);
- await AssertCanPubSubAsync(conn, $"{nameof(AzureManagedRedisConnectsWithoutSubscriptionConnection)}:{protocol}");
+ await AssertCanPubSubAsync(conn, $"{nameof(AzureManagedRedisConnectionCount)}:{protocol}:{sharedSubscriptionConnection}");
}
[Fact]
diff --git a/tests/StackExchange.Redis.Tests/MaintenanceNotificationTests.cs b/tests/StackExchange.Redis.Tests/MaintenanceNotificationTests.cs
index a399312ce..1e197f48c 100644
--- a/tests/StackExchange.Redis.Tests/MaintenanceNotificationTests.cs
+++ b/tests/StackExchange.Redis.Tests/MaintenanceNotificationTests.cs
@@ -174,7 +174,7 @@ public async Task MovingWithAPlaceholderYieldsNoEndpoint(string placeholder)
using (server)
await using (conn)
{
- Assert.Equal(1, server.SendRawPush(null, "MOVING", "1", "15", placeholder));
+ Assert.True(server.SendRawPush(null, "MOVING", "1", "15", placeholder) > 0); // raw pushes ignore opt-in, so the subscription connection gets one too
var evt = await events.NextAsync();
log.WriteLine(evt.RawMessage ?? "(no message)");
@@ -228,7 +228,7 @@ public async Task IntegersMayArriveAsStrings()
using (server)
await using (conn)
{
- Assert.Equal(1, server.SendRawPush(null, "MIGRATING", "42", "7", "[\"shard:2\"]"));
+ Assert.True(server.SendRawPush(null, "MIGRATING", "42", "7", "[\"shard:2\"]") > 0); // raw pushes ignore opt-in, so the subscription connection gets one too
var evt = await events.NextAsync();
log.WriteLine(evt.RawMessage ?? "(no message)");
@@ -263,7 +263,7 @@ public async Task LowercaseTypeNamesAreRecognized()
using (server)
await using (conn)
{
- Assert.Equal(1, server.SendRawPush(null, "failing_over", "5", "9", "[]"));
+ Assert.True(server.SendRawPush(null, "failing_over", "5", "9", "[]") > 0); // raw pushes ignore opt-in, so the subscription connection gets one too
var evt = await events.NextAsync();
log.WriteLine(evt.RawMessage ?? "(no message)");
@@ -279,7 +279,7 @@ public async Task TrailingElementsAreTolerated()
using (server)
await using (conn)
{
- Assert.Equal(1, server.SendRawPush(null, "MIGRATING", "8", "20", "[\"shard:3\"]", "something-new", "and-another"));
+ Assert.True(server.SendRawPush(null, "MIGRATING", "8", "20", "[\"shard:3\"]", "something-new", "and-another") > 0); // raw pushes ignore opt-in, so the subscription connection gets one too
var evt = await events.NextAsync();
log.WriteLine(evt.RawMessage ?? "(no message)");
diff --git a/tests/StackExchange.Redis.Tests/PubSubMultiserverTests.cs b/tests/StackExchange.Redis.Tests/PubSubMultiserverTests.cs
index e4e97e63a..b5bb40e50 100644
--- a/tests/StackExchange.Redis.Tests/PubSubMultiserverTests.cs
+++ b/tests/StackExchange.Redis.Tests/PubSubMultiserverTests.cs
@@ -70,7 +70,7 @@ await sub.SubscribeAsync(channel, (_, val) =>
Log("Connected to: " + initialServer);
conn.AllowConnect = false;
- if (TestContext.Current.IsResp3())
+ if (subscribedServerEndpoint.SharesSubscriptionConnection())
{
subscribedServerEndpoint.SimulateConnectionFailure(SimulatedFailureType.All);
@@ -156,7 +156,7 @@ await sub.SubscribeAsync(
Log("Connected to: " + initialServer);
conn.AllowConnect = false;
- if (TestContext.Current.IsResp3())
+ if (subscribedServerEndpoint.SharesSubscriptionConnection())
{
subscribedServerEndpoint.SimulateConnectionFailure(SimulatedFailureType.All); // need to kill the main connection
Assert.False(subscribedServerEndpoint.IsConnected, "subscribedServerEndpoint.IsConnected");
diff --git a/tests/StackExchange.Redis.Tests/PubSubOutputBufferTests.cs b/tests/StackExchange.Redis.Tests/PubSubOutputBufferTests.cs
new file mode 100644
index 000000000..c7e42e10d
--- /dev/null
+++ b/tests/StackExchange.Redis.Tests/PubSubOutputBufferTests.cs
@@ -0,0 +1,101 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+using Xunit;
+
+namespace StackExchange.Redis.Tests;
+
+///
+/// The server classifies any connection with a live subscription as a pub/sub client, and applies the pub/sub
+/// output-buffer limits to it (client-output-buffer-limit pubsub 32mb 8mb 60 by default) - far tighter
+/// than the (unlimited) normal limits, and well within what a busy interactive connection can legitimately need.
+/// A reply bigger than the hard limit gets the connection closed, taking every in-flight command with it. See #3263.
+///
+///
+/// Hence a dedicated subscription connection by default, even under RESP3; sharing the interactive connection
+/// is opt-in via , and these tests pin both
+/// sides of that trade. The oversized reply is an MGET of the same small value many times over: the
+/// whole reply is buffered (and checked against the limit) before any of it is written, so this needs only a
+/// small value and no change to the server's configuration.
+///
+[RunPerProtocol]
+public class PubSubOutputBufferTests(ITestOutputHelper output) : TestBase(output)
+{
+ private const int ValueSize = 64 * 1024;
+
+ [Fact]
+ public Task LargeReplyWithDedicatedSubscriptionConnection() => LargeReplyAsync(sharedSubscriptionConnection: false);
+
+ [Fact]
+ public Task LargeReplyWithSharedSubscriptionConnection() => LargeReplyAsync(sharedSubscriptionConnection: true); // fatal under RESP3
+
+ private async Task LargeReplyAsync(bool sharedSubscriptionConnection)
+ {
+ await using var conn = Create(
+ allowAdmin: true,
+ shared: false,
+ configuration: sharedSubscriptionConnection ? GetConfiguration() + ",sharedSubscriptionConnection=true" : null);
+ var server = GetServer(conn);
+ var endpoint = conn.GetServerEndPoint(server.EndPoint);
+
+ var hardLimit = await GetPubSubHardLimitAsync(server);
+ Assert.SkipUnless(hardLimit > 0, "Skipping because the server has no pub/sub output-buffer hard limit");
+ var repeat = checked((int)(hardLimit / ValueSize)) + 16;
+ Log($"pub/sub hard limit: {hardLimit} bytes; fetching {repeat} x {ValueSize} bytes");
+
+ // a subscription of our own, so the connection carrying it is a pub/sub client whatever else is going on
+ await conn.GetSubscriber().SubscribeAsync(RedisChannel.Literal(Me()), (_, _) => { });
+
+ var db = conn.GetDatabase();
+ RedisKey key = Me();
+ await db.StringSetAsync(key, new byte[ValueSize]);
+ var keys = Enumerable.Repeat(key, repeat).ToArray();
+ try
+ {
+ var mux = (IInternalConnectionMultiplexer)conn;
+ var interactiveId = mux.GetConnectionId(server.EndPoint, ConnectionType.Interactive);
+ Assert.NotNull(interactiveId);
+
+ if (endpoint.SharesSubscriptionConnection())
+ {
+ SetExpectedAmbientFailureCount(-1);
+ await Assert.ThrowsAsync(() => db.StringGetAsync(keys));
+ }
+ else
+ {
+ var values = await db.StringGetAsync(keys);
+ Assert.Equal(repeat, values.Length);
+ Assert.All(values, value => Assert.Equal(ValueSize, ((byte[])value!).Length));
+
+ // the same connection, which the server still sees as an ordinary client
+ Assert.Equal(interactiveId, mux.GetConnectionId(server.EndPoint, ConnectionType.Interactive));
+ var self = Assert.Single(await server.ClientListAsync(), x => x.Id == interactiveId);
+ Assert.Equal(ClientType.Normal, self.ClientType);
+ }
+ }
+ finally
+ {
+ await db.KeyDeleteAsync(key);
+ }
+ }
+
+ /// The pub/sub hard limit in bytes, or zero if there is none (or we cannot ask).
+ private static async Task GetPubSubHardLimitAsync(IServer server)
+ {
+ KeyValuePair[] config;
+ try
+ {
+ config = await server.ConfigGetAsync("client-output-buffer-limit");
+ }
+ catch (RedisServerException)
+ {
+ return 0; // managed services commonly disallow CONFIG
+ }
+
+ // "normal 0 0 0 slave 268435456 67108864 60 pubsub 33554432 8388608 60"
+ var tokens = config.FirstOrDefault().Value?.Split([' '], StringSplitOptions.RemoveEmptyEntries) ?? [];
+ var index = Array.IndexOf(tokens, "pubsub");
+ return index >= 0 && index + 1 < tokens.Length && long.TryParse(tokens[index + 1], out var hard) ? hard : 0;
+ }
+}
diff --git a/tests/StackExchange.Redis.Tests/Resp3DowngradeTests.cs b/tests/StackExchange.Redis.Tests/Resp3DowngradeTests.cs
index 2694cf430..aa9783469 100644
--- a/tests/StackExchange.Redis.Tests/Resp3DowngradeTests.cs
+++ b/tests/StackExchange.Redis.Tests/Resp3DowngradeTests.cs
@@ -1,4 +1,4 @@
-using System;
+using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
@@ -12,18 +12,22 @@ namespace StackExchange.Redis.Tests;
///
/// A connection can negotiate RESP3 and then, on a later reconnect, fail to (the HELLO times out, a token
/// expires, the endpoint moves to a down-level server, ...). We then downgrade to RESP2, which needs pub/sub on
-/// its own connection - but subscriptions queued while we still expected RESP3 are sitting in the *interactive*
-/// bridge's backlog. Writing those to the interactive connection puts it into subscriber mode, which breaks
+/// its own connection - but subscriptions queued while we still expected RESP3 (and were sharing the interactive
+/// connection with pub/sub; ) are sitting in the
+/// *interactive* bridge's backlog. Writing those to the interactive connection puts it into subscriber mode, which breaks
/// every normal command on it; see issue #3154.
///
public class Resp3DowngradeTests(ITestOutputHelper log)
{
- [Fact]
- public async Task SubscribeQueuedBeforeDowngradeDoesNotPoisonInteractive()
+ [Theory]
+ [InlineData(true)] // the #3154 scenario: subscriptions were queued on the interactive bridge
+ [InlineData(false)] // a dedicated subscription connection all along: nothing to reroute
+ public async Task SubscribeQueuedBeforeDowngradeDoesNotPoisonInteractive(bool sharedSubscriptionConnection)
{
using var server = new DowngradeServer(log);
var config = server.GetClientConfig();
config.Protocol = RedisProtocol.Resp3;
+ config.SharedSubscriptionConnection = sharedSubscriptionConnection;
config.AllowSimulateConnectionFailure = true;
config.AbortOnConnectFail = false;
config.ConnectTimeout = 5000;
@@ -130,8 +134,8 @@ protected override TypedRedisValue Hello(RedisClient client, in RedisRequest req
=> HelloSupported ? base.Hello(client, in request) : request.CommandNotFound();
///
- /// Connections that carried both subscription and ordinary commands; legitimate under RESP3 (one
- /// connection does everything), fatal under RESP2 (subscriber mode rejects ordinary commands).
+ /// Connections that carried both subscription and ordinary commands; legitimate under RESP3 when
+ /// sharing the connection with pub/sub (one connection does everything), fatal under RESP2 (subscriber mode rejects ordinary commands).
///
public List ConnectionsMixingSubscriptionsAndCommands(RedisProtocol protocol) =>
[.. _byClient