test: surface peer startup crashes in dirty-shard test (#9161)

* test: wait for WAL flock release on peer restart in dirty-shard test

The flake addressed by #9124 (bumping `wait_for_peer_online` to 60s) was
misdiagnosed as CPU contention. Logs show the restarted peer panics within
~1s of startup at `consensus_wal.rs:36` with:

    Wal error: Can't init WAL: Kind(WouldBlock)
    Panic: Can't open consensus WAL: Kind(WouldBlock)

`wal::Wal::open` calls `fs4::FileExt::try_lock` (non-blocking `flock`) on
the WAL directory fd. After `p.kill()` (SIGKILL + waitpid) the kernel
normally releases the killed peer's flock immediately, but under
pytest-xdist load there is a small window where it lags. The fresh peer's
startup then races and panics. After the panic `/readyz` never returns
200, so neither 30s nor 60s rescues the test.

Fix: add a `wait_for_wal_unlocked` helper that polls both the consensus
WAL directory and the local-shard WAL directory with the same exclusive
non-blocking flock that qdrant uses, and call it after every `p.kill()`
that is followed by a `start_peer` on the same `peer_dir`. The two
60s timeouts are restored to the default 30s now that the underlying
race is gone.

Co-authored-by: Cursor <cursoragent@cursor.com>

* test: fail fast if sync restart crashes in dirty-shard test

The sync restart in `test_dirty_shard_survives_update_collection` used
plain `wait_for_peer_online(sync_uri)`, which only polls `/readyz`. If
the freshly started peer panics on startup (e.g. WAL `WouldBlock`), the
test waits the full 30s timeout and then reports a `/readyz` timeout
instead of the actual panic message and exit code.

Switch the sync restart to `wait_for_peer_online_or_crash(...)` (same
helper already used for the dirty restart). On crash it dumps the peer
log tail so the next CI failure shows the real reason instead of a
generic timeout.

Co-authored-by: Cursor <cursoragent@cursor.com>

* test: drop speculative WAL flock wait, keep crash-detection switch

Reverts the `wait_for_wal_unlocked` helper that polled the WAL
directories with non-blocking flock. The kernel-side flock race it was
guarding against could not be reproduced in isolation (0/300 iters of
SIGKILL+wait+re-flock on bare Linux), so it was speculative.

Kept:
- Sync restart now uses `wait_for_peer_online_or_crash(...)` instead of
  plain `wait_for_peer_online`, so any startup panic surfaces fast with
  the actual log tail instead of hiding behind a 30s `/readyz` timeout.
- The two 60s timeouts bumped in #9124 are restored to the default 30s.

If the flake recurs in CI, the new failure output will tell us the
actual cause (panic message + exit code), which is more useful than
papering over it.

Co-authored-by: Cursor <cursoragent@cursor.com>

---------

Co-authored-by: Cursor Agent <agent@cursor.com>
Co-authored-by: Cursor <cursoragent@cursor.com>

tests/consensus_tests/test_dirty_shard_crash_loop.py

@@ -85,16 +85,20 @@ def test_dirty_shard_survives_update_collection(tmp_path: pathlib.Path):
     assert r.status_code == 200, f"UpdateCollection failed: {r.text}"
 
     # Restart target briefly (no dirty flag) so it syncs the UpdateCollection entry
+    sync_log = f"peer_sync_{target_idx}.log"
     sync_uri = start_peer(
         peer_dirs[target_idx],
-        f"peer_sync_{target_idx}.log",
+        sync_log,
         bootstrap_uri,
         port=restart_port,
     )
-    # Under parallel pytest-xdist load the leader can take longer than the
-    # default 30s budget to replicate the UpdateCollection entry and let the
-    # rejoining peer report ready. Allow more time to avoid CPU-contention flakes.
-    wait_for_peer_online(sync_uri, wait_for_timeout=60)
+    online, exit_code = wait_for_peer_online_or_crash(sync_uri, processes[-1])
+    if not online:
+        log_content = read_log(sync_log)
+        pytest.fail(
+            f"Sync restart failed to come online (exit code {exit_code}).\n"
+            f"Log tail:\n{log_content[-1000:]}"
+        )
 
     # Kill again — the UpdateCollection entry is now committed in its WAL
     p = processes.pop()
@@ -122,7 +126,7 @@ def test_dirty_shard_survives_update_collection(tmp_path: pathlib.Path):
         port=restart_port,
     )
 
-    online, exit_code = wait_for_peer_online_or_crash(new_uri, processes[-1], timeout=60)
+    online, exit_code = wait_for_peer_online_or_crash(new_uri, processes[-1], timeout=30)
 
     if not online:
         log_content = read_log(restart_log)