Add lease-aware cleanup to DeleteTranscodeFileTask (#25)

* Initial plan * Add GetActiveSessionsAsync to ITranscodeSessionStore and update DeleteTranscodeFileTask for lease-aware cleanup Co-authored-by: ZoltyMat <177592743+ZoltyMat@users.noreply.github.com> * Fix Redis exception propagation in GetActiveSessionsAsync for safe abort behavior Co-authored-by: ZoltyMat <177592743+ZoltyMat@users.noreply.github.com> * fix: use KeysAsync to resolve CA1849 analyzer violation Replace synchronous IServer.Keys() with async IServer.KeysAsync() using await foreach to satisfy CA1849 (TreatWarningsAsErrors). CA1849: 'IServer.Keys()' synchronously blocks. Await 'IServer.KeysAsync()' instead. Line 161 in RedisTranscodeSessionStore.GetActiveSessionsAsync. --------- Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com> Co-authored-by: ZoltyMat <177592743+ZoltyMat@users.noreply.github.com> Co-authored-by: mat <mstrommen@gmail.com>
2026-03-10 01:11:32 -04:00
parent c2a11f3e68
commit 9817185fa3
9 changed files with 412 additions and 6 deletions
@@ -1,4 +1,6 @@
 using System;
+using System.Collections.Generic;
+using System.Linq;
 using System.Text.Json;
 using System.Threading;
 using System.Threading.Tasks;
@@ -35,6 +37,7 @@ 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<RedisTranscodeSessionStore> _logger;
@@ -50,6 +53,7 @@ return 1";
        IOptions<TranscodeStoreOptions> options,
        ILogger<RedisTranscodeSessionStore> logger)
    {
+        _redis = redis;
        _db = redis.GetDatabase();
        _options = options.Value;
        _logger = logger;
@@ -140,4 +144,54 @@ return 1";
    }

    private static string GetKey(string playSessionId) => KeyPrefix + playSessionId;
+
+    /// <inheritdoc />
+    public async Task<IEnumerable<TranscodeSession>> GetActiveSessionsAsync(CancellationToken cancellationToken = default)
+    {
+        var sessions = new List<TranscodeSession>();
+        var servers = _redis.GetServers();
+
+        foreach (var server in servers)
+        {
+            if (!server.IsConnected)
+            {
+                continue;
+            }
+
+            var keys = new List<RedisKey>();
+            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<TranscodeSession>(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;
+    }
 }