Skip to content

Commit ef388bd

Browse files
authored
Fix #2392: Dequeue all timed out messages from the backlog when not connected, even when no completion is needed, to be able to dequeue and complete other timed out messages. (#2397)
When the client is not connected timed out fire and forget messages currently are not removed from the backlog that also results in subsequent timed out messages not being marked as timed out, as described in #2392.
1 parent 1364ef8 commit ef388bd

File tree

3 files changed

+59
-7
lines changed

3 files changed

+59
-7
lines changed

docs/ReleaseNotes.md

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@ Current package versions:
88

99
## Unreleased
1010

11+
- Fix [#2392](https://github.com/StackExchange/StackExchange.Redis/issues/2392): Dequeue *all* timed out messages from the backlog when not connected (including Fire+Forget) ([#2397 by kornelpal](https://github.com/StackExchange/StackExchange.Redis/pull/2397))
1112
- Fix [#2400](https://github.com/StackExchange/StackExchange.Redis/issues/2400): Expose `ChannelMessageQueue` as `IAsyncEnumerable<ChannelMessage>` ([#2402 by mgravell](https://github.com/StackExchange/StackExchange.Redis/pull/2402))
1213
- Add: support for `CLIENT SETINFO` (lib name/version) during handshake; opt-out is via `ConfigurationOptions`; also support read of `resp`, `lib-ver` and `lib-name` via `CLIENT LIST` ([#2414 by mgravell](https://github.com/StackExchange/StackExchange.Redis/pull/2414))
1314

src/StackExchange.Redis/PhysicalBridge.cs

Lines changed: 12 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -869,25 +869,30 @@ private void CheckBacklogForTimeouts()
869869
while (_backlog.TryPeek(out Message? message))
870870
{
871871
// See if the message has pass our async timeout threshold
872-
// or has otherwise been completed (e.g. a sync wait timed out) which would have cleared the ResultBox
873-
if (!message.HasTimedOut(now, timeout, out var _) || message.ResultBox == null) break; // not a timeout - we can stop looking
872+
// Note: All timed out messages must be dequeued, even when no completion is needed, to be able to dequeue and complete other timed out messages.
873+
if (!message.HasTimedOut(now, timeout, out var _)) break; // not a timeout - we can stop looking
874874
lock (_backlog)
875875
{
876876
// Peek again since we didn't have lock before...
877877
// and rerun the exact same checks as above, note that it may be a different message now
878878
if (!_backlog.TryPeek(out message)) break;
879-
if (!message.HasTimedOut(now, timeout, out var _) && message.ResultBox != null) break;
879+
if (!message.HasTimedOut(now, timeout, out var _)) break;
880880

881881
if (!BacklogTryDequeue(out var message2) || (message != message2)) // consume it for real
882882
{
883883
throw new RedisException("Thread safety bug detected! A queue message disappeared while we had the backlog lock");
884884
}
885885
}
886886

887-
// Tell the message it has failed
888-
// Note: Attempting to *avoid* reentrancy/deadlock issues by not holding the lock while completing messages.
889-
var ex = Multiplexer.GetException(WriteResult.TimeoutBeforeWrite, message, ServerEndPoint);
890-
message.SetExceptionAndComplete(ex, this);
887+
// We only handle async timeouts here, synchronous timeouts are handled upstream.
888+
// Those sync timeouts happen in ConnectionMultiplexer.ExecuteSyncImpl() via Monitor.Wait.
889+
if (message.ResultBoxIsAsync)
890+
{
891+
// Tell the message it has failed
892+
// Note: Attempting to *avoid* reentrancy/deadlock issues by not holding the lock while completing messages.
893+
var ex = Multiplexer.GetException(WriteResult.TimeoutBeforeWrite, message, ServerEndPoint);
894+
message.SetExceptionAndComplete(ex, this);
895+
}
891896
}
892897
}
893898

Lines changed: 46 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,46 @@
1+
using System;
2+
using System.Threading.Tasks;
3+
using Xunit;
4+
using Xunit.Abstractions;
5+
6+
namespace StackExchange.Redis.Tests.Issues
7+
{
8+
public class Issue2392Tests : TestBase
9+
{
10+
public Issue2392Tests(ITestOutputHelper output) : base(output) { }
11+
12+
[Fact]
13+
public async Task Execute()
14+
{
15+
var options = new ConfigurationOptions()
16+
{
17+
BacklogPolicy = new()
18+
{
19+
QueueWhileDisconnected = true,
20+
AbortPendingOnConnectionFailure = false,
21+
},
22+
AbortOnConnectFail = false,
23+
ConnectTimeout = 1,
24+
ConnectRetry = 0,
25+
AsyncTimeout = 1,
26+
SyncTimeout = 1,
27+
AllowAdmin = true,
28+
};
29+
options.EndPoints.Add("127.0.0.1:1234");
30+
31+
using var conn = await ConnectionMultiplexer.ConnectAsync(options, Writer);
32+
var key = Me();
33+
var db = conn.GetDatabase();
34+
var server = conn.GetServerSnapshot()[0];
35+
36+
// Fail the connection
37+
conn.AllowConnect = false;
38+
server.SimulateConnectionFailure(SimulatedFailureType.All);
39+
Assert.False(conn.IsConnected);
40+
41+
await db.StringGetAsync(key, flags: CommandFlags.FireAndForget);
42+
var ex = await Assert.ThrowsAnyAsync<Exception>(() => db.StringGetAsync(key).WithTimeout(5000));
43+
Assert.True(ex is RedisTimeoutException or RedisConnectionException);
44+
}
45+
}
46+
}

0 commit comments

Comments
 (0)