safekeeper compiles now

libs/neon_failpoint/src/macros.rs

@@ -28,6 +28,20 @@ macro_rules! fail_point {
             }
         }
     }};
+    ($name:literal, $condition:expr, $closure:expr) => {{
+        if cfg!(feature = "testing") {
+            if $condition {
+                match $crate::failpoint($name, None).await {
+                    $crate::FailpointResult::Continue => {},
+                    $crate::FailpointResult::Return(value) => {
+                        let closure = $closure;
+                        return closure(value.as_str());
+                    },
+                    $crate::FailpointResult::Cancelled => {},
+                }
+            }
+        }
+    }};
 }
 
 /// Failpoint macro with context support
@@ -56,6 +70,20 @@ macro_rules! fail_point_with_context {
             }
         }
     }};
+    ($name:literal, $context:expr, $condition:expr, $closure:expr) => {{
+        if cfg!(feature = "testing") {
+            if $condition {
+                match $crate::failpoint($name, Some($context)).await {
+                    $crate::FailpointResult::Continue => {},
+                    $crate::FailpointResult::Return(value) => {
+                        let closure = $closure;
+                        return closure(value.as_str());
+                    },
+                    $crate::FailpointResult::Cancelled => {},
+                }
+            }
+        }
+    }};
 }
 
 /// Pausable failpoint macro - equivalent to the old pausable_failpoint

safekeeper/src/bin/safekeeper.rs

@@ -65,7 +65,7 @@ const FEATURES: &[&str] = &[
 fn version() -> String {
     format!(
         "{GIT_VERSION} failpoints: {}, features: {:?}",
-        fail::has_failpoints(),
+        neon_failpoint::has_failpoints(),
         FEATURES,
     )
 }

safekeeper/src/send_wal.rs

@@ -872,14 +872,14 @@ impl<IO: AsyncRead + AsyncWrite + Unpin> WalSender<'_, IO> {
     async fn wait_wal(&mut self) -> Result<(), CopyStreamHandlerEnd> {
         loop {
             self.end_pos = self.end_watch.get();
-            let have_something_to_send = (|| {
-                fail::fail_point!(
+            let have_something_to_send = async {
+                neon_failpoint::fail_point!(
                     "sk-pause-send",
                     self.appname.as_deref() != Some("pageserver"),
                     |_| { false }
                 );
                 self.end_pos > self.start_pos
-            })();
+            }.await;
 
             if have_something_to_send {
                 trace!("got end_pos {:?}, streaming", self.end_pos);
@@ -931,14 +931,14 @@ impl<IO: AsyncRead + AsyncWrite + Unpin> WalSender<'_, IO> {
     /// - Err in case of error -- only if 1) term changed while fetching in recovery
     ///   mode 2) watch channel closed, which must never happen.
     async fn wait_for_lsn(&mut self) -> anyhow::Result<Option<Lsn>> {
-        let fp = (|| {
-            fail::fail_point!(
+        let fp = async {
+            neon_failpoint::fail_point!(
                 "sk-pause-send",
                 self.appname.as_deref() != Some("pageserver"),
                 |_| { true }
             );
             false
-        })();
+        }.await;
         if fp {
             tokio::time::sleep(POLL_STATE_TIMEOUT).await;
             return Ok(None);

safekeeper/src/wal_backup.rs

@@ -657,7 +657,7 @@ pub async fn delete_timeline(
 
     pausable_failpoint!("sk-delete-timeline-remote-pause");
 
-    fail::fail_point!("sk-delete-timeline-remote", |_| {
+    neon_failpoint::fail_point!("sk-delete-timeline-remote", |_| {
         Err(anyhow::anyhow!("failpoint: sk-delete-timeline-remote"))
     });