merge: resolve CoreAppHost conflict with main, drop dead redisConnectionString local
This commit is contained in:
@@ -0,0 +1,56 @@
|
||||
using System;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using MediaBrowser.Controller.MediaEncoding;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Hosting;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using StackExchange.Redis;
|
||||
|
||||
namespace Emby.Server.Implementations.MediaEncoding;
|
||||
|
||||
/// <summary>
|
||||
/// Pings the configured Redis transcode session store once at startup so an unreachable store is
|
||||
/// reported there instead of being discovered as a silent loss of cross-pod takeover.
|
||||
/// </summary>
|
||||
public sealed class TranscodeStoreConnectivityProbe : IHostedService
|
||||
{
|
||||
private readonly IServiceProvider _serviceProvider;
|
||||
private readonly ILogger<TranscodeStoreConnectivityProbe> _logger;
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="TranscodeStoreConnectivityProbe"/> class.
|
||||
/// </summary>
|
||||
/// <param name="serviceProvider">The service provider used to resolve the Redis connection.</param>
|
||||
/// <param name="logger">The logger.</param>
|
||||
public TranscodeStoreConnectivityProbe(IServiceProvider serviceProvider, ILogger<TranscodeStoreConnectivityProbe> logger)
|
||||
{
|
||||
_serviceProvider = serviceProvider;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task StartAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
try
|
||||
{
|
||||
// Resolved here rather than injected: connecting must not be able to abort startup.
|
||||
var redis = _serviceProvider.GetRequiredService<IConnectionMultiplexer>();
|
||||
var roundTrip = await redis.GetDatabase().PingAsync().ConfigureAwait(false);
|
||||
|
||||
_logger.LogInformation(
|
||||
"Redis transcode session store is reachable ({RoundTripMs}ms round trip). HA transcode takeover is active.",
|
||||
(long)roundTrip.TotalMilliseconds);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(
|
||||
ex,
|
||||
"Redis transcode session store is configured but UNREACHABLE. HA transcode takeover is not working: sessions stay local to this instance and are lost when it restarts. Check {Key}.",
|
||||
TranscodeStoreOptions.RedisConnectionStringKey);
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task StopAsync(CancellationToken cancellationToken) => Task.CompletedTask;
|
||||
}
|
||||
@@ -256,7 +256,7 @@ namespace Emby.Server.Implementations.Session
|
||||
ArgumentException.ThrowIfNullOrEmpty(deviceId);
|
||||
|
||||
var activityDate = DateTime.UtcNow;
|
||||
var session = GetSessionInfo(appName, appVersion, deviceId, deviceName, remoteEndPoint, user);
|
||||
var session = await GetSessionInfo(appName, appVersion, deviceId, deviceName, remoteEndPoint, user).ConfigureAwait(false);
|
||||
var lastActivityDate = session.LastActivityDate;
|
||||
session.LastActivityDate = activityDate;
|
||||
|
||||
@@ -491,7 +491,7 @@ namespace Emby.Server.Implementations.Session
|
||||
/// <param name="remoteEndPoint">The remote end point.</param>
|
||||
/// <param name="user">The user.</param>
|
||||
/// <returns>SessionInfo.</returns>
|
||||
private SessionInfo GetSessionInfo(
|
||||
private async Task<SessionInfo> GetSessionInfo(
|
||||
string appName,
|
||||
string appVersion,
|
||||
string deviceId,
|
||||
@@ -504,7 +504,7 @@ namespace Emby.Server.Implementations.Session
|
||||
ArgumentException.ThrowIfNullOrEmpty(deviceId);
|
||||
|
||||
var key = GetSessionKey(appName, deviceId, user?.Id ?? Guid.Empty);
|
||||
SessionInfo newSession = CreateSessionInfo(key, appName, appVersion, deviceId, deviceName, remoteEndPoint, user);
|
||||
SessionInfo newSession = await CreateSessionInfo(key, appName, appVersion, deviceId, deviceName, remoteEndPoint, user).ConfigureAwait(false);
|
||||
SessionInfo sessionInfo = _activeConnections.GetOrAdd(key, newSession);
|
||||
if (ReferenceEquals(newSession, sessionInfo))
|
||||
{
|
||||
@@ -532,7 +532,7 @@ namespace Emby.Server.Implementations.Session
|
||||
return sessionInfo;
|
||||
}
|
||||
|
||||
private SessionInfo CreateSessionInfo(
|
||||
private async Task<SessionInfo> CreateSessionInfo(
|
||||
string key,
|
||||
string appName,
|
||||
string appVersion,
|
||||
@@ -562,7 +562,7 @@ namespace Emby.Server.Implementations.Session
|
||||
deviceName = "Network Device";
|
||||
}
|
||||
|
||||
var deviceOptions = _deviceManager.GetDeviceOptions(deviceId) ?? new()
|
||||
var deviceOptions = await _deviceManager.GetDeviceOptions(deviceId).ConfigureAwait(false) ?? new()
|
||||
{
|
||||
DeviceId = deviceId
|
||||
};
|
||||
@@ -1768,12 +1768,12 @@ namespace Emby.Server.Implementations.Session
|
||||
// This should be validated above, but if it isn't don't delete all tokens.
|
||||
ArgumentException.ThrowIfNullOrEmpty(deviceId);
|
||||
|
||||
var existing = _deviceManager.GetDevices(
|
||||
var existing = (await _deviceManager.GetDevices(
|
||||
new DeviceQuery
|
||||
{
|
||||
DeviceId = deviceId,
|
||||
UserId = user.Id
|
||||
}).Items;
|
||||
}).ConfigureAwait(false)).Items;
|
||||
|
||||
foreach (var auth in existing)
|
||||
{
|
||||
@@ -1801,12 +1801,12 @@ namespace Emby.Server.Implementations.Session
|
||||
|
||||
ArgumentException.ThrowIfNullOrEmpty(accessToken);
|
||||
|
||||
var existing = _deviceManager.GetDevices(
|
||||
var existing = (await _deviceManager.GetDevices(
|
||||
new DeviceQuery
|
||||
{
|
||||
Limit = 1,
|
||||
AccessToken = accessToken
|
||||
}).Items;
|
||||
}).ConfigureAwait(false)).Items;
|
||||
|
||||
if (existing.Count > 0)
|
||||
{
|
||||
@@ -1845,10 +1845,10 @@ namespace Emby.Server.Implementations.Session
|
||||
{
|
||||
CheckDisposed();
|
||||
|
||||
var existing = _deviceManager.GetDevices(new DeviceQuery
|
||||
var existing = await _deviceManager.GetDevices(new DeviceQuery
|
||||
{
|
||||
UserId = userId
|
||||
});
|
||||
}).ConfigureAwait(false);
|
||||
|
||||
foreach (var info in existing.Items)
|
||||
{
|
||||
@@ -2045,11 +2045,11 @@ namespace Emby.Server.Implementations.Session
|
||||
/// <inheritdoc />
|
||||
public async Task<SessionInfo> GetSessionByAuthenticationToken(string token, string deviceId, string remoteEndpoint)
|
||||
{
|
||||
var items = _deviceManager.GetDevices(new DeviceQuery
|
||||
var items = (await _deviceManager.GetDevices(new DeviceQuery
|
||||
{
|
||||
AccessToken = token,
|
||||
Limit = 1
|
||||
}).Items;
|
||||
}).ConfigureAwait(false)).Items;
|
||||
|
||||
if (items.Count == 0)
|
||||
{
|
||||
|
||||
@@ -50,10 +50,10 @@ public class DevicesController : BaseJellyfinApiController
|
||||
/// <returns>An <see cref="OkResult"/> containing the list of devices.</returns>
|
||||
[HttpGet]
|
||||
[ProducesResponseType(StatusCodes.Status200OK)]
|
||||
public ActionResult<QueryResult<DeviceInfoDto>> GetDevices([FromQuery] Guid? userId)
|
||||
public async Task<ActionResult<QueryResult<DeviceInfoDto>>> GetDevices([FromQuery] Guid? userId)
|
||||
{
|
||||
userId = RequestHelpers.GetUserId(User, userId);
|
||||
return _deviceManager.GetDevicesForUser(userId);
|
||||
return await _deviceManager.GetDevicesForUser(userId).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
@@ -66,9 +66,9 @@ public class DevicesController : BaseJellyfinApiController
|
||||
[HttpGet("Info")]
|
||||
[ProducesResponseType(StatusCodes.Status200OK)]
|
||||
[ProducesResponseType(StatusCodes.Status404NotFound)]
|
||||
public ActionResult<DeviceInfoDto> GetDeviceInfo([FromQuery, Required] string id)
|
||||
public async Task<ActionResult<DeviceInfoDto>> GetDeviceInfo([FromQuery, Required] string id)
|
||||
{
|
||||
var deviceInfo = _deviceManager.GetDevice(id);
|
||||
var deviceInfo = await _deviceManager.GetDevice(id).ConfigureAwait(false);
|
||||
if (deviceInfo is null)
|
||||
{
|
||||
return NotFound();
|
||||
@@ -87,9 +87,9 @@ public class DevicesController : BaseJellyfinApiController
|
||||
[HttpGet("Options")]
|
||||
[ProducesResponseType(StatusCodes.Status200OK)]
|
||||
[ProducesResponseType(StatusCodes.Status404NotFound)]
|
||||
public ActionResult<DeviceOptionsDto> GetDeviceOptions([FromQuery, Required] string id)
|
||||
public async Task<ActionResult<DeviceOptionsDto>> GetDeviceOptions([FromQuery, Required] string id)
|
||||
{
|
||||
var deviceInfo = _deviceManager.GetDeviceOptions(id);
|
||||
var deviceInfo = await _deviceManager.GetDeviceOptions(id).ConfigureAwait(false);
|
||||
if (deviceInfo is null)
|
||||
{
|
||||
return NotFound();
|
||||
@@ -127,7 +127,12 @@ public class DevicesController : BaseJellyfinApiController
|
||||
[ProducesResponseType(StatusCodes.Status400BadRequest)]
|
||||
public async Task<ActionResult> DeleteDevice([FromQuery] string[] id)
|
||||
{
|
||||
var devices = id.Select(_deviceManager.GetDevice).ToArray();
|
||||
var devices = new List<DeviceInfoDto?>(id.Length);
|
||||
foreach (var deviceId in id)
|
||||
{
|
||||
devices.Add(await _deviceManager.GetDevice(deviceId).ConfigureAwait(false));
|
||||
}
|
||||
|
||||
if (devices.Any(f => f is null))
|
||||
{
|
||||
return BadRequest();
|
||||
@@ -135,7 +140,7 @@ public class DevicesController : BaseJellyfinApiController
|
||||
|
||||
foreach (var device in devices)
|
||||
{
|
||||
var sessions = _deviceManager.GetDevices(new DeviceQuery { DeviceId = device!.Id });
|
||||
var sessions = await _deviceManager.GetDevices(new DeviceQuery { DeviceId = device!.Id }).ConfigureAwait(false);
|
||||
|
||||
foreach (var session in sessions.Items)
|
||||
{
|
||||
|
||||
@@ -31,8 +31,6 @@ namespace Jellyfin.Server.Implementations.Devices
|
||||
private readonly IDbContextFactory<JellyfinDbContext> _dbProvider;
|
||||
private readonly IUserManager _userManager;
|
||||
private readonly ConcurrentDictionary<string, ClientCapabilities> _capabilitiesMap = new();
|
||||
private readonly ConcurrentDictionary<int, Device> _devices;
|
||||
private readonly ConcurrentDictionary<string, DeviceOptions> _deviceOptions;
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="DeviceManager"/> class.
|
||||
@@ -43,23 +41,6 @@ namespace Jellyfin.Server.Implementations.Devices
|
||||
{
|
||||
_dbProvider = dbProvider;
|
||||
_userManager = userManager;
|
||||
_devices = new ConcurrentDictionary<int, Device>();
|
||||
_deviceOptions = new ConcurrentDictionary<string, DeviceOptions>();
|
||||
|
||||
using var dbContext = _dbProvider.CreateDbContext();
|
||||
foreach (var device in dbContext.Devices
|
||||
.OrderBy(d => d.Id)
|
||||
.AsEnumerable())
|
||||
{
|
||||
_devices.TryAdd(device.Id, device);
|
||||
}
|
||||
|
||||
foreach (var deviceOption in dbContext.DeviceOptions
|
||||
.OrderBy(d => d.Id)
|
||||
.AsEnumerable())
|
||||
{
|
||||
_deviceOptions.TryAdd(deviceOption.DeviceId, deviceOption);
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
@@ -89,8 +70,6 @@ namespace Jellyfin.Server.Implementations.Devices
|
||||
await dbContext.SaveChangesAsync().ConfigureAwait(false);
|
||||
}
|
||||
|
||||
_deviceOptions[deviceId] = deviceOptions;
|
||||
|
||||
DeviceOptionsUpdated?.Invoke(this, new GenericEventArgs<Tuple<string, DeviceOptions>>(new Tuple<string, DeviceOptions>(deviceId, deviceOptions)));
|
||||
}
|
||||
|
||||
@@ -102,21 +81,24 @@ namespace Jellyfin.Server.Implementations.Devices
|
||||
{
|
||||
dbContext.Devices.Add(device);
|
||||
await dbContext.SaveChangesAsync().ConfigureAwait(false);
|
||||
_devices.TryAdd(device.Id, device);
|
||||
}
|
||||
|
||||
return device;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public DeviceOptionsDto? GetDeviceOptions(string deviceId)
|
||||
public async Task<DeviceOptionsDto?> GetDeviceOptions(string deviceId)
|
||||
{
|
||||
if (_deviceOptions.TryGetValue(deviceId, out var deviceOptions))
|
||||
var dbContext = await _dbProvider.CreateDbContextAsync().ConfigureAwait(false);
|
||||
await using (dbContext.ConfigureAwait(false))
|
||||
{
|
||||
return ToDeviceOptionsDto(deviceOptions);
|
||||
}
|
||||
var deviceOptions = await dbContext.DeviceOptions
|
||||
.AsNoTracking()
|
||||
.FirstOrDefaultAsync(dev => dev.DeviceId == deviceId)
|
||||
.ConfigureAwait(false);
|
||||
|
||||
return null;
|
||||
return deviceOptions is null ? null : ToDeviceOptionsDto(deviceOptions);
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
@@ -133,43 +115,79 @@ namespace Jellyfin.Server.Implementations.Devices
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public DeviceInfoDto? GetDevice(string id)
|
||||
public async Task<DeviceInfoDto?> GetDevice(string id)
|
||||
{
|
||||
var device = _devices.Values.Where(d => d.DeviceId == id).OrderByDescending(d => d.DateLastActivity).FirstOrDefault();
|
||||
_deviceOptions.TryGetValue(id, out var deviceOption);
|
||||
var dbContext = await _dbProvider.CreateDbContextAsync().ConfigureAwait(false);
|
||||
await using (dbContext.ConfigureAwait(false))
|
||||
{
|
||||
var device = await dbContext.Devices
|
||||
.AsNoTracking()
|
||||
.Where(d => d.DeviceId == id)
|
||||
.OrderByDescending(d => d.DateLastActivity)
|
||||
.FirstOrDefaultAsync()
|
||||
.ConfigureAwait(false);
|
||||
|
||||
var deviceInfo = device is null ? null : ToDeviceInfo(device, deviceOption);
|
||||
return deviceInfo is null ? null : ToDeviceInfoDto(deviceInfo);
|
||||
if (device is null)
|
||||
{
|
||||
return null;
|
||||
}
|
||||
|
||||
var deviceOption = await dbContext.DeviceOptions
|
||||
.AsNoTracking()
|
||||
.FirstOrDefaultAsync(dev => dev.DeviceId == id)
|
||||
.ConfigureAwait(false);
|
||||
|
||||
return ToDeviceInfoDto(ToDeviceInfo(device, deviceOption));
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public QueryResult<Device> GetDevices(DeviceQuery query)
|
||||
public async Task<QueryResult<Device>> GetDevices(DeviceQuery query)
|
||||
{
|
||||
IEnumerable<Device> devices = _devices.Values
|
||||
.Where(device => !query.UserId.HasValue || device.UserId.Equals(query.UserId.Value))
|
||||
.Where(device => query.DeviceId is null || device.DeviceId == query.DeviceId)
|
||||
.Where(device => query.AccessToken is null || device.AccessToken == query.AccessToken)
|
||||
.OrderBy(d => d.Id)
|
||||
.ToList();
|
||||
var count = devices.Count();
|
||||
|
||||
if (query.Skip.HasValue)
|
||||
var dbContext = await _dbProvider.CreateDbContextAsync().ConfigureAwait(false);
|
||||
await using (dbContext.ConfigureAwait(false))
|
||||
{
|
||||
devices = devices.Skip(query.Skip.Value);
|
||||
}
|
||||
IQueryable<Device> filtered = dbContext.Devices.AsNoTracking();
|
||||
|
||||
if (query.Limit.HasValue && query.Limit.Value > 0)
|
||||
{
|
||||
devices = devices.Take(query.Limit.Value);
|
||||
}
|
||||
if (query.UserId.HasValue)
|
||||
{
|
||||
filtered = filtered.Where(device => device.UserId.Equals(query.UserId.Value));
|
||||
}
|
||||
|
||||
return new QueryResult<Device>(query.Skip, count, devices.ToList());
|
||||
if (query.DeviceId is not null)
|
||||
{
|
||||
filtered = filtered.Where(device => device.DeviceId == query.DeviceId);
|
||||
}
|
||||
|
||||
if (query.AccessToken is not null)
|
||||
{
|
||||
filtered = filtered.Where(device => device.AccessToken == query.AccessToken);
|
||||
}
|
||||
|
||||
// Every filter is an exact match on an indexed column and the table holds one row per
|
||||
// user and device, so paging the materialised set costs less than a second round trip.
|
||||
var matched = await filtered.OrderBy(d => d.Id).ToListAsync().ConfigureAwait(false);
|
||||
|
||||
IEnumerable<Device> devices = matched;
|
||||
|
||||
if (query.Skip.HasValue)
|
||||
{
|
||||
devices = devices.Skip(query.Skip.Value);
|
||||
}
|
||||
|
||||
if (query.Limit.HasValue && query.Limit.Value > 0)
|
||||
{
|
||||
devices = devices.Take(query.Limit.Value);
|
||||
}
|
||||
|
||||
return new QueryResult<Device>(query.Skip, matched.Count, devices.ToList());
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public QueryResult<DeviceInfo> GetDeviceInfos(DeviceQuery query)
|
||||
public async Task<QueryResult<DeviceInfo>> GetDeviceInfos(DeviceQuery query)
|
||||
{
|
||||
var devices = GetDevices(query);
|
||||
var devices = await GetDevices(query).ConfigureAwait(false);
|
||||
|
||||
return new QueryResult<DeviceInfo>(
|
||||
devices.StartIndex,
|
||||
@@ -178,38 +196,49 @@ namespace Jellyfin.Server.Implementations.Devices
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public QueryResult<DeviceInfoDto> GetDevicesForUser(Guid? userId)
|
||||
public async Task<QueryResult<DeviceInfoDto>> GetDevicesForUser(Guid? userId)
|
||||
{
|
||||
IEnumerable<Device> devices = _devices.Values
|
||||
.OrderByDescending(d => d.DateLastActivity)
|
||||
.ThenBy(d => d.DeviceId);
|
||||
|
||||
if (!userId.IsNullOrEmpty())
|
||||
var dbContext = await _dbProvider.CreateDbContextAsync().ConfigureAwait(false);
|
||||
await using (dbContext.ConfigureAwait(false))
|
||||
{
|
||||
var user = _userManager.GetUserById(userId.Value);
|
||||
if (user is null)
|
||||
IEnumerable<Device> devices = await dbContext.Devices
|
||||
.AsNoTracking()
|
||||
.OrderByDescending(d => d.DateLastActivity)
|
||||
.ThenBy(d => d.DeviceId)
|
||||
.ToListAsync()
|
||||
.ConfigureAwait(false);
|
||||
|
||||
if (!userId.IsNullOrEmpty())
|
||||
{
|
||||
throw new ResourceNotFoundException();
|
||||
var user = _userManager.GetUserById(userId.Value);
|
||||
if (user is null)
|
||||
{
|
||||
throw new ResourceNotFoundException();
|
||||
}
|
||||
|
||||
devices = devices.Where(i => CanAccessDevice(user, i.DeviceId));
|
||||
}
|
||||
|
||||
devices = devices.Where(i => CanAccessDevice(user, i.DeviceId));
|
||||
var options = await dbContext.DeviceOptions
|
||||
.AsNoTracking()
|
||||
.ToDictionaryAsync(dev => dev.DeviceId)
|
||||
.ConfigureAwait(false);
|
||||
|
||||
var array = devices.Select(device =>
|
||||
{
|
||||
options.TryGetValue(device.DeviceId, out var option);
|
||||
return ToDeviceInfo(device, option);
|
||||
})
|
||||
.Select(ToDeviceInfoDto)
|
||||
.ToArray();
|
||||
|
||||
return new QueryResult<DeviceInfoDto>(array);
|
||||
}
|
||||
|
||||
var array = devices.Select(device =>
|
||||
{
|
||||
_deviceOptions.TryGetValue(device.DeviceId, out var option);
|
||||
return ToDeviceInfo(device, option);
|
||||
})
|
||||
.Select(ToDeviceInfoDto)
|
||||
.ToArray();
|
||||
|
||||
return new QueryResult<DeviceInfoDto>(array);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task DeleteDevice(Device device)
|
||||
{
|
||||
_devices.TryRemove(device.Id, out _);
|
||||
var dbContext = await _dbProvider.CreateDbContextAsync().ConfigureAwait(false);
|
||||
await using (dbContext.ConfigureAwait(false))
|
||||
{
|
||||
@@ -229,8 +258,6 @@ namespace Jellyfin.Server.Implementations.Devices
|
||||
dbContext.Devices.Update(device);
|
||||
await dbContext.SaveChangesAsync().ConfigureAwait(false);
|
||||
}
|
||||
|
||||
_devices[device.Id] = device;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
|
||||
@@ -219,9 +219,11 @@ public sealed partial class BaseItemRepository
|
||||
}
|
||||
else
|
||||
{
|
||||
// The representative is the row no other row in its group sorts before, not MIN(Id):
|
||||
// PostgreSQL has no min(uuid) aggregate, while comparing two uuids is supported everywhere.
|
||||
representativeIds = masterQuery
|
||||
.GroupBy(e => e.PresentationUniqueKey)
|
||||
.Select(g => g.Min(e => e.Id))
|
||||
.Where(e => !masterQuery.Any(o => o.PresentationUniqueKey == e.PresentationUniqueKey && o.Id.CompareTo(e.Id) < 0))
|
||||
.Select(e => e.Id)
|
||||
.ToList();
|
||||
}
|
||||
|
||||
|
||||
@@ -96,22 +96,36 @@ public sealed partial class BaseItemRepository
|
||||
// primary version (PrimaryVersionId is null) so detail pages and actions target it instead
|
||||
// of an arbitrary alternate. Keep the grouped ids as an IQueryable sub-select; materializing
|
||||
// to a List would inline one bound parameter per id and hit SQLite's variable cap.
|
||||
// The representative is the row no other row in its group sorts before, not MIN(Id):
|
||||
// PostgreSQL has no min(uuid) aggregate, while comparing two uuids is supported everywhere.
|
||||
// The anti-join reads the filtered set twice, so it has to close over a local that the
|
||||
// reassignment below cannot reach - capturing dbQuery itself makes the tree self-referential.
|
||||
var candidates = dbQuery;
|
||||
var enableGroupByPresentationUniqueKey = EnableGroupByPresentationUniqueKey(filter);
|
||||
if (enableGroupByPresentationUniqueKey && filter.GroupBySeriesPresentationUniqueKey)
|
||||
{
|
||||
var groupedIds = dbQuery.GroupBy(e => new { e.PresentationUniqueKey, e.SeriesPresentationUniqueKey })
|
||||
.Select(g => g.Where(e => e.PrimaryVersionId == null).Min(e => (Guid?)e.Id) ?? g.Min(e => (Guid?)e.Id));
|
||||
var groupedIds = candidates
|
||||
.Where(e => !candidates.Any(o => o.PresentationUniqueKey == e.PresentationUniqueKey
|
||||
&& o.SeriesPresentationUniqueKey == e.SeriesPresentationUniqueKey
|
||||
&& ((o.PrimaryVersionId == null && e.PrimaryVersionId != null)
|
||||
|| ((o.PrimaryVersionId == null) == (e.PrimaryVersionId == null) && o.Id.CompareTo(e.Id) < 0))))
|
||||
.Select(e => e.Id);
|
||||
dbQuery = context.BaseItems.AsNoTracking().Where(e => groupedIds.Contains(e.Id));
|
||||
}
|
||||
else if (enableGroupByPresentationUniqueKey)
|
||||
{
|
||||
var groupedIds = dbQuery.GroupBy(e => e.PresentationUniqueKey)
|
||||
.Select(g => g.Where(e => e.PrimaryVersionId == null).Min(e => (Guid?)e.Id) ?? g.Min(e => (Guid?)e.Id));
|
||||
var groupedIds = candidates
|
||||
.Where(e => !candidates.Any(o => o.PresentationUniqueKey == e.PresentationUniqueKey
|
||||
&& ((o.PrimaryVersionId == null && e.PrimaryVersionId != null)
|
||||
|| ((o.PrimaryVersionId == null) == (e.PrimaryVersionId == null) && o.Id.CompareTo(e.Id) < 0))))
|
||||
.Select(e => e.Id);
|
||||
dbQuery = context.BaseItems.AsNoTracking().Where(e => groupedIds.Contains(e.Id));
|
||||
}
|
||||
else if (filter.GroupBySeriesPresentationUniqueKey)
|
||||
{
|
||||
var groupedIds = dbQuery.GroupBy(e => e.SeriesPresentationUniqueKey).Select(e => e.Min(x => x.Id));
|
||||
var groupedIds = candidates
|
||||
.Where(e => !candidates.Any(o => o.SeriesPresentationUniqueKey == e.SeriesPresentationUniqueKey && o.Id.CompareTo(e.Id) < 0))
|
||||
.Select(e => e.Id);
|
||||
dbQuery = context.BaseItems.AsNoTracking().Where(e => groupedIds.Contains(e.Id));
|
||||
}
|
||||
else
|
||||
|
||||
@@ -126,95 +126,95 @@ namespace Jellyfin.Server.Implementations.Security
|
||||
return authInfo;
|
||||
}
|
||||
|
||||
var device = (await _deviceManager.GetDevices(
|
||||
new DeviceQuery { AccessToken = token }).ConfigureAwait(false)).Items.FirstOrDefault();
|
||||
|
||||
if (device is not null)
|
||||
{
|
||||
authInfo.IsAuthenticated = true;
|
||||
var updateToken = false;
|
||||
|
||||
// TODO: Remove these checks for IsNullOrWhiteSpace
|
||||
if (string.IsNullOrWhiteSpace(authInfo.Client))
|
||||
{
|
||||
authInfo.Client = device.AppName;
|
||||
}
|
||||
|
||||
if (string.IsNullOrWhiteSpace(authInfo.DeviceId))
|
||||
{
|
||||
authInfo.DeviceId = device.DeviceId;
|
||||
}
|
||||
|
||||
// Temporary. TODO - allow clients to specify that the token has been shared with a casting device
|
||||
var allowTokenInfoUpdate = !authInfo.Client.Contains("chromecast", StringComparison.OrdinalIgnoreCase);
|
||||
|
||||
if (string.IsNullOrWhiteSpace(authInfo.Device))
|
||||
{
|
||||
authInfo.Device = device.DeviceName;
|
||||
}
|
||||
else if (!string.Equals(authInfo.Device, device.DeviceName, StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
if (allowTokenInfoUpdate)
|
||||
{
|
||||
updateToken = true;
|
||||
device.DeviceName = authInfo.Device;
|
||||
}
|
||||
}
|
||||
|
||||
if (string.IsNullOrWhiteSpace(authInfo.Version))
|
||||
{
|
||||
authInfo.Version = device.AppVersion;
|
||||
}
|
||||
else if (!string.Equals(authInfo.Version, device.AppVersion, StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
if (allowTokenInfoUpdate)
|
||||
{
|
||||
updateToken = true;
|
||||
device.AppVersion = authInfo.Version;
|
||||
}
|
||||
}
|
||||
|
||||
if ((DateTime.UtcNow - device.DateLastActivity).TotalMinutes > 3)
|
||||
{
|
||||
device.DateLastActivity = DateTime.UtcNow;
|
||||
updateToken = true;
|
||||
}
|
||||
|
||||
authInfo.User = _userManager.GetUserById(device.UserId);
|
||||
|
||||
if (updateToken)
|
||||
{
|
||||
await _deviceManager.UpdateDevice(device).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
return authInfo;
|
||||
}
|
||||
|
||||
var dbContext = await _jellyfinDbProvider.CreateDbContextAsync().ConfigureAwait(false);
|
||||
await using (dbContext.ConfigureAwait(false))
|
||||
{
|
||||
var device = _deviceManager.GetDevices(
|
||||
new DeviceQuery { AccessToken = token }).Items.FirstOrDefault();
|
||||
|
||||
if (device is not null)
|
||||
var key = await dbContext.ApiKeys.FirstOrDefaultAsync(apiKey => apiKey.AccessToken == token).ConfigureAwait(false);
|
||||
if (key is not null)
|
||||
{
|
||||
authInfo.IsAuthenticated = true;
|
||||
var updateToken = false;
|
||||
|
||||
// TODO: Remove these checks for IsNullOrWhiteSpace
|
||||
if (string.IsNullOrWhiteSpace(authInfo.Client))
|
||||
{
|
||||
authInfo.Client = device.AppName;
|
||||
}
|
||||
|
||||
authInfo.Client = key.Name;
|
||||
authInfo.Token = key.AccessToken;
|
||||
if (string.IsNullOrWhiteSpace(authInfo.DeviceId))
|
||||
{
|
||||
authInfo.DeviceId = device.DeviceId;
|
||||
authInfo.DeviceId = _serverApplicationHost.SystemId;
|
||||
}
|
||||
|
||||
// Temporary. TODO - allow clients to specify that the token has been shared with a casting device
|
||||
var allowTokenInfoUpdate = !authInfo.Client.Contains("chromecast", StringComparison.OrdinalIgnoreCase);
|
||||
|
||||
if (string.IsNullOrWhiteSpace(authInfo.Device))
|
||||
{
|
||||
authInfo.Device = device.DeviceName;
|
||||
}
|
||||
else if (!string.Equals(authInfo.Device, device.DeviceName, StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
if (allowTokenInfoUpdate)
|
||||
{
|
||||
updateToken = true;
|
||||
device.DeviceName = authInfo.Device;
|
||||
}
|
||||
authInfo.Device = _serverApplicationHost.Name;
|
||||
}
|
||||
|
||||
if (string.IsNullOrWhiteSpace(authInfo.Version))
|
||||
{
|
||||
authInfo.Version = device.AppVersion;
|
||||
}
|
||||
else if (!string.Equals(authInfo.Version, device.AppVersion, StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
if (allowTokenInfoUpdate)
|
||||
{
|
||||
updateToken = true;
|
||||
device.AppVersion = authInfo.Version;
|
||||
}
|
||||
authInfo.Version = _serverApplicationHost.ApplicationVersionString;
|
||||
}
|
||||
|
||||
if ((DateTime.UtcNow - device.DateLastActivity).TotalMinutes > 3)
|
||||
{
|
||||
device.DateLastActivity = DateTime.UtcNow;
|
||||
updateToken = true;
|
||||
}
|
||||
|
||||
authInfo.User = _userManager.GetUserById(device.UserId);
|
||||
|
||||
if (updateToken)
|
||||
{
|
||||
await _deviceManager.UpdateDevice(device).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
var key = await dbContext.ApiKeys.FirstOrDefaultAsync(apiKey => apiKey.AccessToken == token).ConfigureAwait(false);
|
||||
if (key is not null)
|
||||
{
|
||||
authInfo.IsAuthenticated = true;
|
||||
authInfo.Client = key.Name;
|
||||
authInfo.Token = key.AccessToken;
|
||||
if (string.IsNullOrWhiteSpace(authInfo.DeviceId))
|
||||
{
|
||||
authInfo.DeviceId = _serverApplicationHost.SystemId;
|
||||
}
|
||||
|
||||
if (string.IsNullOrWhiteSpace(authInfo.Device))
|
||||
{
|
||||
authInfo.Device = _serverApplicationHost.Name;
|
||||
}
|
||||
|
||||
if (string.IsNullOrWhiteSpace(authInfo.Version))
|
||||
{
|
||||
authInfo.Version = _serverApplicationHost.ApplicationVersionString;
|
||||
}
|
||||
|
||||
authInfo.IsApiKey = true;
|
||||
}
|
||||
authInfo.IsApiKey = true;
|
||||
}
|
||||
|
||||
return authInfo;
|
||||
|
||||
@@ -61,10 +61,10 @@ public sealed class DeviceAccessHost : IHostedService
|
||||
|
||||
private async Task UpdateDeviceAccess(User user)
|
||||
{
|
||||
var existing = _deviceManager.GetDevices(new DeviceQuery
|
||||
var existing = (await _deviceManager.GetDevices(new DeviceQuery
|
||||
{
|
||||
UserId = user.Id
|
||||
}).Items;
|
||||
}).ConfigureAwait(false)).Items;
|
||||
|
||||
foreach (var device in existing)
|
||||
{
|
||||
|
||||
@@ -3,6 +3,7 @@ using System.Collections.Generic;
|
||||
using System.Reflection;
|
||||
using Emby.Server.Implementations;
|
||||
using Emby.Server.Implementations.MediaEncoding;
|
||||
using Emby.Server.Implementations.ScheduledTasks;
|
||||
using Emby.Server.Implementations.Session;
|
||||
using Jellyfin.Api.WebSocketListeners;
|
||||
using Jellyfin.Database.Implementations;
|
||||
@@ -34,7 +35,6 @@ using MediaBrowser.Providers.Lyric;
|
||||
using Microsoft.Extensions.Configuration;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using StackExchange.Redis;
|
||||
|
||||
namespace Jellyfin.Server
|
||||
{
|
||||
@@ -106,29 +106,7 @@ namespace Jellyfin.Server
|
||||
serviceCollection.AddScoped<IAuthenticationManager, AuthenticationManager>();
|
||||
|
||||
// Transcode session store: Redis-backed when configured, no-op otherwise.
|
||||
serviceCollection.Configure<TranscodeStoreOptions>(_startupConfig.GetSection("Jellyfin:TranscodeStore"));
|
||||
var redisConnectionString = _startupConfig["Jellyfin:TranscodeStore:RedisConnectionString"];
|
||||
if (!string.IsNullOrEmpty(redisConnectionString))
|
||||
{
|
||||
serviceCollection.AddSingleton<IConnectionMultiplexer>(sp =>
|
||||
{
|
||||
try
|
||||
{
|
||||
return ConnectionMultiplexer.Connect(redisConnectionString);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
sp.GetRequiredService<ILogger<CoreAppHost>>()
|
||||
.LogError(ex, "Failed to connect to Redis. Check the Jellyfin:TranscodeStore:RedisConnectionString configuration.");
|
||||
throw;
|
||||
}
|
||||
});
|
||||
serviceCollection.AddSingleton<ITranscodeSessionStore, RedisTranscodeSessionStore>();
|
||||
}
|
||||
else
|
||||
{
|
||||
serviceCollection.AddSingleton<ITranscodeSessionStore, NullTranscodeSessionStore>();
|
||||
}
|
||||
serviceCollection.AddTranscodeSessionStore(_startupConfig, Logger);
|
||||
|
||||
// Scan-leader lease: gates periodic library-mutating scheduled tasks to a single leader
|
||||
// instance. Active by default once a Redis connection is configured, no-op otherwise.
|
||||
|
||||
@@ -0,0 +1,51 @@
|
||||
using System.Collections.Generic;
|
||||
using System.Linq;
|
||||
using Microsoft.Extensions.Configuration;
|
||||
|
||||
namespace Jellyfin.Server.Extensions;
|
||||
|
||||
/// <summary>
|
||||
/// Extensions for building the application configuration.
|
||||
/// </summary>
|
||||
public static class ConfigurationBuilderExtensions
|
||||
{
|
||||
/// <summary>
|
||||
/// The environment variable prefix that maps onto this fork's own <c>Jellyfin:*</c> configuration
|
||||
/// keys, for example <c>Jellyfin__TranscodeStore__RedisConnectionString</c>.
|
||||
/// </summary>
|
||||
public const string JellyfinSectionEnvironmentPrefix = "Jellyfin__";
|
||||
|
||||
/// <summary>
|
||||
/// The configuration section root this fork keeps its own settings under.
|
||||
/// </summary>
|
||||
public const string JellyfinSectionRoot = "Jellyfin";
|
||||
|
||||
/// <summary>
|
||||
/// Adds environment variables named <c>Jellyfin__Section__Key</c> as the configuration keys
|
||||
/// <c>Jellyfin:Section:Key</c>.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// The base configuration only reads <c>JELLYFIN_</c> prefixed environment variables, so the
|
||||
/// unprefixed form every manifest, chart and document uses would otherwise be dropped and the
|
||||
/// feature it configures would stay off with no error.
|
||||
/// </remarks>
|
||||
/// <param name="builder">The configuration builder.</param>
|
||||
/// <returns>The updated configuration builder.</returns>
|
||||
public static IConfigurationBuilder AddJellyfinSectionEnvironmentVariables(this IConfigurationBuilder builder)
|
||||
{
|
||||
// Read through the framework provider so "__" to ":" normalisation and case handling match the
|
||||
// prefixed form exactly; the prefix it strips is then put back as the section root.
|
||||
var scoped = new ConfigurationBuilder()
|
||||
.AddEnvironmentVariables(JellyfinSectionEnvironmentPrefix)
|
||||
.Build();
|
||||
|
||||
var entries = scoped.AsEnumerable()
|
||||
.Where(entry => entry.Value is not null)
|
||||
.Select(entry => new KeyValuePair<string, string?>(
|
||||
ConfigurationPath.Combine(JellyfinSectionRoot, entry.Key),
|
||||
entry.Value))
|
||||
.ToList();
|
||||
|
||||
return builder.AddInMemoryCollection(entries);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,87 @@
|
||||
using System;
|
||||
using System.Linq;
|
||||
using Emby.Server.Implementations.MediaEncoding;
|
||||
using MediaBrowser.Controller.MediaEncoding;
|
||||
using Microsoft.Extensions.Configuration;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using StackExchange.Redis;
|
||||
|
||||
namespace Jellyfin.Server.Extensions;
|
||||
|
||||
/// <summary>
|
||||
/// Extensions for registering the transcode session store.
|
||||
/// </summary>
|
||||
public static class TranscodeStoreServiceCollectionExtensions
|
||||
{
|
||||
/// <summary>
|
||||
/// Registers the transcode session store, Redis-backed when a connection string is configured and
|
||||
/// no-op otherwise, and reports the selected store at <see cref="LogLevel.Information"/>.
|
||||
/// </summary>
|
||||
/// <param name="serviceCollection">The service collection.</param>
|
||||
/// <param name="configuration">The configuration to read <c>Jellyfin:TranscodeStore</c> from.</param>
|
||||
/// <param name="logger">The logger to report the selected store on.</param>
|
||||
/// <returns>The updated service collection.</returns>
|
||||
public static IServiceCollection AddTranscodeSessionStore(
|
||||
this IServiceCollection serviceCollection,
|
||||
IConfiguration configuration,
|
||||
ILogger logger)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(configuration);
|
||||
ArgumentNullException.ThrowIfNull(logger);
|
||||
|
||||
serviceCollection.Configure<TranscodeStoreOptions>(configuration.GetSection(TranscodeStoreOptions.ConfigurationSection));
|
||||
|
||||
var redisConnectionString = configuration[TranscodeStoreOptions.RedisConnectionStringKey];
|
||||
if (string.IsNullOrEmpty(redisConnectionString))
|
||||
{
|
||||
logger.LogInformation(
|
||||
"Transcode session store: {Store}. Cross-pod transcode takeover is off; set {Key} to enable it.",
|
||||
nameof(NullTranscodeSessionStore),
|
||||
TranscodeStoreOptions.RedisConnectionStringKey);
|
||||
|
||||
return serviceCollection.AddSingleton<ITranscodeSessionStore, NullTranscodeSessionStore>();
|
||||
}
|
||||
|
||||
logger.LogInformation(
|
||||
"Transcode session store: {Store} on {Endpoints}.",
|
||||
nameof(RedisTranscodeSessionStore),
|
||||
DescribeEndpoints(redisConnectionString));
|
||||
|
||||
serviceCollection.AddSingleton<IConnectionMultiplexer>(sp =>
|
||||
{
|
||||
try
|
||||
{
|
||||
return ConnectionMultiplexer.Connect(redisConnectionString);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
sp.GetRequiredService<ILogger<CoreAppHost>>()
|
||||
.LogError(ex, "Failed to connect to Redis. Check the {Key} configuration.", TranscodeStoreOptions.RedisConnectionStringKey);
|
||||
throw;
|
||||
}
|
||||
});
|
||||
serviceCollection.AddSingleton<ITranscodeSessionStore, RedisTranscodeSessionStore>();
|
||||
serviceCollection.AddHostedService<TranscodeStoreConnectivityProbe>();
|
||||
|
||||
return serviceCollection;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Renders the endpoints of a connection string for logging. The connection string itself is never
|
||||
/// logged because it can carry a password.
|
||||
/// </summary>
|
||||
private static string DescribeEndpoints(string redisConnectionString)
|
||||
{
|
||||
try
|
||||
{
|
||||
return string.Join(
|
||||
',',
|
||||
ConfigurationOptions.Parse(redisConnectionString).EndPoints.Select(endpoint => endpoint.ToString()));
|
||||
}
|
||||
catch (ArgumentException)
|
||||
{
|
||||
return "(unparsable connection string)";
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -391,6 +391,8 @@ namespace Jellyfin.Server
|
||||
.AddInMemoryCollection(inMemoryDefaultConfig)
|
||||
.AddJsonFile(LoggingConfigFileDefault, optional: false, reloadOnChange: true)
|
||||
.AddJsonFile(LoggingConfigFileSystem, optional: true, reloadOnChange: true)
|
||||
// Added before the prefixed source so an explicit JELLYFIN_ variable still wins.
|
||||
.AddJellyfinSectionEnvironmentVariables()
|
||||
.AddEnvironmentVariables("JELLYFIN_")
|
||||
.AddInMemoryCollection(commandLineOpts.ConvertToConfig());
|
||||
}
|
||||
|
||||
@@ -47,29 +47,29 @@ public interface IDeviceManager
|
||||
/// Gets the device information.
|
||||
/// </summary>
|
||||
/// <param name="id">The identifier.</param>
|
||||
/// <returns>DeviceInfoDto.</returns>
|
||||
DeviceInfoDto? GetDevice(string id);
|
||||
/// <returns>A <see cref="Task"/> representing the retrieval of the device information.</returns>
|
||||
Task<DeviceInfoDto?> GetDevice(string id);
|
||||
|
||||
/// <summary>
|
||||
/// Gets devices based on the provided query.
|
||||
/// </summary>
|
||||
/// <param name="query">The device query.</param>
|
||||
/// <returns>A <see cref="Task{QueryResult}"/> representing the retrieval of the devices.</returns>
|
||||
QueryResult<Device> GetDevices(DeviceQuery query);
|
||||
Task<QueryResult<Device>> GetDevices(DeviceQuery query);
|
||||
|
||||
/// <summary>
|
||||
/// Gets device information based on the provided query.
|
||||
/// </summary>
|
||||
/// <param name="query">The device query.</param>
|
||||
/// <returns>A <see cref="Task{QueryResult}"/> representing the retrieval of the device information.</returns>
|
||||
QueryResult<DeviceInfo> GetDeviceInfos(DeviceQuery query);
|
||||
Task<QueryResult<DeviceInfo>> GetDeviceInfos(DeviceQuery query);
|
||||
|
||||
/// <summary>
|
||||
/// Gets the device information.
|
||||
/// </summary>
|
||||
/// <param name="userId">The user's id, or <c>null</c>.</param>
|
||||
/// <returns>IEnumerable<DeviceInfoDto>.</returns>
|
||||
QueryResult<DeviceInfoDto> GetDevicesForUser(Guid? userId);
|
||||
/// <returns>A <see cref="Task{QueryResult}"/> representing the retrieval of the device information.</returns>
|
||||
Task<QueryResult<DeviceInfoDto>> GetDevicesForUser(Guid? userId);
|
||||
|
||||
/// <summary>
|
||||
/// Deletes a device.
|
||||
@@ -105,8 +105,8 @@ public interface IDeviceManager
|
||||
/// Gets the options of a device.
|
||||
/// </summary>
|
||||
/// <param name="deviceId">The device id.</param>
|
||||
/// <returns><see cref="DeviceOptions"/> of the device.</returns>
|
||||
DeviceOptionsDto? GetDeviceOptions(string deviceId);
|
||||
/// <returns>A <see cref="Task"/> representing the retrieval of the <see cref="DeviceOptions"/> of the device.</returns>
|
||||
Task<DeviceOptionsDto?> GetDeviceOptions(string deviceId);
|
||||
|
||||
/// <summary>
|
||||
/// Gets the dto for client capabilities.
|
||||
|
||||
@@ -5,6 +5,16 @@ namespace MediaBrowser.Controller.MediaEncoding;
|
||||
/// </summary>
|
||||
public sealed class TranscodeStoreOptions
|
||||
{
|
||||
/// <summary>
|
||||
/// The configuration section these options bind from.
|
||||
/// </summary>
|
||||
public const string ConfigurationSection = "Jellyfin:TranscodeStore";
|
||||
|
||||
/// <summary>
|
||||
/// The configuration key holding the Redis connection string.
|
||||
/// </summary>
|
||||
public const string RedisConnectionStringKey = ConfigurationSection + ":RedisConnectionString";
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets the Redis connection string.
|
||||
/// A <c>null</c> or empty value indicates single-instance mode, where
|
||||
|
||||
@@ -72,7 +72,7 @@ dotnet run --project Jellyfin.Server/Jellyfin.Server.csproj -- \
|
||||
|
||||
### HA mode with Redis
|
||||
|
||||
Set the `Jellyfin:TranscodeStore:RedisConnectionString` configuration key. You can pass it as an environment variable, a `DOTNET_` prefixed env var, or in a JSON config file.
|
||||
Set the `Jellyfin:TranscodeStore:RedisConnectionString` configuration key. You can pass it as a `Jellyfin__TranscodeStore__RedisConnectionString` environment variable, as the equivalent `JELLYFIN_` prefixed variable, or in a JSON config file.
|
||||
|
||||
**Environment variable:**
|
||||
|
||||
@@ -98,7 +98,14 @@ dotnet run --project Jellyfin.Server/Jellyfin.Server.csproj -- \
|
||||
}
|
||||
```
|
||||
|
||||
When `RedisConnectionString` is set, `RedisTranscodeSessionStore` is registered in DI. If the Redis connection fails at startup, the server throws and refuses to start — this is intentional so you don't silently fall back to broken HA behavior.
|
||||
The selected store is logged at startup, so HA transcoding is never on or off without a signal:
|
||||
|
||||
```
|
||||
Transcode session store: RedisTranscodeSessionStore on valkey:6379.
|
||||
Redis transcode session store is reachable (2ms round trip). HA transcode takeover is active.
|
||||
```
|
||||
|
||||
Without a connection string the line reads `Transcode session store: NullTranscodeSessionStore`. A configured but unreachable store is logged at `Error`; the server keeps serving with per-instance sessions rather than refusing to start.
|
||||
|
||||
---
|
||||
|
||||
|
||||
@@ -37,6 +37,9 @@ auth or plugin logic is rewritten.
|
||||
| `MediaBrowser.Controller/MediaEncoding/TranscodeStoreOptions.cs` | `RedisConnectionString`, `LeaseDurationSeconds` and `SessionRetentionSeconds` |
|
||||
| `MediaBrowser.Controller/MediaEncoding/NullTranscodeSessionStore.cs` | No-op store used when no Redis connection is configured |
|
||||
| `Emby.Server.Implementations/MediaEncoding/RedisTranscodeSessionStore.cs` | Redis store; sessions under `jellyfin:transcode:{playSessionId}`, key TTL is the retention window so an orphaned session outlives its lease |
|
||||
| `Emby.Server.Implementations/MediaEncoding/TranscodeStoreConnectivityProbe.cs` | Startup ping; an unreachable configured store is logged at `Error` instead of failing open silently |
|
||||
| `Jellyfin.Server/Extensions/TranscodeStoreServiceCollectionExtensions.cs` | Store selection, logged at `Information` so the active store is visible at startup |
|
||||
| `Jellyfin.Server/Extensions/ConfigurationBuilderExtensions.cs` | Reads bare `Jellyfin__*` environment variables into the `Jellyfin:*` configuration root |
|
||||
|
||||
Lease takeover and renewal each run as a single Lua script, so concurrent pods cannot both
|
||||
claim an expired lease and a renewal cannot revert a takeover. The expiry is stored as unix
|
||||
|
||||
+103
@@ -0,0 +1,103 @@
|
||||
using System;
|
||||
using System.IO;
|
||||
using System.Threading.Tasks;
|
||||
using Emby.Server.Implementations.MediaEncoding;
|
||||
using Jellyfin.Server;
|
||||
using Jellyfin.Server.Extensions;
|
||||
using MediaBrowser.Common.Configuration;
|
||||
using MediaBrowser.Controller.MediaEncoding;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using Moq;
|
||||
using StackExchange.Redis;
|
||||
using Testcontainers.Redis;
|
||||
using Xunit;
|
||||
|
||||
namespace Jellyfin.Server.Implementations.Tests.MediaEncoding;
|
||||
|
||||
/// <summary>
|
||||
/// Drives the whole configuration path a deployment uses: a bare <c>Jellyfin__TranscodeStore__*</c>
|
||||
/// environment variable, the server's own configuration builder, the store registration, and a
|
||||
/// session round-trip against a real Valkey server.
|
||||
/// </summary>
|
||||
[Trait("Category", "RequiresDocker")]
|
||||
public sealed class TranscodeStoreWiringTests : IAsyncLifetime
|
||||
{
|
||||
private const string RedisConnectionStringVariable = "Jellyfin__TranscodeStore__RedisConnectionString";
|
||||
|
||||
private readonly RedisContainer _container;
|
||||
private string _configDirectory = string.Empty;
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="TranscodeStoreWiringTests"/> class.
|
||||
/// </summary>
|
||||
public TranscodeStoreWiringTests()
|
||||
{
|
||||
_container = new RedisBuilder("valkey/valkey:8-alpine").Build();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Starts Valkey and lays out the configuration directory the server reads at startup.
|
||||
/// </summary>
|
||||
/// <returns>A <see cref="ValueTask"/> representing the asynchronous operation.</returns>
|
||||
public async ValueTask InitializeAsync()
|
||||
{
|
||||
await _container.StartAsync().ConfigureAwait(false);
|
||||
_configDirectory = Directory.CreateTempSubdirectory("jellyfin-wiring-test").FullName;
|
||||
await File.WriteAllTextAsync(Path.Combine(_configDirectory, "logging.default.json"), "{}").ConfigureAwait(false);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Removes the environment variable, configuration directory and container.
|
||||
/// </summary>
|
||||
/// <returns>A <see cref="ValueTask"/> representing the asynchronous operation.</returns>
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
Environment.SetEnvironmentVariable(RedisConnectionStringVariable, null);
|
||||
|
||||
if (_configDirectory.Length > 0)
|
||||
{
|
||||
Directory.Delete(_configDirectory, true);
|
||||
}
|
||||
|
||||
await _container.DisposeAsync().ConfigureAwait(false);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The variable form deployments set selects the Redis store and that store really talks to Valkey.
|
||||
/// </summary>
|
||||
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task ManifestStyleEnvironmentVariable_Should_Reach_Valkey()
|
||||
{
|
||||
Environment.SetEnvironmentVariable(
|
||||
RedisConnectionStringVariable,
|
||||
_container.GetConnectionString() + ",abortConnect=false");
|
||||
|
||||
var appPaths = new Mock<IApplicationPaths>();
|
||||
appPaths.Setup(paths => paths.ConfigurationDirectoryPath).Returns(_configDirectory);
|
||||
var configuration = Program.CreateAppConfiguration(new StartupOptions(), appPaths.Object);
|
||||
|
||||
var services = new ServiceCollection();
|
||||
services.AddLogging();
|
||||
services.AddTranscodeSessionStore(configuration, NullLogger.Instance);
|
||||
|
||||
await using var provider = services.BuildServiceProvider();
|
||||
|
||||
var store = provider.GetRequiredService<ITranscodeSessionStore>();
|
||||
Assert.IsType<RedisTranscodeSessionStore>(store);
|
||||
|
||||
var playSessionId = Guid.NewGuid().ToString("N");
|
||||
await store.SetAsync(
|
||||
TranscodeSession.CreateForPlaylist(playSessionId, "media-1", "pod-a", "/transcodes/abc.m3u8", TimeSpan.FromSeconds(30)),
|
||||
TestContext.Current.CancellationToken);
|
||||
|
||||
var stored = await store.TryGetAsync(playSessionId, TestContext.Current.CancellationToken);
|
||||
|
||||
Assert.NotNull(stored);
|
||||
Assert.Equal("pod-a", stored.OwnerPod);
|
||||
|
||||
var redis = provider.GetRequiredService<IConnectionMultiplexer>();
|
||||
Assert.True(await redis.GetDatabase().KeyExistsAsync("jellyfin:transcode:" + playSessionId));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,197 @@
|
||||
using System;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using Jellyfin.Data.Queries;
|
||||
using Jellyfin.Database.Implementations;
|
||||
using Jellyfin.Database.Implementations.DbConfiguration;
|
||||
using Jellyfin.Database.Implementations.Entities;
|
||||
using Jellyfin.Database.Implementations.Entities.Security;
|
||||
using Jellyfin.Database.Implementations.Locking;
|
||||
using Jellyfin.Database.Providers.PostgreSQL;
|
||||
using Jellyfin.Server.Implementations.Devices;
|
||||
using Jellyfin.Server.Tests.Migrations;
|
||||
using MediaBrowser.Controller.Library;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using Moq;
|
||||
using Npgsql;
|
||||
using Xunit;
|
||||
|
||||
namespace Jellyfin.Server.Tests.Devices;
|
||||
|
||||
/// <summary>
|
||||
/// Two independently constructed <see cref="DeviceManager"/> instances over one PostgreSQL database are the
|
||||
/// in-process stand-in for two replicas sharing one database: what either of them writes, the other has to
|
||||
/// see on its very next read.
|
||||
/// </summary>
|
||||
[Trait("Category", "RequiresDocker")]
|
||||
public sealed class DeviceManagerReplicaTests : IAsyncLifetime
|
||||
{
|
||||
private PostgreSqlTestServer _server = null!;
|
||||
|
||||
/// <inheritdoc/>
|
||||
public async ValueTask InitializeAsync()
|
||||
{
|
||||
_server = await PostgreSqlTestServer.StartAsync().ConfigureAwait(false);
|
||||
}
|
||||
|
||||
/// <inheritdoc/>
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
await _server.DisposeAsync().ConfigureAwait(false);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A client that logs in against one replica has to be authenticated by every other replica that was
|
||||
/// already running when the token was minted - a rolling update or a scale-up must not 401 it.
|
||||
/// </summary>
|
||||
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task TokenMintedOnOneReplica_IsValidOnAnother()
|
||||
{
|
||||
var cancellationToken = TestContext.Current.CancellationToken;
|
||||
var connectionString = await _server.CreateDatabaseAsync("device_replica_create", cancellationToken);
|
||||
|
||||
await using var dataSource = new NpgsqlDataSourceBuilder(connectionString).Build();
|
||||
var user = await CreateSchemaWithUserAsync(dataSource, cancellationToken);
|
||||
|
||||
// Both replicas start before the login, so neither has the device in hand when it is created.
|
||||
var replicaA = CreateManager(dataSource, user);
|
||||
var replicaB = CreateManager(dataSource, user);
|
||||
|
||||
var device = await replicaA.CreateDevice(new Device(user.Id, "Jellyfin Web", "1.0.0", "Living Room TV", "device-1"));
|
||||
|
||||
var seenByB = await replicaB.GetDevices(new DeviceQuery { AccessToken = device.AccessToken });
|
||||
|
||||
Assert.Equal(device.Id, Assert.Single(seenByB.Items).Id);
|
||||
Assert.Equal(user.Id, seenByB.Items[0].UserId);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Revocation has to propagate at least as fast as creation: a token logged out on one replica must not
|
||||
/// still authenticate on another.
|
||||
/// </summary>
|
||||
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task TokenRevokedOnOneReplica_IsInvalidOnAnother()
|
||||
{
|
||||
var cancellationToken = TestContext.Current.CancellationToken;
|
||||
var connectionString = await _server.CreateDatabaseAsync("device_replica_revoke", cancellationToken);
|
||||
|
||||
await using var dataSource = new NpgsqlDataSourceBuilder(connectionString).Build();
|
||||
var user = await CreateSchemaWithUserAsync(dataSource, cancellationToken);
|
||||
|
||||
var replicaA = CreateManager(dataSource, user);
|
||||
var device = await replicaA.CreateDevice(new Device(user.Id, "Jellyfin Web", "1.0.0", "Living Room TV", "device-1"));
|
||||
|
||||
// Replica B comes up after the login, so it starts out agreeing that the token is valid.
|
||||
var replicaB = CreateManager(dataSource, user);
|
||||
Assert.Single((await replicaB.GetDevices(new DeviceQuery { AccessToken = device.AccessToken })).Items);
|
||||
|
||||
await replicaA.DeleteDevice(device);
|
||||
|
||||
Assert.Empty((await replicaB.GetDevices(new DeviceQuery { AccessToken = device.AccessToken })).Items);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A device renamed on one replica has to be reported under its new name by the others, and the rename has
|
||||
/// to survive being read back through a replica that never saw the write.
|
||||
/// </summary>
|
||||
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task DeviceOptionsWrittenOnOneReplica_AreReadOnAnother()
|
||||
{
|
||||
var cancellationToken = TestContext.Current.CancellationToken;
|
||||
var connectionString = await _server.CreateDatabaseAsync("device_replica_options", cancellationToken);
|
||||
|
||||
await using var dataSource = new NpgsqlDataSourceBuilder(connectionString).Build();
|
||||
var user = await CreateSchemaWithUserAsync(dataSource, cancellationToken);
|
||||
|
||||
var replicaA = CreateManager(dataSource, user);
|
||||
var replicaB = CreateManager(dataSource, user);
|
||||
|
||||
await replicaA.CreateDevice(new Device(user.Id, "Jellyfin Web", "1.0.0", "Living Room TV", "device-1"));
|
||||
await replicaA.UpdateDeviceOptions("device-1", "Kitchen TV");
|
||||
|
||||
var options = await replicaB.GetDeviceOptions("device-1");
|
||||
Assert.Equal("Kitchen TV", options?.CustomName);
|
||||
|
||||
var info = await replicaB.GetDevice("device-1");
|
||||
Assert.Equal("Kitchen TV", info?.CustomName);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Activity written by the replica serving the request has to be visible to the others, because the next
|
||||
/// request from the same client can land anywhere.
|
||||
/// </summary>
|
||||
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task DeviceUpdatedOnOneReplica_IsReadOnAnother()
|
||||
{
|
||||
var cancellationToken = TestContext.Current.CancellationToken;
|
||||
var connectionString = await _server.CreateDatabaseAsync("device_replica_update", cancellationToken);
|
||||
|
||||
await using var dataSource = new NpgsqlDataSourceBuilder(connectionString).Build();
|
||||
var user = await CreateSchemaWithUserAsync(dataSource, cancellationToken);
|
||||
|
||||
var replicaA = CreateManager(dataSource, user);
|
||||
var replicaB = CreateManager(dataSource, user);
|
||||
|
||||
var device = await replicaA.CreateDevice(new Device(user.Id, "Jellyfin Web", "1.0.0", "Living Room TV", "device-1"));
|
||||
device.AppVersion = "2.0.0";
|
||||
await replicaA.UpdateDevice(device);
|
||||
|
||||
var seenByB = Assert.Single((await replicaB.GetDevices(new DeviceQuery { DeviceId = "device-1" })).Items);
|
||||
Assert.Equal("2.0.0", seenByB.AppVersion);
|
||||
}
|
||||
|
||||
private static DeviceManager CreateManager(NpgsqlDataSource dataSource, User user)
|
||||
{
|
||||
var userManager = new Mock<IUserManager>();
|
||||
userManager.Setup(manager => manager.GetUserById(user.Id)).Returns(user);
|
||||
return new DeviceManager(new DataSourceContextFactory(dataSource), userManager.Object);
|
||||
}
|
||||
|
||||
private static async Task<User> CreateSchemaWithUserAsync(NpgsqlDataSource dataSource, CancellationToken cancellationToken)
|
||||
{
|
||||
var context = CreateContext(dataSource);
|
||||
await using (context.ConfigureAwait(false))
|
||||
{
|
||||
await context.Database.EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
|
||||
|
||||
var user = new User("replica-user", "provider", "provider");
|
||||
context.Users.Add(user);
|
||||
await context.SaveChangesAsync(cancellationToken).ConfigureAwait(false);
|
||||
|
||||
return user;
|
||||
}
|
||||
}
|
||||
|
||||
private static JellyfinDbContext CreateContext(NpgsqlDataSource dataSource)
|
||||
{
|
||||
var optionsBuilder = new DbContextOptionsBuilder<JellyfinDbContext>();
|
||||
var provider = new PostgreSqlDatabaseProvider(dataSource);
|
||||
provider.Initialise(optionsBuilder, new DatabaseConfigurationOptions { DatabaseType = "PostgreSQL" });
|
||||
return new JellyfinDbContext(
|
||||
optionsBuilder.Options,
|
||||
NullLogger<JellyfinDbContext>.Instance,
|
||||
provider,
|
||||
new NoLockBehavior(NullLogger<NoLockBehavior>.Instance));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Hands every <see cref="DeviceManager"/> its own context over the one shared database, the way the
|
||||
/// pooled factory does in the server.
|
||||
/// </summary>
|
||||
private sealed class DataSourceContextFactory : IDbContextFactory<JellyfinDbContext>
|
||||
{
|
||||
private readonly NpgsqlDataSource _dataSource;
|
||||
|
||||
public DataSourceContextFactory(NpgsqlDataSource dataSource)
|
||||
{
|
||||
_dataSource = dataSource;
|
||||
}
|
||||
|
||||
public JellyfinDbContext CreateDbContext() => CreateContext(_dataSource);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,95 @@
|
||||
using System;
|
||||
using System.IO;
|
||||
using MediaBrowser.Common.Configuration;
|
||||
using Microsoft.Extensions.Configuration;
|
||||
using Moq;
|
||||
using Xunit;
|
||||
|
||||
namespace Jellyfin.Server.Tests.HighAvailability;
|
||||
|
||||
/// <summary>
|
||||
/// The fork's own settings live under the <c>Jellyfin:*</c> configuration root and every manifest,
|
||||
/// chart and document sets them as bare <c>Jellyfin__Section__Key</c> environment variables. The
|
||||
/// startup configuration the host reads them from must therefore accept that form; when it does not,
|
||||
/// a correctly set variable is dropped and the feature it configures stays off without any error.
|
||||
/// </summary>
|
||||
public sealed class JellyfinSectionConfigurationTests : IDisposable
|
||||
{
|
||||
private const string RedisKey = "Jellyfin:TranscodeStore:RedisConnectionString";
|
||||
private const string UnprefixedVariable = "Jellyfin__TranscodeStore__RedisConnectionString";
|
||||
private const string PrefixedVariable = "JELLYFIN_Jellyfin__TranscodeStore__RedisConnectionString";
|
||||
private const string LeaseUnprefixedVariable = "Jellyfin__TranscodeStore__LeaseDurationSeconds";
|
||||
|
||||
private readonly string _configDirectory;
|
||||
|
||||
public JellyfinSectionConfigurationTests()
|
||||
{
|
||||
_configDirectory = Directory.CreateTempSubdirectory("jellyfin-config-test").FullName;
|
||||
File.WriteAllText(Path.Combine(_configDirectory, "logging.default.json"), "{}");
|
||||
}
|
||||
|
||||
public void Dispose()
|
||||
{
|
||||
Environment.SetEnvironmentVariable(UnprefixedVariable, null);
|
||||
Environment.SetEnvironmentVariable(PrefixedVariable, null);
|
||||
Environment.SetEnvironmentVariable(LeaseUnprefixedVariable, null);
|
||||
Directory.Delete(_configDirectory, true);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void CreateAppConfiguration_Should_Read_UnprefixedVariable()
|
||||
{
|
||||
Environment.SetEnvironmentVariable(UnprefixedVariable, "valkey-cheeztv-valkey:6379,abortConnect=false");
|
||||
|
||||
var config = CreateConfiguration();
|
||||
|
||||
Assert.Equal("valkey-cheeztv-valkey:6379,abortConnect=false", config[RedisKey]);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void CreateAppConfiguration_Should_Read_UnprefixedNonStringValue()
|
||||
{
|
||||
Environment.SetEnvironmentVariable(LeaseUnprefixedVariable, "45");
|
||||
|
||||
var config = CreateConfiguration();
|
||||
|
||||
Assert.Equal("45", config["Jellyfin:TranscodeStore:LeaseDurationSeconds"]);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void CreateAppConfiguration_Should_Read_PrefixedVariable()
|
||||
{
|
||||
Environment.SetEnvironmentVariable(PrefixedVariable, "redis:6379");
|
||||
|
||||
var config = CreateConfiguration();
|
||||
|
||||
Assert.Equal("redis:6379", config[RedisKey]);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void CreateAppConfiguration_Should_Prefer_PrefixedVariable()
|
||||
{
|
||||
Environment.SetEnvironmentVariable(UnprefixedVariable, "unprefixed:6379");
|
||||
Environment.SetEnvironmentVariable(PrefixedVariable, "prefixed:6379");
|
||||
|
||||
var config = CreateConfiguration();
|
||||
|
||||
Assert.Equal("prefixed:6379", config[RedisKey]);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void CreateAppConfiguration_Should_Leave_Key_Unset_Without_Variables()
|
||||
{
|
||||
var config = CreateConfiguration();
|
||||
|
||||
Assert.Null(config[RedisKey]);
|
||||
}
|
||||
|
||||
private IConfiguration CreateConfiguration()
|
||||
{
|
||||
var appPaths = new Mock<IApplicationPaths>();
|
||||
appPaths.Setup(paths => paths.ConfigurationDirectoryPath).Returns(_configDirectory);
|
||||
|
||||
return Program.CreateAppConfiguration(new StartupOptions(), appPaths.Object);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,42 @@
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Linq;
|
||||
using Microsoft.Extensions.Logging;
|
||||
|
||||
namespace Jellyfin.Server.Tests.HighAvailability;
|
||||
|
||||
/// <summary>
|
||||
/// An <see cref="ILogger"/> that keeps every formatted entry so tests can assert on the startup
|
||||
/// signals operators rely on.
|
||||
/// </summary>
|
||||
/// <typeparam name="T">The category type.</typeparam>
|
||||
internal sealed class RecordingLogger<T> : ILogger<T>
|
||||
{
|
||||
private readonly List<(LogLevel Level, string Message, Exception? Exception)> _entries = new();
|
||||
|
||||
public IReadOnlyList<(LogLevel Level, string Message, Exception? Exception)> Entries => _entries;
|
||||
|
||||
public IDisposable BeginScope<TState>(TState state)
|
||||
where TState : notnull
|
||||
=> NoopScope.Instance;
|
||||
|
||||
public bool IsEnabled(LogLevel logLevel) => true;
|
||||
|
||||
public void Log<TState>(LogLevel logLevel, EventId eventId, TState state, Exception? exception, Func<TState, Exception?, string> formatter)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(formatter);
|
||||
_entries.Add((logLevel, formatter(state, exception), exception));
|
||||
}
|
||||
|
||||
public bool HasEntry(LogLevel level, string substring)
|
||||
=> _entries.Any(entry => entry.Level == level && entry.Message.Contains(substring, StringComparison.Ordinal));
|
||||
|
||||
private sealed class NoopScope : IDisposable
|
||||
{
|
||||
public static readonly NoopScope Instance = new();
|
||||
|
||||
public void Dispose()
|
||||
{
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,75 @@
|
||||
using System;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using Emby.Server.Implementations.MediaEncoding;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Moq;
|
||||
using StackExchange.Redis;
|
||||
using Xunit;
|
||||
|
||||
namespace Jellyfin.Server.Tests.HighAvailability;
|
||||
|
||||
/// <summary>
|
||||
/// A Redis store that cannot be reached degrades silently: the client is configured not to abort the
|
||||
/// connection and every call site swallows failures. The probe is the only startup signal, so both of
|
||||
/// its outcomes are pinned here.
|
||||
/// </summary>
|
||||
public sealed class TranscodeStoreConnectivityProbeTests
|
||||
{
|
||||
[Fact]
|
||||
public async Task StartAsync_Should_Log_Information_When_Reachable()
|
||||
{
|
||||
var database = new Mock<IDatabase>();
|
||||
database.Setup(db => db.PingAsync(It.IsAny<CommandFlags>())).ReturnsAsync(TimeSpan.FromMilliseconds(3));
|
||||
|
||||
var (logger, probe) = CreateProbe(database.Object);
|
||||
|
||||
await probe.StartAsync(CancellationToken.None);
|
||||
|
||||
Assert.True(logger.HasEntry(LogLevel.Information, "reachable"));
|
||||
Assert.DoesNotContain(logger.Entries, entry => entry.Level >= LogLevel.Warning);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task StartAsync_Should_Log_Error_When_Unreachable()
|
||||
{
|
||||
var database = new Mock<IDatabase>();
|
||||
database.Setup(db => db.PingAsync(It.IsAny<CommandFlags>()))
|
||||
.ThrowsAsync(new RedisConnectionException(ConnectionFailureType.UnableToConnect, "no route to host"));
|
||||
|
||||
var (logger, probe) = CreateProbe(database.Object);
|
||||
|
||||
await probe.StartAsync(CancellationToken.None);
|
||||
|
||||
Assert.True(logger.HasEntry(LogLevel.Error, "UNREACHABLE"));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task StartAsync_Should_Log_Error_Instead_Of_Aborting_Startup()
|
||||
{
|
||||
var services = new ServiceCollection();
|
||||
services.AddSingleton<IConnectionMultiplexer>(_ => throw new RedisConnectionException(ConnectionFailureType.UnableToConnect, "no route to host"));
|
||||
using var provider = services.BuildServiceProvider();
|
||||
|
||||
var logger = new RecordingLogger<TranscodeStoreConnectivityProbe>();
|
||||
var probe = new TranscodeStoreConnectivityProbe(provider, logger);
|
||||
|
||||
await probe.StartAsync(CancellationToken.None);
|
||||
|
||||
Assert.True(logger.HasEntry(LogLevel.Error, "UNREACHABLE"));
|
||||
}
|
||||
|
||||
private static (RecordingLogger<TranscodeStoreConnectivityProbe> Logger, TranscodeStoreConnectivityProbe Probe) CreateProbe(IDatabase database)
|
||||
{
|
||||
var multiplexer = new Mock<IConnectionMultiplexer>();
|
||||
multiplexer.Setup(redis => redis.GetDatabase(It.IsAny<int>(), It.IsAny<object>())).Returns(database);
|
||||
|
||||
var services = new ServiceCollection();
|
||||
services.AddSingleton(multiplexer.Object);
|
||||
var provider = services.BuildServiceProvider();
|
||||
|
||||
var logger = new RecordingLogger<TranscodeStoreConnectivityProbe>();
|
||||
return (logger, new TranscodeStoreConnectivityProbe(provider, logger));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,104 @@
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Linq;
|
||||
using Emby.Server.Implementations.MediaEncoding;
|
||||
using Jellyfin.Server.Extensions;
|
||||
using MediaBrowser.Controller.MediaEncoding;
|
||||
using Microsoft.Extensions.Configuration;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Hosting;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Microsoft.Extensions.Options;
|
||||
using Xunit;
|
||||
|
||||
namespace Jellyfin.Server.Tests.HighAvailability;
|
||||
|
||||
/// <summary>
|
||||
/// Which transcode session store was selected is invisible at runtime — the Redis client does not
|
||||
/// abort on an unreachable server and every call site swallows failures — so the selection is
|
||||
/// asserted here together with the startup log line that reports it.
|
||||
/// </summary>
|
||||
public sealed class TranscodeStoreRegistrationTests
|
||||
{
|
||||
[Fact]
|
||||
public void AddTranscodeSessionStore_Should_Select_RedisStore_From_UnprefixedEnvironmentKeyShape()
|
||||
{
|
||||
var services = new ServiceCollection();
|
||||
var logger = new RecordingLogger<TranscodeStoreRegistrationTests>();
|
||||
|
||||
services.AddTranscodeSessionStore(BuildConfiguration("valkey-cheeztv-valkey:6379,abortConnect=false"), logger);
|
||||
|
||||
Assert.Equal(typeof(RedisTranscodeSessionStore), StoreImplementation(services));
|
||||
Assert.True(logger.HasEntry(LogLevel.Information, nameof(RedisTranscodeSessionStore)));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void AddTranscodeSessionStore_Should_Select_NullStore_Without_ConnectionString()
|
||||
{
|
||||
var services = new ServiceCollection();
|
||||
var logger = new RecordingLogger<TranscodeStoreRegistrationTests>();
|
||||
|
||||
services.AddTranscodeSessionStore(BuildConfiguration(null), logger);
|
||||
|
||||
Assert.Equal(typeof(NullTranscodeSessionStore), StoreImplementation(services));
|
||||
Assert.True(logger.HasEntry(LogLevel.Information, nameof(NullTranscodeSessionStore)));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void AddTranscodeSessionStore_Should_Log_Endpoints_Without_Password()
|
||||
{
|
||||
var services = new ServiceCollection();
|
||||
var logger = new RecordingLogger<TranscodeStoreRegistrationTests>();
|
||||
|
||||
services.AddTranscodeSessionStore(BuildConfiguration("valkey:6379,password=hunter2"), logger);
|
||||
|
||||
var message = Assert.Single(logger.Entries, entry => entry.Level == LogLevel.Information).Message;
|
||||
Assert.Contains("valkey:6379", message, StringComparison.Ordinal);
|
||||
Assert.DoesNotContain("hunter2", message, StringComparison.Ordinal);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void AddTranscodeSessionStore_Should_Register_ConnectivityProbe_Only_With_Redis()
|
||||
{
|
||||
var withRedis = new ServiceCollection();
|
||||
withRedis.AddTranscodeSessionStore(BuildConfiguration("valkey:6379"), new RecordingLogger<TranscodeStoreRegistrationTests>());
|
||||
|
||||
var withoutRedis = new ServiceCollection();
|
||||
withoutRedis.AddTranscodeSessionStore(BuildConfiguration(null), new RecordingLogger<TranscodeStoreRegistrationTests>());
|
||||
|
||||
Assert.Contains(withRedis, descriptor => descriptor.ImplementationType == typeof(TranscodeStoreConnectivityProbe));
|
||||
Assert.DoesNotContain(withoutRedis, descriptor => descriptor.ServiceType == typeof(IHostedService));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void AddTranscodeSessionStore_Should_Bind_Options()
|
||||
{
|
||||
var services = new ServiceCollection();
|
||||
var configuration = new ConfigurationBuilder()
|
||||
.AddInMemoryCollection(new Dictionary<string, string?>
|
||||
{
|
||||
[TranscodeStoreOptions.RedisConnectionStringKey] = "valkey:6379",
|
||||
["Jellyfin:TranscodeStore:LeaseDurationSeconds"] = "45"
|
||||
})
|
||||
.Build();
|
||||
|
||||
services.AddTranscodeSessionStore(configuration, new RecordingLogger<TranscodeStoreRegistrationTests>());
|
||||
|
||||
using var provider = services.BuildServiceProvider();
|
||||
var options = provider.GetRequiredService<IOptions<TranscodeStoreOptions>>().Value;
|
||||
|
||||
Assert.Equal(45, options.LeaseDurationSeconds);
|
||||
Assert.Equal("valkey:6379", options.RedisConnectionString);
|
||||
}
|
||||
|
||||
private static IConfiguration BuildConfiguration(string? redisConnectionString)
|
||||
=> new ConfigurationBuilder()
|
||||
.AddInMemoryCollection(new Dictionary<string, string?>
|
||||
{
|
||||
[TranscodeStoreOptions.RedisConnectionStringKey] = redisConnectionString
|
||||
})
|
||||
.Build();
|
||||
|
||||
private static Type? StoreImplementation(IServiceCollection services)
|
||||
=> services.Single(descriptor => descriptor.ServiceType == typeof(ITranscodeSessionStore)).ImplementationType;
|
||||
}
|
||||
@@ -0,0 +1,213 @@
|
||||
using System;
|
||||
using System.Linq;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using Emby.Server.Implementations.Data;
|
||||
using Jellyfin.Database.Implementations;
|
||||
using Jellyfin.Database.Implementations.DbConfiguration;
|
||||
using Jellyfin.Database.Implementations.Entities;
|
||||
using Jellyfin.Database.Implementations.Locking;
|
||||
using Jellyfin.Database.Providers.PostgreSQL;
|
||||
using Jellyfin.Server.Implementations.Item;
|
||||
using Jellyfin.Server.Tests.Migrations;
|
||||
using MediaBrowser.Controller;
|
||||
using MediaBrowser.Controller.Configuration;
|
||||
using MediaBrowser.Controller.Entities;
|
||||
using MediaBrowser.Model.Configuration;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using Moq;
|
||||
using Npgsql;
|
||||
using Xunit;
|
||||
using BaseItemKind = Jellyfin.Data.Enums.BaseItemKind;
|
||||
using User = Jellyfin.Database.Implementations.Entities.User;
|
||||
|
||||
namespace Jellyfin.Server.Tests.Item;
|
||||
|
||||
/// <summary>
|
||||
/// Runs the presentation-key collapse that library browse, search and the by-name endpoints all go
|
||||
/// through against a real PostgreSQL. SQLite accepts <c>MIN</c> over any column type, PostgreSQL has no
|
||||
/// <c>min(uuid)</c> aggregate, so a representative picked with an aggregate over the id only ever fails
|
||||
/// here - with <c>42883 function min(uuid) does not exist</c>.
|
||||
/// </summary>
|
||||
[Trait("Category", "RequiresDocker")]
|
||||
public sealed class PostgreSqlPresentationKeyGroupingTests : IAsyncLifetime
|
||||
{
|
||||
// The alternate sorts before the primary, and the second duplicate genre before the first, so a
|
||||
// representative that ignores the primary-version preference or the id order picks the wrong row.
|
||||
private static readonly Guid _primaryMovieId = Guid.Parse("eeeeeeee-0000-0000-0000-000000000001");
|
||||
private static readonly Guid _alternateMovieId = Guid.Parse("11111111-0000-0000-0000-000000000002");
|
||||
private static readonly Guid _secondAlternateMovieId = Guid.Parse("22222222-0000-0000-0000-000000000003");
|
||||
private static readonly Guid _firstOrphanId = Guid.Parse("33333333-0000-0000-0000-000000000004");
|
||||
private static readonly Guid _secondOrphanId = Guid.Parse("dddddddd-0000-0000-0000-000000000005");
|
||||
private static readonly Guid _orphanPrimaryId = Guid.Parse("cccccccc-0000-0000-0000-000000000006");
|
||||
private static readonly Guid _standaloneMovieId = Guid.Parse("44444444-0000-0000-0000-000000000007");
|
||||
|
||||
private static readonly Guid _firstGenreId = Guid.Parse("bbbbbbbb-0000-0000-0000-000000000001");
|
||||
private static readonly Guid _secondGenreId = Guid.Parse("aaaaaaaa-0000-0000-0000-000000000002");
|
||||
private static readonly Guid _genreMovieId = Guid.Parse("66666666-0000-0000-0000-000000000003");
|
||||
private static readonly Guid _genreValueId = Guid.Parse("77777777-0000-0000-0000-000000000004");
|
||||
|
||||
private readonly ItemTypeLookup _itemTypeLookup = new();
|
||||
|
||||
private PostgreSqlTestServer _server = null!;
|
||||
private NpgsqlDataSource _dataSource = null!;
|
||||
private BaseItemRepository _repository = null!;
|
||||
|
||||
public async ValueTask InitializeAsync()
|
||||
{
|
||||
_server = await PostgreSqlTestServer.StartAsync().ConfigureAwait(false);
|
||||
var connectionString = await _server.CreateDatabaseAsync("presentation_key_grouping", TestContext.Current.CancellationToken).ConfigureAwait(false);
|
||||
_dataSource = new NpgsqlDataSourceBuilder(connectionString).Build();
|
||||
|
||||
var context = CreateDbContext();
|
||||
await using (context.ConfigureAwait(false))
|
||||
{
|
||||
await context.Database.EnsureCreatedAsync(TestContext.Current.CancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
var serverConfigurationManager = new Mock<IServerConfigurationManager>();
|
||||
serverConfigurationManager.Setup(c => c.Configuration).Returns(new ServerConfiguration());
|
||||
|
||||
var factory = new Mock<IDbContextFactory<JellyfinDbContext>>();
|
||||
factory.Setup(f => f.CreateDbContext()).Returns(CreateDbContext);
|
||||
factory.Setup(f => f.CreateDbContextAsync(It.IsAny<CancellationToken>())).ReturnsAsync(CreateDbContext);
|
||||
|
||||
_repository = new BaseItemRepository(
|
||||
factory.Object,
|
||||
new Mock<IServerApplicationHost>().Object,
|
||||
_itemTypeLookup,
|
||||
serverConfigurationManager.Object,
|
||||
NullLogger<BaseItemRepository>.Instance);
|
||||
|
||||
await SeedAsync().ConfigureAwait(false);
|
||||
}
|
||||
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
await _dataSource.DisposeAsync().ConfigureAwait(false);
|
||||
await _server.DisposeAsync().ConfigureAwait(false);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The collapse behind library browse and search: one row per presentation key, the primary version
|
||||
/// when the group has one, the lowest id otherwise.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void GetItemList_CollapsesPresentationKeyGroups()
|
||||
{
|
||||
var items = _repository.GetItemList(new InternalItemsQuery(new User("grouping", "auth", "reset"))
|
||||
{
|
||||
IncludeItemTypes = [BaseItemKind.Movie],
|
||||
IncludeOwnedItems = true
|
||||
});
|
||||
|
||||
Assert.Equal(
|
||||
new[] { _primaryMovieId, _firstOrphanId, _standaloneMovieId, _genreMovieId }.OrderBy(id => id).ToArray(),
|
||||
items.Select(i => i.Id).OrderBy(id => id).ToArray());
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The same collapse on the by-name path, which picks the lowest id per group without a
|
||||
/// primary-version preference.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void GetGenres_CollapsesPresentationKeyGroups()
|
||||
{
|
||||
var result = _repository.GetGenres(new InternalItemsQuery(new User("genres", "auth", "reset")));
|
||||
|
||||
var item = Assert.Single(result.Items);
|
||||
Assert.Equal(_secondGenreId, item.Item.Id);
|
||||
Assert.Equal(1, result.TotalRecordCount);
|
||||
}
|
||||
|
||||
private JellyfinDbContext CreateDbContext()
|
||||
{
|
||||
var optionsBuilder = new DbContextOptionsBuilder<JellyfinDbContext>();
|
||||
var provider = new PostgreSqlDatabaseProvider(_dataSource);
|
||||
provider.Initialise(optionsBuilder, new DatabaseConfigurationOptions { DatabaseType = "PostgreSQL" });
|
||||
return new JellyfinDbContext(
|
||||
optionsBuilder.Options,
|
||||
NullLogger<JellyfinDbContext>.Instance,
|
||||
provider,
|
||||
new NoLockBehavior(NullLogger<NoLockBehavior>.Instance));
|
||||
}
|
||||
|
||||
private async Task SeedAsync()
|
||||
{
|
||||
var context = CreateDbContext();
|
||||
await using (context.ConfigureAwait(false))
|
||||
{
|
||||
// One version group: a primary plus two alternates, both sorting before it by id.
|
||||
var versionKey = _primaryMovieId.ToString("N");
|
||||
context.BaseItems.Add(CreateMovie(_primaryMovieId, "Movie", versionKey, null));
|
||||
context.BaseItems.Add(CreateMovie(_alternateMovieId, "Movie - 1080p", versionKey, _primaryMovieId));
|
||||
context.BaseItems.Add(CreateMovie(_secondAlternateMovieId, "Movie - 4K", versionKey, _primaryMovieId));
|
||||
|
||||
// One group whose primary is not in the result set, so the lowest id represents it.
|
||||
var orphanKey = _orphanPrimaryId.ToString("N");
|
||||
context.BaseItems.Add(CreateMovie(_firstOrphanId, "Orphan - 1080p", orphanKey, _orphanPrimaryId));
|
||||
context.BaseItems.Add(CreateMovie(_secondOrphanId, "Orphan - 4K", orphanKey, _orphanPrimaryId));
|
||||
|
||||
context.BaseItems.Add(CreateMovie(_standaloneMovieId, "Standalone", _standaloneMovieId.ToString("N"), null));
|
||||
|
||||
// Two genre entities sharing a presentation key, credited on one movie so the by-name
|
||||
// item-value join sees them.
|
||||
var genreKey = _firstGenreId.ToString("N");
|
||||
context.BaseItems.Add(CreateGenre(_firstGenreId, "Action", genreKey));
|
||||
context.BaseItems.Add(CreateGenre(_secondGenreId, "Action", genreKey));
|
||||
|
||||
var genreMovie = CreateMovie(_genreMovieId, "Genre Movie", _genreMovieId.ToString("N"), null);
|
||||
genreMovie.CleanName = "genre movie";
|
||||
context.BaseItems.Add(genreMovie);
|
||||
|
||||
var genreValue = new ItemValue
|
||||
{
|
||||
ItemValueId = _genreValueId,
|
||||
Type = ItemValueType.Genre,
|
||||
Value = "Action",
|
||||
CleanValue = "action"
|
||||
};
|
||||
context.ItemValues.Add(genreValue);
|
||||
context.ItemValuesMap.Add(new ItemValueMap
|
||||
{
|
||||
ItemId = _genreMovieId,
|
||||
ItemValueId = _genreValueId,
|
||||
Item = genreMovie,
|
||||
ItemValue = genreValue
|
||||
});
|
||||
|
||||
await context.SaveChangesAsync(TestContext.Current.CancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
|
||||
private BaseItemEntity CreateMovie(Guid id, string name, string presentationKey, Guid? primaryVersionId)
|
||||
{
|
||||
return new BaseItemEntity
|
||||
{
|
||||
Id = id,
|
||||
Type = _itemTypeLookup.BaseItemKindNames[BaseItemKind.Movie],
|
||||
Name = name,
|
||||
PresentationUniqueKey = presentationKey,
|
||||
PrimaryVersionId = primaryVersionId,
|
||||
MediaType = "Video",
|
||||
IsMovie = true,
|
||||
IsFolder = false,
|
||||
IsVirtualItem = false
|
||||
};
|
||||
}
|
||||
|
||||
private BaseItemEntity CreateGenre(Guid id, string name, string presentationKey)
|
||||
{
|
||||
return new BaseItemEntity
|
||||
{
|
||||
Id = id,
|
||||
Type = _itemTypeLookup.BaseItemKindNames[BaseItemKind.Genre],
|
||||
Name = name,
|
||||
CleanName = "action",
|
||||
PresentationUniqueKey = presentationKey,
|
||||
IsFolder = false,
|
||||
IsVirtualItem = false
|
||||
};
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user