test(session): cover the maintenance sweeps and capabilities routing

tests/Jellyfin.Server.Implementations.Tests/SessionManager/SessionManagerTests.cs

@@ -223,7 +223,7 @@ public class SessionManagerTests
         var victimSession = await LogSessionActivity(sessionManager, victim);
         var attackerSession = await LogSessionActivity(sessionManager, attacker);
 
-        Assert.Throws<SecurityException>(() => sessionManager.ReportCapabilities(attackerSession.Id, victimSession.Id, new ClientCapabilities()));
+        await Assert.ThrowsAsync<SecurityException>(() => sessionManager.ReportCapabilities(attackerSession.Id, victimSession.Id, new ClientCapabilities()));
     }
 
     private static Emby.Server.Implementations.Session.SessionManager CreateSessionManager(params User[] users)

tests/Jellyfin.Server.Tests/HighAvailability/SessionDirectoryReplicaTests.cs

@@ -3,12 +3,16 @@ using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Globalization;
 using System.Linq;
+using System.Reflection;
 using System.Text.Json;
 using System.Threading;
 using System.Threading.Tasks;
+using Jellyfin.Data;
+using Jellyfin.Data.Enums;
 using Jellyfin.Database.Implementations;
 using Jellyfin.Database.Implementations.DbConfiguration;
 using Jellyfin.Database.Implementations.Entities;
+using Jellyfin.Database.Implementations.Enums;
 using Jellyfin.Database.Implementations.Locking;
 using Jellyfin.Database.Providers.PostgreSQL;
 using Jellyfin.Server.Implementations.Devices;
@@ -18,11 +22,14 @@ using MediaBrowser.Controller;
 using MediaBrowser.Controller.Configuration;
 using MediaBrowser.Controller.Drawing;
 using MediaBrowser.Controller.Dto;
+using MediaBrowser.Controller.Entities;
+using MediaBrowser.Controller.Entities.Movies;
 using MediaBrowser.Controller.Events;
 using MediaBrowser.Controller.Library;
 using MediaBrowser.Controller.Session;
 using MediaBrowser.Model.Configuration;
 using MediaBrowser.Model.Dto;
+using MediaBrowser.Model.Entities;
 using MediaBrowser.Model.Session;
 using MediaBrowser.Model.SyncPlay;
 using Microsoft.EntityFrameworkCore;
@@ -480,11 +487,15 @@ public sealed class SessionDirectoryReplicaTests : IAsyncLifetime
 
         var recorded = await ReadOwnerEpoch(session.Id, cancellationToken);
         var allocated = await _directory.AllocateConnectionEpochAsync(session.Id, cancellationToken);
+        var other = await _directory.AllocateConnectionEpochAsync(session.Id + "-other", cancellationToken);
 
-        // A counter the store owns, not a reading of any replica's clock.
-        Assert.Equal(1, recorded);
-        Assert.Equal(2, allocated);
-        Assert.Equal(1, await _directory.AllocateConnectionEpochAsync(session.Id + "-other", cancellationToken));
+        // A per-session counter the store hands out in order, orders of magnitude below any tick count,
+        // so it cannot be a reading of a replica's clock.
+        Assert.True(recorded > 0);
+        Assert.True(allocated > recorded);
+        Assert.True(other > 0);
+        Assert.True(other < await _directory.AllocateConnectionEpochAsync(session.Id + "-other", cancellationToken));
+        Assert.True(allocated < TimeSpan.TicksPerSecond);
     }
 
     /// <summary>
@@ -570,6 +581,157 @@ public sealed class SessionDirectoryReplicaTests : IAsyncLifetime
         await WaitUntil(() => !session.AdditionalUsers.Any(i => i.UserId.Equals(_guest.Id)), cancellationToken);
     }
 
+    /// <summary>
+    /// The idle sweep runs on every replica but stops playback and rewrites the resume position. A
+    /// replica holding a copy it does not own sees a check-in that froze when reports started routing
+    /// away, so an unguarded sweep would stop a film the other replica is still playing.
+    /// </summary>
+    /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
+    [Fact]
+    public async Task IdleSweepOnANonOwner_LeavesTheOwnersPlaybackAlone()
+    {
+        var cancellationToken = TestContext.Current.CancellationToken;
+        var stopped = new TaskCompletionSource<PlaybackStopEventArgs>(TaskCreationOptions.RunContinuationsAsynchronously);
+        var events = new Mock<IEventManager>();
+        events.Setup(i => i.PublishAsync(It.IsAny<PlaybackStopEventArgs>()))
+            .Callback<PlaybackStopEventArgs>(args => stopped.TrySetResult(args))
+            .Returns(Task.CompletedTask);
+
+        var movie = new Movie { Id = Guid.NewGuid(), Name = "Live Movie" };
+        var libraryManager = new Mock<ILibraryManager>();
+        libraryManager.Setup(i => i.GetItemById(movie.Id)).Returns(movie);
+
+        var userDataManager = new Mock<IUserDataManager>();
+        userDataManager.Setup(i => i.GetUserData(It.IsAny<User>(), It.IsAny<BaseItem>())).Returns(new UserItemData { Key = "test" });
+
+        await using var replicaA = CreateReplica(
+            "pod-a",
+            userDataManager: userDataManager.Object,
+            libraryManager: libraryManager.Object,
+            eventManager: events.Object);
+        await using var replicaB = CreateReplica("pod-b");
+
+        var owned = await Request(replicaA, "device-idle-sweep");
+        owned.AddController(new RecordingSessionController());
+        await replicaA.OnSessionControllerConnected(owned);
+
+        await replicaA.OnPlaybackStart(new PlaybackStartInfo
+        {
+            SessionId = owned.Id,
+            ItemId = movie.Id,
+            MediaSourceId = movie.Id.ToString("N", CultureInfo.InvariantCulture),
+            Item = new BaseItemDto { Id = movie.Id, Name = movie.Name },
+            PositionTicks = 0
+        });
+
+        // The copy a non-owner is left with: it still remembers a now playing item, and its check-in
+        // froze the moment reports started going to the owner instead.
+        var copy = await Request(replicaB, "device-idle-sweep");
+        copy.NowPlayingItem = new BaseItemDto { Id = movie.Id, Name = movie.Name };
+        copy.StartAutomaticProgress(new PlaybackProgressInfo { IsPaused = true, PositionTicks = 123456789 });
+        copy.StopAutomaticProgress();
+        copy.LastPlaybackCheckIn = DateTime.UtcNow.AddHours(-1);
+
+        InvokeSweep(replicaB, "CheckForIdlePlayback");
+
+        await Assert.ThrowsAsync<TimeoutException>(() => stopped.Task.WaitAsync(TimeSpan.FromSeconds(5), cancellationToken));
+
+        Assert.Equal(movie.Name, owned.NowPlayingItem?.Name);
+        Assert.Equal(movie.Name, Single(await replicaB.GetSessions(_user.Id, null, null, null, false, cancellationToken), owned.Id).NowPlayingItem?.Name);
+        userDataManager.Verify(
+            i => i.SaveUserData(It.IsAny<User>(), It.IsAny<BaseItem>(), It.IsAny<UserItemData>(), UserDataSaveReason.PlaybackFinished, It.IsAny<CancellationToken>()),
+            Times.Never);
+
+        // The same sweep on the owner does stop it, so the assertions above are not vacuous.
+        owned.LastPlaybackCheckIn = DateTime.UtcNow.AddHours(-1);
+        InvokeSweep(replicaA, "CheckForIdlePlayback");
+
+        await stopped.Task.WaitAsync(TimeSpan.FromSeconds(30), cancellationToken);
+
+        Assert.Null(owned.NowPlayingItem);
+    }
+
+    /// <summary>
+    /// The inactive sweep decides from the local copy's paused state and then writes a real Stop to
+    /// whichever replica holds the connection. A replica that only holds a copy has no business
+    /// stopping the session the other one is serving.
+    /// </summary>
+    /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
+    [Fact]
+    public async Task InactiveSweepOnANonOwner_DoesNotStopTheOwnersSession()
+    {
+        var cancellationToken = TestContext.Current.CancellationToken;
+        var configuration = new ServerConfiguration { InactiveSessionThreshold = 1 };
+
+        await using var replicaA = CreateReplica("pod-a", configuration: configuration);
+        await using var replicaB = CreateReplica("pod-b", configuration: configuration);
+
+        var owned = await Request(replicaA, "device-inactive-sweep");
+        var controller = new RecordingSessionController();
+        owned.AddController(controller);
+        await replicaA.OnSessionControllerConnected(owned);
+
+        owned.NowPlayingItem = new BaseItemDto { Id = Guid.NewGuid(), Name = "Paused Movie" };
+        owned.PlayState.IsPaused = true;
+        owned.LastPausedDate = DateTime.UtcNow.AddHours(-1);
+
+        var copy = await Request(replicaB, "device-inactive-sweep");
+        copy.NowPlayingItem = owned.NowPlayingItem;
+        copy.PlayState.IsPaused = true;
+        copy.LastPausedDate = DateTime.UtcNow.AddHours(-1);
+
+        InvokeSweep(replicaB, "CheckForInactiveSteams");
+
+        await Task.Delay(TimeSpan.FromSeconds(3), cancellationToken);
+
+        Assert.False(controller.HasMessage);
+
+        // The same sweep on the owner does write the Stop, so the assertion above is not vacuous.
+        InvokeSweep(replicaA, "CheckForInactiveSteams");
+
+        var (messageType, _) = await controller.WaitForMessageAsync(cancellationToken);
+        Assert.Equal(SessionMessageType.Playstate, messageType);
+    }
+
+    /// <summary>
+    /// Without sticky sessions a device posts its capabilities to either replica, and a session is only
+    /// offered for remote control while the replica that publishes it knows they support media control.
+    /// A report kept by the replica that served it drops the device from the cast list everywhere.
+    /// </summary>
+    /// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
+    [Fact]
+    public async Task CapabilitiesReportedToTheNonOwner_KeepTheDeviceControllable()
+    {
+        var cancellationToken = TestContext.Current.CancellationToken;
+        _user.SetPermission(PermissionKind.EnableAllDevices, true);
+
+        await using var replicaA = CreateReplica("pod-a");
+        await using var replicaB = CreateReplica("pod-b");
+
+        var session = await Request(replicaA, "device-capabilities");
+        session.AddController(new RecordingSessionController());
+        await replicaA.OnSessionControllerConnected(session);
+
+        // The web client posts its capabilities to whichever replica the load balancer picked.
+        await Request(replicaB, "device-capabilities");
+        await replicaB.ReportCapabilities(string.Empty, session.Id, new ClientCapabilities
+        {
+            PlayableMediaTypes = [MediaType.Video],
+            SupportedCommands = [GeneralCommandType.DisplayMessage],
+            SupportsMediaControl = true,
+            SupportsPersistentIdentifier = true
+        });
+
+        Assert.True(session.SupportsRemoteControl);
+
+        await WaitUntilAsync(
+            async () => (await ControllableIds(replicaB, cancellationToken)).Contains(session.Id),
+            cancellationToken);
+
+        Assert.Contains(session.Id, await ControllableIds(replicaA, cancellationToken));
+        Assert.Contains(session.Id, await ControllableIds(replicaB, cancellationToken));
+    }
+
     /// <summary>
     /// A replica that dies stops refreshing its entries, and the sessions it held have to leave the
     /// directory rather than linger in every other replica's session list forever.
@@ -633,6 +795,21 @@ public sealed class SessionDirectoryReplicaTests : IAsyncLifetime
         }
     }
 
+    private static async Task WaitUntilAsync(Func<Task<bool>> condition, CancellationToken cancellationToken)
+    {
+        var deadline = DateTime.UtcNow.AddSeconds(30);
+        while (!await condition())
+        {
+            Assert.True(DateTime.UtcNow < deadline, "The expected change never arrived.");
+            await Task.Delay(100, cancellationToken);
+        }
+    }
+
+    private static void InvokeSweep(SessionManager replica, string name)
+        => typeof(SessionManager)
+            .GetMethod(name, BindingFlags.Instance | BindingFlags.NonPublic)!
+            .Invoke(replica, [null]);
+
     private static JellyfinDbContext CreateContext(NpgsqlDataSource dataSource)
     {
         var optionsBuilder = new DbContextOptionsBuilder<JellyfinDbContext>();
@@ -648,7 +825,19 @@ public sealed class SessionDirectoryReplicaTests : IAsyncLifetime
     private Task<SessionInfo> Request(SessionManager replica, string deviceId)
         => replica.LogSessionActivity(AppName, AppVersion, deviceId, DeviceName, RemoteEndPoint, _user);
 
-    private SessionManager CreateReplica(string podId, SessionDirectoryOptions? options = null, ILogger<SessionManager>? logger = null)
+    private async Task<IReadOnlyList<string>> ControllableIds(SessionManager replica, CancellationToken cancellationToken)
+        => (await replica.GetSessions(_user.Id, null, null, _user.Id, false, cancellationToken))
+            .Select(i => i.Id ?? string.Empty)
+            .ToList();
+
+    private SessionManager CreateReplica(
+        string podId,
+        SessionDirectoryOptions? options = null,
+        ILogger<SessionManager>? logger = null,
+        IUserDataManager? userDataManager = null,
+        ILibraryManager? libraryManager = null,
+        IEventManager? eventManager = null,
+        ServerConfiguration? configuration = null)
     {
         options ??= new SessionDirectoryOptions { EntryTtlSeconds = 60, RefreshIntervalSeconds = 3600 };
 
@@ -657,7 +846,7 @@ public sealed class SessionDirectoryReplicaTests : IAsyncLifetime
             Options.Create(options),
             NullLogger<RedisSessionDirectory>.Instance);
 
-        return CreateReplica(podId, options, directory, CreateBus(podId, options), logger);
+        return CreateReplica(podId, options, directory, CreateBus(podId, options), logger, userDataManager, libraryManager, eventManager, configuration);
     }
 
     private SessionManager CreateReplica(string podId, ISessionDirectory directory)
@@ -666,7 +855,16 @@ public sealed class SessionDirectoryReplicaTests : IAsyncLifetime
     private SessionManager CreateReplica(string podId, ISessionDirectory directory, IPodMessageBus bus)
         => CreateReplica(podId, new SessionDirectoryOptions(), directory, bus, null);
 
-    private SessionManager CreateReplica(string podId, SessionDirectoryOptions options, ISessionDirectory directory, IPodMessageBus bus, ILogger<SessionManager>? logger)
+    private SessionManager CreateReplica(
+        string podId,
+        SessionDirectoryOptions options,
+        ISessionDirectory directory,
+        IPodMessageBus bus,
+        ILogger<SessionManager>? logger,
+        IUserDataManager? userDataManager = null,
+        ILibraryManager? libraryManager = null,
+        IEventManager? eventManager = null,
+        ServerConfiguration? configuration = null)
     {
         var userManager = new Mock<IUserManager>();
         userManager.Setup(i => i.GetUserById(_user.Id)).Returns(_user);
@@ -676,14 +874,14 @@ public sealed class SessionDirectoryReplicaTests : IAsyncLifetime
         appHost.SetupGet(i => i.SystemId).Returns("server-" + podId);
 
         var configurationManager = new Mock<IServerConfigurationManager>();
-        configurationManager.SetupGet(i => i.Configuration).Returns(new ServerConfiguration());
+        configurationManager.SetupGet(i => i.Configuration).Returns(configuration ?? new ServerConfiguration());
 
         return new SessionManager(
             logger ?? NullLogger<SessionManager>.Instance,
-            Mock.Of<IEventManager>(),
-            Mock.Of<IUserDataManager>(),
+            eventManager ?? Mock.Of<IEventManager>(),
+            userDataManager ?? Mock.Of<IUserDataManager>(),
             configurationManager.Object,
-            Mock.Of<ILibraryManager>(),
+            libraryManager ?? Mock.Of<ILibraryManager>(),
             userManager.Object,
             Mock.Of<IMusicManager>(),
             Mock.Of<IDtoService>(),