Align decoder startup deadlines and deploy resilient 7.0.8

Tests/AdaptiveProbeTests.cs

@@ -50,6 +50,38 @@ public static class AdaptiveProbeTests
         }
         check(Deleted(snapshots), "successful adaptive probe deletes capture and snapshots");
 
+        snapshots.Clear(); attempts = 0;
+        var rejectedSampleStream = new ProbePacketStream(delay: 1);
+        using (var handler = new ProbeHttpHandler(rejectedSampleStream))
+        using (var client = new HttpClient(handler) { Timeout = Timeout.InfiniteTimeSpan })
+        {
+            var probe = new LiveStreamProbe(client, (path, programme, extended, token) =>
+            {
+                snapshots.Add(path); attempts++;
+                if (!extended) throw new LiveStreamProbe.ProbeProcessException(1);
+                return Task.FromResult(Metadata(Sid, 1920));
+            }, limits, TimeSpan.FromSeconds(2));
+            check((await probe.Get("http://relay/channel", CancellationToken.None, Sid))[0].Width == 1920 && attempts == 2 && handler.Calls == 1,
+                "a first ffprobe sample rejection retries analysis on the same HTTP capture");
+        }
+        check(rejectedSampleStream.Disposed && Deleted(snapshots), "retried process rejection cleans up subscription and sample files");
+
+        snapshots.Clear(); attempts = 0;
+        var failedAnalysisStream = new ProbePacketStream(delay: 1);
+        using (var handler = new ProbeHttpHandler(failedAnalysisStream))
+        using (var client = new HttpClient(handler) { Timeout = Timeout.InfiniteTimeSpan })
+        {
+            var probe = new LiveStreamProbe(client, (path, programme, extended, token) =>
+            {
+                snapshots.Add(path); attempts++;
+                throw new InvalidOperationException("Analysis implementation failed.");
+            }, limits, TimeSpan.FromSeconds(2));
+            var error = await Failure(probe.Get("http://relay/channel", CancellationToken.None, Sid));
+            check(attempts == 1 && handler.Calls == 1 && error.Contains("cause=probe_analysis_failure"),
+                "an unrelated first analysis failure is not retried");
+        }
+        check(failedAnalysisStream.Disposed && Deleted(snapshots), "unrelated analysis failure cleans up subscription and sample files");
+
         snapshots.Clear(); attempts = 0;
         var initiallyQuietStream = new ProbePacketStream(delay: 150);
         using (var handler = new ProbeHttpHandler(initiallyQuietStream))
@@ -125,7 +157,7 @@ public static class AdaptiveProbeTests
                 finally { cancelledAnalysis = true; }
             }, limits, TimeSpan.FromMilliseconds(180));
             var error = await Failure(probe.Get("http://relay/channel", CancellationToken.None, Sid));
-            check(cancelledAnalysis && handler.Calls == 1 && analysingStream.Disposed && Deleted(snapshots) && error.Contains("cause=deadline"), "deadline cancels analysis, joins the reader and removes temporary files");
+            check(cancelledAnalysis && handler.Calls == 1 && snapshots.Count == 1 && analysingStream.Disposed && Deleted(snapshots) && error.Contains("cause=deadline"), "deadline cancels analysis without retry, joins the reader and removes temporary files");
         }
     }
 

Tests/Program.cs

@@ -107,6 +107,7 @@ Check(true,"empty source rejected before playback");
 var liveVideo=LiveStreamProbe.ParseLiveMetadata(probes[0][0]);
 Check(liveVideo.Index==-1&&liveVideo.IsInterlaced&&liveVideo.AverageFrameRate==25&&probes[0][0].Index==0,"live automatic mapping preserves metadata without mutating cached probe");
 await AdaptiveProbeTests.Run(Check);
+await RelayStartupTests.Run(Check);
 var relayHandler=new RelayHandler();
 using(var relay=new LiveStreamRelay(new System.Net.Http.HttpClient(relayHandler),8096,TimeSpan.FromMilliseconds(100)))
 {

Tests/RelayStartupTests.cs

@@ -0,0 +1,135 @@
+using System;
+using System.Diagnostics;
+using System.IO;
+using System.Net;
+using System.Net.Http;
+using System.Text.Json;
+using System.Threading;
+using System.Threading.Tasks;
+using Jellyfin.Plugin.Enigma2;
+
+public static class RelayStartupTests
+{
+    public static async Task Run(Action<bool, string> check)
+    {
+        var delayed = new StartupPacketStream(TimeSpan.FromMilliseconds(120));
+        using (var handler = new StartupHttpHandler(delayed, TimeSpan.FromMilliseconds(70)))
+        using (var relay = new LiveStreamRelay(new HttpClient(handler), 8096, TimeSpan.FromMilliseconds(10), TimeSpan.FromMilliseconds(300), TimeSpan.FromMilliseconds(50)))
+        {
+            var id = relay.Register("http://receiver/delayed").Split('/')[^1];
+            var subscriptions = await Task.WhenAll(relay.Subscribe(id, CancellationToken.None), relay.Subscribe(id, CancellationToken.None));
+            await using var first = subscriptions[0];
+            await using var second = subscriptions[1];
+            var one = await first.Reader.ReadAsync();
+            var two = await second.Reader.ReadAsync();
+            check(handler.Calls == 1 && one[0] == 0x47 && two[0] == 0x47 && delayed.Delivered > 0,
+                "two subscribers share delayed startup beyond the steady-state read timeout");
+        }
+
+        var slowTotal = new StartupPacketStream(TimeSpan.FromMilliseconds(180));
+        using (var handler = new StartupHttpHandler(slowTotal, TimeSpan.FromMilliseconds(180)))
+        using (var relay = new LiveStreamRelay(new HttpClient(handler), 8096, TimeSpan.FromMilliseconds(10), TimeSpan.FromMilliseconds(300), TimeSpan.FromMilliseconds(50)))
+        {
+            var id = relay.Register("http://receiver/combined-delay").Split('/')[^1];
+            var watch = Stopwatch.StartNew();
+            await ExpectFailure(relay.Subscribe(id, CancellationToken.None));
+            check(handler.Calls == 1 && handler.HeadersReturned && slowTotal.ReadCancelled && slowTotal.Delivered == 0 && slowTotal.Disposed && watch.Elapsed < TimeSpan.FromSeconds(2),
+                "HTTP headers and first packet use one cumulative startup deadline");
+            for (var i = 0; i < 20 && !JsonSerializer.Serialize(relay.Status()).Contains("\"ActiveSources\":0"); i++) await Task.Delay(5);
+            check(JsonSerializer.Serialize(relay.Status()).Contains("\"ActiveSources\":0"), "startup timeout closes its source");
+        }
+
+        var noHeaders = new StartupPacketStream(TimeSpan.Zero);
+        using (var handler = new StartupHttpHandler(noHeaders, TimeSpan.FromMilliseconds(180)))
+        using (var relay = new LiveStreamRelay(new HttpClient(handler), 8096, TimeSpan.FromMilliseconds(10), TimeSpan.FromMilliseconds(60), TimeSpan.FromMilliseconds(30)))
+        {
+            var id = relay.Register("http://receiver/silent-headers").Split('/')[^1];
+            await ExpectFailure(relay.Subscribe(id, CancellationToken.None));
+            check(handler.Cancelled && !handler.HeadersReturned && noHeaders.Delivered == 0,
+                "silent HTTP headers obey the same bounded startup deadline");
+        }
+
+        var stalled = new StartupPacketStream(TimeSpan.Zero, stallAfterFirst: true);
+        using (var handler = new StartupHttpHandler(stalled, TimeSpan.Zero))
+        using (var relay = new LiveStreamRelay(new HttpClient(handler), 8096, TimeSpan.FromMilliseconds(10), TimeSpan.FromSeconds(2), TimeSpan.FromMilliseconds(50)))
+        {
+            var id = relay.Register("http://receiver/steady-stall").Split('/')[^1];
+            await using var subscription = await relay.Subscribe(id, CancellationToken.None);
+            check((await subscription.Reader.ReadAsync()).Length > 0, "initial packet makes relay ready");
+            var watch = Stopwatch.StartNew();
+            await ExpectFailure(subscription.Reader.ReadAsync().AsTask());
+            check(stalled.ReadCancelled && stalled.Disposed && stalled.Delivered == 1 && watch.Elapsed < TimeSpan.FromSeconds(1),
+                "a stall after startup retains the shorter steady-state timeout");
+        }
+
+        using (var client = new HttpClient(new StartupHttpHandler(new StartupPacketStream(TimeSpan.Zero), TimeSpan.Zero)))
+        {
+            var startupRejected = false;
+            var readRejected = false;
+            try { using var invalid = new LiveStreamRelay(client, 8096, startupTimeout: TimeSpan.Zero); }
+            catch (ArgumentOutOfRangeException) { startupRejected = true; }
+            try { using var invalid = new LiveStreamRelay(client, 8096, readTimeout: TimeSpan.FromMilliseconds(-1)); }
+            catch (ArgumentOutOfRangeException) { readRejected = true; }
+            check(startupRejected && readRejected, "relay rejects non-positive startup and read budgets");
+        }
+    }
+
+    private static async Task ExpectFailure(Task task)
+    {
+        try { await task.WaitAsync(TimeSpan.FromSeconds(2)); }
+        catch (IOException) { return; }
+        catch (OperationCanceledException) { return; }
+        throw new Exception("Expected bounded relay failure.");
+    }
+}
+
+public sealed class StartupHttpHandler : HttpMessageHandler
+{
+    private readonly Stream stream;
+    private readonly TimeSpan headerDelay;
+    public int Calls;
+    public bool HeadersReturned;
+    public bool Cancelled;
+    public StartupHttpHandler(Stream stream, TimeSpan headerDelay) { this.stream = stream; this.headerDelay = headerDelay; }
+    protected override async Task<HttpResponseMessage> SendAsync(HttpRequestMessage request, CancellationToken token)
+    {
+        Interlocked.Increment(ref Calls);
+        try { await Task.Delay(headerDelay, token); }
+        catch (OperationCanceledException) { Cancelled = true; throw; }
+        HeadersReturned = true;
+        return new HttpResponseMessage(HttpStatusCode.OK) { Content = new StreamContent(stream) };
+    }
+    protected override void Dispose(bool disposing) { if (disposing) stream.Dispose(); base.Dispose(disposing); }
+}
+
+public sealed class StartupPacketStream : Stream
+{
+    private readonly TimeSpan firstDelay;
+    private readonly bool stallAfterFirst;
+    public int Delivered;
+    public bool ReadCancelled;
+    public bool Disposed;
+    public StartupPacketStream(TimeSpan firstDelay, bool stallAfterFirst = false) { this.firstDelay = firstDelay; this.stallAfterFirst = stallAfterFirst; }
+    public override async ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken token = default)
+    {
+        try { await Task.Delay(Delivered == 0 ? firstDelay : stallAfterFirst ? Timeout.InfiniteTimeSpan : TimeSpan.FromMilliseconds(10), token); }
+        catch (OperationCanceledException) { ReadCancelled = true; throw; }
+        var count = Math.Min(buffer.Length, 188 * 256);
+        count -= count % 188;
+        buffer.Span[..count].Clear();
+        for (var i = 0; i < count; i += 188) buffer.Span[i] = 0x47;
+        Delivered++;
+        return count;
+    }
+    protected override void Dispose(bool disposing) { Disposed = true; base.Dispose(disposing); }
+    public override bool CanRead => true;
+    public override bool CanSeek => false;
+    public override bool CanWrite => false;
+    public override long Length => throw new NotSupportedException();
+    public override long Position { get => 0; set => throw new NotSupportedException(); }
+    public override int Read(byte[] buffer, int offset, int count) => throw new NotSupportedException();
+    public override void Flush() { }
+    public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException();
+    public override void SetLength(long value) => throw new NotSupportedException();
+    public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException();
+}