Merge pull request #6 from ZenosInteractive/reader/fix-stale-prefetch-reentry

fix(reader): respawn prefetch when cancelled chunk re-enters window b…

Author: virtexalejandro

CHANGELOG.md

@@ -5,6 +5,14 @@ All notable changes to the VTX SDK will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased] - 2026-04-23
+
+### Fixed
+- **reader**: `ReplayReader::UpdateCacheWindow` no longer leaves a chunk permanently stuck when it is cancelled by a window shift and then immediately re-requested before the worker thread has started running.  The §1.A cancellation path (PR #4) flags out-of-window prefetches via `request_stop()` on their per-`PendingLoad` stop_source but leaves the map entry in place.  If the chunk re-entered the window before its worker was scheduled, the `trigger()` lambda saw the stale entry, assumed a load was already in flight, and skipped spawning a new task.  The original worker then ran, observed `stop_requested() == true` at its entry check in `PerformHeavyLoading`, returned an empty `CachedChunk`, and `AsyncLoadTask` skipped the cache write (correctly, because the stop was still requested).  The future resolved cleanly, but the cache stayed empty.  A synchronous caller waiting on that future in `GetFramePtrSync` then read the empty cache and returned `nullptr` -- manifesting as a spurious load failure under random-seek patterns.  Fix: `trigger()` now detects a pending entry whose stop is already requested and replaces it with a fresh `PendingLoad`; the orphaned worker exits on its own and its `stop_requested()`-gated cache write still cannot pollute anything.  Exposed by the TSan CI job on `ReaderApiFlatBuffers.RandomAccessSkipsLateralPrefetches`; the scheduling overhead of the ThreadSanitizer runtime makes the "cancelled before scheduled" race far more likely to manifest.  Stock release builds had been papering over it by the workers happening to get past the entry check before cancellation landed
+
+### Added
+- **tests**: `ReaderApiFlatBuffers.CancelledPrefetchReEntersWindow` -- focused regression for the bug above.  Runs the cancel + re-enter pattern (prime chunks 0..2, jump to chunk 10 to cancel, jump to chunk 2 to revive) 50 times against a fresh reader each iteration.  Pre-fix this fails ~every run under TSan and flakes at single-digit-% on stock release; post-fix it is deterministic green on both
+
 ## [Unreleased] - 2026-04-22
 
 ### Changed

sdk/include/vtx/reader/core/vtx_reader.h

@@ -443,6 +443,29 @@ namespace VTX {
         }
 
         void UpdateCacheWindow(int32_t current_idx) {
+            // Orphaned PendingLoads moved out of pending_loads_ by trigger()
+            // when it respawns a stale-cancelled entry.  Declared BEFORE the
+            // lock_guard so the vector -- and therefore each orphan's
+            // std::shared_future destructor -- runs AFTER cache_mutex_ has
+            // been released on function exit.
+            //
+            // libstdc++'s std::shared_future dtor blocks until the async
+            // task completes when it holds the last reference to a state
+            // created by std::async(std::launch::async, ...).  The task
+            // itself acquires cache_mutex_ in AsyncLoadTask to check
+            // stop_requested() and (optionally) write chunk_cache_, so
+            // destroying a to-be-cancelled shared_future WHILE holding
+            // cache_mutex_ would deadlock: we would wait on a task that
+            // cannot make progress without the mutex we hold.
+            //
+            // By moving the stale entry into orphans[] before overwriting
+            // it in pending_loads_, we defer the shared_future dtor until
+            // the lock_guard below has released the mutex.  The task can
+            // then observe its already-requested stop, bail at the entry
+            // check in PerformHeavyLoading, skip the cache write, and
+            // resolve its future -- unblocking the orphans[] destructor.
+            std::vector<PendingLoad> orphans;
+
             std::lock_guard<std::mutex> lock(cache_mutex_);
 
             // §1.B -- access-pattern detection.
@@ -535,26 +558,62 @@ namespace VTX {
             const size_t max_concurrent_loads = 3;
 
             auto trigger = [&](int32_t i) {
-                if (!chunk_cache_.contains(i) && !pending_loads_.contains(i)) {
-                    if (pending_loads_.size() >= max_concurrent_loads) return;
-
-                    if (evts.OnChunkLoadStarted) evts.OnChunkLoadStarted(i);
-
-                    // Give each prefetch its own stop_source so we can
-                    // cancel it independently from the "window changed"
-                    // path above.  The token is captured by value into the
-                    // lambda -- it's ref-counted, cheap to copy, and stays
-                    // valid even if the map entry is later erased while the
-                    // worker thread is still running (it won't be, because
-                    // we wait on the future first, but the invariant is
-                    // useful for reasoning about shutdown paths).
-                    PendingLoad pl;
-                    auto token = pl.stop.get_token();
-                    pl.future = std::async(std::launch::async, [this, i, token]() {
-                        this->AsyncLoadTask(i, token);
-                    }).share();
-                    pending_loads_[i] = std::move(pl);
+                if (chunk_cache_.contains(i))
+                    return;
+
+                // A prefetch that was cancelled by a previous window shift
+                // (line ~509) leaves its entry in pending_loads_ with the
+                // stop_token already requested.  If the chunk re-enters the
+                // window before the worker thread has even started running
+                // (easy under TSan's scheduling overhead, or any burst of
+                // random seeks), the worker will observe stop_requested()
+                // on entry and bail without populating chunk_cache_.  The
+                // future resolves cleanly, but GetFramePtrSync() then reads
+                // an empty cache and returns nullptr -- a latent bug from
+                // the §1.A cancellation work (PR #4).
+                //
+                // Detect that case here and replace the stale entry with a
+                // fresh PendingLoad.  The orphaned task will exit on its
+                // own; its cache write is gated by a stop_requested() check
+                // inside cache_mutex_ (see AsyncLoadTask) so it cannot
+                // pollute the cache with empty data either.
+                const bool stale = pending_loads_.contains(i) && pending_loads_[i].stop.get_token().stop_requested();
+
+                if (pending_loads_.contains(i) && !stale)
+                    return;
+
+                // The concurrency cap guards *new* slots.  Respawning a
+                // stale entry replaces an existing slot rather than adding
+                // one, so don't gate the resurrection path on the cap --
+                // otherwise we could refuse to revive a chunk the caller is
+                // about to synchronously wait on.
+                if (!stale && pending_loads_.size() >= max_concurrent_loads)
+                    return;
+
+                if (evts.OnChunkLoadStarted)
+                    evts.OnChunkLoadStarted(i);
+
+                // Relocate the stale entry to orphans[] before overwriting.
+                // See the long comment at the top of UpdateCacheWindow for
+                // why destroying its shared_future inside cache_mutex_
+                // would deadlock under TSan.
+                if (stale) {
+                    orphans.push_back(std::move(pending_loads_[i]));
                 }
+
+                // Give each prefetch its own stop_source so we can
+                // cancel it independently from the "window changed"
+                // path above.  The token is captured by value into the
+                // lambda -- it's ref-counted, cheap to copy, and stays
+                // valid even if the map entry is later erased while the
+                // worker thread is still running (it won't be, because
+                // we wait on the future first, but the invariant is
+                // useful for reasoning about shutdown paths).
+                PendingLoad pl;
+                auto token = pl.stop.get_token();
+                pl.future =
+                    std::async(std::launch::async, [this, i, token]() { this->AsyncLoadTask(i, token); }).share();
+                pending_loads_[i] = std::move(pl); // overwrites moved-from entry if stale
             };
 
             trigger(current_idx);

tests/reader/test_reader_api.cpp

@@ -265,6 +265,51 @@ TEST(ReaderApiFlatBuffers, RandomAccessSkipsLateralPrefetches)
         << jump_frames.size() << " jumps.";
 }
 
+// Regression for the "stale-cancelled prefetch blocks re-entry" bug.
+//
+// Sequence that hits the race:
+//   1. GetFrameSync(frame_in_chunk_0) -- triggers chunk 0 (sync) plus
+//      lateral prefetches of chunks 1 and 2.  Workers for 1 and 2 are
+//      queued on std::async but may not have started yet.
+//   2. GetFrameSync(frame_in_chunk_10) -- window shifts away from 0.
+//      The UpdateCacheWindow cancel loop calls request_stop() on the
+//      PendingLoads for chunks 1 and 2; their entries remain in
+//      pending_loads_ until the workers exit and the next reap sweep
+//      picks them up.
+//   3. GetFrameSync(frame_in_chunk_2) -- chunk 2 is back in the window.
+//      Pre-fix: trigger() saw pending_loads_[2] and skipped spawning a
+//      new task; worker 2 eventually ran, observed stop_requested(),
+//      bailed at its entry check, and the chunk_cache_ write was
+//      skipped.  GetFramePtrSync waited on the future, it resolved,
+//      cache was empty, returned nullptr.
+//   4. Post-fix: trigger() detects pending_loads_[2] has its stop
+//      already requested and respawns with a fresh stop_source; the
+//      orphaned worker exits on its own, the new worker populates the
+//      cache, GetFramePtrSync returns the frame.
+//
+// The race is timing-dependent, so we run the pattern 50 iterations.
+// Under TSan's scheduler overhead a single iteration suffices; under
+// stock release it is a flaky single-digit-% race and 50 reps push
+// the miss probability below the CI flake floor.
+TEST(ReaderApiFlatBuffers, CancelledPrefetchReEntersWindow) {
+    const auto path = VtxTest::OutputPath("ReaderApiFlatBuffers_CancelledPrefetchReEntersWindow.vtx");
+    WriteReplay(VTX::VtxFormat::FlatBuffers, path, 100, 5); // 20 chunks * 5 frames
+
+    constexpr int kIters = 50;
+    for (int iter = 0; iter < kIters; ++iter) {
+        auto ctx = VTX::OpenReplayFile(path);
+        ASSERT_TRUE(ctx) << "iter=" << iter << " " << ctx.error;
+        ctx.reader->SetCacheWindow(2, 2);
+
+        // Step 1: prime chunks 0..2 (chunk 0 sync + 1,2 as laterals).
+        ASSERT_NE(ctx.reader->GetFrameSync(0), nullptr) << "iter=" << iter << " step=1";
+        // Step 2: jump far away -> cancels 1 and 2 before they run.
+        ASSERT_NE(ctx.reader->GetFrameSync(50), nullptr) << "iter=" << iter << " step=2";
+        // Step 3: jump back to a cancelled chunk.  Pre-fix returns null.
+        ASSERT_NE(ctx.reader->GetFrameSync(10), nullptr) << "iter=" << iter << " step=3";
+    }
+}
+
 // §3.A regression coverage.  WarmAt must trigger an asynchronous load
 // of the chunk containing `frame_index` without blocking the caller,
 // and without requiring a subsequent GetFrame to fire the load.