using System;
using System.Collections.Generic;
using System.Linq;
using System.Text.Json;
using System.Threading;
using System.Threading.Tasks;
using MediaBrowser.Controller.MediaEncoding;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using StackExchange.Redis;
namespace Emby.Server.Implementations.MediaEncoding;
///
/// A Redis-backed implementation of that provides
/// durable, distributed session tracking with lease-based ownership between pods.
///
public sealed class RedisTranscodeSessionStore : ITranscodeSessionStore
{
private const string KeyPrefix = "jellyfin:transcode:";
private const string LiveStreamKeyPrefix = "jellyfin:livestream:";
///
/// Lua script for atomic takeover: reads the stored session, checks whether the lease has
/// expired (comparing LeaseExpiresUtc.Ticks against the caller-supplied current ticks),
/// and if expired, updates the owner and expiry before returning 1; returns 0 otherwise.
///
private const string TakeoverScript = @"
local raw = redis.call('GET', KEYS[1])
if not raw then return 0 end
local session = cjson.decode(raw)
local currentTicks = tonumber(ARGV[1])
if session['LeaseExpiresUtc'] > currentTicks then return 0 end
session['OwnerPod'] = ARGV[2]
local leaseDurationMs = tonumber(ARGV[3])
local newTicks = currentTicks + (leaseDurationMs * 10000)
session['LeaseExpiresUtc'] = newTicks
redis.call('SET', KEYS[1], cjson.encode(session), 'PX', leaseDurationMs)
return 1";
private readonly IConnectionMultiplexer _redis;
private readonly IDatabase _db;
private readonly TranscodeStoreOptions _options;
private readonly ILogger _logger;
///
/// Initializes a new instance of the class.
///
/// The Redis connection multiplexer.
/// The transcode store configuration options.
/// The logger.
public RedisTranscodeSessionStore(
IConnectionMultiplexer redis,
IOptions options,
ILogger logger)
{
_redis = redis;
_db = redis.GetDatabase();
_options = options.Value;
_logger = logger;
}
///
public async Task SetAsync(TranscodeSession session, CancellationToken cancellationToken = default)
{
var key = GetKey(session.PlaySessionId);
var json = JsonSerializer.Serialize(session);
var leaseDurationMs = (long)_options.LeaseDurationSeconds * 1000;
await _db.StringSetAsync(key, json, TimeSpan.FromMilliseconds(leaseDurationMs)).ConfigureAwait(false);
_logger.LogDebug("Set transcode session {PlaySessionId} in Redis.", session.PlaySessionId);
}
///
public async Task TryGetAsync(string playSessionId, CancellationToken cancellationToken = default)
{
var key = GetKey(playSessionId);
var raw = await _db.StringGetAsync(key).ConfigureAwait(false);
if (!raw.HasValue)
{
return null;
}
var session = JsonSerializer.Deserialize(raw.ToString());
// Check LeaseExpiresUtc in addition to Redis TTL to guard against the window between
// Redis TTL evaluation and the GET result being returned to the caller.
if (session is null || session.LeaseExpiresUtc <= DateTime.UtcNow)
{
return null;
}
return session;
}
///
public async Task RenewLeaseAsync(string playSessionId, CancellationToken cancellationToken = default)
{
var key = GetKey(playSessionId);
var raw = await _db.StringGetAsync(key).ConfigureAwait(false);
if (!raw.HasValue)
{
return;
}
var session = JsonSerializer.Deserialize(raw.ToString());
if (session is null)
{
return;
}
var leaseDurationMs = (long)_options.LeaseDurationSeconds * 1000;
session.LeaseExpiresUtc = DateTime.UtcNow.AddMilliseconds(leaseDurationMs);
var json = JsonSerializer.Serialize(session);
await _db.StringSetAsync(key, json, TimeSpan.FromMilliseconds(leaseDurationMs)).ConfigureAwait(false);
_logger.LogDebug("Renewed lease for transcode session {PlaySessionId}.", playSessionId);
}
///
public async Task DeleteAsync(string playSessionId, CancellationToken cancellationToken = default)
{
var key = GetKey(playSessionId);
await _db.KeyDeleteAsync(key).ConfigureAwait(false);
_logger.LogDebug("Deleted transcode session {PlaySessionId} from Redis.", playSessionId);
}
///
public async Task TryTakeoverAsync(string playSessionId, string claimingPod, CancellationToken cancellationToken = default)
{
var key = GetKey(playSessionId);
var leaseDurationMs = (long)_options.LeaseDurationSeconds * 1000;
var currentTicks = DateTime.UtcNow.Ticks;
var result = (long?)await _db.ScriptEvaluateAsync(
TakeoverScript,
keys: new RedisKey[] { key },
values: new RedisValue[] { currentTicks, claimingPod, leaseDurationMs }).ConfigureAwait(false);
var succeeded = result == 1;
if (succeeded)
{
_logger.LogInformation("Pod {ClaimingPod} successfully took over transcode session {PlaySessionId}.", claimingPod, playSessionId);
}
return succeeded;
}
private static string GetKey(string playSessionId) => KeyPrefix + playSessionId;
private static string GetLiveStreamKey(string liveStreamId, string sessionIdOrPlaySessionId)
=> LiveStreamKeyPrefix + liveStreamId + ":" + sessionIdOrPlaySessionId;
///
public async Task> GetActiveSessionsAsync(CancellationToken cancellationToken = default)
{
var sessions = new List();
var servers = _redis.GetServers();
foreach (var server in servers)
{
if (!server.IsConnected)
{
continue;
}
var keys = new List();
await foreach (var key in server.KeysAsync(database: _db.Database, pattern: KeyPrefix + "*", pageSize: 1000).WithCancellation(cancellationToken).ConfigureAwait(false))
{
keys.Add(key);
}
var tasks = keys.Select(key => _db.StringGetAsync(key)).ToList();
var values = await Task.WhenAll(tasks).ConfigureAwait(false);
foreach (var raw in values)
{
if (!raw.HasValue)
{
continue;
}
TranscodeSession? session;
try
{
session = JsonSerializer.Deserialize(raw.ToString());
}
catch (Exception ex)
{
_logger.LogWarning(ex, "Failed to deserialize transcode session from Redis.");
continue;
}
if (session is not null)
{
sessions.Add(session);
}
}
}
return sessions;
}
///
public async Task SetLiveStreamAsync(LiveStreamSession session, CancellationToken cancellationToken = default)
{
var key = GetLiveStreamKey(session.LiveStreamId, session.SessionId);
var json = JsonSerializer.Serialize(session);
// Live stream records use the same lease duration as transcode sessions.
var leaseDurationMs = (long)_options.LeaseDurationSeconds * 1000;
await _db.StringSetAsync(key, json, TimeSpan.FromMilliseconds(leaseDurationMs)).ConfigureAwait(false);
// Also index by play session id so the caller can look up by either key.
if (!string.IsNullOrEmpty(session.PlaySessionId))
{
var playKey = GetLiveStreamKey(session.LiveStreamId, session.PlaySessionId);
await _db.StringSetAsync(playKey, json, TimeSpan.FromMilliseconds(leaseDurationMs)).ConfigureAwait(false);
}
_logger.LogDebug(
"Set live stream session {LiveStreamId}/{SessionId} in Redis.",
session.LiveStreamId,
session.SessionId);
}
///
public async Task TryGetLiveStreamAsync(string liveStreamId, string sessionIdOrPlaySessionId, CancellationToken cancellationToken = default)
{
var key = GetLiveStreamKey(liveStreamId, sessionIdOrPlaySessionId);
var raw = await _db.StringGetAsync(key).ConfigureAwait(false);
if (!raw.HasValue)
{
return null;
}
return JsonSerializer.Deserialize(raw.ToString());
}
///
public async Task DeleteLiveStreamAsync(string liveStreamId, string sessionIdOrPlaySessionId, CancellationToken cancellationToken = default)
{
var key = GetLiveStreamKey(liveStreamId, sessionIdOrPlaySessionId);
var raw = await _db.StringGetAsync(key).ConfigureAwait(false);
if (raw.HasValue)
{
var session = JsonSerializer.Deserialize(raw.ToString());
if (session is not null)
{
// Remove both the session-id key and the play-session-id key if present.
var keysToDelete = new System.Collections.Generic.List
{
GetLiveStreamKey(liveStreamId, session.SessionId)
};
if (!string.IsNullOrEmpty(session.PlaySessionId))
{
keysToDelete.Add(GetLiveStreamKey(liveStreamId, session.PlaySessionId));
}
await _db.KeyDeleteAsync(keysToDelete.ToArray()).ConfigureAwait(false);
_logger.LogDebug(
"Deleted live stream session {LiveStreamId}/{SessionId} from Redis.",
liveStreamId,
session.SessionId);
return;
}
}
// Fallback: delete just the key that was supplied.
await _db.KeyDeleteAsync(key).ConfigureAwait(false);
}
}