feat(33): the step-down acts — one encoder relaunch, one toast, same URL
Slice 3. 589 green, 0 warnings. - **`FramePump.ChangeEncoderAsync`** is a full pump stop/start, not an encoder swap. A swap leaves an in-flight submit pointing at a disposed process, and that faults the loop — which ends the broadcast, the one thing a quality step-down must never do. `StopAsync` already awaits the loop and flushes, so this reuses the tested path and costs one short freeze, which the ladder accepts. - **`MainViewModel.StepDown`**: one planner per broadcast, subscribed to `StatsReported`, unsubscribed on every stop path. The pump thread only decides; the restart is posted to the dispatcher, because the act stops the loop that raised the event. - **The creator's choice is not rewritten**: `_encoderTierOverride` changes what the encoder emits while the dropdown keeps showing what they picked, and the badge adds `· lowered` so the two never disagree silently. - Copy comes from `StepDownPlanner.Describe` (machine cause owns it, upload cause blames the upload); the log gets the cause-tagged line for support. Test: `EncoderRestartTests` runs a real pump with a fake encoder, changes the tier, and asserts a NEW encoder started at the new fps/bitrate and produced a frame — the regression that would otherwise ship as "it lowered the badge and the stream died". Remaining for 33: shed our own capture cost before the fps step, and the session-report line.
This commit is contained in:
@@ -248,6 +248,21 @@ public sealed class FramePump : IDisposable
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>TASK 33: relaunch the pump and encoder at a new tier, over the same URL and audio pipe.
|
||||
///
|
||||
/// <para>Full stop + start rather than an encoder swap, deliberately: <see cref="StopAsync"/> awaits
|
||||
/// the loop and flushes the encoder, so no in-flight submit can land on a disposed process. A bare
|
||||
/// swap races exactly that and would fault the loop — which ends the broadcast, the one thing a
|
||||
/// quality step-down must never do. The cost is a short freeze, already accepted by the ladder.</para>
|
||||
/// </summary>
|
||||
public async Task<bool> ChangeEncoderAsync(CancellationToken cancellationToken = default)
|
||||
{
|
||||
await StopAsync(cancellationToken);
|
||||
if (cancellationToken.IsCancellationRequested) return false;
|
||||
await StartAsync(cancellationToken);
|
||||
return IsRunning;
|
||||
}
|
||||
|
||||
public async Task StopAsync(CancellationToken cancellationToken = default)
|
||||
{
|
||||
IFfmpegEncoder? encoder;
|
||||
|
||||
@@ -49,7 +49,7 @@ cited first), then 35 / 37 / 40·A/C/D / 55·2–3. TASK 32 is complete, Back on
|
||||
| 40·A/C/D | ☐ | App Settings: active-camera picker; persist output resolution + surface transition defaults + confirm-before-End; accent consolidation | [`task-40`](TASKS/task-40-app-settings-round.md) |
|
||||
| 3·20 | ✅ | Session report — **derived from `superChatEvents.list` on demand, not persisted** (YouTube is the authority; no ledger). Per-currency totals, money off by default (done 2026-11-05) | [`task-03`](TASKS/task-03-capture-pipeline.md) |
|
||||
| 32 | ✅ | Stream resilience: fragmented MP4, blip retry, poll-driven RECONNECTING (no countdown — YouTube's cutoff is not a constant), ingest pre-flight, one-click Back on air (`BackOnAirTests`) | [`task-32`](TASKS/task-32-stream-resilience.md) |
|
||||
| 33 | 🔶 | Auto step-down — **`StepDownPolicy` + 12 unit tests done 2026-11-09** (two ladders, budgets, floors, cooldown, ambiguity, recovery behind a flag); policy + `StepDownPlanner` + 16 tests done; remaining: VM subscribe + encoder restart + toast, capture-cost shed | [`task-33`](TASKS/task-33-bandwidth-step-down.md) |
|
||||
| 33 | 🔶 | Auto step-down — **`StepDownPolicy` + 12 unit tests done 2026-11-09** (two ladders, budgets, floors, cooldown, ambiguity, recovery behind a flag); policy, planner, VM watch, encoder relaunch and the toast are in (17 tests); remaining: shed our own capture cost before the fps step, and the session-report line | [`task-33`](TASKS/task-33-bandwidth-step-down.md) |
|
||||
| 34 | ☐ | Scheduled streams (Text-drawer version) | [`task-34`](TASKS/task-34-scheduled-streams.md) |
|
||||
| 35 | ☐ | Scene-linked audio | [`task-35`](TASKS/task-35-scene-linked-audio.md) |
|
||||
| 37 | ☐ | Defaults vs current layout split (build-id); enables one-click revert to defaults | [`task-37`](TASKS/task-37-defaults-current-split.md) |
|
||||
|
||||
@@ -0,0 +1,94 @@
|
||||
using System;
|
||||
using System.Windows;
|
||||
using ytLive.Helpers;
|
||||
using ytLive.Models;
|
||||
using ytLive.Services;
|
||||
using ytLive.Services.Encoder;
|
||||
|
||||
namespace ytLive.ViewModels;
|
||||
|
||||
/// <summary>
|
||||
/// TASK 33, slice 3: the session's step-down watch — policy feed, one encoder relaunch, one toast.
|
||||
///
|
||||
/// <para>Kept in its own file because it is the only place that both reads the pump's stats and acts on
|
||||
/// the session; the pure decision lives in <see cref="StepDownPolicy"/>, the sentence in
|
||||
/// <see cref="StepDownPlanner"/>.</para>
|
||||
/// </summary>
|
||||
public partial class MainViewModel
|
||||
{
|
||||
private StepDownPlanner? _stepDownPlanner;
|
||||
private (int Width, int Height, int Fps, double BitrateMbps)? _encoderTierOverride;
|
||||
private bool _stepDownBusy;
|
||||
|
||||
/// <summary>Starts the watch for this broadcast. Called once per session, before the pump starts.</summary>
|
||||
private void StartStepDownWatch()
|
||||
{
|
||||
var quality = OutputProfile.Resolve(SelectedQuality, QualityOptions, HasProOutput);
|
||||
_encoderTierOverride = null;
|
||||
_stepDownPlanner = new StepDownPlanner(
|
||||
new StepDownPolicy(quality.Width, quality.Height, quality.Fps, quality.Bitrate),
|
||||
_sessionStartedAtUtc ?? DateTime.UtcNow);
|
||||
_framePump.StatsReported += OnPumpStatsReported;
|
||||
}
|
||||
|
||||
/// <summary>Ends the watch. Called from StopStream on every path, so a stale planner can never
|
||||
/// carry a spent budget into the next broadcast.</summary>
|
||||
private void StopStepDownWatch()
|
||||
{
|
||||
_framePump.StatsReported -= OnPumpStatsReported;
|
||||
_stepDownPlanner = null;
|
||||
_encoderTierOverride = null;
|
||||
_stepDownBusy = false;
|
||||
}
|
||||
|
||||
/// <summary>Runs on the PUMP thread, so it must never block: the restart it may trigger stops the
|
||||
/// very loop that raised this event. Decide here (pure and fast), act on the dispatcher.</summary>
|
||||
private void OnPumpStatsReported(object? sender, FramePumpStats stats)
|
||||
{
|
||||
if (_stepDownPlanner is not { } planner) return;
|
||||
|
||||
var decision = planner.Observe(stats, DateTime.UtcNow);
|
||||
if (!decision.Acted) return;
|
||||
|
||||
var dispatcher = Application.Current?.Dispatcher;
|
||||
if (dispatcher == null) return;
|
||||
_ = dispatcher.InvokeAsync(() => ApplyStepDownAsync(decision));
|
||||
}
|
||||
|
||||
private async void ApplyStepDownAsync(StepDownDecision decision)
|
||||
{
|
||||
var planner = _stepDownPlanner;
|
||||
if (planner == null || _stepDownBusy) return;
|
||||
|
||||
_stepDownBusy = true;
|
||||
var landed = planner.Current;
|
||||
try
|
||||
{
|
||||
_encoderTierOverride = landed;
|
||||
AppLog.Write($"Quality step-down: {decision.Why}");
|
||||
await _framePump.ChangeEncoderAsync();
|
||||
|
||||
// Show what is actually going out. The dropdown keeps the creator's choice — the policy
|
||||
// does not get to rewrite their intent, only the encoder's output.
|
||||
CurrentHealth.FPS = landed.Fps;
|
||||
CurrentHealth.CurrentBitrate = landed.BitrateMbps;
|
||||
ResolutionBadgeText = $"{landed.Width}×{landed.Height} @ {landed.Fps} fps · lowered";
|
||||
OnPropertyChanged(nameof(CurrentHealth));
|
||||
OnPropertyChanged(nameof(ResolutionBadgeText));
|
||||
|
||||
var copy = StepDownPlanner.Describe(decision, landed);
|
||||
if (copy.Length > 0)
|
||||
_notifications.Info("Quality lowered", copy);
|
||||
|
||||
AppLog.Write($"Quality step-down complete: {StepDownPlanner.ReportLine(DateTime.Now, decision)}");
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
AppLog.Write($"Quality step-down failed: {ex.Message}", AppLogMessageType.Error, AppLogErrorClass.Class2);
|
||||
}
|
||||
finally
|
||||
{
|
||||
_stepDownBusy = false;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -163,6 +163,7 @@ public partial class MainViewModel : ViewModelBase
|
||||
if (!TryStartAudioPipe()) return;
|
||||
_liveElapsed = TimeSpan.Zero;
|
||||
_recordLength = TimeSpan.Zero;
|
||||
StartStepDownWatch();
|
||||
|
||||
var recordStarted = false;
|
||||
if (alsoRecord)
|
||||
@@ -405,6 +406,7 @@ public partial class MainViewModel : ViewModelBase
|
||||
private async void StopStream()
|
||||
{
|
||||
CancelStreamRecovery(); // a stop wins over a relaunch in flight
|
||||
StopStepDownWatch();
|
||||
RemoveAttentionAlert("stream-recovery");
|
||||
var wasLive = IsLive;
|
||||
var wasRecording = IsRecording;
|
||||
@@ -494,6 +496,11 @@ public partial class MainViewModel : ViewModelBase
|
||||
{
|
||||
var url = _rtmpUrlProvider();
|
||||
var quality = OutputProfile.Resolve(SelectedQuality, QualityOptions, HasProOutput);
|
||||
// TASK 33: while a session is live, a step-down owns the emitted tier even though the dropdown
|
||||
// still shows what the creator picked. Their intent is not edited by the policy.
|
||||
if (_encoderTierOverride is { } landed)
|
||||
quality = new QualityOption($"{landed.Width}p{landed.Fps}", landed.Fps,
|
||||
landed.BitrateMbps, landed.Width, landed.Height);
|
||||
var compositor = BuildCompositorOptions();
|
||||
return new EncoderOptions
|
||||
{
|
||||
|
||||
@@ -0,0 +1,104 @@
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using Xunit;
|
||||
using ytLive.Models;
|
||||
using ytLive.Services;
|
||||
using ytLive.Services.Compositor;
|
||||
using ytLive.Services.Encoder;
|
||||
|
||||
namespace ytLive.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// TASK 33 slice 3 — the step-down's one dangerous act: relaunching the encoder mid-broadcast.
|
||||
///
|
||||
/// <para>The restart is a full pump stop/start rather than an encoder swap, and this test is why: a swap
|
||||
/// leaves an in-flight submit pointing at a disposed process, which faults the loop and ends the
|
||||
/// broadcast. Here the pump must come back and produce a frame with a <b>new</b> encoder instance and
|
||||
/// the tier the policy landed on.</para>
|
||||
/// </summary>
|
||||
public sealed class EncoderRestartTests
|
||||
{
|
||||
[Fact]
|
||||
public async Task ChangeEncoder_RelauchesWithANewEncoderAtTheNewTier()
|
||||
{
|
||||
var tier = (Width: 1920, Height: 1080, Fps: 60, Bitrate: 8.0);
|
||||
var started = new List<EncoderOptions>();
|
||||
var frames = new List<CapturingEncoder>();
|
||||
|
||||
var scene = new Scene { Name = "Live" };
|
||||
scene.Elements.Add(new Source { Name = "Background", Type = SourceType.Background, IsBackground = true });
|
||||
var black = new VideoFrame(1920, 1080, new byte[1920 * 1080 * 4]);
|
||||
|
||||
using var pump = new FramePump(
|
||||
sceneProvider: () => scene,
|
||||
frameResolver: _ => black,
|
||||
compositorOptions: () => new CompositorOptions
|
||||
{
|
||||
SourceRectX = 0, SourceRectY = 0,
|
||||
SourceRectWidth = 1920, SourceRectHeight = 1080,
|
||||
OutputWidth = 1920, OutputHeight = 1080,
|
||||
},
|
||||
encoderOptions: () => new EncoderOptions
|
||||
{
|
||||
RtmpUrl = "rtmp://a.rtmp.youtube.com/live2/abc",
|
||||
Width = tier.Width, Height = tier.Height, Fps = tier.Fps,
|
||||
BitrateKbps = (int)(tier.Bitrate * 1000),
|
||||
},
|
||||
encoderFactory: () =>
|
||||
{
|
||||
var encoder = new CapturingEncoder();
|
||||
encoder.Started += options => started.Add(options);
|
||||
frames.Add(encoder);
|
||||
return encoder;
|
||||
},
|
||||
log: _ => { },
|
||||
pacingDelay: (_, _) => Task.Delay(5));
|
||||
|
||||
await pump.StartAsync();
|
||||
Assert.True(await frames[0].FirstFrame.Task.WaitAsync(TimeSpan.FromSeconds(5)), "first session never produced a frame");
|
||||
|
||||
// The policy lands on 1080p30 @ 6.4 Mbps; the VM has already written the override, so the
|
||||
// options func now reports the new tier — which the restart must re-read.
|
||||
tier = (1920, 1080, 30, 6.4);
|
||||
|
||||
Assert.True(await pump.ChangeEncoderAsync());
|
||||
Assert.True(await frames[1].FirstFrame.Task.WaitAsync(TimeSpan.FromSeconds(5)), "restarted session never produced a frame");
|
||||
|
||||
Assert.Equal(2, started.Count);
|
||||
Assert.Equal(60, started[0].Fps);
|
||||
Assert.Equal(30, started[1].Fps);
|
||||
Assert.Equal(6400, started[1].BitrateKbps);
|
||||
Assert.NotSame(frames[0], frames[1]);
|
||||
|
||||
await pump.StopAsync();
|
||||
}
|
||||
|
||||
private sealed class CapturingEncoder : IFfmpegEncoder
|
||||
{
|
||||
public readonly TaskCompletionSource<bool> FirstFrame = new(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
public event Action<EncoderOptions>? Started;
|
||||
|
||||
public int DroppedFrames => 0;
|
||||
|
||||
public event EventHandler<StreamHealth>? HealthUpdated { add { } remove { } }
|
||||
public event EventHandler<string>? ProcessFailed { add { } remove { } }
|
||||
|
||||
public Task StartAsync(EncoderOptions options, CancellationToken cancellationToken = default)
|
||||
{
|
||||
Started?.Invoke(options);
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public Task SubmitFrameAsync(VideoFrame frame, CancellationToken cancellationToken = default)
|
||||
{
|
||||
FirstFrame.TrySetResult(true);
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public Task StopAsync(CancellationToken cancellationToken = default) => Task.CompletedTask;
|
||||
|
||||
public void Dispose() { }
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user