test(plain): stop the registry probe from starving the waiter it waits for

PublishedWaiterDoesNotTouchTheRegistryAgain failed on ubuntu-latest at exactly
10000ms with "the waiter's exit guard never reaped the planted entry", and had
failed once before on macOS at 13.6s with "the waiter never published". Both are
the same defect, and it is in the test rather than in the code under test.

tryWithRegistry() declared its try-lock in the same scope as the 200us sleep at
the bottom of the loop, so the probe HELD m_waiterMu across that sleep and gave
it up only for the handful of nanoseconds between the unlock and the next
try_to_lock. Every condition the probe waits for is produced by the waiter under
that same mutex — reapFinishedWaiters() erases the bait under it,
publishFinishedWaiter() appends the id under it — so the loop was not polling the
waiter, it was BLOCKING it, and std::mutex hands off by barging rather than FIFO.
The waiter got in only when it happened to be running on another core in that
nanosecond-wide window, which is a lottery with no bound on it.

MEASURED, with iteration counters added to the probe: it acquired the mutex on
essentially every iteration (455/455, 567/567, ~0 try-lock misses) while the
waiter needed between 1 and 700+ attempts to land a single acquisition. The wall
clock is that count times the cost of one iteration, and the second factor is
what CI supplies: on an idle box an iteration costs ~290us, under CPU
oversubscription ~20ms. A few hundred attempts is then the whole 10s budget.
That is also why it never reproduced locally — with cores to spare the woken
waiter is dispatched fast enough to win within a few dozen attempts — and why
the sibling half of this suite never flaked: waitForPublish() takes and releases
the same mutex inside publishedCount() and sleeps OUTSIDE it.

REPRODUCED before changing anything, on Linux in the nix sandbox under 600x CPU
oversubscription: 3 failures in 25 runs, both CI messages verbatim — tookBait1
false at 10001ms after 455 probe iterations, published false at 10002ms after
311 and after 567.

THE FIX IS ONE SCOPE: release the try-lock before the sleep. The waiter then
blocks on the mutex and takes it the moment the probe lets go, so each phase
completes in one or two iterations instead of hundreds (measured: 25-232
iterations and 6-60ms become 2 iterations and 0ms). Nothing about what is being
tested moves — the window this test needs is held open by gate1, not by timing —
and the 10s budget goes back to being a backstop instead of the mechanism.
The budget is deliberately NOT raised: that would have hidden the cause.

STILL A DETECTOR, checked by rebuilding with the defect this half exists to
catch (a second reapFinishedWaiters() below publishFinishedWaiter()): 10/10
caught on Linux, 10/10 on macOS, and 10/10 on Linux under the same 600x load, so
the reliability fix did not narrow the window. Every one of them fired on
bait2Gone — the registry-half invariant — not on a precondition.

Verified: nix build .#tests 289/289 on aarch64-linux and on aarch64-darwin; the
full gtest binary 3 times, 289/289 each; this test 50/50 on Linux (slowest
268ms), 50/50 on macOS (slowest 261ms), and 50/50 under the 600x load that
produced 3 failures in 25 before. Against CI's 10737ms pass and 10000ms failure,
the test now runs in ~260ms.

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

tests/protocol/test_plain_waiter_publish_is_last.cpp

@@ -932,15 +932,50 @@ TEST_F(PlainWaiterPublishIsLastTest, PublishedWaiterDoesNotTouchTheRegistryAgain
     // Every registry read below is a TRY-lock: if the reap ever started joining
     // with m_waiterMu held, a blocking lock here would hang the suite instead of
     // reporting it.
+    //
+    // THE LOCK IS DROPPED BEFORE THE SLEEP, and that inner scope is the whole
+    // reason this probe terminates. Every condition it waits for is one only the
+    // WAITER can produce, and the waiter produces all of them under this very
+    // mutex — reapFinishedWaiters() takes it to erase the bait,
+    // publishFinishedWaiter() takes it to append the id. Sleeping inside the
+    // lock's scope therefore does not poll the waiter, it BLOCKS it: the probe
+    // would hold m_waiterMu for the whole 200us and yield it only for the few
+    // nanoseconds between the unlock and the next try_to_lock, and std::mutex
+    // hands off by barging rather than FIFO, so the waiter only gets in when it
+    // happens to be running on another core at that instant. That is a lottery
+    // with no bound on it, and it is what made this test flaky:
+    //
+    //   * measured with the lock held across the sleep, by counting iterations:
+    //     the probe acquired the mutex on essentially EVERY iteration (455/455,
+    //     567/567, ~0 try-lock misses) while the waiter needed anywhere from 1
+    //     to 700+ attempts to get a single acquisition through;
+    //   * the wall clock is that count times the cost of an iteration, and on a
+    //     loaded or virtualised runner the 200us sleep really costs ~20ms, so a
+    //     few hundred attempts is the 10s budget. Under deliberate CPU
+    //     oversubscription both CI failures reproduced from this one cause, 3
+    //     runs in 25: "the waiter's exit guard never reaped the planted entry"
+    //     (the reap's acquisition lost) and "the waiter never published" (the
+    //     publish's did);
+    //   * on an idle many-core box the woken waiter is dispatched fast enough to
+    //     win within a few dozen attempts, which is why this only ever failed in
+    //     CI and never locally.
+    //
+    // With the lock released the waiter simply blocks on it and takes it the
+    // moment the probe lets go, so each phase completes in one or two
+    // iterations instead of hundreds. Nothing about what is being tested moves:
+    // the window this test needs is held open by gate1, not by timing, and the
+    // budget below stays a backstop rather than the mechanism.
     auto tryWithRegistry = [&](const std::function<bool()>& fn, int budgetMs,
                                bool* lockedAtLeastOnce) -> bool {
         QElapsedTimer t;
         t.start();
         while (t.elapsed() < budgetMs) {
-            std::unique_lock<std::mutex> lk(mu, std::try_to_lock);
-            if (lk.owns_lock()) {
-                if (lockedAtLeastOnce) *lockedAtLeastOnce = true;
-                if (fn()) return true;
+            {
+                std::unique_lock<std::mutex> lk(mu, std::try_to_lock);
+                if (lk.owns_lock()) {
+                    if (lockedAtLeastOnce) *lockedAtLeastOnce = true;
+                    if (fn()) return true;
+                }
             }
             QThread::usleep(200);
         }