diff --git a/Emby.Server.Implementations/Session/RedisPodMessageBus.cs b/Emby.Server.Implementations/Session/RedisPodMessageBus.cs
index 3db25ab013..5b6a48063c 100644
--- a/Emby.Server.Implementations/Session/RedisPodMessageBus.cs
+++ b/Emby.Server.Implementations/Session/RedisPodMessageBus.cs
@@ -52,14 +52,9 @@ public sealed class RedisPodMessageBus : IPodMessageBus
_timeout = TimeSpan.FromSeconds(Math.Max(1, options.Value.OperationTimeoutSeconds));
PodId = podId;
- try
- {
- _subscriber.Subscribe(RedisChannel.Literal(ChannelPrefix + PodId), (_, value) => Dispatch(value));
- }
- catch (Exception ex)
- {
- _logger.LogWarning(ex, "Failed to subscribe to {PodId}; messages routed here are dropped.", PodId);
- }
+ // A bus that cannot subscribe can only send, so every request it makes waits out the timeout and
+ // nothing routed here is ever answered. The caller degrades the pair to single-instance instead.
+ _subscriber.Subscribe(RedisChannel.Literal(ChannelPrefix + PodId), (_, value) => Dispatch(value));
}
///
@@ -77,20 +72,25 @@ public sealed class RedisPodMessageBus : IPodMessageBus
var acknowledged = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
_pending[message.CorrelationId] = acknowledged;
+ // Publish and acknowledgement share one deadline, so a request is bounded by the timeout rather
+ // than by twice it.
+ using var deadline = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
+ deadline.CancelAfter(_timeout);
+
try
{
var subscribers = await _subscriber.PublishAsync(
RedisChannel.Literal(ChannelPrefix + targetPod),
- JsonSerializer.Serialize(message, _jsonOptions)).WaitAsync(_timeout, cancellationToken).ConfigureAwait(false);
+ JsonSerializer.Serialize(message, _jsonOptions)).WaitAsync(deadline.Token).ConfigureAwait(false);
if (subscribers == 0)
{
return false;
}
- return await acknowledged.Task.WaitAsync(_timeout, cancellationToken).ConfigureAwait(false);
+ return await acknowledged.Task.WaitAsync(deadline.Token).ConfigureAwait(false);
}
- catch (TimeoutException)
+ catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested)
{
_logger.LogWarning("Instance {TargetPod} did not acknowledge a {Kind} message within {Timeout}.", targetPod, message.Kind, _timeout);
return false;