fix(session): keep the maintenance sweeps and capabilities with the session owner

Emby.Server.Implementations/Session/RemoteSessionController.cs

@@ -30,17 +30,19 @@ public sealed class RemoteSessionController : ISessionController
     /// <param name="ownerPod">The instance holding the connection.</param>
     /// <param name="sessionId">The session identifier.</param>
     /// <param name="supportsMediaControl">Whether the owner reported the session as controllable.</param>
-    public RemoteSessionController(IPodMessageBus bus, ILogger logger, string ownerPod, string sessionId, bool supportsMediaControl)
+    /// <param name="holdsConnection">Whether the owner reported that it holds the session's connection.</param>
+    public RemoteSessionController(IPodMessageBus bus, ILogger logger, string ownerPod, string sessionId, bool supportsMediaControl, bool holdsConnection)
     {
         _bus = bus;
         _logger = logger;
         _ownerPod = ownerPod;
         _sessionId = sessionId;
         SupportsMediaControl = supportsMediaControl;
+        IsSessionActive = holdsConnection;
     }
 
     /// <inheritdoc />
-    public bool IsSessionActive => true;
+    public bool IsSessionActive { get; }
 
     /// <inheritdoc />
     public bool SupportsMediaControl { get; }

Emby.Server.Implementations/Session/SessionManager.cs

@@ -523,7 +523,7 @@ namespace Emby.Server.Implementations.Session
                 Capabilities = dto.Capabilities?.ToClientCapabilities()
             };
 
-            session.AddController(new RemoteSessionController(_podMessageBus, _logger, entry.OwnerPod, dto.Id, dto.SupportsMediaControl));
+            session.AddController(new RemoteSessionController(_podMessageBus, _logger, entry.OwnerPod, dto.Id, dto.SupportsMediaControl, entry.HoldsConnection));
 
             return session;
         }
@@ -540,6 +540,8 @@ namespace Emby.Server.Implementations.Session
                     return Task.FromResult(OnRoutedAdditionalUserChange(message));
                 case RoutedNowViewingItem.Kind:
                     return Task.FromResult(OnRoutedNowViewingItem(message));
+                case RoutedCapabilities.Kind:
+                    return Task.FromResult(OnRoutedCapabilities(message));
                 case RoutedPlaybackReport.StartKind:
                 case RoutedPlaybackReport.ProgressKind:
                 case RoutedPlaybackReport.StoppedKind:
@@ -621,6 +623,23 @@ namespace Emby.Server.Implementations.Session
             return true;
         }
 
+        private bool OnRoutedCapabilities(PodMessage message)
+        {
+            var routed = JsonSerializer.Deserialize<RoutedCapabilities>(message.Payload, JsonDefaults.Options);
+            var session = routed?.Capabilities is null
+                ? null
+                : Sessions.FirstOrDefault(i => string.Equals(i.Id, routed.SessionId, StringComparison.Ordinal));
+
+            if (session is null)
+            {
+                return false;
+            }
+
+            ReportCapabilities(session, routed.Capabilities, true);
+
+            return true;
+        }
+
         private async Task<bool> OnRoutedPlaybackReport(PodMessage message)
         {
             try
@@ -677,7 +696,14 @@ namespace Emby.Server.Implementations.Session
 
                 if (!routed)
                 {
-                    _logger.LogWarning("Instance {OwnerPod} did not apply the {Kind} report for session {Session}; it is applied here instead.", entry.OwnerPod, kind, sessionId);
+                    // Delivery is at-least-once: an owner that applied the report and then failed to
+                    // acknowledge it has it applied here as well, so a stop can be applied twice. A
+                    // duplicate re-saves the same position; dropping the report loses it outright.
+                    _logger.LogWarning(
+                        "Instance {OwnerPod} did not acknowledge the {Kind} report for session {Session}; it is applied here as well, which may repeat one the owner already applied.",
+                        entry.OwnerPod,
+                        kind,
+                        sessionId);
                 }
 
                 return routed;
@@ -989,6 +1015,29 @@ namespace Emby.Server.Implementations.Session
             return users;
         }
 
+        // The sweeps stop playback and save user data, so they act only on sessions this instance owns.
+        // A copy left on a non-owner stops checking in as soon as reports route away, and stopping it would
+        // end the owner's live playback and save the stale position the copy last saw.
+        private async Task<bool> OwnsSessionAsync(SessionInfo session)
+        {
+            if (!_directoryEnabled)
+            {
+                return true;
+            }
+
+            try
+            {
+                var entry = await _sessionDirectory.GetAsync(session.Id).ConfigureAwait(false);
+
+                return entry is null || string.Equals(entry.OwnerPod, _podMessageBus.PodId, StringComparison.Ordinal);
+            }
+            catch (Exception ex)
+            {
+                _logger.LogDebug(ex, "Could not read the owner of session {Session}; it is left alone.", session.Id);
+                return false;
+            }
+        }
+
         private void StartCheckTimers()
         {
             _idleTimer ??= new Timer(CheckForIdlePlayback, null, TimeSpan.FromMinutes(5), TimeSpan.FromMinutes(5));
@@ -1033,6 +1082,11 @@ namespace Emby.Server.Implementations.Session
 
                 foreach (var session in idle)
                 {
+                    if (!await OwnsSessionAsync(session).ConfigureAwait(false))
+                    {
+                        continue;
+                    }
+
                     _logger.LogDebug("Session {0} has gone idle while playing", session.Id);
 
                     try
@@ -1067,6 +1121,11 @@ namespace Emby.Server.Implementations.Session
 
             foreach (var session in inactiveSessions)
             {
+                if (!await OwnsSessionAsync(session).ConfigureAwait(false))
+                {
+                    continue;
+                }
+
                 _logger.LogDebug("Session {Session} has been inactive for {InactiveTime} minutes. Stopping it.", session.Id, _config.Configuration.InactiveSessionThreshold);
 
                 try
@@ -2399,19 +2458,34 @@ namespace Emby.Server.Implementations.Session
         /// <param name="controllingSessionId">The controlling session identifier.</param>
         /// <param name="sessionId">The session identifier.</param>
         /// <param name="capabilities">The capabilities.</param>
+        /// <returns>A task representing the operation.</returns>
         /// <exception cref="SecurityException">The controlling user is not allowed to control the session.</exception>
-        public void ReportCapabilities(string controllingSessionId, string sessionId, ClientCapabilities capabilities)
+        public async Task ReportCapabilities(string controllingSessionId, string sessionId, ClientCapabilities capabilities)
         {
             CheckDisposed();
 
-            var session = GetSession(sessionId);
+            var session = await GetSessionForControl(sessionId).ConfigureAwait(false);
 
             if (!string.IsNullOrEmpty(controllingSessionId))
             {
-                AssertCanControl(session, GetSession(controllingSessionId));
+                AssertCanControl(session, await GetSessionForControl(controllingSessionId).ConfigureAwait(false));
             }
 
-            ReportCapabilities(session, capabilities, true);
+            // Capabilities decide whether the session is offered for remote control, and they are held in
+            // the owner's process, so the report has to reach the instance everyone else is shown.
+            var payload = new RoutedCapabilities { SessionId = sessionId, Capabilities = capabilities };
+            var routed = await TryRouteToOwnerAsync(sessionId, RoutedCapabilities.Kind, payload, CancellationToken.None).ConfigureAwait(false);
+
+            var local = GetSession(sessionId, false);
+            if (local is not null)
+            {
+                ReportCapabilities(local, capabilities, true);
+            }
+            else if (!routed)
+            {
+                throw new ResourceNotFoundException(
+                    string.Format(CultureInfo.InvariantCulture, "Session {0} not found.", sessionId));
+            }
         }
 
         private void ReportCapabilities(
@@ -2431,6 +2505,10 @@ namespace Emby.Server.Implementations.Session
                     });
 
                 _deviceManager.SaveCapabilities(session.DeviceId, capabilities);
+
+                // Capabilities decide whether the session is listed as controllable, so the change is not
+                // left to the throttle.
+                _ = PublishToDirectoryNowAsync(session);
             }
         }
 
@@ -2523,11 +2601,11 @@ namespace Emby.Server.Implementations.Session
                 return;
             }
 
-            var local = GetSession(sessionId, false);
-            if (local is not null)
-            {
-                SetNowViewingItem(local, itemId);
-            }
+            var local = GetSession(sessionId, false)
+                ?? throw new ResourceNotFoundException(
+                    string.Format(CultureInfo.InvariantCulture, "Session {0} not found.", sessionId));
+
+            SetNowViewingItem(local, itemId);
         }
 
         private void SetNowViewingItem(SessionInfo session, string itemId)

Jellyfin.Api/Controllers/SessionController.cs

@@ -365,13 +365,13 @@ public class SessionController : BaseJellyfinApiController
             id = currentSessionId;
         }
 
-        _sessionManager.ReportCapabilities(currentSessionId, id, new ClientCapabilities
+        await _sessionManager.ReportCapabilities(currentSessionId, id, new ClientCapabilities
         {
             PlayableMediaTypes = playableMediaTypes,
             SupportedCommands = supportedCommands,
             SupportsMediaControl = supportsMediaControl,
             SupportsPersistentIdentifier = supportsPersistentIdentifier
-        });
+        }).ConfigureAwait(false);
         return NoContent();
     }
 
@@ -395,7 +395,7 @@ public class SessionController : BaseJellyfinApiController
             id = currentSessionId;
         }
 
-        _sessionManager.ReportCapabilities(currentSessionId, id, capabilities.ToClientCapabilities());
+        await _sessionManager.ReportCapabilities(currentSessionId, id, capabilities.ToClientCapabilities()).ConfigureAwait(false);
 
         return NoContent();
     }

MediaBrowser.Controller/Session/ISessionManager.cs

@@ -278,7 +278,8 @@ namespace MediaBrowser.Controller.Session
         /// <param name="controllingSessionId">The controlling session identifier.</param>
         /// <param name="sessionId">The session identifier.</param>
         /// <param name="capabilities">The capabilities.</param>
-        void ReportCapabilities(string controllingSessionId, string sessionId, ClientCapabilities capabilities);
+        /// <returns>A task representing the operation.</returns>
+        Task ReportCapabilities(string controllingSessionId, string sessionId, ClientCapabilities capabilities);
 
         /// <summary>
         /// Reports the transcoding information.

MediaBrowser.Controller/Session/RoutedCapabilities.cs

@@ -0,0 +1,25 @@
+using MediaBrowser.Model.Session;
+
+namespace MediaBrowser.Controller.Session;
+
+/// <summary>
+/// A capabilities report for a session held by another instance, carried as a <see cref="PodMessage"/>.
+/// The calling instance has already authorized it.
+/// </summary>
+public sealed class RoutedCapabilities
+{
+    /// <summary>
+    /// The <see cref="PodMessage.Kind"/> this payload travels under.
+    /// </summary>
+    public const string Kind = "Capabilities";
+
+    /// <summary>
+    /// Gets or sets the session the report applies to.
+    /// </summary>
+    public string SessionId { get; set; } = string.Empty;
+
+    /// <summary>
+    /// Gets or sets the reported capabilities.
+    /// </summary>
+    public ClientCapabilities? Capabilities { get; set; }
+}