ErsatzTV.Application/Streaming/HlsSessionWorker.cs

@@ -26,7 +26,9 @@ namespace ErsatzTV.Application.Streaming;
 
 public class HlsSessionWorker : IHlsSessionWorker
 {
-    private static int _workAheadCount;
+    // process-wide, shared by every session — the work-ahead limit is a global resource budget
+    private static readonly WorkAheadSlots _workAheadSlots = new();
+
     private readonly OutputFormatKind _outputFormatKind;
     private readonly IHlsInitSegmentCache _hlsInitSegmentCache;
     private readonly Dictionary<long, int> _discontinuityMap = [];
@@ -243,10 +245,12 @@ public class HlsSessionWorker : IHlsSessionWorker
                     cancellationToken);
             }
 
-            bool initialWorkAhead = Volatile.Read(ref _workAheadCount) < await GetWorkAheadLimit(cancellationToken);
+            // claim the slot here rather than checking here and claiming inside Transcode: the check
+            // and the claim have to be one atomic step or every simultaneous tune-in wins (#536)
+            bool initialWorkAhead = _workAheadSlots.TryAcquire(await GetWorkAheadLimit(cancellationToken));
             _state = initialWorkAhead ? HlsSessionState.SeekAndWorkAhead : HlsSessionState.SeekAndRealtime;
 
-            if (!await Transcode(!initialWorkAhead, cancellationToken))
+            if (!await Transcode(initialWorkAhead, cancellationToken))
             {
                 return;
             }
@@ -269,8 +273,8 @@ public class HlsSessionWorker : IHlsSessionWorker
                     // only use realtime encoding when we're at least 30 seconds ahead
                     bool realtime = transcodedBuffer >= TimeSpan.FromSeconds(30);
                     bool subsequentWorkAhead =
-                        !realtime && Volatile.Read(ref _workAheadCount) < await GetWorkAheadLimit(cancellationToken);
-                    if (!await Transcode(!subsequentWorkAhead, cancellationToken))
+                        !realtime && _workAheadSlots.TryAcquire(await GetWorkAheadLimit(cancellationToken));
+                    if (!await Transcode(subsequentWorkAhead, cancellationToken))
                     {
                         return;
                     }
@@ -456,15 +460,23 @@ public class HlsSessionWorker : IHlsSessionWorker
         return result;
     }
 
-    private async Task<bool> Transcode(bool realtime, CancellationToken cancellationToken)
+    /// <summary>
+    /// Runs one transcode. The caller is the one that races for a work-ahead slot, so ownership is
+    /// passed IN: <paramref name="ownsWorkAheadSlot" /> means the caller already claimed a slot from
+    /// <see cref="_workAheadSlots" />, and this method releases it in its <c>finally</c> — acquire
+    /// and release stay one-for-one (#536).
+    /// </summary>
+    private async Task<bool> Transcode(bool ownsWorkAheadSlot, CancellationToken cancellationToken)
     {
+        // a session works ahead exactly when it holds a slot; everything else runs realtime (throttled)
+        bool realtime = !ownsWorkAheadSlot;
+
         try
         {
             bool wasSeekAndWorkAhead = _state is HlsSessionState.SeekAndWorkAhead;
 
             if (!realtime)
             {
-                Interlocked.Increment(ref _workAheadCount);
                 _logger.LogDebug("HLS segmenter will work ahead for channel {Channel}", _channelNumber);
 
                 HlsSessionState nextState = _state switch
@@ -747,9 +759,9 @@ public class HlsSessionWorker : IHlsSessionWorker
                 // do nothing
             }
 
-            if (!realtime)
+            if (ownsWorkAheadSlot)
             {
-                Interlocked.Decrement(ref _workAheadCount);
+                _workAheadSlots.Release();
             }
         }
 

ErsatzTV.Application/Streaming/WorkAheadSlots.cs

@@ -0,0 +1,91 @@
+namespace ErsatzTV.Application.Streaming;
+
+/// <summary>
+/// The process-wide pool of work-ahead slots shared by every HLS session (ersatztv#536).
+/// </summary>
+/// <remarks>
+/// <para>
+/// <c>workAheadSegmenterLimit</c> is a resource guarantee, not a tuning knob: it bounds how many
+/// transcodes may run unthrottled (no <c>-readrate</c>) at once, and the QSV hardware-frame pool
+/// sizing from ersatztv#529 assumes that bound holds.
+/// </para>
+/// <para>
+/// Acquisition must therefore be atomic. The previous shape — <c>Volatile.Read(count) &lt; limit</c>
+/// in the caller, <c>Interlocked.Increment</c> later inside the transcode — is a check-then-act
+/// TOCTOU separated by at least one <c>await</c> (the limit is a DB-backed config read), so N
+/// simultaneous tune-ins all observed <c>0 &lt; limit</c> and all ran unthrottled. Same class as
+/// ersatztv#231 / #250.
+/// </para>
+/// </remarks>
+public sealed class WorkAheadSlots
+{
+    private int _count;
+    private int _unbalancedReleases;
+
+    /// <summary>
+    /// Gets the number of slots currently held. For diagnostics and tests only — never branch on
+    /// this to decide whether to work ahead; that is exactly the race <see cref="TryAcquire" /> exists to close.
+    /// </summary>
+    public int Count => Volatile.Read(ref _count);
+
+    /// <summary>
+    /// Atomically claims one slot if fewer than <paramref name="limit" /> are held.
+    /// </summary>
+    /// <returns><c>true</c> when a slot was claimed; the caller then owns it and MUST
+    /// <see cref="Release" /> it exactly once.</returns>
+    public bool TryAcquire(int limit)
+    {
+        while (true)
+        {
+            int current = Volatile.Read(ref _count);
+            if (current >= limit)
+            {
+                return false;
+            }
+
+            // only the thread whose compare-exchange observes the value it read wins the slot, so
+            // the count can never transiently exceed the limit and two racers can never both claim
+            if (Interlocked.CompareExchange(ref _count, current + 1, current) == current)
+            {
+                return true;
+            }
+        }
+    }
+
+    /// <summary>
+    /// Gets the number of releases that were not matched by a successful acquire. Always zero in a
+    /// correct program; a non-zero value means the ownership contract was broken somewhere.
+    /// </summary>
+    public int UnbalancedReleases => Volatile.Read(ref _unbalancedReleases);
+
+    /// <summary>
+    /// Returns a slot claimed by <see cref="TryAcquire" />. Only ever called by the owner of that slot.
+    /// </summary>
+    /// <remarks>
+    /// Ownership is a discipline, not a token — the same call-once contract as `EntityLocker` (#231).
+    /// The one failure this does defend against is an unbalanced release driving the count negative:
+    /// this pool is process-wide and lives for the life of the app, so a count of -1 would silently
+    /// and permanently admit one extra unthrottled transcode, re-opening the #529 QSV pool
+    /// exhaustion with nothing in the logs to find it by. Clamping keeps the resource guarantee
+    /// intact and records the breakage in <see cref="UnbalancedReleases" /> instead of throwing —
+    /// the single caller releases from a `finally`, where a throw would swallow the real exception.
+    /// </remarks>
+    public void Release()
+    {
+        if (Interlocked.Decrement(ref _count) >= 0)
+        {
+            return;
+        }
+
+        Interlocked.Increment(ref _unbalancedReleases);
+
+        while (true)
+        {
+            int current = Volatile.Read(ref _count);
+            if (current >= 0 || Interlocked.CompareExchange(ref _count, 0, current) == current)
+            {
+                return;
+            }
+        }
+    }
+}

ErsatzTV.Core.Tests/Streaming/WorkAheadSlotsTests.cs

@@ -0,0 +1,200 @@
+using ErsatzTV.Application.Streaming;
+using NUnit.Framework;
+using Shouldly;
+
+namespace ErsatzTV.Core.Tests.Streaming;
+
+// Verifies the atomic-claim contract of the work-ahead slot pool (ersatztv#536): at most
+// `limit` sessions may hold a slot at once, no matter how many race for one simultaneously.
+//
+// NEGATIVE CONTROL (per the ersatztv#231/#250 lesson — a one-shot Barrier + Task.WhenAll does NOT
+// catch this class on this hardware): to prove these tests are non-vacuous, temporarily replace the
+// compare-exchange in WorkAheadSlots.TryAcquire with the check-then-act shape this issue fixed —
+//
+//     int current = Volatile.Read(ref _count);
+//     if (current >= limit) return false;
+//     Interlocked.Increment(ref _count);
+//     return true;
+//
+// — and TryAcquire_ParallelCallers_NeverExceedsLimit must FAIL (badRounds > 0). Do NOT "break" it by
+// stubbing `if (true)`: that leaves values assigned-but-never-read, and CS0219 under
+// warnings-as-errors fails the build silently, so `dotnet test --no-build` then runs the STALE
+// (fixed) dll and the control falsely passes. Always grep the build output for `error CS` first.
+[TestFixture]
+public class WorkAheadSlotsTests
+{
+    [Test]
+    public void TryAcquire_BelowLimit_Succeeds()
+    {
+        var slots = new WorkAheadSlots();
+
+        slots.TryAcquire(2).ShouldBeTrue();
+        slots.TryAcquire(2).ShouldBeTrue();
+        slots.TryAcquire(2).ShouldBeFalse();
+        slots.Count.ShouldBe(2);
+    }
+
+    [Test]
+    public void TryAcquire_DoesNotConsumeASlotWhenItFails()
+    {
+        var slots = new WorkAheadSlots();
+
+        slots.TryAcquire(1).ShouldBeTrue();
+        slots.TryAcquire(1).ShouldBeFalse();
+        slots.Count.ShouldBe(1);
+
+        // the failed attempt must not have leaked a slot: releasing the one real holder frees the pool
+        slots.Release();
+        slots.Count.ShouldBe(0);
+        slots.TryAcquire(1).ShouldBeTrue();
+    }
+
+    [Test]
+    public void TryAcquire_WithNonPositiveLimit_NeverSucceeds()
+    {
+        var slots = new WorkAheadSlots();
+
+        slots.TryAcquire(0).ShouldBeFalse();
+        slots.TryAcquire(-1).ShouldBeFalse();
+        slots.Count.ShouldBe(0);
+    }
+
+    // The pool is process-wide and never recreated, so a negative count would not self-heal: it would
+    // permanently admit more than `limit` unthrottled transcodes, with nothing in the logs to find it by.
+    [Test]
+    public void Release_WithoutAcquire_ClampsAtZeroAndIsRecorded()
+    {
+        var slots = new WorkAheadSlots();
+
+        slots.Release();
+
+        slots.Count.ShouldBe(0);
+        slots.UnbalancedReleases.ShouldBe(1);
+
+        // the budget is intact: a limit of 1 still admits exactly one holder, not two
+        slots.TryAcquire(1).ShouldBeTrue();
+        slots.TryAcquire(1).ShouldBeFalse();
+    }
+
+    [Test]
+    public void UnbalancedReleases_IsZeroUnderCorrectUse()
+    {
+        var slots = new WorkAheadSlots();
+
+        slots.TryAcquire(1).ShouldBeTrue();
+        slots.Release();
+
+        slots.UnbalancedReleases.ShouldBe(0);
+    }
+
+    // Hammers the acquire race over many rounds rather than a single simultaneous burst: the
+    // read->increment window is far too narrow to collide reliably when threads release once.
+    [Test]
+    [TestCase(1)]
+    [TestCase(2)]
+    [TestCase(3)]
+    public void TryAcquire_ParallelCallers_NeverExceedsLimit(int limit)
+    {
+        const int threads = 8;
+        const int rounds = 20_000;
+
+        var slots = new WorkAheadSlots();
+        var winners = 0;
+        var badRounds = 0;
+
+        using var startRound = new Barrier(threads);
+        using var endRound = new Barrier(
+            threads,
+            _ =>
+            {
+                int held = Volatile.Read(ref winners);
+                if (held > limit || held != slots.Count)
+                {
+                    Interlocked.Increment(ref badRounds);
+                }
+
+                // reset for the next round: release every slot claimed this round
+                for (var i = 0; i < held; i++)
+                {
+                    slots.Release();
+                }
+
+                Volatile.Write(ref winners, 0);
+            });
+
+        var workers = new Thread[threads];
+        for (var t = 0; t < threads; t++)
+        {
+            workers[t] = new Thread(() =>
+            {
+                for (var r = 0; r < rounds; r++)
+                {
+                    startRound.SignalAndWait();
+                    if (slots.TryAcquire(limit))
+                    {
+                        Interlocked.Increment(ref winners);
+                    }
+
+                    endRound.SignalAndWait();
+                }
+            });
+            workers[t].Start();
+        }
+
+        foreach (Thread worker in workers)
+        {
+            worker.Join();
+        }
+
+        badRounds.ShouldBe(0);
+        slots.Count.ShouldBe(0);
+        slots.UnbalancedReleases.ShouldBe(0);
+    }
+
+    // The release path is the fiddly half: a slot freed by its owner must become available again,
+    // and a burst of acquire/release cycles must not drift the count in either direction.
+    [Test]
+    public void AcquireAndRelease_UnderContention_LeavesNoLeakedOrDoubleFreedSlots()
+    {
+        const int threads = 8;
+        const int rounds = 20_000;
+        const int limit = 3;
+
+        var slots = new WorkAheadSlots();
+        var overLimit = 0;
+        var live = 0;
+
+        var workers = new Thread[threads];
+        for (var t = 0; t < threads; t++)
+        {
+            workers[t] = new Thread(() =>
+            {
+                for (var r = 0; r < rounds; r++)
+                {
+                    if (!slots.TryAcquire(limit))
+                    {
+                        continue;
+                    }
+
+                    if (Interlocked.Increment(ref live) > limit)
+                    {
+                        Interlocked.Increment(ref overLimit);
+                    }
+
+                    Interlocked.Decrement(ref live);
+                    slots.Release();
+                }
+            });
+            workers[t].Start();
+        }
+
+        foreach (Thread worker in workers)
+        {
+            worker.Join();
+        }
+
+        overLimit.ShouldBe(0);
+        slots.Count.ShouldBe(0);
+        slots.UnbalancedReleases.ShouldBe(0);
+    }
+}

docs/decisions.md

@@ -132,6 +132,7 @@ in-file entries.
 - [2026-07-21 — Parallel orientation + selection is the startup protocol; #237 retired (#520)](#2026-07-21--parallel-orientation--selection-is-the-startup-protocol-237-retired-520)
 - [2026-07-21 — External channel-logo URLs are downloaded and cached at save time; the render path never fetches a logo (#525)](#2026-07-21--external-channel-logo-urls-are-downloaded-and-cached-at-save-time-the-render-path-never-fetches-a-logo-525)
 - [2026-07-21 — Check the worked issue before the decision corpus; a closed tracker's comments need no retrofit (#524)](#2026-07-21--check-the-worked-issue-before-the-decision-corpus-a-closed-trackers-comments-need-no-retrofit-524)
+- [2026-07-21 — Work-ahead slots are claimed atomically by the caller, released by the transcode it hands them to (#536)](#2026-07-21--work-ahead-slots-are-claimed-atomically-by-the-caller-released-by-the-transcode-it-hands-them-to-536)
 
 ---
 
@@ -2855,6 +2856,14 @@ again — the duplication is what let the defect exist in triplicate. Covered by
 > made the failure near-deterministic. See `ffmpeg.qsv-extra-hw-frames-floor`; the decision recorded
 > here still stands, with that floor in place.
 
+> **Correction (2026-07-21, #536):** the bullet below stating that "every concurrent tune-in falls
+> back to the throttled path" described the *intent*, not the behaviour. The slot check and its
+> increment straddled an `await`, so N simultaneous tune-ins all read `0 < limit` and all started
+> unthrottled; the "one winner, two throttled" observation held only because those tunes were
+> effectively staggered. The measurement and the conclusions drawn from it stand — slot availability
+> *was* the variable behind the bimodality — but the guarantee itself was not enforced until
+> `ffmpeg.work-ahead-slot-atomic`.
+
 - **`-readrate` throttles from the first read, so it sets a floor on time-to-first-segment.** The
   realtime playback path pins input reading to 1.05× wall clock so a channel behaves like live TV.
   Since `OutputFormatHls.SegmentSeconds` is 4 and the segmenter serves the playlist only once the
@@ -3316,3 +3325,40 @@ that skips the lore bucket will leave real knowledge on the floor.
 **A zero result is a legitimate outcome**, not evidence the triage was done wrong. What would change
 the answer is a tracker whose sessions did *not* also write per-issue closing records — one where the
 tracker genuinely was the primary rather than the narration layer.
+
+## 2026-07-21 — Work-ahead slots are claimed atomically by the caller, released by the transcode it hands them to (#536)
+
+`key: ffmpeg.work-ahead-slot-atomic` · `status: active` · `since: 2026-07-21` · `supersedes: none` · `superseded-by: none`
+**Rule:** `workAheadSegmenterLimit` is enforced by a single compare-exchange claim on a shared `WorkAheadSlots` pool taken by the *caller* of `Transcode`, which then passes ownership in and gets the release in `Transcode`'s `finally` — never a `Volatile.Read` compare in one place and an `Interlocked.Increment` in another.
+**Signals:** work-ahead slot, workAheadSegmenterLimit, check-then-act across an await, unthrottled tune-in, TOCTOU · paths: `ErsatzTV.Application/Streaming/WorkAheadSlots.cs`, `HlsSessionWorker.Run`/`Transcode`, `ErsatzTV.Core.Tests/Streaming/WorkAheadSlotsTests.cs` · issues: #536, #350, #529, #231, #250
+**Mechanics:** `WorkAheadSlots.TryAcquire(limit)` CAS loop; `Transcode(bool ownsWorkAheadSlot, …)`
+
+**The defect was structural, not a missing `Interlocked`.** The write side already used
+`Interlocked.Increment`, which is why the code read as thread-safe. The read side was a separate,
+earlier `Volatile.Read(ref _workAheadCount) < await GetWorkAheadLimit(...)` in `Run`, and the
+increment happened later inside `Transcode` — with at least one `await` (a DB-backed config read) in
+between. `Interlocked` on one half of a check-then-act buys nothing. Three simultaneous tune-ins on
+prod with `workAheadSegmenterLimit: 1` all observed `0 < 1` and all ran with no `-readrate`.
+
+**Why the caller claims and the callee releases.** Moving the whole acquire/release pair inside
+`Transcode` would be more symmetric, but `Run` needs the outcome *before* the call: it sets
+`_state = SeekAndWorkAhead | SeekAndRealtime`, and `Transcode` reads that state on entry
+(`wasSeekAndWorkAhead`) to decide whether the item starts at `DateTimeOffset.Now` or at
+`_transcodedUntil`. Folding acquisition inward would have silently changed that branch. Instead the
+parameter was inverted — `Transcode(bool ownsWorkAheadSlot, …)` with `realtime = !ownsWorkAheadSlot`
+derived on the first line — so the contract is stated in the signature rather than implied by a
+double negative, and there is exactly one release site guarded by the same flag.
+
+**Compare-exchange rather than increment-then-back-off.** The issue proposed
+`Interlocked.Increment` followed by a decrement when the post-increment value exceeds the limit.
+That is correct on holder count, but the counter transiently overshoots, so a concurrent reader of
+`Count` can observe a value above the limit. The CAS loop never publishes a state that violates the
+invariant, which matters because the QSV hardware-frame pool sizing (`ffmpeg.qsv-extra-hw-frames-floor`)
+is derived from that bound.
+
+**Testing shape is inherited from #231/#250.** A single `Barrier(N)` + `Task.WhenAll` round does not
+reliably collide on this hardware; the tests hammer 8 threads × 20 000 rounds with a per-round
+barrier that validates the winner count and resets the pool. The negative control is documented in
+the test file: reinstate the check-then-act body (**not** `if (true)`, which trips CS0219 under
+warnings-as-errors and leaves `--no-build` running a stale, still-fixed dll). Verified: with the
+pre-fix shape, 15 912 of 20 000 rounds over-claimed.

docs/decisions/README.md

@@ -53,6 +53,7 @@ the link for rationale. Superseded/retired history lives in `archive/`. Regenera
 | `ffmpeg.qsv-decode-encode-split` | QSV decode is decoupled from QSV encode via a single `FFmpegProfile.QsvPreferNativeDecoder` bool (default ON, Linux-only), so a QSV encode profile can decode with the more tolerant native VA-API decoder instead of the QSV decoder, mirroring Jellyfin's hybrid decode/encode toggle instead of a general decode-family enum. | 2026-07-20 | [link](../decisions.md#2026-07-20-498--qsv-decode-is-split-from-qsv-encode-via-a-single-qsvprefernativedecoder-bool) |
 | `ffmpeg.qsv-extra-hw-frames-floor` | a QSV upload never emits `extra_hw_frames` below `FFmpegState.MinimumQsvExtraHardwareFrames` (64); a stored `0` or negative value is treated as "no pool configured" rather than honored literally, because with no headroom any unthrottled read exhausts the pool and the transcode writes nothing at all. | 2026-07-21 | [link](../decisions.md#2026-07-21--qsv-hardware-frame-headroom-is-a-floor-not-an-operator-preference-529) |
 | `ffmpeg.remote-image-fetcher-bounded` | remote graphics-engine images are fetched through `IRemoteImageFetcher` with a pooled `HttpClientFactory` client, a body-covering deadline, a wire-transfer size cap, and a decoder-enforced `DecoderOptions.MaxFrames` bound re-verified post-decode — never cached, re-fetched per element init. | 2026-07-20 | [link](../decisions.md#2026-07-20--remote-graphics-engine-images-are-fetched-through-a-bounded-pooled-iremoteimagefetcher-re-fetched-per-element-init-not-cached-511) |
+| `ffmpeg.work-ahead-slot-atomic` | `workAheadSegmenterLimit` is enforced by a single compare-exchange claim on a shared `WorkAheadSlots` pool taken by the *caller* of `Transcode`, which then passes ownership in and gets the release in `Transcode`'s `finally` — never a `Volatile.Read` compare in one place and an `Interlocked.Increment` in another. | 2026-07-21 | [link](../decisions.md#2026-07-21--work-ahead-slots-are-claimed-atomically-by-the-caller-released-by-the-transcode-it-hands-them-to-536) |
 | `graphics.channel-logo-caching` | An external `http(s)` channel-logo URL is fetched, decode-budget-validated, and stored in the image cache under a content-hash name at SAVE time — becoming byte-identical to an uploaded logo — so the render path never fetches a logo over HTTP; a bad URL fails the save with a 422 (BaseError → ValidationProblemDetails). | 2026-07-21 | [link](../decisions.md#2026-07-21--external-channel-logo-urls-are-downloaded-and-cached-at-save-time-the-render-path-never-fetches-a-logo-525) |
 | `iptv.base-url` | An optional advertised base URL (`iptv.base_url`) is resolved centrally via a pure Core helper (`AdvertisedBaseUrl`) inside the two IPTV generation handlers (M3U + XMLTV); unset/malformed values fall back byte-identical to the request-derived host, and it's a new `iptv` settings group distinct from `ETV_BASE_URL` and out of scope for HDHomeRun. | 2026-07-16 | [link](../decisions.md#2026-07-16--optional-advertised-iptv-base-url-iptvbase_url-resolved-centrally-in-the-two-generators-340) |
 | `iptv.logo-drives-bug-preset` | One uploaded channel logo drives both the listing logo and the on-screen bug via a shared, seeded `ChannelLogo`-sourced watermark preset (`Channel Bug`), not new per-channel schema. | 2026-07-20 | [link](../decisions.md#2026-07-20--one-logo-drives-the-bug-via-a-shared-channellogo-preset-not-new-schema-67) |