From ce7e9f0c3d74a4bbae4af13e840e544ab6098da9 Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Sat, 26 Sep 2026 22:11:03 +1000 Subject: [PATCH] fix(session): bound a routed request and stop hiding a failed subscribe --- .../Session/RedisPodMessageBus.cs | 22 +++++++++---------- 1 file changed, 11 insertions(+), 11 deletions(-) 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;