Fix durability test flake (#5233)

# Description of Changes

Removes a race(sleep) in
`local_durability_crashes_on_new_segment_if_not_enough_space()` where we
might not wait long enough for `fallocate` to fail and crash the
durability actor.

# API and ABI breaking changes

N/A

# Expected complexity level and risk

1

# Testing

The test should no longer flake.

Author: joshua-spacetime

crates/durability/tests/io/fallocate.rs

@@ -38,7 +38,7 @@ use spacetimedb_commitlog::{
 use spacetimedb_durability::{local::OpenError, Durability, Transaction, Txdata};
 use spacetimedb_paths::{server::ReplicaDir, FromPathUnchecked};
 use tempfile::{NamedTempFile, TempDir};
-use tokio::{sync::watch, time::sleep};
+use tokio::{sync::watch, time::timeout};
 
 const MB: u64 = 1024 * 1024;
 
@@ -92,6 +92,7 @@ async fn local_durability_crashes_on_new_segment_if_not_enough_space() {
                 new_segment_tx.send_replace(());
             });
             let durability = local_durability(replica_dir, 256 * MB, Some(on_new_segment)).await?;
+            let mut durable_offset = durability.durable_tx_offset();
             let txdata = txdata();
 
             // Mark initial segment as seen.
@@ -102,8 +103,13 @@ async fn local_durability_crashes_on_new_segment_if_not_enough_space() {
             }
             // Ensure new segment is created.
             new_segment_rx.changed().await?;
-            // Yield to give fallocate a chance to run (and fail).
-            sleep(Duration::from_millis(5)).await;
+            // Wait for fallocate to run (and fail).
+            match timeout(Duration::from_secs(10), durable_offset.wait_for(256)).await {
+                Ok(Err(_)) => {}
+                Ok(Ok(offset)) => return Err(anyhow!("durability unexpectedly reached offset {offset}")),
+                Err(_) => return Err(anyhow!("timed out waiting for durability actor to exit")),
+            }
+
             // Durability actor should have crashed, so this should panic.
             info!("trying append on crashed durability");
             durability.append_tx(Box::new(Transaction::from((256, txdata.clone()))));