Hold host calls inside the bridge so the quiesce is tested, not assumed

The closerace scenario pinned the order of the count and the notify but never
that a host call is counted at all. Deleting the call scope from
`cultmesh_quic_next_event` outright leaves every configuration green — the
pollers finish before the teardown reaches them — and so does inverting the
comparison that makes a positive timeout block, which empties the library before
the close even begins. A scenario that only sleeps and hopes cannot tell any of
that apart from the real thing.

So the bridge says it itself, in development builds only (section 7 of the
header, behind CULTMESH_QUIC_DEBUG_ASSERTS, absent from every shipped binary):
the peak number of host calls counted inside the library, the number counted
when the close began its wait, and a hold that parks a call inside the library
until the scenario releases it. A held call returns the 0 an empty queue would
have and touches the runtime no further, so a bridge that frees the runtime
under it is reported by the scenario rather than crashing inside it.

`holdclose` drives the quiesce through that seam: the calls are counted, they
were still counted when the close began, and the close did not return while they
were held. `closerace` keeps hunting the race and now also refuses to pass when
a poller left before the close began, which is a scenario that raced nothing.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

native/GameCult.Mesh.Quic.Native/CMakeLists.txt

@@ -73,6 +73,12 @@ if(CULTMESH_QUIC_BUILD_TESTS)
   # The header only. The scenarios drive the bridge as a host does, so they get
   # no more of it than a host has.
   target_link_libraries(cultmesh_quic_native_tests PRIVATE gamecult_mesh_quic_native)
+  # The scenarios see the development seam (section 7 of the header) exactly when
+  # the library has one, so `holdclose` compiles out of the same builds the
+  # bridge's own assertions compile out of.
+  if(CULTMESH_QUIC_DEBUG_ASSERTS)
+    target_compile_definitions(cultmesh_quic_native_tests PRIVATE CULTMESH_QUIC_DEBUG_ASSERTS)
+  endif()
   set_target_properties(cultmesh_quic_native_tests PROPERTIES
     RUNTIME_OUTPUT_DIRECTORY "${CMAKE_BINARY_DIR}/bin")
   if(NOT WIN32)

native/GameCult.Mesh.Quic.Native/cultmesh_quic_native.cpp

@@ -71,6 +71,31 @@
 #define CULTMESH_QUIC_ASSERT_QUIESCED(count) ((void)0)
 #endif
 
+// Development builds only, and never in a shipped binary: the seam the scenario
+// runner drives the quiesce through. A scenario that only sleeps and hopes its
+// pollers are inside the library cannot tell a bridge that counts host calls
+// from one that does not, so the bridge says so itself:
+//
+//  - PEAK raises the high-water mark of calls counted inside, so a scenario can
+//    assert that the calls it launched were counted rather than assumed.
+//  - AT_CLOSE records how many were counted when the close began its wait, which
+//    is the number the wait exists for.
+//  - HELD parks a call inside the library until the scenario releases it, so the
+//    close races calls that are genuinely there for as long as the scenario
+//    wants them there, instead of for as long as the schedule happens to give.
+//    It returns true when it parked, and touches nothing afterwards: a bridge
+//    that frees the runtime out from under a held call is then reported by the
+//    scenario rather than crashing inside it.
+#if defined(CULTMESH_QUIC_DEBUG_ASSERTS)
+#define CULTMESH_QUIC_DEBUG_PEAK(count) DebugRaisePeak(count)
+#define CULTMESH_QUIC_DEBUG_AT_CLOSE(count) DebugRecordCallsAtClose(count)
+#define CULTMESH_QUIC_DEBUG_HELD(lock) DebugHoldCall(lock)
+#else
+#define CULTMESH_QUIC_DEBUG_PEAK(count) ((void)0)
+#define CULTMESH_QUIC_DEBUG_AT_CLOSE(count) ((void)0)
+#define CULTMESH_QUIC_DEBUG_HELD(lock) (false)
+#endif
+
 namespace {
 
 constexpr uint32_t kApiVersion = 2;
@@ -203,6 +228,36 @@ struct Runtime {
     int active_calls = 0;
 };
 
+#if defined(CULTMESH_QUIC_DEBUG_ASSERTS)
+// The development seam behind the three CULTMESH_QUIC_DEBUG_ macros. None of it
+// is compiled into a shipped library.
+std::atomic<int> debug_peak_calls{0};
+std::atomic<int> debug_calls_at_close{-1};
+std::mutex debug_hold_gate;
+std::condition_variable debug_hold_signal;
+bool debug_hold_armed = false;
+
+void DebugRaisePeak(int count) {
+    int peak = debug_peak_calls.load(std::memory_order_relaxed);
+    while (count > peak && !debug_peak_calls.compare_exchange_weak(peak, count)) {
+    }
+}
+
+void DebugRecordCallsAtClose(int count) { debug_calls_at_close.store(count); }
+
+// Parks the calling host thread inside the library until the scenario releases
+// it. The runtime gate is dropped first, so nothing else is blocked behind the
+// held call, and it is never retaken: the call is still counted, because its
+// CallScope is still alive, and that is the whole point.
+bool DebugHoldCall(std::unique_lock<std::mutex>& lock) {
+    std::unique_lock<std::mutex> hold(debug_hold_gate);
+    if (!debug_hold_armed) return false;
+    lock.unlock();
+    debug_hold_signal.wait(hold, [] { return !debug_hold_armed; });
+    return true;
+}
+#endif
+
 // Every host-facing export enters through this. It refuses once
 // `cultmesh_quic_runtime_close` has started, and otherwise holds the runtime
 // open for the length of the call.
@@ -213,6 +268,7 @@ public:
         std::lock_guard<std::mutex> lock(runtime_->gate);
         if (runtime_->closing) return;
         ++runtime_->active_calls;
+        CULTMESH_QUIC_DEBUG_PEAK(runtime_->active_calls);
         entered_ = true;
     }
     // The notify stays under `gate`, and that is the whole point of it. Released
@@ -817,6 +873,7 @@ CULTMESH_API void cultmesh_quic_runtime_close(void* handle) {
         if (runtime->closing) return;
         runtime->closing = true;
         runtime->signal.notify_all();
+        CULTMESH_QUIC_DEBUG_AT_CLOSE(runtime->active_calls);
         runtime->signal.wait(lock, [runtime] { return runtime->active_calls == 0; });
         CULTMESH_QUIC_ASSERT_QUIESCED(runtime->active_calls);
         for (auto& entry : runtime->streams) streams.push_back(std::move(entry.second));
@@ -1117,6 +1174,10 @@ CULTMESH_API int32_t cultmesh_quic_next_event(
     if (runtime->events.empty() && timeout_ms > 0) {
         runtime->signal.wait_for(lock, std::chrono::milliseconds(timeout_ms),
             [runtime] { return !runtime->events.empty() || runtime->closing; });
+        // Development builds only; a no-op unless a scenario has armed the hold.
+        // A held call returns the same 0 an empty queue would have, and touches
+        // the runtime no further.
+        if (CULTMESH_QUIC_DEBUG_HELD(lock)) return 0;
     }
     if (runtime->events.empty()) return 0;
 
@@ -1371,3 +1432,28 @@ CULTMESH_API void cultmesh_quic_close(V1Client* client) {
 }
 
 #endif // _WIN32
+
+#if defined(CULTMESH_QUIC_DEBUG_ASSERTS)
+// The development seam's three exports. They exist only in a build configured
+// with CULTMESH_QUIC_DEBUG_ASSERTS; a shipped library exports neither these nor
+// the counters behind them. See section 7 of the header.
+CULTMESH_API void cultmesh_quic_debug_hold_calls(int32_t hold) {
+    if (hold != 0) {
+        debug_peak_calls.store(0);
+        debug_calls_at_close.store(-1);
+    }
+    {
+        std::lock_guard<std::mutex> lock(debug_hold_gate);
+        debug_hold_armed = hold != 0;
+    }
+    debug_hold_signal.notify_all();
+}
+
+CULTMESH_API int32_t cultmesh_quic_debug_peak_calls(void) {
+    return debug_peak_calls.load();
+}
+
+CULTMESH_API int32_t cultmesh_quic_debug_calls_at_close(void) {
+    return debug_calls_at_close.load();
+}
+#endif

native/GameCult.Mesh.Quic.Native/include/cultmesh_quic_native.h

@@ -308,6 +308,38 @@ CULTMESH_QUIC_API int32_t cultmesh_quic_last_error(
  * when nothing has failed. */
 CULTMESH_QUIC_API int32_t cultmesh_quic_last_status(void* runtime);
 
+/* ---------------------------------------------------------------------------
+ * 7. The development seam
+ *
+ * Present only in a library configured with `CULTMESH_QUIC_DEBUG_ASSERTS`, which
+ * a shipped build never is. It exists for the runtime-lifetime scenarios: the
+ * quiesce is a promise about host calls that are inside the library, and a
+ * scenario that can only sleep and hope cannot tell a bridge that counts them
+ * from one that does not.
+ *
+ * It is process-wide, not per-runtime, because the numbers outlive the runtime
+ * they describe: `cultmesh_quic_runtime_close` frees the runtime, and the count
+ * it waited on is read afterwards. The scenarios open one runtime at a time.
+ * --------------------------------------------------------------------------- */
+#if defined(CULTMESH_QUIC_DEBUG_ASSERTS)
+
+/* Non-zero arms the hold and resets both counters: every call that reaches the
+ * blocking wait in `cultmesh_quic_next_event` parks inside the library instead
+ * of returning. Zero releases every held call and disarms it. A held call
+ * returns 0 and touches the runtime no further. */
+CULTMESH_QUIC_API void cultmesh_quic_debug_hold_calls(int32_t hold);
+
+/* The high-water mark of host calls counted inside the library since the hold
+ * was armed. Zero means no call was ever counted. */
+CULTMESH_QUIC_API int32_t cultmesh_quic_debug_peak_calls(void);
+
+/* How many host calls were counted inside the library when the last
+ * `cultmesh_quic_runtime_close` began its wait, or -1 if no close has begun one
+ * since the hold was armed. This is the number the wait exists for. */
+CULTMESH_QUIC_API int32_t cultmesh_quic_debug_calls_at_close(void);
+
+#endif /* CULTMESH_QUIC_DEBUG_ASSERTS */
+
 #if defined(__cplusplus)
 }
 #endif

native/GameCult.Mesh.Quic.Native/tests/cultmesh_quic_native_tests.cpp

@@ -18,6 +18,14 @@
 // be caught by an unlucky thread schedule.
 //
 //   cultmesh_quic_native_tests closerace [iterations] [pollers]
+//   cultmesh_quic_native_tests holdclose [iterations] [pollers]
+//
+// `closerace` hunts the race: many pollers, many iterations, and whatever the
+// schedule gives. `holdclose` takes the guessing out of it — the bridge's
+// development seam (section 7 of the header) holds the pollers inside the
+// library until this scenario releases them, and reports how many host calls the
+// bridge actually counted, so the scenario asserts the quiesce instead of
+// assuming its own fixture worked. It needs CULTMESH_QUIC_DEBUG_ASSERTS.
 //
 // Exit code 0 means every assertion held. Anything else, including a sanitizer
 // abort or the bridge's own assertion, is a failure. Under ThreadSanitizer run
@@ -73,11 +81,18 @@ std::string CloseRaceOnce(int pollers) {
     std::atomic<int> returned{0};
     std::atomic<int> undefined_return{0};
     std::atomic<int> timed_out{0};
+    // Set just before the close, and read by each poller as it returns. A poller
+    // that is already out by then was never in the race, and a scenario full of
+    // those passes without testing anything: this is what catches a bridge whose
+    // blocking crossing has stopped blocking.
+    std::atomic<bool> closing_started{false};
+    std::atomic<int> left_before_the_close{0};
 
     std::vector<std::thread> threads;
     threads.reserve(static_cast<size_t>(pollers));
     for (int index = 0; index < pollers; ++index) {
-        threads.emplace_back([runtime, &entering, &returned, &undefined_return, &timed_out] {
+        threads.emplace_back([runtime, &entering, &returned, &undefined_return, &timed_out,
+                              &closing_started, &left_before_the_close] {
             std::vector<uint8_t> payload(static_cast<size_t>(kPayloadCapacity));
             cultmesh_quic_event event{};
             int32_t required = 0;
@@ -96,6 +111,7 @@ std::string CloseRaceOnce(int pollers) {
             // The close is what ends this wait. Sitting out the whole timeout
             // instead means the wake never reached it.
             if (elapsed >= kPollTimeoutMs / 2) timed_out.fetch_add(1);
+            if (!closing_started.load()) left_before_the_close.fetch_add(1);
             returned.fetch_add(1);
         });
     }
@@ -106,9 +122,13 @@ std::string CloseRaceOnce(int pollers) {
     // close, which is the host's to avoid, not the bridge's to survive.
     std::this_thread::sleep_for(std::chrono::milliseconds(50));
 
+    closing_started.store(true);
     cultmesh_quic_runtime_close(runtime);
     for (auto& thread : threads) thread.join();
 
+    if (left_before_the_close.load() != 0)
+        return std::to_string(left_before_the_close.load()) + " of " + std::to_string(pollers) +
+            " pollers returned before the close began, so they were never in the race";
     if (undefined_return.load() != 0)
         return std::to_string(undefined_return.load()) + " poller(s) saw a return the header does not define";
     if (timed_out.load() != 0)
@@ -119,6 +139,119 @@ std::string CloseRaceOnce(int pollers) {
     return {};
 }
 
+#if defined(CULTMESH_QUIC_DEBUG_ASSERTS)
+
+// How long a poll that does not block is given to prove it by returning, after
+// the calls have been counted and before the close begins. A bridge whose
+// blocking wait no longer blocks has emptied itself of host calls by the time
+// the close starts, and `calls_at_close` then says 0.
+constexpr int kSettleMs = 100;
+
+// How long the close is required to stay inside `cultmesh_quic_runtime_close`
+// while the pollers are held. The held calls are what it is waiting for, and
+// nothing releases them until this elapses, so a close that returns first gave
+// up on a call that was still inside. This bounds what the scenario can see: a
+// close that waits for longer than this and then gives up is not distinguished
+// from one that waits properly, which is the honest limit of the check.
+constexpr int kHeldWindowMs = 500;
+
+// How long the pollers are given to be counted inside the library before the
+// close begins. Reaching `pollers` is the normal path; the deadline only stops
+// a bridge that never counts them from hanging the scenario.
+constexpr int kCountedDeadlineMs = 2000;
+
+// One iteration of the quiesce itself, with the guesswork taken out: the pollers
+// are held inside the library by the bridge's own development seam until this
+// scenario lets them go, and the bridge says how many calls it counted rather
+// than the scenario assuming the schedule put them there.
+//
+// Three promises, and a mutation that breaks any one of them is seen here:
+//  - the calls are counted (`peak`), so a blocking crossing outside a call scope
+//    is not merely uncounted-but-lucky, it reads as zero;
+//  - they were still counted when the close began its wait (`calls_at_close`),
+//    so a poll that stopped blocking cannot leave the scenario testing an empty
+//    library;
+//  - the close did not return while they were held, so a wait that gives up
+//    after any interval shorter than the hold is not a wait.
+std::string HoldCloseOnce(int pollers) {
+    cultmesh_quic_debug_hold_calls(1);
+    void* runtime = nullptr;
+    const int32_t opened = cultmesh_quic_runtime_open("cultmesh-quic-native-tests", &runtime);
+    if (opened != 0 || runtime == nullptr) {
+        cultmesh_quic_debug_hold_calls(0);
+        return "cultmesh_quic_runtime_open returned " + std::to_string(opened);
+    }
+
+    std::atomic<int> returned{0};
+    std::atomic<int> undefined_return{0};
+    std::vector<std::thread> threads;
+    threads.reserve(static_cast<size_t>(pollers));
+    for (int index = 0; index < pollers; ++index) {
+        threads.emplace_back([runtime, &returned, &undefined_return] {
+            std::vector<uint8_t> payload(static_cast<size_t>(kPayloadCapacity));
+            cultmesh_quic_event event{};
+            int32_t required = 0;
+            const int32_t result = cultmesh_quic_next_event(
+                runtime, kPollTimeoutMs, &event, payload.data(), kPayloadCapacity, &required);
+            if (result != 0 && result != 1 && result != 2) undefined_return.fetch_add(1);
+            returned.fetch_add(1);
+        });
+    }
+
+    const auto counted_deadline =
+        std::chrono::steady_clock::now() + std::chrono::milliseconds(kCountedDeadlineMs);
+    while (cultmesh_quic_debug_peak_calls() < pollers &&
+           std::chrono::steady_clock::now() < counted_deadline)
+        std::this_thread::sleep_for(std::chrono::milliseconds(1));
+    std::this_thread::sleep_for(std::chrono::milliseconds(kSettleMs));
+
+    std::atomic<bool> closed{false};
+    std::thread closer([runtime, &closed] {
+        cultmesh_quic_runtime_close(runtime);
+        closed.store(true);
+    });
+    const auto held_until =
+        std::chrono::steady_clock::now() + std::chrono::milliseconds(kHeldWindowMs);
+    while (!closed.load() && std::chrono::steady_clock::now() < held_until)
+        std::this_thread::sleep_for(std::chrono::milliseconds(5));
+
+    const bool closed_while_held = closed.load();
+    const int32_t peak = cultmesh_quic_debug_peak_calls();
+    const int32_t at_close = cultmesh_quic_debug_calls_at_close();
+
+    cultmesh_quic_debug_hold_calls(0);
+    closer.join();
+    for (auto& thread : threads) thread.join();
+
+    if (peak != pollers)
+        return "the library counted a peak of " + std::to_string(peak) + " host call(s) inside it, not " +
+            std::to_string(pollers) + "; the blocking crossing is not inside a call scope";
+    if (at_close != pollers)
+        return "the close began its wait with " + std::to_string(at_close) +
+            " host call(s) counted inside, not " + std::to_string(pollers) +
+            "; the pollers were not inside the library when the close started";
+    if (closed_while_held)
+        return "the close returned while " + std::to_string(pollers) +
+            " host call(s) were still held inside the library";
+    if (undefined_return.load() != 0)
+        return std::to_string(undefined_return.load()) + " poller(s) saw a return the header does not define";
+    if (returned.load() != pollers)
+        return std::to_string(returned.load()) + " of " + std::to_string(pollers) + " pollers returned";
+    return {};
+}
+
+int HoldClose(int iterations, int pollers) {
+    for (int iteration = 0; iteration < iterations; ++iteration) {
+        const std::string failure = HoldCloseOnce(pollers);
+        if (!failure.empty())
+            return Fail("holdclose iteration " + std::to_string(iteration) + ": " + failure);
+    }
+    std::printf("holdclose %dx%d: ok\n", iterations, pollers);
+    return 0;
+}
+
+#endif  // CULTMESH_QUIC_DEBUG_ASSERTS
+
 int CloseRace(int iterations, int pollers) {
     for (int iteration = 0; iteration < iterations; ++iteration) {
         const std::string failure = CloseRaceOnce(pollers);
@@ -133,18 +266,26 @@ int CloseRace(int iterations, int pollers) {
 
 int main(int argc, char** argv) {
     if (argc < 2) {
-        std::fprintf(stderr, "usage: %s closerace [iterations] [pollers]\n", argv[0]);
+        std::fprintf(stderr, "usage: %s closerace|holdclose [iterations] [pollers]\n", argv[0]);
         return 2;
     }
     const std::string scenario = argv[1];
-    if (scenario == "closerace") {
-        const int iterations = argc > 2 ? std::atoi(argv[2]) : 20;
-        const int pollers = argc > 3 ? std::atoi(argv[3]) : 256;
-        if (iterations <= 0 || pollers <= 0) {
-            std::fprintf(stderr, "closerace needs a positive iteration and poller count\n");
-            return 2;
-        }
-        return CloseRace(iterations, pollers);
+    const int iterations = argc > 2 ? std::atoi(argv[2]) : 20;
+    const int pollers = argc > 3 ? std::atoi(argv[3]) : 256;
+    if (iterations <= 0 || pollers <= 0) {
+        std::fprintf(stderr, "%s needs a positive iteration and poller count\n", scenario.c_str());
+        return 2;
+    }
+    if (scenario == "closerace") return CloseRace(iterations, pollers);
+    if (scenario == "holdclose") {
+#if defined(CULTMESH_QUIC_DEBUG_ASSERTS)
+        return HoldClose(iterations, pollers);
+#else
+        std::fprintf(stderr,
+            "holdclose needs the development seam: configure with "
+            "-DCULTMESH_QUIC_DEBUG_ASSERTS=ON\n");
+        return 2;
+#endif
     }
     std::fprintf(stderr, "unknown scenario '%s'\n", scenario.c_str());
     return 2;