Close a change batch on its own window so a library scan cannot grow it without bound

Emby.Server.Implementations/EntryPoints/UserDataChangeNotifier.cs

@@ -18,15 +18,17 @@ namespace Emby.Server.Implementations.EntryPoints
     public sealed class UserDataChangeNotifier : IHostedService, IDisposable
     {
         private const int UpdateDuration = 500;
+        internal const int MaxBatchSize = 2000;
 
         private readonly ISessionManager _sessionManager;
         private readonly IUserDataManager _userDataManager;
         private readonly IUserManager _userManager;
 
-        private readonly Dictionary<Guid, List<BaseItem>> _changedItems = new();
+        private readonly Dictionary<Guid, Dictionary<Guid, BaseItem>> _changedItems = [];
         private readonly Lock _syncLock = new();
 
         private Timer? _updateTimer;
+        private int _changedItemCount;
 
         /// <summary>
         /// Initializes a new instance of the <see cref="UserDataChangeNotifier"/> class.
@@ -69,50 +71,64 @@ namespace Emby.Server.Implementations.EntryPoints
 
             lock (_syncLock)
             {
-                if (_updateTimer is null)
-                {
-                    _updateTimer = new Timer(
-                        UpdateTimerCallback,
-                        null,
-                        UpdateDuration,
-                        Timeout.Infinite);
-                }
-                else
-                {
-                    _updateTimer.Change(UpdateDuration, Timeout.Infinite);
-                }
+                // The window runs from the first change of a batch and is never extended, so a stream
+                // of changes that never pauses - a library scan - still closes its batches instead of
+                // holding every item it touched alive until the stream stops.
+                _updateTimer ??= new Timer(
+                    UpdateTimerCallback,
+                    null,
+                    UpdateDuration,
+                    Timeout.Infinite);
 
-                if (!_changedItems.TryGetValue(e.UserId, out List<BaseItem>? keys))
+                if (!_changedItems.TryGetValue(e.UserId, out Dictionary<Guid, BaseItem>? keys))
                 {
-                    keys = new List<BaseItem>();
+                    keys = [];
                     _changedItems[e.UserId] = keys;
                 }
 
-                keys.Add(e.Item);
-
                 var baseItem = e.Item;
 
                 // Go up one level for indicators
                 if (baseItem is not null)
                 {
+                    Track(keys, baseItem);
+
                     var parent = baseItem.GetOwner() ?? baseItem.GetParent();
 
                     if (parent is not null)
                     {
-                        keys.Add(parent);
+                        Track(keys, parent);
                     }
                 }
+
+                // A window long enough to cover a burst still has to give way once the batch is
+                // large enough to be worth sending on its own.
+                if (_changedItemCount >= MaxBatchSize)
+                {
+                    _updateTimer.Change(0, Timeout.Infinite);
+                }
+            }
+        }
+
+        private void Track(Dictionary<Guid, BaseItem> keys, BaseItem item)
+        {
+            var before = keys.Count;
+            keys[item.Id] = item;
+
+            if (keys.Count != before)
+            {
+                _changedItemCount++;
             }
         }
 
         private async void UpdateTimerCallback(object? state)
         {
-            List<KeyValuePair<Guid, List<BaseItem>>> changes;
+            List<KeyValuePair<Guid, Dictionary<Guid, BaseItem>>> changes;
             lock (_syncLock)
             {
-                // Remove dupes in case some were saved multiple times
                 changes = _changedItems.ToList();
                 _changedItems.Clear();
+                _changedItemCount = 0;
 
                 if (_updateTimer is not null)
                 {
@@ -121,17 +137,22 @@ namespace Emby.Server.Implementations.EntryPoints
                 }
             }
 
+            if (changes.Count == 0)
+            {
+                return;
+            }
+
             foreach (var (userId, changedItems) in changes)
             {
                 await _sessionManager.SendMessageToUserSessions(
                     [userId],
                     SessionMessageType.UserDataChanged,
-                    () => GetUserDataChangeInfo(userId, changedItems),
+                    () => GetUserDataChangeInfo(userId, changedItems.Values),
                     default).ConfigureAwait(false);
             }
         }
 
-        private UserDataChangeInfo GetUserDataChangeInfo(Guid userId, List<BaseItem> changedItems)
+        private UserDataChangeInfo GetUserDataChangeInfo(Guid userId, IEnumerable<BaseItem> changedItems)
         {
             var user = _userManager.GetUserById(userId)
                 ?? throw new ArgumentException("Invalid user ID", nameof(userId));
@@ -140,7 +161,6 @@ namespace Emby.Server.Implementations.EntryPoints
             {
                 UserId = userId,
                 UserDataList = changedItems
-                    .DistinctBy(x => x.Id)
                     .Select(i =>
                     {
                         var dto = _userDataManager.GetUserDataDto(i, user);