unkin/jellyfin-ha-src
4
0
Fork 0
Code Issues 12 Pull Requests 2 Packages Projects Releases Wiki Activity

fix(userdata): read user data through to the database
ci/woodpecker/pr/ci Pipeline was successful
Details
ci/woodpecker/push/ci Pipeline was successful
Details

Browse Source 
The user data cache and the item's in-memory rows were both filled once per
replica and never invalidated, so a pod serving a playback tick read its own
stale resume position and wrote it back over the position another pod had just
saved, silently losing resume points, played state, favourites and ratings.

- drop the user item data cache
- read single and batched user data from the database on every query
- prefetch user data for in-memory sorts and filters so each stays one query
- cover the lost update against real PostgreSQL with two manager instances
This commit is contained in:
Unkin Agent
2026-09-20 23:51:29 +10:00
parent ec581b5e5f
commit a7919b9bac
14 changed files with 470 additions and 136 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Emby.Server.Implementations/Library/UserDataManager.cs
+31 -74
View File
@@ -2,10 +2,8 @@
using System;
using System.Collections.Generic;
using System.Globalization;
using System.Linq;
using System.Threading;
using BitFaster.Caching.Lru;
using Jellyfin.Database.Implementations;
using Jellyfin.Database.Implementations.Entities;
using MediaBrowser.Controller.Configuration;
@@ -27,7 +25,6 @@ namespace Emby.Server.Implementations.Library
{
private readonly IServerConfigurationManager _config;
private readonly IDbContextFactory<JellyfinDbContext> _repository;
private readonly FastConcurrentLru<string, UserItemData> _cache;
/// <summary>
/// Initializes a new instance of the <see cref="UserDataManager"/> class.
@@ -40,7 +37,6 @@ namespace Emby.Server.Implementations.Library
{
_config = config;
_repository = repository;
_cache = new FastConcurrentLru<string, UserItemData>(Environment.ProcessorCount, _config.Configuration.CacheSize, StringComparer.OrdinalIgnoreCase);
}
/// <inheritdoc />
@@ -77,11 +73,6 @@ namespace Emby.Server.Implementations.Library
dbContext.SaveChanges();
transaction.Commit();
var userId = user.InternalId;
var cacheKey = GetCacheKey(userId, item.Id);
_cache.AddOrUpdate(cacheKey, userData);
item.UserData = dbContext.UserData.Where(e => e.ItemId == item.Id).AsNoTracking().ToArray(); // rehydrate the cached userdata
UserDataSaved?.Invoke(this, new UserDataSaveEventArgs
{
Keys = keys,
@@ -180,64 +171,41 @@ namespace Emby.Server.Implementations.Library
/// <inheritdoc />
public Dictionary<Guid, UserItemData> GetUserDataBatch(IReadOnlyList<BaseItem> items, User user)
{
ArgumentNullException.ThrowIfNull(items);
ArgumentNullException.ThrowIfNull(user);
var result = new Dictionary<Guid, UserItemData>(items.Count);
var itemsNeedingQuery = new List<(BaseItem Item, List<string> Keys)>();
foreach (var item in items)
{
var cacheKey = GetCacheKey(user.InternalId, item.Id);
if (_cache.TryGet(cacheKey, out var cachedData))
{
result[item.Id] = cachedData;
}
else
{
var userDataRow = ResolveUserDataRow(item, item.UserData?.Where(e => e.UserId.Equals(user.Id)));
var userData = userDataRow is not null ? Map(userDataRow) : null;
if (userData is not null)
{
result[item.Id] = userData;
_cache.AddOrUpdate(cacheKey, userData);
}
else
{
var keys = item.GetUserDataKeys();
itemsNeedingQuery.Add((item, keys));
}
}
}
if (itemsNeedingQuery.Count == 0)
if (items.Count == 0)
{
return result;
}
// Build a single query for all missing items. Fetch rows by item alone so rows kept
// under keys from older metadata resolve the same way as the in-memory path.
var allItemIds = itemsNeedingQuery.Select(x => x.Item.Id).ToList();
// Fetch rows by item alone so rows kept under keys from older metadata resolve the same
// way as the single item path.
var itemIds = items.Select(e => e.Id).Distinct().ToList();
using var context = _repository.CreateDbContext();
var userDataArray = context.UserData
var userDataByItem = context.UserData
.AsNoTracking()
.Where(e => e.UserId.Equals(user.Id))
.WhereOneOrMany(allItemIds, e => e.ItemId)
.ToArray();
.WhereOneOrMany(itemIds, e => e.ItemId)
.ToArray()
.GroupBy(e => e.ItemId)
.ToDictionary(g => g.Key, g => g.ToArray());
var userDataByItem = userDataArray.GroupBy(e => e.ItemId).ToDictionary(g => g.Key, g => g.ToArray());
foreach (var (item, keys) in itemsNeedingQuery)
foreach (var item in items)
{
UserItemData userData;
if (userDataByItem.TryGetValue(item.Id, out var itemUserData) && itemUserData.Length > 0)
if (result.ContainsKey(item.Id))
{
userData = Map(ResolveUserDataRow(item, itemUserData)!);
}
else
{
userData = new UserItemData { Key = keys.Count > 0 ? keys[0] : string.Empty };
continue;
}
result[item.Id] = userData;
var cacheKey = GetCacheKey(user.InternalId, item.Id);
_cache.AddOrUpdate(cacheKey, userData);
var row = userDataByItem.TryGetValue(item.Id, out var itemUserData)
? ResolveUserDataRow(item, itemUserData)
: null;
result[item.Id] = row is not null
? Map(row)
: new UserItemData { Key = item.GetUserDataKeys().FirstOrDefault() ?? string.Empty };
}
return result;
@@ -340,20 +308,19 @@ namespace Emby.Server.Implementations.Library
return result;
}
/// <summary>
/// Gets the internal key.
/// </summary>
/// <returns>System.String.</returns>
private static string GetCacheKey(long internalUserId, Guid itemId)
{
return internalUserId.ToString(CultureInfo.InvariantCulture) + "-" + itemId.ToString("N", CultureInfo.InvariantCulture);
}
/// <inheritdoc />
public UserItemData? GetUserData(User user, BaseItem item)
{
ArgumentNullException.ThrowIfNull(user);
var row = ResolveUserDataRow(item, item.UserData?.Where(e => e.UserId.Equals(user.Id)));
ArgumentNullException.ThrowIfNull(item);
using var dbContext = _repository.CreateDbContext();
var rows = dbContext.UserData
.AsNoTracking()
.Where(e => e.ItemId == item.Id && e.UserId == user.Id)
.ToArray();
var row = ResolveUserDataRow(item, rows);
return row is not null ? Map(row) : new UserItemData()
{
Key = item.GetUserDataKeys()[0],
@@ -536,16 +503,6 @@ namespace Emby.Server.Implementations.Library
}
dbContext.SaveChanges();
var cacheKey = GetCacheKey(user.InternalId, item.Id);
if (_cache.TryGet(cacheKey, out var cached))
{
cached.AudioStreamIndex = null;
cached.SubtitleStreamIndex = null;
_cache.AddOrUpdate(cacheKey, cached);
}
item.UserData = dbContext.UserData.Where(e => e.ItemId == item.Id).AsNoTracking().ToArray();
}
}
}