feat(stream): retry a dropped push instead of ending the session (TASK 32·1)

A dead encoder used to run the whole stop path, so a one-second blip ended the broadcast. Now a dropped
push is relaunched against the same ingest while the broadcast is still alive.

- `Services/StreamRetryPolicy.cs` — immediate, +1 s, +2 s, +3 s. Minimum waits on purpose (creator
  ruling): YouTube's grace is short, so every second not pushing is grace spent.
- `Services/StreamRecovery.cs` — the loop, with the audio pipe recreated before every attempt (its
  server is single-instance and unusable after the old client dies, so a naive relaunch fails on a
  missing input). After a push comes up it asks the API whether the broadcast survived: "complete"
  ends recovery as `BroadcastComplete` instead of showing a live sign over a dead stream; an API
  hiccup counts as recovered rather than ending a healthy session.
- `FramePump.EncoderFailed` — a dead push is recoverable; a compositor fault still goes through
  `Failed` and ends the session, because retrying it would loop.
- `MainViewModel.Recovery.cs` — wires the steps, reports each episode into the left panel alert stack
  (one row per episode, updated in place, log range painted), and cancels on Stop.
- Alert stack polish: a rule and a gap now separate the stack from the scene layers (they read as one
  list before), and the licence indicator uses the creator's wording — `STEAM ID NOT PRESENT` /
  `Steam ID Limit: 30fps` / `Click for More Info`, becoming `STEAM ID PRESENT` / `60fps` with a close
  control once owned.

Tests: `StreamRecoveryTests` (schedule, recover on attempt 2 with the pipe restarted each time,
complete-broadcast honesty, unknown liveness, give-up, cancel, audio-restart fault),
`FramePumpTests` now pin that process death raises `EncoderFailed` and not `Failed`.
FAST gate: 538 passed (one documented pre-existing audio failure), 0 warnings.
This commit is contained in:
2026-11-05 12:09:38 -08:00
parent a80108dd96
commit 72cb3ad55a
16 changed files with 583 additions and 35 deletions
+7 -1
View File
@@ -129,6 +129,10 @@ public sealed class FramePump : IDisposable
/// <summary>Raised when the encoder cannot start or dies mid-stream. The pump stops itself.</summary>
public event EventHandler<string>? Failed;
/// <summary>TASK 32·1: the ENCODER died (the push dropped) — recoverable, unlike a compositor fault.
/// Raised before the pump stops, so the session can relaunch the push instead of ending.</summary>
public event EventHandler<string>? EncoderFailed;
public FramePump(
Func<Scene?> sceneProvider,
Func<SceneElement, VideoFrame?> frameResolver,
@@ -969,7 +973,9 @@ public sealed class FramePump : IDisposable
private void OnProcessFailed(object? sender, string message)
{
_log?.Invoke($"FramePump: encoder process failed: {message}");
Failed?.Invoke(this, message);
// A dead push is not a dead session: the broadcast may still be live, so this is the
// recoverable signal (TASK 32·1). Fatal pump faults still go through Failed.
EncoderFailed?.Invoke(this, message);
_ = StopAsync();
}
}
+150
View File
@@ -0,0 +1,150 @@
using System;
using System.Threading;
using System.Threading.Tasks;
namespace ytLive.Services;
/// <summary>What YouTube says about the broadcast once a push is back.</summary>
public enum BroadcastLiveness
{
/// <summary>Could not tell (no token, network, API hiccup) — treat as recovered.</summary>
Unknown,
/// <summary>The broadcast is still live/ready: the push landed and the show continues.</summary>
Live,
/// <summary>The broadcast is complete. It cannot be resumed; the push is talking to a dead ingest.</summary>
Complete,
}
/// <summary>Everything the recovery loop needs from the app, as delegates so it can be tested
/// without ffmpeg, without a network, and without a window.</summary>
public sealed class StreamRecoverySteps
{
/// <summary>Recreates the audio pipe server. The old one is unusable after its client died
/// (single-instance server), and the encoder cannot start without it.</summary>
public required Func<CancellationToken, Task> RestartAudio { get; init; }
/// <summary>Starts the push. True when the encoder came up.</summary>
public required Func<CancellationToken, Task<bool>> StartPush { get; init; }
/// <summary>Asks YouTube whether the broadcast survived. Only called after a push starts.</summary>
public required Func<CancellationToken, Task<BroadcastLiveness>> CheckBroadcast { get; init; }
/// <summary>State for the alert stack: attempt numbers and the outcome.</summary>
public required Action<string> Report { get; init; }
/// <summary>The record: one line per attempt, into the app log.</summary>
public required Action<string> Log { get; init; }
public Func<TimeSpan, CancellationToken, Task> Delay { get; init; }
= (delay, ct) => delay <= TimeSpan.Zero ? Task.CompletedTask : Task.Delay(delay, ct);
}
public enum RecoveryOutcome
{
Recovered,
GaveUp,
Cancelled,
BroadcastComplete,
}
public sealed record RecoveryResult(RecoveryOutcome Outcome, int Attempts, TimeSpan Elapsed);
/// <summary>
/// TASK 32·1 — the blip retry. When the encoder dies while the session is live, relaunch the same push
/// against the same ingest, restarting the audio pipe each time (its server is single-instance and the
/// dead client leaves it unusable).
/// <para><b>The check that stops us lying:</b> a reconnect can succeed at the socket level after
/// YouTube has already ended the broadcast. So each successful start asks the API whether the broadcast
/// is still there; "complete" ends recovery instead of showing a live sign over a dead stream.</para>
/// </summary>
public sealed class StreamRecovery
{
private readonly StreamRetryPolicy _policy;
private readonly StreamRecoverySteps _steps;
public StreamRecovery(StreamRetryPolicy policy, StreamRecoverySteps steps)
{
_policy = policy;
_steps = steps;
}
public async Task<RecoveryResult> RunAsync(CancellationToken cancellationToken = default)
{
var clock = System.Diagnostics.Stopwatch.StartNew();
var attempts = 0;
for (var attempt = 0; attempt < _policy.Attempts; attempt++)
{
if (cancellationToken.IsCancellationRequested) return Result(RecoveryOutcome.Cancelled);
attempts = attempt + 1;
var wait = _policy.WaitBefore(attempt);
if (wait > TimeSpan.Zero)
{
try { await _steps.Delay(wait, cancellationToken).ConfigureAwait(false); }
catch (OperationCanceledException) { return Result(RecoveryOutcome.Cancelled); }
}
if (cancellationToken.IsCancellationRequested) return Result(RecoveryOutcome.Cancelled);
_steps.Report($"Stream dropped — reconnecting (attempt {attempts} of {_policy.Attempts})");
_steps.Log($"RECONNECT: attempt {attempts} of {_policy.Attempts} — restarting the audio pipe and the push");
try
{
await _steps.RestartAudio(cancellationToken).ConfigureAwait(false);
}
catch (Exception ex)
{
_steps.Log($"RECONNECT: audio pipe restart failed on attempt {attempts}: {ex.Message}");
continue;
}
bool started;
try
{
started = await _steps.StartPush(cancellationToken).ConfigureAwait(false);
}
catch (Exception ex)
{
_steps.Log($"RECONNECT: push restart failed on attempt {attempts}: {ex.Message}");
continue;
}
if (!started)
{
_steps.Log($"RECONNECT: attempt {attempts} of {_policy.Attempts} failed — the encoder did not come up");
continue;
}
var liveness = BroadcastLiveness.Unknown;
try
{
liveness = await _steps.CheckBroadcast(cancellationToken).ConfigureAwait(false);
}
catch (Exception ex)
{
_steps.Log($"RECONNECT: could not check the broadcast on attempt {attempts}: {ex.Message}");
}
if (liveness == BroadcastLiveness.Complete)
{
_steps.Report("Stream ended — YouTube reports this broadcast complete. It cannot be resumed.");
_steps.Log("RECONNECT: push came back, but the broadcast is complete — cannot resume");
return Result(RecoveryOutcome.BroadcastComplete);
}
_steps.Report($"Stream recovered after {clock.Elapsed.TotalSeconds:0.0} s");
_steps.Log($"RECONNECT: push re-established after {clock.Elapsed.TotalSeconds:0.0} s — still live");
return Result(RecoveryOutcome.Recovered);
}
_steps.Report($"Stream dropped — {_policy.Attempts} attempts failed. The stream is off air.");
_steps.Log($"RECONNECT: gave up after {_policy.Attempts} attempts — stream is off air");
return Result(RecoveryOutcome.GaveUp);
RecoveryResult Result(RecoveryOutcome outcome) => new(outcome, attempts, clock.Elapsed);
}
}
+38
View File
@@ -0,0 +1,38 @@
using System;
using System.Collections.Generic;
namespace ytLive.Services;
/// <summary>
/// TASK 32·1: how many times to relaunch the push, and how long to wait between tries.
/// <para><b>Minimum waits, deliberately</b> (creator ruling 2026-11-05): YouTube is not generous about
/// holding a broadcast whose ingest has gone quiet, so every second spent not pushing is grace spent.
/// The first attempt is immediate — there is nothing to wait for when the socket has just died.</para>
/// <para>Pure so the schedule is asserted without waiting on it.</para>
/// </summary>
public sealed class StreamRetryPolicy
{
/// <summary>Immediate, then +1 s, +2 s, +3 s — four attempts inside about six seconds.</summary>
public static readonly StreamRetryPolicy Default = new(new[]
{
TimeSpan.Zero,
TimeSpan.FromSeconds(1),
TimeSpan.FromSeconds(2),
TimeSpan.FromSeconds(3),
});
public IReadOnlyList<TimeSpan> Waits { get; }
public StreamRetryPolicy(IReadOnlyList<TimeSpan> waits)
{
if (waits == null || waits.Count == 0)
throw new ArgumentException("A retry policy needs at least one attempt.", nameof(waits));
Waits = waits;
}
public int Attempts => Waits.Count;
/// <summary>The wait BEFORE <paramref name="attemptIndex"/> (0-based).</summary>
public TimeSpan WaitBefore(int attemptIndex)
=> attemptIndex < 0 || attemptIndex >= Waits.Count ? TimeSpan.Zero : Waits[attemptIndex];
}
+19
View File
@@ -182,6 +182,25 @@ public class YouTubeStreamService
/// "public"), read from YouTube rather than trusted from local state — the creator can
/// change visibility mid-stream from the broadcast form. Null when signed out or the
/// read fails (the caller then does nothing destructive).</summary>
/// <summary>TASK 32·1: the broadcast's lifecycle status (<c>created</c>/<c>ready</c>/<c>testing</c>/
/// <c>live</c>/<c>complete</c>), or null when it cannot be read. Used after a reconnect: a push can
/// succeed at the socket level while YouTube has already ended the broadcast.</summary>
public async Task<string?> GetBroadcastStatusAsync(string broadcastId)
{
if (!await EnsureToken()) return null;
_http.DefaultRequestHeaders.Authorization = new("Bearer", _auth.CurrentChannel!.AccessToken);
var response = await _http.GetAsync(
$"{ApiBase}/liveBroadcasts?part=status&id={Uri.EscapeDataString(broadcastId)}");
if (!response.IsSuccessStatusCode) return null;
var json = JsonSerializer.Deserialize<JsonElement>(await response.Content.ReadAsStringAsync());
if (!json.TryGetProperty("items", out var items) || items.GetArrayLength() == 0) return null;
if (!items[0].TryGetProperty("status", out var status)) return null;
if (!status.TryGetProperty("lifeCycleStatus", out var lifeCycle)) return null;
return lifeCycle.ValueKind == JsonValueKind.String ? lifeCycle.GetString() : null;
}
public async Task<string?> GetBroadcastPrivacyAsync(string broadcastId)
{
if (!await EnsureToken()) return null;