test: add some logging for compat tests

This log is too noisy. Instead of warning on every retry, let's log only
on the final failure.

Cargo.lock

@@ -1330,6 +1330,7 @@ dependencies = [
  "chrono",
  "clap",
  "compute_api",
+ "fail",
  "flate2",
  "futures",
  "hostname-validator",
@@ -1338,7 +1339,6 @@ dependencies = [
  "itertools 0.10.5",
  "jsonwebtoken",
  "metrics",
- "neon_failpoint",
  "nix 0.30.1",
  "notify",
  "num_cpus",
@@ -2891,13 +2891,13 @@ dependencies = [
  "arc-swap",
  "bytes",
  "camino",
+ "fail",
  "futures",
  "hyper 0.14.30",
  "itertools 0.10.5",
  "jemalloc_pprof",
  "jsonwebtoken",
  "metrics",
- "neon_failpoint",
  "once_cell",
  "pprof",
  "regex",
@@ -3852,23 +3852,6 @@ dependencies = [
  "workspace_hack",
 ]
 
-[[package]]
-name = "neon_failpoint"
-version = "0.1.0"
-dependencies = [
- "anyhow",
- "either",
- "once_cell",
- "parking_lot 0.12.1",
- "rand 0.8.5",
- "regex",
- "serde",
- "tokio",
- "tokio-util",
- "tracing",
- "tracing-subscriber",
-]
-
 [[package]]
 name = "never-say-never"
 version = "6.6.666"
@@ -4374,6 +4357,7 @@ dependencies = [
  "either",
  "enum-map",
  "enumset",
+ "fail",
  "futures",
  "hashlink",
  "hex",
@@ -4388,7 +4372,6 @@ dependencies = [
  "jsonwebtoken",
  "md5",
  "metrics",
- "neon_failpoint",
  "nix 0.30.1",
  "num-traits",
  "num_cpus",
@@ -6208,6 +6191,7 @@ dependencies = [
  "criterion",
  "desim",
  "env_logger",
+ "fail",
  "futures",
  "hex",
  "http 1.1.0",
@@ -6217,7 +6201,6 @@ dependencies = [
  "itertools 0.10.5",
  "jsonwebtoken",
  "metrics",
- "neon_failpoint",
  "once_cell",
  "pageserver_api",
  "parking_lot 0.12.1",
@@ -6904,6 +6887,7 @@ dependencies = [
  "diesel",
  "diesel-async",
  "diesel_migrations",
+ "fail",
  "futures",
  "governor",
  "hex",
@@ -6916,7 +6900,6 @@ dependencies = [
  "lasso",
  "measured",
  "metrics",
- "neon_failpoint",
  "once_cell",
  "pageserver_api",
  "pageserver_client",
@@ -7008,6 +6991,7 @@ dependencies = [
  "pageserver_api",
  "pageserver_client",
  "reqwest",
+ "safekeeper_api",
  "serde_json",
  "storage_controller_client",
  "tokio",
@@ -8181,7 +8165,7 @@ dependencies = [
  "const_format",
  "criterion",
  "diatomic-waker",
- "either",
+ "fail",
  "futures",
  "git-version",
  "hex",
@@ -8189,7 +8173,6 @@ dependencies = [
  "humantime",
  "jsonwebtoken",
  "metrics",
- "neon_failpoint",
  "nix 0.30.1",
  "once_cell",
  "pem",

Cargo.toml

@@ -21,7 +21,6 @@ members = [
     "workspace_hack",
     "libs/compute_api",
     "libs/http-utils",
-    "libs/neon_failpoint",
     "libs/pageserver_api",
     "libs/postgres_ffi",
     "libs/postgres_ffi_types",
@@ -98,6 +97,7 @@ diatomic-waker = { version = "0.2.3" }
 either = "1.8"
 enum-map = "2.4.2"
 enumset = "1.0.12"
+fail = "0.5.0"
 fallible-iterator = "0.2"
 framed-websockets = { version = "0.1.0", git = "https://github.com/neondatabase/framed-websockets" }
 futures = "0.3"
@@ -258,7 +258,6 @@ desim = { version = "0.1", path = "./libs/desim" }
 endpoint_storage = { version = "0.0.1", path = "./endpoint_storage/" }
 http-utils = { version = "0.1", path = "./libs/http-utils/" }
 metrics = { version = "0.1", path = "./libs/metrics/" }
-neon_failpoint = { version = "0.1", path = "./libs/neon_failpoint/" }
 neon-shmem = { version = "0.1", path = "./libs/neon-shmem/" }
 pageserver = { path = "./pageserver" }
 pageserver_api = { version = "0.1", path = "./libs/pageserver_api/" }

compute_tools/Cargo.toml

@@ -7,7 +7,7 @@ license.workspace = true
 [features]
 default = []
 # Enables test specific features.
-testing = ["neon_failpoint/testing"]
+testing = ["fail/failpoints"]
 
 [dependencies]
 async-compression.workspace = true
@@ -23,7 +23,7 @@ camino.workspace = true
 chrono.workspace = true
 cfg-if.workspace = true
 clap.workspace = true
-neon_failpoint.workspace = true
+fail.workspace = true
 flate2.workspace = true
 futures.workspace = true
 http.workspace = true

compute_tools/src/bin/compute_ctl.rs

@@ -154,7 +154,7 @@ impl Cli {
 fn main() -> Result<()> {
     let cli = Cli::parse();
 
-    failpoint_support::init().unwrap();
+    let scenario = failpoint_support::init();
 
     // For historical reasons, the main thread that processes the config and launches postgres
     // is synchronous, but we always have this tokio runtime available and we "enter" it so
@@ -201,6 +201,8 @@ fn main() -> Result<()> {
 
     let exit_code = compute_node.run()?;
 
+    scenario.teardown();
+
     deinit_and_exit(exit_code);
 }
 

compute_tools/src/http/routes/failpoints.rs

@@ -1,9 +1,8 @@
 use axum::response::{IntoResponse, Response};
 use http::StatusCode;
-use neon_failpoint::{configure_failpoint, configure_failpoint_with_context};
 use serde::{Deserialize, Serialize};
-use std::collections::HashMap;
 use tracing::info;
+use utils::failpoint_support::apply_failpoint;
 
 pub type ConfigureFailpointsRequest = Vec<FailpointConfig>;
 
@@ -12,16 +11,10 @@ pub type ConfigureFailpointsRequest = Vec<FailpointConfig>;
 pub struct FailpointConfig {
     /// Name of the fail point
     pub name: String,
-    /// List of actions to take, using the format described in neon_failpoint
+    /// List of actions to take, using the format described in `fail::cfg`
     ///
-    /// We support actions: "pause", "sleep(N)", "return", "return(value)", "exit", "off", "panic(message)"
-    /// Plus probability-based actions: "N%return(value)", "N%M*return(value)", "N%action", "N%M*action"
+    /// We also support `actions = "exit"` to cause the fail point to immediately exit.
     pub actions: String,
-    /// Optional context matching rules for conditional failpoints
-    /// Each key-value pair specifies a context key and a regex pattern to match against
-    /// All context matchers must match for the failpoint to trigger
-    #[serde(default, skip_serializing_if = "Option::is_none")]
-    pub context_matchers: Option<HashMap<String, String>>,
 }
 
 use crate::http::JsonResponse;
@@ -31,7 +24,7 @@ use crate::http::extract::Json;
 pub(in crate::http) async fn configure_failpoints(
     failpoints: Json<ConfigureFailpointsRequest>,
 ) -> Response {
-    if !neon_failpoint::has_failpoints() {
+    if !fail::has_failpoints() {
         return JsonResponse::error(
             StatusCode::PRECONDITION_FAILED,
             "Cannot manage failpoints because neon was compiled without failpoints support",
@@ -39,21 +32,16 @@ pub(in crate::http) async fn configure_failpoints(
     }
 
     for fp in &*failpoints {
-        info!(
-            "cfg failpoint: {} {} (context: {:?})",
-            fp.name, fp.actions, fp.context_matchers
-        );
+        info!("cfg failpoint: {} {}", fp.name, fp.actions);
 
-        let cfg_result = if let Some(context_matchers) = fp.context_matchers.clone() {
-            configure_failpoint_with_context(&fp.name, &fp.actions, context_matchers)
-        } else {
-            configure_failpoint(&fp.name, &fp.actions)
-        };
+        // We recognize one extra "action" that's not natively recognized
+        // by the failpoints crate: exit, to immediately kill the process
+        let cfg_result = apply_failpoint(&fp.name, &fp.actions);
 
         if let Err(e) = cfg_result {
             return JsonResponse::error(
                 StatusCode::BAD_REQUEST,
-                format!("failed to configure failpoint '{}': {e}", fp.name),
+                format!("failed to configure failpoints: {e}"),
             );
         }
     }

compute_tools/src/migration.rs

@@ -1,5 +1,5 @@
 use anyhow::{Context, Result};
-use neon_failpoint::fail_point;
+use fail::fail_point;
 use tokio_postgres::{Client, Transaction};
 use tracing::{error, info};
 
@@ -40,14 +40,13 @@ impl<'m> MigrationRunner<'m> {
         // middle of applying a series of migrations fails in an expected
         // manner
         if cfg!(feature = "testing") {
-            let fail = async {
-                fail_point!("compute-migration", |fail_migration_id: Option<String>| {
+            let fail = (|| {
+                fail_point!("compute-migration", |fail_migration_id| {
                     migration_id == fail_migration_id.unwrap().parse::<i64>().unwrap()
                 });
 
                 false
-            }
-            .await;
+            })();
 
             if fail {
                 return Err(anyhow::anyhow!(format!(

control_plane/src/broker.rs

@@ -36,7 +36,7 @@ impl StorageBroker {
     pub async fn start(&self, retry_timeout: &Duration) -> anyhow::Result<()> {
         let broker = &self.env.broker;
 
-        print!("Starting neon broker at {}", broker.client_url());
+        println!("Starting neon broker at {}", broker.client_url());
 
         let mut args = Vec::new();
 

control_plane/src/pageserver.rs

@@ -303,7 +303,7 @@ impl PageServerNode {
     async fn start_node(&self, retry_timeout: &Duration) -> anyhow::Result<()> {
         // TODO: using a thread here because start_process() is not async but we need to call check_status()
         let datadir = self.repo_path();
-        print!(
+        println!(
             "Starting pageserver node {} at '{}' in {:?}, retrying for {:?}",
             self.conf.id,
             self.pg_connection_config.raw_address(),

control_plane/src/safekeeper.rs

@@ -127,7 +127,7 @@ impl SafekeeperNode {
         extra_opts: &[String],
         retry_timeout: &Duration,
     ) -> anyhow::Result<()> {
-        print!(
+        println!(
             "Starting safekeeper at '{}' in '{}', retrying for {:?}",
             self.pg_connection_config.raw_address(),
             self.datadir_path().display(),

control_plane/src/storage_controller.rs

@@ -660,7 +660,7 @@ impl StorageController {
             ));
         }
 
-        println!("Starting storage controller");
+        println!("Starting storage controller at {scheme}://{host}:{listen_port}");
 
         background_process::start_process(
             COMMAND,

control_plane/storcon_cli/Cargo.toml

@@ -14,6 +14,7 @@ humantime.workspace = true
 pageserver_api.workspace = true
 pageserver_client.workspace = true
 reqwest.workspace = true
+safekeeper_api.workspace=true
 serde_json = { workspace = true, features = ["raw_value"] }
 storage_controller_client.workspace = true
 tokio.workspace = true

control_plane/storcon_cli/src/main.rs

@@ -11,7 +11,7 @@ use pageserver_api::controller_api::{
     PlacementPolicy, SafekeeperDescribeResponse, SafekeeperSchedulingPolicyRequest,
     ShardSchedulingPolicy, ShardsPreferredAzsRequest, ShardsPreferredAzsResponse,
     SkSchedulingPolicy, TenantCreateRequest, TenantDescribeResponse, TenantPolicyRequest,
-    TenantShardMigrateRequest, TenantShardMigrateResponse,
+    TenantShardMigrateRequest, TenantShardMigrateResponse, TimelineSafekeeperMigrateRequest,
 };
 use pageserver_api::models::{
     EvictionPolicy, EvictionPolicyLayerAccessThreshold, ShardParameters, TenantConfig,
@@ -21,6 +21,7 @@ use pageserver_api::models::{
 use pageserver_api::shard::{ShardStripeSize, TenantShardId};
 use pageserver_client::mgmt_api::{self};
 use reqwest::{Certificate, Method, StatusCode, Url};
+use safekeeper_api::models::TimelineLocateResponse;
 use storage_controller_client::control_api::Client;
 use utils::id::{NodeId, TenantId, TimelineId};
 
@@ -279,6 +280,23 @@ enum Command {
         #[arg(long)]
         concurrency: Option<usize>,
     },
+    /// Locate safekeepers for a timeline from the storcon DB.
+    TimelineLocate {
+        #[arg(long)]
+        tenant_id: TenantId,
+        #[arg(long)]
+        timeline_id: TimelineId,
+    },
+    /// Migrate a timeline to a new set of safekeepers
+    TimelineSafekeeperMigrate {
+        #[arg(long)]
+        tenant_id: TenantId,
+        #[arg(long)]
+        timeline_id: TimelineId,
+        /// Example: --new-sk-set 1,2,3
+        #[arg(long, required = true, value_delimiter = ',')]
+        new_sk_set: Vec<NodeId>,
+    },
 }
 
 #[derive(Parser)]
@@ -1324,7 +1342,7 @@ async fn main() -> anyhow::Result<()> {
             concurrency,
         } => {
             let mut path = format!(
-                "/v1/tenant/{tenant_shard_id}/timeline/{timeline_id}/download_heatmap_layers",
+                "v1/tenant/{tenant_shard_id}/timeline/{timeline_id}/download_heatmap_layers",
             );
 
             if let Some(c) = concurrency {
@@ -1335,6 +1353,41 @@ async fn main() -> anyhow::Result<()> {
                 .dispatch::<(), ()>(Method::POST, path, None)
                 .await?;
         }
+        Command::TimelineLocate {
+            tenant_id,
+            timeline_id,
+        } => {
+            let path = format!("debug/v1/tenant/{tenant_id}/timeline/{timeline_id}/locate");
+
+            let resp = storcon_client
+                .dispatch::<(), TimelineLocateResponse>(Method::GET, path, None)
+                .await?;
+
+            let sk_set = resp.sk_set.iter().map(|id| id.0 as i64).collect::<Vec<_>>();
+            let new_sk_set = resp
+                .new_sk_set
+                .as_ref()
+                .map(|ids| ids.iter().map(|id| id.0 as i64).collect::<Vec<_>>());
+
+            println!("generation = {}", resp.generation);
+            println!("sk_set = {sk_set:?}");
+            println!("new_sk_set = {new_sk_set:?}");
+        }
+        Command::TimelineSafekeeperMigrate {
+            tenant_id,
+            timeline_id,
+            new_sk_set,
+        } => {
+            let path = format!("v1/tenant/{tenant_id}/timeline/{timeline_id}/safekeeper_migrate");
+
+            storcon_client
+                .dispatch::<_, ()>(
+                    Method::POST,
+                    path,
+                    Some(TimelineSafekeeperMigrateRequest { new_sk_set }),
+                )
+                .await?;
+        }
     }
 
     Ok(())

libs/http-utils/Cargo.toml

@@ -9,7 +9,7 @@ anyhow.workspace = true
 arc-swap.workspace = true
 bytes.workspace = true
 camino.workspace = true
-neon_failpoint.workspace = true
+fail.workspace = true
 futures.workspace = true
 hyper0.workspace = true
 itertools.workspace = true

libs/http-utils/src/failpoints.rs

@@ -1,8 +1,7 @@
 use hyper::{Body, Request, Response, StatusCode};
-use neon_failpoint::{configure_failpoint, configure_failpoint_with_context, has_failpoints};
 use serde::{Deserialize, Serialize};
-use std::collections::HashMap;
 use tokio_util::sync::CancellationToken;
+use utils::failpoint_support::apply_failpoint;
 
 use crate::error::ApiError;
 use crate::json::{json_request, json_response};
@@ -14,16 +13,10 @@ pub type ConfigureFailpointsRequest = Vec<FailpointConfig>;
 pub struct FailpointConfig {
     /// Name of the fail point
     pub name: String,
-    /// List of actions to take, using the format described in neon_failpoint
+    /// List of actions to take, using the format described in `fail::cfg`
     ///
-    /// We support actions: "pause", "sleep(N)", "return", "return(value)", "exit", "off", "panic(message)"
-    /// Plus probability-based actions: "N%return(value)", "N%M*return(value)", "N%action", "N%M*action"
+    /// We also support `actions = "exit"` to cause the fail point to immediately exit.
     pub actions: String,
-    /// Optional context matching rules for conditional failpoints
-    /// Each key-value pair specifies a context key and a regex pattern to match against
-    /// All context matchers must match for the failpoint to trigger
-    #[serde(default, skip_serializing_if = "Option::is_none")]
-    pub context_matchers: Option<HashMap<String, String>>,
 }
 
 /// Configure failpoints through http.
@@ -31,7 +24,7 @@ pub async fn failpoints_handler(
     mut request: Request<Body>,
     _cancel: CancellationToken,
 ) -> Result<Response<Body>, ApiError> {
-    if !has_failpoints() {
+    if !fail::has_failpoints() {
         return Err(ApiError::BadRequest(anyhow::anyhow!(
             "Cannot manage failpoints because neon was compiled without failpoints support"
         )));
@@ -39,24 +32,15 @@ pub async fn failpoints_handler(
 
     let failpoints: ConfigureFailpointsRequest = json_request(&mut request).await?;
     for fp in failpoints {
-        tracing::info!(
-            "cfg failpoint: {} {} (context: {:?})",
-            fp.name,
-            fp.actions,
-            fp.context_matchers
-        );
+        tracing::info!("cfg failpoint: {} {}", fp.name, fp.actions);
 
-        let cfg_result = if let Some(context_matchers) = fp.context_matchers {
-            configure_failpoint_with_context(&fp.name, &fp.actions, context_matchers)
-        } else {
-            configure_failpoint(&fp.name, &fp.actions)
-        };
+        // We recognize one extra "action" that's not natively recognized
+        // by the failpoints crate: exit, to immediately kill the process
+        let cfg_result = apply_failpoint(&fp.name, &fp.actions);
 
-        if let Err(err) = cfg_result {
+        if let Err(err_msg) = cfg_result {
             return Err(ApiError::BadRequest(anyhow::anyhow!(
-                "Failed to configure failpoint '{}': {}",
-                fp.name,
-                err
+                "Failed to configure failpoints: {err_msg}"
             )));
         }
     }

libs/neon_failpoint/Cargo.toml

@@ -1,27 +0,0 @@
-[package]
-name = "neon_failpoint"
-version = "0.1.0"
-edition = "2021"
-
-[dependencies]
-tokio = { workspace = true, features = ["time", "sync", "rt-multi-thread"] }
-tokio-util = { workspace = true }
-tracing = { workspace = true }
-serde = { workspace = true, features = ["derive"] }
-anyhow = { workspace = true }
-regex = { workspace = true }
-once_cell = { workspace = true }
-parking_lot = { workspace = true }
-rand = { workspace = true }
-either = { workspace = true }
-
-[dev-dependencies]
-tracing-subscriber = { workspace = true, features = ["fmt"] }
-
-[features]
-default = []
-testing = []
-
-[[example]]
-name = "context_demo"
-required-features = ["testing"]

libs/neon_failpoint/README.md

@@ -1,460 +0,0 @@
-# Neon Failpoint Library
-
-A modern, async-first failpoint library for Neon, replacing the `fail` crate with enhanced functionality.
-
-## Features
-
-- **Async-first**: All failpoint operations are async and don't require `spawn_blocking`
-- **Context matching**: Failpoints can be configured to trigger only when specific context conditions are met
-- **Regex support**: Context values can be matched using regular expressions
-- **Cancellation support**: All operations support cancellation tokens
-- **Dynamic reconfiguration**: Paused and sleeping tasks automatically resume when failpoint configurations change
-- **Backward compatibility**: Drop-in replacement for existing `fail` crate usage
-
-## Supported Actions
-
-- `off` - Disable the failpoint
-- `pause` - Pause indefinitely until disabled, reconfigured, or cancelled
-- `sleep(N)` - Sleep for N milliseconds (can be interrupted by reconfiguration)
-- `return` - Return early (empty value)
-- `return(value)` - Return early with a specific value
-- `exit` - Exit the process immediately
-- `panic(message)` - Panic the process with a custom message
-- `N%return(value)` - Return with a specific value N% of the time (probability-based)
-- `N%M*return(value)` - Return with a specific value N% of the time, maximum M times
-- `N%action` - Execute any action N% of the time (probability-based)
-- `N%M*action` - Execute any action N% of the time, maximum M times
-
-## Probability-Based Actions
-
-The library supports probability-based failpoints that trigger only a percentage of the time:
-
-```rust
-// 50% chance to return a value
-configure_failpoint("random_failure", "50%return(error)").unwrap();
-
-// 10% chance to sleep, maximum 3 times
-configure_failpoint("occasional_delay", "10%3*sleep(1000)").unwrap();
-
-// 25% chance to panic
-configure_failpoint("rare_panic", "25%panic(critical error)").unwrap();
-```
-
-The probability system uses a counter to track how many times a probability-based action has been triggered, allowing for precise control over test scenarios.
-
-## Dynamic Behavior
-
-When a failpoint is reconfigured while tasks are waiting on it:
-
-- **Paused tasks** will immediately resume and continue normal execution
-- **Sleeping tasks** will wake up early and continue normal execution
-- **Removed failpoints** will cause all waiting tasks to resume normally
-
-The new configuration only applies to future hits of the failpoint, not to tasks that are already waiting. This allows for flexible testing scenarios where you can pause execution, inspect state, and then resume execution dynamically.
-
-## Example: Dynamic Reconfiguration
-
-```rust
-use neon_failpoint::{configure_failpoint, failpoint, FailpointResult};
-use tokio::time::Duration;
-
-// Start a task that will hit a failpoint
-let task = tokio::spawn(async {
-    println!("About to hit failpoint");
-    match failpoint("test_pause", None).await {
-        FailpointResult::Return(value) => println!("Returned: {}", value),
-        FailpointResult::Continue => println!("Continued normally"),
-        FailpointResult::Cancelled => println!("Cancelled"),
-    }
-});
-
-// Configure the failpoint to pause
-configure_failpoint("test_pause", "pause").unwrap();
-
-// Let the task hit the failpoint and pause
-tokio::time::sleep(Duration::from_millis(10)).await;
-
-// Change the failpoint configuration - this will wake up the paused task
-// The task will resume and continue normally (not apply the new config)
-configure_failpoint("test_pause", "return(not_applied)").unwrap();
-
-// The task will complete with Continue, not Return
-let result = task.await.unwrap();
-```
-
-## Basic Usage
-
-```rust
-use neon_failpoint::{configure_failpoint, failpoint, FailpointResult};
-
-// Configure a failpoint
-configure_failpoint("my_failpoint", "return(42)").unwrap();
-
-// Use the failpoint
-match failpoint("my_failpoint", None).await {
-    FailpointResult::Return(value) => {
-        println!("Failpoint returned: {}", value);
-        return value.parse().unwrap_or_default();
-    }
-    FailpointResult::Continue => {
-        // Continue normal execution
-    }
-    FailpointResult::Cancelled => {
-        // Handle cancellation
-    }
-}
-```
-
-## Context-Based Failpoint Configuration
-
-Context allows you to create **conditional failpoints** that only trigger when specific runtime conditions are met. This is particularly useful for testing scenarios where you want to inject failures only for specific tenants, operations, or other contextual conditions.
-
-### Configuring Context-Based Failpoints
-
-Use `configure_failpoint_with_context()` to set up failpoints with context matching:
-
-```rust
-use neon_failpoint::configure_failpoint_with_context;
-use std::collections::HashMap;
-
-let mut context_matchers = HashMap::new();
-context_matchers.insert("tenant_id".to_string(), "test_.*".to_string());
-context_matchers.insert("operation".to_string(), "backup".to_string());
-
-configure_failpoint_with_context(
-    "backup_operation",           // failpoint name
-    "return(simulated_failure)",  // action to take
-    context_matchers             // context matching rules
-).unwrap();
-```
-
-### Context Matching Rules
-
-The context matching system works as follows:
-
-1. **Key-Value Matching**: Each entry in `context_matchers` specifies a key that must exist in the runtime context
-2. **Regex Support**: Values in `context_matchers` are treated as regular expressions first
-3. **Fallback to Exact Match**: If the regex compilation fails, it falls back to exact string matching
-4. **ALL Must Match**: All context matchers must match for the failpoint to trigger
-
-### Runtime Context Usage
-
-When code hits a failpoint, it provides context using a `HashMap<String, String>`:
-
-```rust
-use neon_failpoint::{failpoint, FailpointResult};
-use std::collections::HashMap;
-
-let mut context = HashMap::new();
-context.insert("tenant_id".to_string(), "test_123".to_string());
-context.insert("operation".to_string(), "backup".to_string());
-context.insert("user_id".to_string(), "user_456".to_string());
-
-match failpoint("backup_operation", Some(&context)) {
-    either::Either::Left(result) => {
-        match result {
-            FailpointResult::Return(value) => {
-                // This will only trigger if ALL context matchers match
-                println!("Backup failed: {}", value);
-            }
-            FailpointResult::Continue => {
-                // Continue with normal backup operation
-            }
-            FailpointResult::Cancelled => {}
-        }
-    }
-    either::Either::Right(future) => {
-        match future.await {
-            FailpointResult::Return(value) => {
-                // This will only trigger if ALL context matchers match
-                println!("Backup failed: {}", value);
-            }
-            FailpointResult::Continue => {
-                // Continue with normal backup operation
-            }
-            FailpointResult::Cancelled => {}
-        }
-    }
-}
-```
-
-### Context Matching Examples
-
-#### Regex Matching
-```rust
-// Configure to match test tenants only
-let mut matchers = HashMap::new();
-matchers.insert("tenant_id".to_string(), "test_.*".to_string());
-
-configure_failpoint_with_context("test_failpoint", "pause", matchers).unwrap();
-
-// This will match
-let mut context = HashMap::new();
-context.insert("tenant_id".to_string(), "test_123".to_string());
-// This will NOT match
-let mut context = HashMap::new();
-context.insert("tenant_id".to_string(), "prod_123".to_string());
-```
-
-#### Multiple Conditions
-```rust
-// Must match BOTH tenant pattern AND operation
-let mut matchers = HashMap::new();
-matchers.insert("tenant_id".to_string(), "test_.*".to_string());
-matchers.insert("operation".to_string(), "backup".to_string());
-
-configure_failpoint_with_context("backup_test", "return(failed)", matchers).unwrap();
-
-// This will match (both conditions met)
-let mut context = HashMap::new();
-context.insert("tenant_id".to_string(), "test_123".to_string());
-context.insert("operation".to_string(), "backup".to_string());
-
-// This will NOT match (missing operation)
-let mut context = HashMap::new();
-context.insert("tenant_id".to_string(), "test_123".to_string());
-context.insert("operation".to_string(), "restore".to_string());
-```
-
-#### Exact String Matching
-```rust
-// If regex compilation fails, falls back to exact match
-let mut matchers = HashMap::new();
-matchers.insert("env".to_string(), "staging".to_string());
-
-configure_failpoint_with_context("env_specific", "sleep(1000)", matchers).unwrap();
-
-// This will match
-let mut context = HashMap::new();
-context.insert("env".to_string(), "staging".to_string());
-// This will NOT match
-let mut context = HashMap::new();
-context.insert("env".to_string(), "production".to_string());
-```
-
-### Benefits of Context-Based Failpoints
-
-1. **Selective Testing**: Only inject failures for specific tenants, environments, or operations
-2. **Production Safety**: Avoid accidentally triggering failpoints in production by using context filters
-3. **Complex Scenarios**: Test interactions between different components with targeted failures
-4. **Debugging**: Isolate issues to specific contexts without affecting the entire system
-
-### Context vs. Non-Context Failpoints
-
-- **Without context**: `configure_failpoint("name", "action")` - triggers for ALL hits
-- **With context**: `configure_failpoint_with_context("name", "action", matchers)` - triggers only when context matches
-
-## Context-Specific Failpoints
-
-```rust
-use neon_failpoint::{configure_failpoint_with_context, failpoint};
-use std::collections::HashMap;
-
-// Configure a failpoint that only triggers for specific tenants
-let mut context_matchers = HashMap::new();
-context_matchers.insert("tenant_id".to_string(), "test_.*".to_string());
-context_matchers.insert("operation".to_string(), "backup".to_string());
-
-configure_failpoint_with_context(
-    "backup_operation",
-    "return(simulated_failure)",
-    context_matchers
-).unwrap();
-
-// Use with context
-let mut context = HashMap::new();
-context.insert("tenant_id".to_string(), "test_123".to_string());
-context.insert("operation".to_string(), "backup".to_string());
-
-match failpoint("backup_operation", Some(&context)) {
-    either::Either::Left(result) => {
-        match result {
-            FailpointResult::Return(value) => {
-                // This will trigger for tenant_id matching "test_.*"
-                println!("Backup failed: {}", value);
-            }
-            FailpointResult::Continue => {
-                // Continue with backup
-            }
-            FailpointResult::Cancelled => {}
-        }
-    }
-    either::Either::Right(future) => {
-        match future.await {
-            FailpointResult::Return(value) => {
-                // This will trigger for tenant_id matching "test_.*"
-                println!("Backup failed: {}", value);
-            }
-            FailpointResult::Continue => {
-                // Continue with backup
-            }
-            FailpointResult::Cancelled => {}
-        }
-    }
-}
-```
-
-## Macros
-
-The library provides convenient macros for common patterns:
-
-### `fail_point!` - Basic Failpoint Macro
-
-The `fail_point!` macro has three variants:
-
-1. **Simple failpoint** - `fail_point!(name)`
-   - Just checks the failpoint and continues or returns early (no value)
-   - Panics if the failpoint is configured with `return(value)` since no closure is provided
-
-2. **Failpoint with return handler** - `fail_point!(name, closure)`
-   - Provides a closure to handle return values from the failpoint
-   - The closure receives `Option<String>` and should return the appropriate value
-
-3. **Conditional failpoint** - `fail_point!(name, condition, closure)`
-   - Only checks the failpoint if the condition is true
-   - Provides a closure to handle return values (receives `&str`)
-
-```rust
-use neon_failpoint::fail_point;
-
-// Simple failpoint - just continue or return early
-fail_point!("my_failpoint");
-
-// Failpoint with return value handling
-fail_point!("my_failpoint", |value: Option<String>| {
-    match value {
-        Some(v) => {
-            println!("Got value: {}", v);
-            return Ok(v.parse().unwrap_or_default());
-        }
-        None => return Ok(42), // Default return value
-    }
-});
-
-// Conditional failpoint - only check if condition is met
-let should_fail = some_condition();
-fail_point!("conditional_failpoint", should_fail, |value: &str| {
-    println!("Conditional failpoint triggered with: {}", value);
-    return Err(anyhow::anyhow!("Simulated failure"));
-});
-```
-
-### `fail_point_with_context!` - Context-Aware Failpoint Macro
-
-The `fail_point_with_context!` macro has three variants that mirror `fail_point!` but include context:
-
-1. **Simple with context** - `fail_point_with_context!(name, context)`
-2. **With context and return handler** - `fail_point_with_context!(name, context, closure)`
-3. **Conditional with context** - `fail_point_with_context!(name, context, condition, closure)`
-
-```rust
-use neon_failpoint::{fail_point_with_context};
-use std::collections::HashMap;
-
-let mut context = HashMap::new();
-context.insert("tenant_id".to_string(), "test_123".to_string());
-context.insert("operation".to_string(), "backup".to_string());
-
-// Simple context failpoint
-fail_point_with_context!("backup_failpoint", &context);
-
-// Context failpoint with return handler
-fail_point_with_context!("backup_failpoint", &context, |value: Option<String>| {
-    match value {
-        Some(v) => return Err(anyhow::anyhow!("Backup failed: {}", v)),
-        None => return Err(anyhow::anyhow!("Backup failed")),
-    }
-});
-
-// Conditional context failpoint
-let is_test_tenant = tenant_id.starts_with("test_");
-fail_point_with_context!("backup_failpoint", &context, is_test_tenant, |value: Option<String>| {
-    // Only triggers for test tenants
-    return Err(anyhow::anyhow!("Test tenant backup failure"));
-});
-```
-
-### Other Utility Macros
-
-```rust
-use neon_failpoint::{pausable_failpoint, sleep_millis_async};
-
-// Pausable failpoint with cancellation
-let cancel_token = CancellationToken::new();
-if let Err(()) = pausable_failpoint!("pause_here", &cancel_token).await {
-    println!("Failpoint was cancelled");
-}
-
-// Sleep failpoint
-sleep_millis_async!("sleep_here", &cancel_token).await;
-
-// Context creation helper
-let mut context = HashMap::new();
-context.insert("key1".to_string(), "value1".to_string());
-context.insert("key2".to_string(), "value2".to_string());
-```
-
-### Argument Reference
-
-- **`name`**: String literal - the name of the failpoint
-- **`context`**: Expression that evaluates to `&HashMap<String, String>` - context for matching
-- **`condition`**: Boolean expression - only check failpoint if true
-- **`closure`**: Closure that handles return values:
-  - For `fail_point!` with closure: receives `Option<String>`
-  - For conditional variants: receives `&str`
-  - For `fail_point_with_context!` with closure: receives `Option<String>`
-- **`cancel`**: `&CancellationToken` - for cancellation support
-
-## Migration from `fail` crate
-
-The library provides a compatibility layer in `libs/utils/src/failpoint_support.rs`. Most existing code should work without changes, but you can migrate to the new async APIs for better performance:
-
-### Before (with `fail` crate):
-```rust
-use utils::failpoint_support::pausable_failpoint;
-
-// This used spawn_blocking internally
-pausable_failpoint!("my_failpoint", &cancel_token).await?;
-```
-
-### After (with `neon_failpoint`):
-```rust
-use neon_failpoint::{failpoint_with_cancellation, FailpointResult};
-
-// This is fully async
-match failpoint_with_cancellation("my_failpoint", None, &cancel_token).await {
-    FailpointResult::Continue => {},
-    FailpointResult::Cancelled => return Err(()),
-    FailpointResult::Return(_) => {},
-}
-```
-
-## Environment Variable Support
-
-Failpoints can be configured via the `FAILPOINTS` environment variable:
-
-```bash
-FAILPOINTS="failpoint1=return(42);failpoint2=sleep(1000);failpoint3=exit"
-```
-
-## Testing
-
-The library includes comprehensive tests and examples. Run them with:
-
-```bash
-cargo test --features testing
-cargo run --example context_demo --features testing
-```
-
-## HTTP Configuration
-
-The library integrates with the existing HTTP failpoint configuration API. Send POST requests to `/v1/failpoints` with:
-
-```json
-[
-  {
-    "name": "my_failpoint",
-    "actions": "return(42)"
-  }
-]
-```

libs/neon_failpoint/examples/context_demo.rs

@@ -1,82 +0,0 @@
-use neon_failpoint::{configure_failpoint_with_context, failpoint, FailpointResult};
-use std::collections::HashMap;
-
-#[tokio::main]
-async fn main() {
-    // Initialize tracing for better output
-    tracing_subscriber::fmt::init();
-
-    // Set up a context-specific failpoint
-    let mut context_matchers = HashMap::new();
-    context_matchers.insert("tenant_id".to_string(), "test_.*".to_string());
-    context_matchers.insert("operation".to_string(), "backup".to_string());
-
-    configure_failpoint_with_context(
-        "backup_operation",
-        "return(simulated_failure)",
-        context_matchers,
-    )
-    .unwrap();
-
-    // Test with matching context
-    let mut context = HashMap::new();
-    context.insert("tenant_id".to_string(), "test_123".to_string());
-    context.insert("operation".to_string(), "backup".to_string());
-
-    println!("Testing with matching context...");
-    match failpoint("backup_operation", Some(&context)) {
-        either::Either::Left(result) => match result {
-            FailpointResult::Return(value) => {
-                println!("Failpoint triggered with value: {value:?}");
-            }
-            FailpointResult::Continue => {
-                println!("Failpoint not triggered");
-            }
-            FailpointResult::Cancelled => {
-                println!("Failpoint cancelled");
-            }
-        },
-        either::Either::Right(future) => match future.await {
-            FailpointResult::Return(value) => {
-                println!("Failpoint triggered with value: {value:?}");
-            }
-            FailpointResult::Continue => {
-                println!("Failpoint not triggered");
-            }
-            FailpointResult::Cancelled => {
-                println!("Failpoint cancelled");
-            }
-        },
-    }
-
-    // Test with non-matching context
-    let mut context = HashMap::new();
-    context.insert("tenant_id".to_string(), "prod_456".to_string());
-    context.insert("operation".to_string(), "backup".to_string());
-
-    println!("Testing with non-matching context...");
-    match failpoint("backup_operation", Some(&context)) {
-        either::Either::Left(result) => match result {
-            FailpointResult::Return(value) => {
-                println!("Failpoint triggered with value: {value:?}");
-            }
-            FailpointResult::Continue => {
-                println!("Failpoint not triggered (expected)");
-            }
-            FailpointResult::Cancelled => {
-                println!("Failpoint cancelled");
-            }
-        },
-        either::Either::Right(future) => match future.await {
-            FailpointResult::Return(value) => {
-                println!("Failpoint triggered with value: {value:?}");
-            }
-            FailpointResult::Continue => {
-                println!("Failpoint not triggered (expected)");
-            }
-            FailpointResult::Cancelled => {
-                println!("Failpoint cancelled");
-            }
-        },
-    }
-}

libs/neon_failpoint/src/macros.rs

@@ -1,356 +0,0 @@
-//! Macros for convenient failpoint usage
-
-/// Simple failpoint macro - async version that doesn't require spawn_blocking
-#[macro_export]
-macro_rules! fail_point {
-    ($name:literal) => {{
-        if cfg!(feature = "testing") {
-            match $crate::failpoint($name, None) {
-                $crate::either::Either::Left(result) => {
-                    match result {
-                        $crate::FailpointResult::Continue => {},
-                        $crate::FailpointResult::Return(_) => {
-                            panic!("failpoint was configured with return(X) but Rust code does not pass a closure to map X to a return value");
-                        },
-                        $crate::FailpointResult::Cancelled => {},
-                    }
-                },
-                $crate::either::Either::Right(future) => {
-                    match future.await {
-                        $crate::FailpointResult::Continue => {},
-                        $crate::FailpointResult::Return(_) => {
-                            panic!("failpoint was configured with return(X) but Rust code does not pass a closure to map X to a return value");
-                        },
-                        $crate::FailpointResult::Cancelled => {},
-                    }
-                },
-            }
-        }
-    }};
-    ($name:literal, $closure:expr) => {{
-        if cfg!(feature = "testing") {
-            match $crate::failpoint($name, None) {
-                $crate::either::Either::Left(result) => {
-                    match result {
-                        $crate::FailpointResult::Continue => {},
-                        $crate::FailpointResult::Return(value) => {
-                            let closure = $closure;
-                            return closure(value);
-                        },
-                        $crate::FailpointResult::Cancelled => {},
-                    }
-                },
-                $crate::either::Either::Right(future) => {
-                    match future.await {
-                        $crate::FailpointResult::Continue => {},
-                        $crate::FailpointResult::Return(value) => {
-                            let closure = $closure;
-                            return closure(value);
-                        },
-                        $crate::FailpointResult::Cancelled => {},
-                    }
-                },
-            }
-        }
-    }};
-    ($name:literal, $condition:expr, $closure:expr) => {{
-        if cfg!(feature = "testing") {
-            if $condition {
-                match $crate::failpoint($name, None) {
-                    $crate::either::Either::Left(result) => {
-                        match result {
-                            $crate::FailpointResult::Continue => {},
-                            $crate::FailpointResult::Return(value) => {
-                                let closure = $closure;
-                                return closure(value);
-                            },
-                            $crate::FailpointResult::Cancelled => {},
-                        }
-                    },
-                    $crate::either::Either::Right(future) => {
-                        match future.await {
-                            $crate::FailpointResult::Continue => {},
-                            $crate::FailpointResult::Return(value) => {
-                                let closure = $closure;
-                                return closure(value);
-                            },
-                            $crate::FailpointResult::Cancelled => {},
-                        }
-                    },
-                }
-            }
-        }
-    }};
-}
-
-/// Simple failpoint macro - sync version that panics if async action is triggered
-#[macro_export]
-macro_rules! fail_point_sync {
-    ($name:literal) => {{
-        if cfg!(feature = "testing") {
-            match $crate::failpoint($name, None) {
-                $crate::either::Either::Left(result) => {
-                    match result {
-                        $crate::FailpointResult::Continue => {},
-                        $crate::FailpointResult::Return(_) => {
-                            panic!("failpoint was configured with return(X) but Rust code does not pass a closure to map X to a return value");
-                        },
-                        $crate::FailpointResult::Cancelled => {},
-                    }
-                },
-                $crate::either::Either::Right(_) => {
-                    panic!("failpoint '{}' triggered an async action (sleep/pause) but fail_point_sync! was used. Use fail_point! instead.", $name);
-                },
-            }
-        }
-    }};
-    ($name:literal, $closure:expr) => {{
-        if cfg!(feature = "testing") {
-            match $crate::failpoint($name, None) {
-                $crate::either::Either::Left(result) => {
-                    match result {
-                        $crate::FailpointResult::Continue => {},
-                        $crate::FailpointResult::Return(value) => {
-                            let closure = $closure;
-                            return closure(value);
-                        },
-                        $crate::FailpointResult::Cancelled => {},
-                    }
-                },
-                $crate::either::Either::Right(_) => {
-                    panic!("failpoint '{}' triggered an async action (sleep/pause) but fail_point_sync! was used. Use fail_point! instead.", $name);
-                },
-            }
-        }
-    }};
-    ($name:literal, $condition:expr, $closure:expr) => {{
-        if cfg!(feature = "testing") {
-            if $condition {
-                match $crate::failpoint($name, None) {
-                    $crate::either::Either::Left(result) => {
-                        match result {
-                            $crate::FailpointResult::Continue => {},
-                            $crate::FailpointResult::Return(value) => {
-                                let closure = $closure;
-                                return closure(value);
-                            },
-                            $crate::FailpointResult::Cancelled => {},
-                        }
-                    },
-                    $crate::either::Either::Right(_) => {
-                        panic!("failpoint '{}' triggered an async action (sleep/pause) but fail_point_sync! was used. Use fail_point! instead.", $name);
-                    },
-                }
-            }
-        }
-    }};
-}
-
-/// Failpoint macro with context support
-#[macro_export]
-macro_rules! fail_point_with_context {
-    ($name:literal, $context:expr) => {{
-        if cfg!(feature = "testing") {
-            match $crate::failpoint($name, Some($context)) {
-                $crate::either::Either::Left(result) => {
-                    match result {
-                        $crate::FailpointResult::Continue => {},
-                        $crate::FailpointResult::Return(_) => {
-                            panic!("failpoint was configured with return(X) but Rust code does not pass a closure to map X to a return value");
-                        },
-                        $crate::FailpointResult::Cancelled => {},
-                    }
-                },
-                $crate::either::Either::Right(future) => {
-                    match future.await {
-                        $crate::FailpointResult::Continue => {},
-                        $crate::FailpointResult::Return(_) => {
-                            panic!("failpoint was configured with return(X) but Rust code does not pass a closure to map X to a return value");
-                        },
-                        $crate::FailpointResult::Cancelled => {},
-                    }
-                },
-            }
-        }
-    }};
-    ($name:literal, $context:expr, $closure:expr) => {{
-        if cfg!(feature = "testing") {
-            match $crate::failpoint($name, Some($context)) {
-                $crate::either::Either::Left(result) => {
-                    match result {
-                        $crate::FailpointResult::Continue => {},
-                        $crate::FailpointResult::Return(value) => {
-                            let closure = $closure;
-                            return closure(value);
-                        },
-                        $crate::FailpointResult::Cancelled => {},
-                    }
-                },
-                $crate::either::Either::Right(future) => {
-                    match future.await {
-                        $crate::FailpointResult::Continue => {},
-                        $crate::FailpointResult::Return(value) => {
-                            let closure = $closure;
-                            return closure(value);
-                        },
-                        $crate::FailpointResult::Cancelled => {},
-                    }
-                },
-            }
-        }
-    }};
-    ($name:literal, $context:expr, $condition:expr, $closure:expr) => {{
-        if cfg!(feature = "testing") {
-            if $condition {
-                match $crate::failpoint($name, Some($context)) {
-                    $crate::either::Either::Left(result) => {
-                        match result {
-                            $crate::FailpointResult::Continue => {},
-                            $crate::FailpointResult::Return(value) => {
-                                let closure = $closure;
-                                return closure(value);
-                            },
-                            $crate::FailpointResult::Cancelled => {},
-                        }
-                    },
-                    $crate::either::Either::Right(future) => {
-                        match future.await {
-                            $crate::FailpointResult::Continue => {},
-                            $crate::FailpointResult::Return(value) => {
-                                let closure = $closure;
-                                return closure(value);
-                            },
-                            $crate::FailpointResult::Cancelled => {},
-                        }
-                    },
-                }
-            }
-        }
-    }};
-}
-
-/// Failpoint macro with context support - sync version
-#[macro_export]
-macro_rules! fail_point_with_context_sync {
-    ($name:literal, $context:expr) => {{
-        if cfg!(feature = "testing") {
-            match $crate::failpoint($name, Some($context)) {
-                $crate::either::Either::Left(result) => {
-                    match result {
-                        $crate::FailpointResult::Continue => {},
-                        $crate::FailpointResult::Return(_) => {
-                            panic!("failpoint was configured with return(X) but Rust code does not pass a closure to map X to a return value");
-                        },
-                        $crate::FailpointResult::Cancelled => {},
-                    }
-                },
-                $crate::either::Either::Right(_) => {
-                    panic!("failpoint '{}' triggered an async action (sleep/pause) but fail_point_with_context_sync! was used. Use fail_point_with_context! instead.", $name);
-                },
-            }
-        }
-    }};
-    ($name:literal, $context:expr, $closure:expr) => {{
-        if cfg!(feature = "testing") {
-            match $crate::failpoint($name, Some($context)) {
-                $crate::either::Either::Left(result) => {
-                    match result {
-                        $crate::FailpointResult::Continue => {},
-                        $crate::FailpointResult::Return(value) => {
-                            let closure = $closure;
-                            return closure(value);
-                        },
-                        $crate::FailpointResult::Cancelled => {},
-                    }
-                },
-                $crate::either::Either::Right(_) => {
-                    panic!("failpoint '{}' triggered an async action (sleep/pause) but fail_point_with_context_sync! was used. Use fail_point_with_context! instead.", $name);
-                },
-            }
-        }
-    }};
-    ($name:literal, $context:expr, $condition:expr, $closure:expr) => {{
-        if cfg!(feature = "testing") {
-            if $condition {
-                match $crate::failpoint($name, Some($context)) {
-                    $crate::either::Either::Left(result) => {
-                        match result {
-                            $crate::FailpointResult::Continue => {},
-                            $crate::FailpointResult::Return(value) => {
-                                let closure = $closure;
-                                return closure(value);
-                            },
-                            $crate::FailpointResult::Cancelled => {},
-                        }
-                    },
-                    $crate::either::Either::Right(_) => {
-                        panic!("failpoint '{}' triggered an async action (sleep/pause) but fail_point_with_context_sync! was used. Use fail_point_with_context! instead.", $name);
-                    },
-                }
-            }
-        }
-    }};
-}
-
-/// Pausable failpoint macro - equivalent to the old pausable_failpoint
-#[macro_export]
-macro_rules! pausable_failpoint {
-    ($name:literal) => {{
-        if cfg!(feature = "testing") {
-            let cancel = ::tokio_util::sync::CancellationToken::new();
-            let _ = $crate::pausable_failpoint!($name, &cancel);
-        }
-    }};
-    ($name:literal, $cancel:expr) => {{
-        if cfg!(feature = "testing") {
-            ::tracing::info!("at failpoint {}", $name); // tests rely on this
-            match $crate::failpoint_with_cancellation($name, None, $cancel) {
-                $crate::either::Either::Left(result) => match result {
-                    $crate::FailpointResult::Continue => Ok(()),
-                    $crate::FailpointResult::Return(_) => Ok(()),
-                    $crate::FailpointResult::Cancelled => Err(()),
-                },
-                $crate::either::Either::Right(future) => match future.await {
-                    $crate::FailpointResult::Continue => Ok(()),
-                    $crate::FailpointResult::Return(_) => Ok(()),
-                    $crate::FailpointResult::Cancelled => Err(()),
-                },
-            }
-        } else {
-            Ok(())
-        }
-    }};
-}
-
-/// Sleep failpoint macro - for async sleep operations
-#[macro_export]
-macro_rules! sleep_millis_async {
-    ($name:literal) => {{
-        if cfg!(feature = "testing") {
-            match $crate::failpoint($name, None) {
-                $crate::either::Either::Left(_) => {}
-                $crate::either::Either::Right(future) => {
-                    future.await;
-                }
-            }
-        }
-    }};
-    ($name:literal, $cancel:expr) => {{
-        if cfg!(feature = "testing") {
-            match $crate::failpoint_with_cancellation($name, None, $cancel) {
-                $crate::either::Either::Left(_) => {}
-                $crate::either::Either::Right(future) => {
-                    future.await;
-                }
-            }
-        }
-    }};
-}
-
-// Re-export for convenience
-pub use fail_point;
-pub use fail_point_sync;
-pub use fail_point_with_context;
-pub use fail_point_with_context_sync;
-pub use pausable_failpoint;
-pub use sleep_millis_async;

libs/pageserver_api/src/controller_api.rs

@@ -10,7 +10,7 @@ use serde::{Deserialize, Serialize};
 use utils::id::{NodeId, TenantId, TimelineId};
 use utils::lsn::Lsn;
 
-use crate::models::{PageserverUtilization, ShardParameters, TenantConfig};
+use crate::models::{PageserverUtilization, ShardParameters, TenantConfig, TimelineInfo};
 use crate::shard::{ShardStripeSize, TenantShardId};
 
 #[derive(Serialize, Deserialize, Debug)]
@@ -126,6 +126,13 @@ pub struct TenantDescribeResponse {
     pub config: TenantConfig,
 }
 
+#[derive(Serialize, Deserialize, Debug)]
+pub struct TenantTimelineDescribeResponse {
+    pub shards: Vec<TimelineInfo>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub image_consistent_lsn: Option<Lsn>,
+}
+
 #[derive(Serialize, Deserialize, Debug)]
 pub struct NodeShardResponse {
     pub node_id: NodeId,

libs/pageserver_api/src/models.rs

@@ -1622,6 +1622,9 @@ pub struct TimelineInfo {
 
     /// Whether the timeline is invisible in synthetic size calculations.
     pub is_invisible: Option<bool>,
+    // HADRON: the largest LSN below which all page updates have been included in the image layers.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub image_consistent_lsn: Option<Lsn>,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]

libs/safekeeper_api/src/models.rs

@@ -11,7 +11,7 @@ use utils::id::{NodeId, TenantId, TenantTimelineId, TimelineId};
 use utils::lsn::Lsn;
 use utils::pageserver_feedback::PageserverFeedback;
 
-use crate::membership::Configuration;
+use crate::membership::{Configuration, SafekeeperGeneration};
 use crate::{ServerInfo, Term};
 
 #[derive(Debug, Serialize, Deserialize)]
@@ -311,3 +311,12 @@ pub struct PullTimelineResponse {
     pub safekeeper_host: Option<String>,
     // TODO: add more fields?
 }
+
+/// Response to a timeline locate request.
+/// Storcon-only API.
+#[derive(Serialize, Deserialize, Clone, Debug)]
+pub struct TimelineLocateResponse {
+    pub generation: SafekeeperGeneration,
+    pub sk_set: Vec<NodeId>,
+    pub new_sk_set: Option<Vec<NodeId>>,
+}

libs/utils/Cargo.toml

@@ -9,7 +9,7 @@ default = ["rename_noreplace"]
 rename_noreplace = []
 # Enables test-only APIs, incuding failpoints. In particular, enables the `fail_point!` macro,
 # which adds some runtime cost to run tests on outage conditions
-testing = ["neon_failpoint/testing"]
+testing = ["fail/failpoints"]
 
 [dependencies]
 arc-swap.workspace = true
@@ -21,11 +21,10 @@ bytes.workspace = true
 camino.workspace = true
 chrono.workspace = true
 diatomic-waker.workspace = true
-either.workspace = true
 git-version.workspace = true
 hex = { workspace = true, features = ["serde"] }
 humantime.workspace = true
-neon_failpoint.workspace = true
+fail.workspace = true
 futures = { workspace = true }
 jsonwebtoken.workspace = true
 nix = { workspace = true, features = ["ioctl"] }

libs/utils/src/failpoint_support.rs

@@ -1,22 +1,59 @@
 //! Failpoint support code shared between pageserver and safekeepers.
-//!
-//! This module provides a compatibility layer over the new neon_failpoint crate.
 
-pub use neon_failpoint::{configure_failpoint as apply_failpoint, has_failpoints, init};
 use tokio_util::sync::CancellationToken;
 
-/// Mere forward to neon_failpoint::pausable_failpoint
+/// Declare a failpoint that can use to `pause` failpoint action.
+/// We don't want to block the executor thread, hence, spawn_blocking + await.
+///
+/// Optionally pass a cancellation token, and this failpoint will drop out of
+/// its pause when the cancellation token fires. This is useful for testing
+/// cases where we would like to block something, but test its clean shutdown behavior.
+/// The macro evaluates to a Result in that case, where Ok(()) is the case
+/// where the failpoint was not paused, and Err() is the case where cancellation
+/// token fired while evaluating the failpoint.
+///
+/// Remember to unpause the failpoint in the test; until that happens, one of the
+/// limited number of spawn_blocking thread pool threads is leaked.
 #[macro_export]
 macro_rules! pausable_failpoint {
-    ($name:literal) => {
-        ::neon_failpoint::pausable_failpoint!($name)
-    };
-    ($name:literal, $cancel:expr) => {
-        ::neon_failpoint::pausable_failpoint!($name, $cancel)
-    };
+    ($name:literal) => {{
+        if cfg!(feature = "testing") {
+            let cancel = ::tokio_util::sync::CancellationToken::new();
+            let _ = $crate::pausable_failpoint!($name, &cancel);
+        }
+    }};
+    ($name:literal, $cancel:expr) => {{
+        if cfg!(feature = "testing") {
+            let failpoint_fut = ::tokio::task::spawn_blocking({
+                let current = ::tracing::Span::current();
+                move || {
+                    let _entered = current.entered();
+                    ::tracing::info!("at failpoint {}", $name);
+                    ::fail::fail_point!($name);
+                }
+            });
+            let cancel_fut = async move {
+                $cancel.cancelled().await;
+            };
+            ::tokio::select! {
+                res = failpoint_fut => {
+                    res.expect("spawn_blocking");
+                    // continue with execution
+                    Ok(())
+                },
+                _ = cancel_fut => {
+                    Err(())
+                }
+            }
+        } else {
+            Ok(())
+        }
+    }};
 }
 
-/// DEPRECATED! - use with fail::cfg("$name", "return(2000)")
+pub use pausable_failpoint;
+
+/// use with fail::cfg("$name", "return(2000)")
 ///
 /// The effect is similar to a "sleep(2000)" action, i.e. we sleep for the
 /// specified time (in milliseconds). The main difference is that we use async
@@ -32,7 +69,7 @@ macro_rules! __failpoint_sleep_millis_async {
         // If the failpoint is used with a "return" action, set should_sleep to the
         // returned value (as string). Otherwise it's set to None.
         let should_sleep = (|| {
-            ::neon_failpoint::fail_point_sync!($name, |x| x);
+            ::fail::fail_point!($name, |x| x);
             ::std::option::Option::None
         })();
 
@@ -45,7 +82,7 @@ macro_rules! __failpoint_sleep_millis_async {
         // If the failpoint is used with a "return" action, set should_sleep to the
         // returned value (as string). Otherwise it's set to None.
         let should_sleep = (|| {
-            ::neon_failpoint::fail_point_sync!($name, |x| x);
+            ::fail::fail_point!($name, |x| x);
             ::std::option::Option::None
         })();
 
@@ -89,3 +126,60 @@ pub async fn failpoint_sleep_cancellable_helper(
     tokio::time::timeout(d, cancel.cancelled()).await.ok();
     tracing::info!("failpoint {:?}: sleep done", name);
 }
+
+/// Initialize the configured failpoints
+///
+/// You must call this function before any concurrent threads do operations.
+pub fn init() -> fail::FailScenario<'static> {
+    // The failpoints lib provides support for parsing the `FAILPOINTS` env var.
+    // We want non-default behavior for `exit`, though, so, we handle it separately.
+    //
+    // Format for FAILPOINTS is "name=actions" separated by ";".
+    let actions = std::env::var("FAILPOINTS");
+    if actions.is_ok() {
+        // SAFETY: this function should before any threads start and access env vars concurrently
+        unsafe {
+            std::env::remove_var("FAILPOINTS");
+        }
+    } else {
+        // let the library handle non-utf8, or nothing for not present
+    }
+
+    let scenario = fail::FailScenario::setup();
+
+    if let Ok(val) = actions {
+        val.split(';')
+            .enumerate()
+            .map(|(i, s)| s.split_once('=').ok_or((i, s)))
+            .for_each(|res| {
+                let (name, actions) = match res {
+                    Ok(t) => t,
+                    Err((i, s)) => {
+                        panic!(
+                            "startup failpoints: missing action on the {}th failpoint; try `{s}=return`",
+                            i + 1,
+                        );
+                    }
+                };
+                if let Err(e) = apply_failpoint(name, actions) {
+                    panic!("startup failpoints: failed to apply failpoint {name}={actions}: {e}");
+                }
+            });
+    }
+
+    scenario
+}
+
+pub fn apply_failpoint(name: &str, actions: &str) -> Result<(), String> {
+    if actions == "exit" {
+        fail::cfg_callback(name, exit_failpoint)
+    } else {
+        fail::cfg(name, actions)
+    }
+}
+
+#[inline(never)]
+fn exit_failpoint() {
+    tracing::info!("Exit requested by failpoint");
+    std::process::exit(1);
+}

pageserver/Cargo.toml

@@ -8,7 +8,7 @@ license.workspace = true
 default = []
 # Enables test-only APIs, incuding failpoints. In particular, enables the `fail_point!` macro,
 # which adds some runtime cost to run tests on outage conditions
-testing = ["neon_failpoint/testing", "pageserver_api/testing", "wal_decoder/testing", "pageserver_client/testing"]
+testing = ["fail/failpoints", "pageserver_api/testing", "wal_decoder/testing", "pageserver_client/testing"]
 
 fuzz-read-path = ["testing"]
 
@@ -33,7 +33,7 @@ crc32c.workspace = true
 either.workspace = true
 enum-map.workspace = true
 enumset = { workspace = true, features = ["serde"]}
-neon_failpoint.workspace = true
+fail.workspace = true
 futures.workspace = true
 hashlink.workspace = true
 hex.workspace = true

pageserver/src/basebackup.rs

@@ -17,7 +17,6 @@ use anyhow::{Context, anyhow};
 use async_compression::tokio::write::GzipEncoder;
 use bytes::{BufMut, Bytes, BytesMut};
 use fail::fail_point;
-use neon_failpoint as fail;
 use pageserver_api::key::{Key, rel_block_to_key};
 use pageserver_api::reltag::{RelTag, SlruKind};
 use postgres_ffi::pg_constants::{PG_HBA, PGDATA_SPECIAL_FILES};

pageserver/src/bin/pageserver.rs

@@ -68,7 +68,7 @@ const FEATURES: &[&str] = &[
 fn version() -> String {
     format!(
         "{GIT_VERSION} failpoints: {}, features: {:?}",
-        neon_failpoint::has_failpoints(),
+        fail::has_failpoints(),
         FEATURES,
     )
 }
@@ -84,7 +84,7 @@ fn main() -> anyhow::Result<()> {
     }
 
     // Initialize up failpoints support
-    failpoint_support::init().unwrap();
+    let scenario = failpoint_support::init();
 
     let workdir = arg_matches
         .get_one::<String>("workdir")
@@ -221,6 +221,7 @@ fn main() -> anyhow::Result<()> {
 
     start_pageserver(launch_ts, conf, ignored, otel_guard).context("Failed to start pageserver")?;
 
+    scenario.teardown();
     Ok(())
 }
 
@@ -365,9 +366,16 @@ fn start_pageserver(
 
     // If any failpoints were set from FAILPOINTS environment variable,
     // print them to the log for debugging purposes
-    let failpoints = neon_failpoint::list();
-    for (name, actions) in failpoints {
-        info!("starting with failpoint: {name} {actions}");
+    let failpoints = fail::list();
+    if !failpoints.is_empty() {
+        info!(
+            "started with failpoints: {}",
+            failpoints
+                .iter()
+                .map(|(name, actions)| format!("{name}={actions}"))
+                .collect::<Vec<String>>()
+                .join(";")
+        )
     }
 
     // Create and lock PID file. This ensures that there cannot be more than one

pageserver/src/consumption_metrics/disk_cache.rs

@@ -6,8 +6,6 @@ use camino::{Utf8Path, Utf8PathBuf};
 use super::{NewMetricsRoot, NewRawMetric, RawMetric};
 use crate::consumption_metrics::NewMetricsRefRoot;
 
-use neon_failpoint as fail;
-
 pub(super) fn read_metrics_from_serde_value(
     json_value: serde_json::Value,
 ) -> anyhow::Result<Vec<NewRawMetric>> {
@@ -131,7 +129,7 @@ pub(super) async fn flush_metrics_to_disk(
             tempfile.flush()?;
             tempfile.as_file().sync_all()?;
 
-            fail::fail_point_sync!("before-persist-last-metrics-collected");
+            fail::fail_point!("before-persist-last-metrics-collected");
 
             drop(tempfile.persist(&*path).map_err(|e| e.error)?);
 

pageserver/src/deletion_queue/deleter.rs

@@ -8,7 +8,6 @@
 
 use std::time::Duration;
 
-use neon_failpoint as fail;
 use remote_storage::{GenericRemoteStorage, RemotePath, TimeoutOrCancel};
 use tokio_util::sync::CancellationToken;
 use tracing::{info, warn};

pageserver/src/http/routes.rs

@@ -28,7 +28,6 @@ use http_utils::{RequestExt, RouterBuilder};
 use humantime::format_rfc3339;
 use hyper::{Body, Request, Response, StatusCode, Uri, header};
 use metrics::launch_timestamp::LaunchTimestamp;
-use neon_failpoint as fail;
 use pageserver_api::models::virtual_file::IoMode;
 use pageserver_api::models::{
     DetachBehavior, DownloadRemoteLayersTaskSpawnRequest, IngestAuxFilesRequest,
@@ -398,6 +397,7 @@ async fn build_timeline_info(
     timeline: &Arc<Timeline>,
     include_non_incremental_logical_size: bool,
     force_await_initial_logical_size: bool,
+    include_image_consistent_lsn: bool,
     ctx: &RequestContext,
 ) -> anyhow::Result<TimelineInfo> {
     crate::tenant::debug_assert_current_span_has_tenant_and_timeline_id();
@@ -422,6 +422,10 @@ async fn build_timeline_info(
                 .await?,
         );
     }
+    // HADRON
+    if include_image_consistent_lsn {
+        info.image_consistent_lsn = Some(timeline.compute_image_consistent_lsn().await?);
+    }
     Ok(info)
 }
 
@@ -511,6 +515,8 @@ async fn build_timeline_info_common(
         is_invisible: Some(is_invisible),
 
         walreceiver_status,
+        // HADRON
+        image_consistent_lsn: None,
     };
     Ok(info)
 }
@@ -713,6 +719,8 @@ async fn timeline_list_handler(
         parse_query_param(&request, "include-non-incremental-logical-size")?;
     let force_await_initial_logical_size: Option<bool> =
         parse_query_param(&request, "force-await-initial-logical-size")?;
+    let include_image_consistent_lsn: Option<bool> =
+        parse_query_param(&request, "include-image-consistent-lsn")?;
     check_permission(&request, Some(tenant_shard_id.tenant_id))?;
 
     let state = get_state(&request);
@@ -733,6 +741,7 @@ async fn timeline_list_handler(
                 &timeline,
                 include_non_incremental_logical_size.unwrap_or(false),
                 force_await_initial_logical_size.unwrap_or(false),
+                include_image_consistent_lsn.unwrap_or(false),
                 &ctx,
             )
             .instrument(info_span!("build_timeline_info", timeline_id = %timeline.timeline_id))
@@ -761,6 +770,9 @@ async fn timeline_and_offloaded_list_handler(
         parse_query_param(&request, "include-non-incremental-logical-size")?;
     let force_await_initial_logical_size: Option<bool> =
         parse_query_param(&request, "force-await-initial-logical-size")?;
+    let include_image_consistent_lsn: Option<bool> =
+        parse_query_param(&request, "include-image-consistent-lsn")?;
+
     check_permission(&request, Some(tenant_shard_id.tenant_id))?;
 
     let state = get_state(&request);
@@ -781,6 +793,7 @@ async fn timeline_and_offloaded_list_handler(
                 &timeline,
                 include_non_incremental_logical_size.unwrap_or(false),
                 force_await_initial_logical_size.unwrap_or(false),
+                include_image_consistent_lsn.unwrap_or(false),
                 &ctx,
             )
             .instrument(info_span!("build_timeline_info", timeline_id = %timeline.timeline_id))
@@ -965,6 +978,9 @@ async fn timeline_detail_handler(
         parse_query_param(&request, "include-non-incremental-logical-size")?;
     let force_await_initial_logical_size: Option<bool> =
         parse_query_param(&request, "force-await-initial-logical-size")?;
+    // HADRON
+    let include_image_consistent_lsn: Option<bool> =
+        parse_query_param(&request, "include-image-consistent-lsn")?;
     check_permission(&request, Some(tenant_shard_id.tenant_id))?;
 
     // Logical size calculation needs downloading.
@@ -985,6 +1001,7 @@ async fn timeline_detail_handler(
             &timeline,
             include_non_incremental_logical_size.unwrap_or(false),
             force_await_initial_logical_size.unwrap_or(false),
+            include_image_consistent_lsn.unwrap_or(false),
             ctx,
         )
         .await
@@ -3644,6 +3661,7 @@ async fn activate_post_import_handler(
         let timeline_info = build_timeline_info(
             &timeline, false, // include_non_incremental_logical_size,
             false, // force_await_initial_logical_size
+            false, // include_image_consistent_lsn
             &ctx,
         )
         .await
@@ -3976,7 +3994,7 @@ pub fn make_router(
         .get("/profile/cpu", |r| request_span(r, profile_cpu_handler))
         .get("/profile/heap", |r| request_span(r, profile_heap_handler))
         .get("/v1/status", |r| api_handler(r, status_handler))
-        .post("/v1/failpoints", |r| {
+        .put("/v1/failpoints", |r| {
             testing_api_handler("manage failpoints", r, failpoints_handler)
         })
         .post("/v1/reload_auth_validation_keys", |r| {

pageserver/src/page_service.rs

@@ -19,7 +19,6 @@ use futures::future::BoxFuture;
 use futures::{FutureExt, Stream};
 use itertools::Itertools;
 use jsonwebtoken::TokenData;
-use neon_failpoint as fail;
 use once_cell::sync::OnceCell;
 use pageserver_api::config::{
     GetVectoredConcurrentIo, PageServicePipeliningConfig, PageServicePipeliningConfigPipelined,
@@ -336,21 +335,18 @@ async fn page_service_conn_main(
 
     let default_timeout_ms = 10 * 60 * 1000; // 10 minutes by default
     let socket_timeout_ms = (|| {
-        fail::fail_point_sync!(
-            "simulated-bad-compute-connection",
-            |avg_timeout_ms: Option<String>| {
-                // Exponential distribution for simulating
-                // poor network conditions, expect about avg_timeout_ms to be around 15
-                // in tests
-                if let Some(avg_timeout_ms) = avg_timeout_ms {
-                    let avg = avg_timeout_ms.parse::<i64>().unwrap() as f32;
-                    let u = rand::random::<f32>();
-                    ((1.0 - u).ln() / (-avg)) as u64
-                } else {
-                    default_timeout_ms
-                }
+        fail::fail_point!("simulated-bad-compute-connection", |avg_timeout_ms| {
+            // Exponential distribution for simulating
+            // poor network conditions, expect about avg_timeout_ms to be around 15
+            // in tests
+            if let Some(avg_timeout_ms) = avg_timeout_ms {
+                let avg = avg_timeout_ms.parse::<i64>().unwrap() as f32;
+                let u = rand::random::<f32>();
+                ((1.0 - u).ln() / (-avg)) as u64
+            } else {
+                default_timeout_ms
             }
-        );
+        });
         default_timeout_ms
     })();
 
@@ -3047,7 +3043,7 @@ where
         _pgb: &mut PostgresBackend<IO>,
         sm: &FeStartupPacket,
     ) -> Result<(), QueryError> {
-        fail::fail_point_sync!("ps::connection-start::startup-packet");
+        fail::fail_point!("ps::connection-start::startup-packet");
 
         if let FeStartupPacket::StartupMessage { params, .. } = sm {
             if let Some(app_name) = params.get("application_name") {

pageserver/src/pgdatadir_mapping.rs

@@ -14,7 +14,6 @@ use crate::{PERF_TRACE_TARGET, ensure_walingest};
 use anyhow::Context;
 use bytes::{Buf, Bytes, BytesMut};
 use enum_map::Enum;
-use neon_failpoint as fail;
 use pageserver_api::key::{
     AUX_FILES_KEY, CHECKPOINT_KEY, CONTROLFILE_KEY, CompactKey, DBDIR_KEY, Key, RelDirExists,
     TWOPHASEDIR_KEY, dbdir_key_range, rel_block_to_key, rel_dir_to_key, rel_key_range,

pageserver/src/tenant.rs

@@ -30,7 +30,6 @@ use enumset::EnumSet;
 use futures::StreamExt;
 use futures::stream::FuturesUnordered;
 use itertools::Itertools as _;
-use neon_failpoint as fail;
 use once_cell::sync::Lazy;
 pub use pageserver_api::models::TenantState;
 use pageserver_api::models::{self, RelSizeMigration};
@@ -9572,7 +9571,7 @@ mod tests {
         writer.finish_write(Lsn(0x30));
         drop(writer);
 
-        neon_failpoint::configure_failpoint(
+        fail::cfg(
             "flush-layer-before-update-remote-consistent-lsn",
             "return()",
         )
@@ -12817,6 +12816,40 @@ mod tests {
                 },
             ]
         );
+
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_get_force_image_creation_lsn() -> anyhow::Result<()> {
+        let tenant_conf = pageserver_api::models::TenantConfig {
+            pitr_interval: Some(Duration::from_secs(7 * 3600)),
+            image_layer_force_creation_period: Some(Duration::from_secs(3600)),
+            ..Default::default()
+        };
+
+        let tenant_id = TenantId::generate();
+
+        let harness = TenantHarness::create_custom(
+            "test_get_force_image_creation_lsn",
+            tenant_conf,
+            tenant_id,
+            ShardIdentity::unsharded(),
+            Generation::new(1),
+        )
+        .await?;
+        let (tenant, ctx) = harness.load().await;
+        let timeline = tenant
+            .create_test_timeline(TIMELINE_ID, Lsn(0x10), DEFAULT_PG_VERSION, &ctx)
+            .await?;
+        timeline.gc_info.write().unwrap().cutoffs.time = Some(Lsn(100));
+        {
+            let writer = timeline.writer().await;
+            writer.finish_write(Lsn(5000));
+        }
+
+        let image_creation_lsn = timeline.get_force_image_creation_lsn().unwrap();
+        assert_eq!(image_creation_lsn, Lsn(4300));
         Ok(())
     }
 }

pageserver/src/tenant/layer_map.rs

@@ -46,10 +46,11 @@
 mod historic_layer_coverage;
 mod layer_coverage;
 
-use std::collections::{HashMap, VecDeque};
+use std::collections::{BTreeMap, HashMap, VecDeque};
 use std::iter::Peekable;
 use std::ops::Range;
 use std::sync::Arc;
+use std::time::Instant;
 
 use anyhow::Result;
 use historic_layer_coverage::BufferedHistoricLayerCoverage;
@@ -904,6 +905,103 @@ impl LayerMap {
         max_stacked_deltas
     }
 
+    /* BEGIN_HADRON */
+    /**
+     * Compute the image consistent LSN, the largest LSN below which all pages have been redone successfully.
+     * It works by first finding the latest image layers and store them into a map. Then for each delta layer,
+     * find all overlapping image layers in order to potentially increase the image LSN in case there are gaps
+     * (e.g., if an image is created at LSN 100 but the delta layer spans LSN [150, 200], then we can increase
+     * image LSN to 150 because there is no WAL record in between).
+     * Finally, the image consistent LSN is computed by taking the minimum of all image layers.
+     */
+    pub fn compute_image_consistent_lsn(&self, disk_consistent_lsn: Lsn) -> Lsn {
+        struct ImageLayerInfo {
+            // creation LSN of the image layer
+            image_lsn: Lsn,
+            // the current minimum LSN of newer delta layers with overlapping key ranges
+            min_delta_lsn: Lsn,
+        }
+        let started_at = Instant::now();
+
+        let min_l0_deltas_lsn = {
+            let l0_deltas = self.level0_deltas();
+            l0_deltas
+                .iter()
+                .map(|layer| layer.get_lsn_range().start)
+                .min()
+                .unwrap_or(disk_consistent_lsn)
+        };
+        let global_key_range = Key::MIN..Key::MAX;
+
+        // step 1: collect all most recent image layers into a map
+        // map: end key to image_layer_info
+        let mut image_map: BTreeMap<Key, ImageLayerInfo> = BTreeMap::new();
+        for (img_range, img) in self.image_coverage(&global_key_range, disk_consistent_lsn) {
+            let img_lsn = img.map(|layer| layer.get_lsn_range().end).unwrap_or(Lsn(0));
+            image_map.insert(
+                img_range.end,
+                ImageLayerInfo {
+                    image_lsn: img_lsn,
+                    min_delta_lsn: min_l0_deltas_lsn,
+                },
+            );
+        }
+
+        // step 2: go through all delta layers, and update the image layer info with overlapping
+        // key ranges
+        for layer in self.historic.iter() {
+            if !layer.is_delta {
+                continue;
+            }
+            let delta_key_range = layer.get_key_range();
+            let delta_lsn_range = layer.get_lsn_range();
+            for (img_end_key, img_info) in image_map.range_mut(delta_key_range.start..Key::MAX) {
+                debug_assert!(img_end_key >= &delta_key_range.start);
+                if delta_lsn_range.end > img_info.image_lsn {
+                    // the delta layer includes WAL records after the image
+                    // it's possibel that the delta layer's start LSN < image LSN, which will be simply ignored by step 3
+                    img_info.min_delta_lsn =
+                        std::cmp::min(img_info.min_delta_lsn, delta_lsn_range.start);
+                }
+                if img_end_key >= &delta_key_range.end {
+                    // we have fully processed all overlapping image layers
+                    break;
+                }
+            }
+        }
+
+        // step 3, go through all image layers and find the image consistent LSN
+        let mut img_consistent_lsn = min_l0_deltas_lsn.checked_sub(Lsn(1)).unwrap();
+        let mut prev_key = Key::MIN;
+        for (img_key, img_info) in image_map {
+            tracing::debug!(
+                "Image layer {:?}:{} has min delta lsn {}",
+                Range {
+                    start: prev_key,
+                    end: img_key,
+                },
+                img_info.image_lsn,
+                img_info.min_delta_lsn,
+            );
+            let image_lsn = std::cmp::max(
+                img_info.image_lsn,
+                img_info.min_delta_lsn.checked_sub(Lsn(1)).unwrap_or(Lsn(0)),
+            );
+            img_consistent_lsn = std::cmp::min(img_consistent_lsn, image_lsn);
+            prev_key = img_key;
+        }
+        tracing::info!(
+            "computed image_consistent_lsn {} for disk_consistent_lsn {} in {}ms. Processed {} layrs in total.",
+            img_consistent_lsn,
+            disk_consistent_lsn,
+            started_at.elapsed().as_millis(),
+            self.historic.len()
+        );
+        img_consistent_lsn
+    }
+
+    /* END_HADRON */
+
     /// Return all L0 delta layers
     pub fn level0_deltas(&self) -> &Vec<Arc<PersistentLayerDesc>> {
         &self.l0_delta_layers
@@ -1579,6 +1677,138 @@ mod tests {
             LayerVisibilityHint::Visible
         ));
     }
+
+    /* BEGIN_HADRON */
+    #[test]
+    fn test_compute_image_consistent_lsn() {
+        let mut layer_map = LayerMap::default();
+
+        let disk_consistent_lsn = Lsn(1000);
+        // case 1: empty layer map
+        let image_consistent_lsn = layer_map.compute_image_consistent_lsn(disk_consistent_lsn);
+        assert_eq!(
+            disk_consistent_lsn.checked_sub(Lsn(1)).unwrap(),
+            image_consistent_lsn
+        );
+
+        // case 2: only L0 delta layer
+        {
+            let mut updates = layer_map.batch_update();
+            updates.insert_historic(PersistentLayerDesc::new_test(
+                Key::from_i128(0)..Key::from_i128(100),
+                Lsn(900)..Lsn(990),
+                true,
+            ));
+
+            updates.insert_historic(PersistentLayerDesc::new_test(
+                Key::from_i128(0)..Key::from_i128(100),
+                Lsn(850)..Lsn(899),
+                true,
+            ));
+        }
+
+        // should use min L0 delta LSN - 1 as image consistent LSN
+        let image_consistent_lsn = layer_map.compute_image_consistent_lsn(disk_consistent_lsn);
+        assert_eq!(Lsn(849), image_consistent_lsn);
+
+        // case 3: 3 images, no L1 delta
+        {
+            let mut updates = layer_map.batch_update();
+            updates.insert_historic(PersistentLayerDesc::new_test(
+                Key::from_i128(0)..Key::from_i128(40),
+                Lsn(100)..Lsn(100),
+                false,
+            ));
+
+            updates.insert_historic(PersistentLayerDesc::new_test(
+                Key::from_i128(40)..Key::from_i128(70),
+                Lsn(200)..Lsn(200),
+                false,
+            ));
+
+            updates.insert_historic(PersistentLayerDesc::new_test(
+                Key::from_i128(70)..Key::from_i128(100),
+                Lsn(150)..Lsn(150),
+                false,
+            ));
+        }
+        // should use min L0 delta LSN - 1 as image consistent LSN
+        let image_consistent_lsn = layer_map.compute_image_consistent_lsn(disk_consistent_lsn);
+        assert_eq!(Lsn(849), image_consistent_lsn);
+
+        // case 4: 3 images with 1 L1 delta
+        {
+            let mut updates = layer_map.batch_update();
+            updates.insert_historic(PersistentLayerDesc::new_test(
+                Key::from_i128(0)..Key::from_i128(50),
+                Lsn(300)..Lsn(350),
+                true,
+            ));
+        }
+        let image_consistent_lsn = layer_map.compute_image_consistent_lsn(disk_consistent_lsn);
+        assert_eq!(Lsn(299), image_consistent_lsn);
+
+        // case 5: 3 images with 1 more L1 delta with smaller LSN
+        {
+            let mut updates = layer_map.batch_update();
+            updates.insert_historic(PersistentLayerDesc::new_test(
+                Key::from_i128(50)..Key::from_i128(72),
+                Lsn(200)..Lsn(300),
+                true,
+            ));
+        }
+        let image_consistent_lsn = layer_map.compute_image_consistent_lsn(disk_consistent_lsn);
+        assert_eq!(Lsn(199), image_consistent_lsn);
+
+        // case 6: 3 images with more newer L1 deltas (no impact on final results)
+        {
+            let mut updates = layer_map.batch_update();
+            updates.insert_historic(PersistentLayerDesc::new_test(
+                Key::from_i128(0)..Key::from_i128(30),
+                Lsn(400)..Lsn(500),
+                true,
+            ));
+            updates.insert_historic(PersistentLayerDesc::new_test(
+                Key::from_i128(35)..Key::from_i128(100),
+                Lsn(450)..Lsn(600),
+                true,
+            ));
+        }
+        let image_consistent_lsn = layer_map.compute_image_consistent_lsn(disk_consistent_lsn);
+        assert_eq!(Lsn(199), image_consistent_lsn);
+
+        // case 7: 3 images with more older L1 deltas (no impact on final results)
+        {
+            let mut updates = layer_map.batch_update();
+            updates.insert_historic(PersistentLayerDesc::new_test(
+                Key::from_i128(0)..Key::from_i128(40),
+                Lsn(0)..Lsn(50),
+                true,
+            ));
+
+            updates.insert_historic(PersistentLayerDesc::new_test(
+                Key::from_i128(50)..Key::from_i128(100),
+                Lsn(10)..Lsn(60),
+                true,
+            ));
+        }
+        let image_consistent_lsn = layer_map.compute_image_consistent_lsn(disk_consistent_lsn);
+        assert_eq!(Lsn(199), image_consistent_lsn);
+
+        // case 8: 3 images with one more L1 delta with overlapping LSN range
+        {
+            let mut updates = layer_map.batch_update();
+            updates.insert_historic(PersistentLayerDesc::new_test(
+                Key::from_i128(0)..Key::from_i128(50),
+                Lsn(50)..Lsn(250),
+                true,
+            ));
+        }
+        let image_consistent_lsn = layer_map.compute_image_consistent_lsn(disk_consistent_lsn);
+        assert_eq!(Lsn(100), image_consistent_lsn);
+    }
+
+    /* END_HADRON */
 }
 
 #[cfg(test)]

pageserver/src/tenant/mgr.rs

@@ -12,7 +12,6 @@ use anyhow::Context;
 use camino::{Utf8DirEntry, Utf8Path, Utf8PathBuf};
 use futures::StreamExt;
 use itertools::Itertools;
-use neon_failpoint as fail;
 use pageserver_api::key::Key;
 use pageserver_api::models::{DetachBehavior, LocationConfigMode};
 use pageserver_api::shard::{

pageserver/src/tenant/remote_timeline_client.rs

@@ -194,7 +194,6 @@ pub(crate) use download::{
 };
 use index::GcCompactionState;
 pub(crate) use index::LayerFileMetadata;
-use neon_failpoint as fail;
 use pageserver_api::models::{RelSizeMigration, TimelineArchivalState, TimelineVisibilityState};
 use pageserver_api::shard::{ShardIndex, TenantShardId};
 use regex::Regex;

pageserver/src/tenant/remote_timeline_client/download.rs

@@ -11,7 +11,6 @@ use std::time::SystemTime;
 
 use anyhow::{Context, anyhow};
 use camino::{Utf8Path, Utf8PathBuf};
-use neon_failpoint as fail;
 use pageserver_api::shard::TenantShardId;
 use remote_storage::{
     DownloadError, DownloadKind, DownloadOpts, GenericRemoteStorage, ListingMode, RemotePath,

pageserver/src/tenant/remote_timeline_client/upload.rs

@@ -8,7 +8,6 @@ use anyhow::{Context, bail};
 use bytes::Bytes;
 use camino::Utf8Path;
 use fail::fail_point;
-use neon_failpoint as fail;
 use pageserver_api::shard::TenantShardId;
 use remote_storage::{GenericRemoteStorage, RemotePath, TimeTravelError};
 use tokio::fs::{self, File};

pageserver/src/tenant/timeline.rs

@@ -39,7 +39,6 @@ use layer_manager::{
     LayerManagerLockHolder, LayerManagerReadGuard, LayerManagerWriteGuard, LockedLayerManager,
     Shutdown,
 };
-use neon_failpoint as fail;
 
 use once_cell::sync::Lazy;
 use pageserver_api::config::tenant_conf_defaults::DEFAULT_PITR_INTERVAL;
@@ -352,13 +351,6 @@ pub struct Timeline {
     last_image_layer_creation_check_at: AtomicLsn,
     last_image_layer_creation_check_instant: std::sync::Mutex<Option<Instant>>,
 
-    // HADRON
-    /// If a key range has writes with LSN > force_image_creation_lsn, then we should force image layer creation
-    /// on this key range.
-    force_image_creation_lsn: AtomicLsn,
-    /// The last time instant when force_image_creation_lsn is computed.
-    force_image_creation_lsn_computed_at: std::sync::Mutex<Option<Instant>>,
-
     /// Current logical size of the "datadir", at the last LSN.
     current_logical_size: LogicalSize,
 
@@ -2855,7 +2847,7 @@ impl Timeline {
     }
 
     // HADRON
-    fn get_image_creation_timeout(&self) -> Option<Duration> {
+    fn get_image_layer_force_creation_period(&self) -> Option<Duration> {
         let tenant_conf = self.tenant_conf.load();
         tenant_conf
             .tenant_conf
@@ -3135,9 +3127,6 @@ impl Timeline {
                 repartition_threshold: 0,
                 last_image_layer_creation_check_at: AtomicLsn::new(0),
                 last_image_layer_creation_check_instant: Mutex::new(None),
-                // HADRON
-                force_image_creation_lsn: AtomicLsn::new(0),
-                force_image_creation_lsn_computed_at: std::sync::Mutex::new(None),
                 last_received_wal: Mutex::new(None),
                 rel_size_latest_cache: RwLock::new(HashMap::new()),
                 rel_size_snapshot_cache: Mutex::new(LruCache::new(relsize_snapshot_cache_capacity)),
@@ -5184,9 +5173,7 @@ impl Timeline {
             *self.applied_gc_cutoff_lsn.read(),
         );
 
-        neon_failpoint::fail_point_sync!("checkpoint-before-saving-metadata", |x: Option<
-            String,
-        >| bail!(
+        fail_point!("checkpoint-before-saving-metadata", |x| bail!(
             "{}",
             x.unwrap()
         ));
@@ -5384,13 +5371,16 @@ impl Timeline {
         }
 
         // HADRON
+        // for child timelines, we consider all pages up to ancestor_LSN are redone successfully by the parent timeline
+        min_image_lsn = min_image_lsn.max(self.get_ancestor_lsn());
         if min_image_lsn < force_image_creation_lsn.unwrap_or(Lsn(0)) && max_deltas > 0 {
             info!(
-                "forcing image creation for partitioned range {}-{}. Min image LSN: {}, force image creation LSN: {}",
+                "forcing image creation for partitioned range {}-{}. Min image LSN: {}, force image creation LSN: {}, num deltas: {}",
                 partition.ranges[0].start,
                 partition.ranges[0].end,
                 min_image_lsn,
-                force_image_creation_lsn.unwrap()
+                force_image_creation_lsn.unwrap(),
+                max_deltas
             );
             return true;
         }
@@ -7156,6 +7146,19 @@ impl Timeline {
             .unwrap()
             .clone()
     }
+
+    /* BEGIN_HADRON */
+    pub(crate) async fn compute_image_consistent_lsn(&self) -> anyhow::Result<Lsn> {
+        let guard = self
+            .layers
+            .read(LayerManagerLockHolder::ComputeImageConsistentLsn)
+            .await;
+        let layer_map = guard.layer_map()?;
+        let disk_consistent_lsn = self.get_disk_consistent_lsn();
+
+        Ok(layer_map.compute_image_consistent_lsn(disk_consistent_lsn))
+    }
+    /* END_HADRON */
 }
 
 impl Timeline {

pageserver/src/tenant/timeline/compaction.rs

@@ -8,7 +8,7 @@ use std::cmp::min;
 use std::collections::{BinaryHeap, HashMap, HashSet, VecDeque};
 use std::ops::{Deref, Range};
 use std::sync::Arc;
-use std::time::{Duration, Instant, SystemTime};
+use std::time::{Duration, Instant};
 
 use super::layer_manager::LayerManagerLockHolder;
 use super::{
@@ -26,7 +26,6 @@ use enumset::EnumSet;
 use fail::fail_point;
 use futures::FutureExt;
 use itertools::Itertools;
-use neon_failpoint as fail;
 use once_cell::sync::Lazy;
 use pageserver_api::config::tenant_conf_defaults::DEFAULT_CHECKPOINT_DISTANCE;
 use pageserver_api::key::{KEY_SIZE, Key};
@@ -35,7 +34,6 @@ use pageserver_api::models::{CompactInfoResponse, CompactKeyRange};
 use pageserver_api::shard::{ShardCount, ShardIdentity, TenantShardId};
 use pageserver_compaction::helpers::{fully_contains, overlaps_with};
 use pageserver_compaction::interface::*;
-use postgres_ffi::to_pg_timestamp;
 use serde::Serialize;
 use tokio::sync::{OwnedSemaphorePermit, Semaphore};
 use tokio_util::sync::CancellationToken;
@@ -48,7 +46,6 @@ use wal_decoder::models::value::Value;
 
 use crate::context::{AccessStatsBehavior, RequestContext, RequestContextBuilder};
 use crate::page_cache;
-use crate::pgdatadir_mapping::LsnForTimestamp;
 use crate::statvfs::Statvfs;
 use crate::tenant::checks::check_valid_layermap;
 use crate::tenant::gc_block::GcBlock;
@@ -1272,10 +1269,7 @@ impl Timeline {
         // Define partitioning schema if needed
 
         // HADRON
-        let force_image_creation_lsn = self
-            .get_or_compute_force_image_creation_lsn(cancel, ctx)
-            .await
-            .map_err(CompactionError::Other)?;
+        let force_image_creation_lsn = self.get_force_image_creation_lsn();
 
         // 1. L0 Compact
         let l0_outcome = {
@@ -1485,59 +1479,37 @@ impl Timeline {
     }
 
     /* BEGIN_HADRON */
-    // Get the force image creation LSN. Compute it if the last computed LSN is too old.
-    async fn get_or_compute_force_image_creation_lsn(
-        self: &Arc<Self>,
-        cancel: &CancellationToken,
-        ctx: &RequestContext,
-    ) -> anyhow::Result<Option<Lsn>> {
-        const FORCE_IMAGE_CREATION_LSN_COMPUTE_INTERVAL: Duration = Duration::from_secs(10 * 60); // 10 minutes
-        let image_layer_force_creation_period = self.get_image_creation_timeout();
-        if image_layer_force_creation_period.is_none() {
-            return Ok(None);
+    // Get the force image creation LSN based on gc_cutoff_lsn.
+    // Note that this is an estimation and the workload rate may suddenly change. When that happens,
+    // the force image creation may be too early or too late, but eventually it should be able to catch up.
+    pub(crate) fn get_force_image_creation_lsn(self: &Arc<Self>) -> Option<Lsn> {
+        let image_creation_period = self.get_image_layer_force_creation_period()?;
+        let current_lsn = self.get_last_record_lsn();
+        let pitr_lsn = self.gc_info.read().unwrap().cutoffs.time?;
+        let pitr_interval = self.get_pitr_interval();
+        if pitr_lsn == Lsn::INVALID || pitr_interval.is_zero() {
+            tracing::warn!(
+                "pitr LSN/interval not found, skipping force image creation LSN calculation"
+            );
+            return None;
         }
 
-        let image_layer_force_creation_period = image_layer_force_creation_period.unwrap();
-        let force_image_creation_lsn_computed_at =
-            *self.force_image_creation_lsn_computed_at.lock().unwrap();
-        if force_image_creation_lsn_computed_at.is_none()
-            || force_image_creation_lsn_computed_at.unwrap().elapsed()
-                > FORCE_IMAGE_CREATION_LSN_COMPUTE_INTERVAL
-        {
-            let now: SystemTime = SystemTime::now();
-            let timestamp = now
-                .checked_sub(image_layer_force_creation_period)
-                .ok_or_else(|| {
-                    anyhow::anyhow!(
-                        "image creation timeout is too large: {image_layer_force_creation_period:?}"
-                    )
-                })?;
-            let timestamp = to_pg_timestamp(timestamp);
-            let force_image_creation_lsn = match self
-                .find_lsn_for_timestamp(timestamp, cancel, ctx)
-                .await?
-            {
-                LsnForTimestamp::Present(lsn) | LsnForTimestamp::Future(lsn) => lsn,
-                _ => {
-                    let gc_lsn = *self.get_applied_gc_cutoff_lsn();
-                    tracing::info!(
-                        "no LSN found for timestamp {timestamp:?}, using latest GC cutoff LSN {}",
-                        gc_lsn
-                    );
-                    gc_lsn
-                }
-            };
-            self.force_image_creation_lsn
-                .store(force_image_creation_lsn);
-            *self.force_image_creation_lsn_computed_at.lock().unwrap() = Some(Instant::now());
-            tracing::info!(
-                "computed force image creation LSN: {}",
-                force_image_creation_lsn
-            );
-            Ok(Some(force_image_creation_lsn))
-        } else {
-            Ok(Some(self.force_image_creation_lsn.load()))
-        }
+        let delta_lsn = current_lsn.checked_sub(pitr_lsn).unwrap().0
+            * image_creation_period.as_secs()
+            / pitr_interval.as_secs();
+        let force_image_creation_lsn = current_lsn.checked_sub(delta_lsn).unwrap_or(Lsn(0));
+
+        tracing::info!(
+            "Tenant shard {} computed force_image_creation_lsn: {}. Current lsn: {}, image_layer_force_creation_period: {:?}, GC cutoff: {}, PITR interval: {:?}",
+            self.tenant_shard_id,
+            force_image_creation_lsn,
+            current_lsn,
+            image_creation_period,
+            pitr_lsn,
+            pitr_interval
+        );
+
+        Some(force_image_creation_lsn)
     }
     /* END_HADRON */
 

pageserver/src/tenant/timeline/delete.rs

@@ -2,7 +2,6 @@ use std::ops::{Deref, DerefMut};
 use std::sync::Arc;
 
 use anyhow::Context;
-use neon_failpoint as fail;
 use pageserver_api::models::TimelineState;
 use pageserver_api::shard::TenantShardId;
 use remote_storage::DownloadError;

pageserver/src/tenant/timeline/detach_ancestor.rs

@@ -4,7 +4,6 @@ use std::sync::Arc;
 use anyhow::Context;
 use bytes::Bytes;
 use http_utils::error::ApiError;
-use neon_failpoint as fail;
 use pageserver_api::key::Key;
 use pageserver_api::keyspace::KeySpace;
 use pageserver_api::models::DetachBehavior;
@@ -1114,7 +1113,7 @@ pub(super) async fn detach_and_reparent(
     // others will fail as if those timelines had been stopped for whatever reason.
     #[cfg(feature = "testing")]
     let failpoint_sem = || -> Option<Arc<Semaphore>> {
-        fail::fail_point_sync!("timeline-detach-ancestor::allow_one_reparented", |_| Some(
+        fail::fail_point!("timeline-detach-ancestor::allow_one_reparented", |_| Some(
             Arc::new(Semaphore::new(1))
         ));
         None

pageserver/src/tenant/timeline/handle.rs

@@ -359,14 +359,14 @@ impl<T: Types> Cache<T> {
                 Err(e) => {
                     // Retry on tenant manager error to handle tenant split more gracefully
                     if attempt < GET_MAX_RETRIES {
-                        tracing::warn!(
-                            "Fail to resolve tenant shard in attempt {}: {:?}. Retrying...",
-                            attempt,
-                            e
-                        );
                         tokio::time::sleep(RETRY_BACKOFF).await;
                         continue;
                     } else {
+                        tracing::warn!(
+                            "Failed to resolve tenant shard after {} attempts: {:?}",
+                            GET_MAX_RETRIES,
+                            e
+                        );
                         return Err(e);
                     }
                 }

pageserver/src/tenant/timeline/layer_manager.rs

@@ -47,6 +47,7 @@ pub(crate) enum LayerManagerLockHolder {
     ImportPgData,
     DetachAncestor,
     Eviction,
+    ComputeImageConsistentLsn,
     #[cfg(test)]
     Testing,
 }

pageserver/src/tenant/timeline/uninit.rs

@@ -5,7 +5,6 @@ use std::sync::Arc;
 
 use anyhow::Context;
 use camino::Utf8PathBuf;
-use neon_failpoint as fail;
 use tracing::{error, info, info_span};
 use utils::fs_ext;
 use utils::id::TimelineId;

pageserver/src/tenant/timeline/walreceiver/walreceiver_connection.rs

@@ -11,7 +11,6 @@ use bytes::BytesMut;
 use chrono::{NaiveDateTime, Utc};
 use fail::fail_point;
 use futures::StreamExt;
-use neon_failpoint as fail;
 use postgres_backend::is_expected_io_error;
 use postgres_connection::PgConnectionConfig;
 use postgres_ffi::WAL_SEGMENT_SIZE;

safekeeper/Cargo.toml

@@ -6,7 +6,9 @@ license.workspace = true
 
 [features]
 default = []
-testing = ["neon_failpoint/testing"]
+# Enables test-only APIs, incuding failpoints. In particular, enables the `fail_point!` macro,
+# which adds some runtime cost to run tests on outage conditions
+testing = ["fail/failpoints"]
 benchmarking = []
 
 [dependencies]
@@ -19,7 +21,7 @@ camino-tempfile.workspace = true
 chrono.workspace = true
 clap = { workspace = true, features = ["derive"] }
 crc32c.workspace = true
-neon_failpoint.workspace = true
+fail.workspace = true
 hex.workspace = true
 humantime.workspace = true
 http.workspace = true

safekeeper/src/bin/safekeeper.rs

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

safekeeper/src/http/routes.rs

@@ -717,7 +717,7 @@ pub fn make_router(
         .get("/profile/cpu", |r| request_span(r, profile_cpu_handler))
         .get("/profile/heap", |r| request_span(r, profile_heap_handler))
         .get("/v1/status", |r| request_span(r, status_handler))
-        .post("/v1/failpoints", |r| {
+        .put("/v1/failpoints", |r| {
             request_span(r, move |r| async {
                 check_permission(&r, None)?;
                 let cancel = CancellationToken::new();

safekeeper/src/send_wal.rs

@@ -872,15 +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 = async {
-                neon_failpoint::fail_point!(
+            let have_something_to_send = (|| {
+                fail::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);
@@ -932,15 +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 = async {
-            neon_failpoint::fail_point!(
+        let fp = (|| {
+            fail::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");
 
-    neon_failpoint::fail_point!("sk-delete-timeline-remote", |_| {
+    fail::fail_point!("sk-delete-timeline-remote", |_| {
         Err(anyhow::anyhow!("failpoint: sk-delete-timeline-remote"))
     });
 

safekeeper/src/wal_storage.rs

@@ -301,7 +301,7 @@ impl PhysicalStorage {
                 format!("Failed to open tmp wal file {:?}", &tmp_path)
             })?;
 
-            neon_failpoint::fail_point!("sk-zero-segment", |_| {
+            fail::fail_point!("sk-zero-segment", |_| {
                 info!("sk-zero-segment failpoint hit");
                 Err(anyhow::anyhow!("failpoint: sk-zero-segment"))
             });

storage_controller/Cargo.toml

@@ -22,7 +22,7 @@ clap.workspace = true
 clashmap.workspace = true
 compute_api.workspace = true
 cron.workspace = true
-neon_failpoint.workspace = true
+fail.workspace = true
 futures.workspace = true
 governor.workspace = true
 hex.workspace = true

storage_controller/src/http.rs

@@ -850,6 +850,31 @@ async fn handle_tenant_describe(
     json_response(StatusCode::OK, service.tenant_describe(tenant_id)?)
 }
 
+/* BEGIN_HADRON */
+async fn handle_tenant_timeline_describe(
+    service: Arc<Service>,
+    req: Request<Body>,
+) -> Result<Response<Body>, ApiError> {
+    check_permissions(&req, Scope::Scrubber)?;
+
+    let tenant_id: TenantId = parse_request_param(&req, "tenant_id")?;
+    let timeline_id: TimelineId = parse_request_param(&req, "timeline_id")?;
+    match maybe_forward(req).await {
+        ForwardOutcome::Forwarded(res) => {
+            return res;
+        }
+        ForwardOutcome::NotForwarded(_req) => {}
+    };
+
+    json_response(
+        StatusCode::OK,
+        service
+            .tenant_timeline_describe(tenant_id, timeline_id)
+            .await?,
+    )
+}
+/* END_HADRON */
+
 async fn handle_tenant_list(
     service: Arc<Service>,
     req: Request<Body>,
@@ -2480,6 +2505,13 @@ pub fn make_router(
             )
         })
         // Timeline operations
+        .get("/control/v1/tenant/:tenant_id/timeline/:timeline_id", |r| {
+            tenant_service_handler(
+                r,
+                handle_tenant_timeline_describe,
+                RequestName("v1_tenant_timeline_describe"),
+            )
+        })
         .delete("/v1/tenant/:tenant_id/timeline/:timeline_id", |r| {
             tenant_service_handler(
                 r,

storage_controller/src/pageserver_client.rs

@@ -86,6 +86,23 @@ impl PageserverClient {
         )
     }
 
+    /* BEGIN_HADRON */
+    pub(crate) async fn tenant_timeline_describe(
+        &self,
+        tenant_shard_id: &TenantShardId,
+        timeline_id: &TimelineId,
+    ) -> Result<TimelineInfo> {
+        measured_request!(
+            "tenant_timeline_describe",
+            crate::metrics::Method::Get,
+            &self.node_id_label,
+            self.inner
+                .tenant_timeline_describe(tenant_shard_id, timeline_id,)
+                .await
+        )
+    }
+    /* END_HADRON */
+
     pub(crate) async fn tenant_scan_remote_storage(
         &self,
         tenant_id: TenantId,

storage_controller/src/service.rs

@@ -25,7 +25,6 @@ use futures::stream::FuturesUnordered;
 use http_utils::error::ApiError;
 use hyper::Uri;
 use itertools::Itertools;
-use neon_failpoint as fail;
 use pageserver_api::config::PostHogConfig;
 use pageserver_api::controller_api::{
     AvailabilityZone, MetadataHealthRecord, MetadataHealthUpdateRequest, NodeAvailability,
@@ -33,7 +32,7 @@ use pageserver_api::controller_api::{
     ShardSchedulingPolicy, ShardsPreferredAzsRequest, ShardsPreferredAzsResponse,
     SkSchedulingPolicy, TenantCreateRequest, TenantCreateResponse, TenantCreateResponseShard,
     TenantDescribeResponse, TenantDescribeResponseShard, TenantLocateResponse, TenantPolicyRequest,
-    TenantShardMigrateRequest, TenantShardMigrateResponse,
+    TenantShardMigrateRequest, TenantShardMigrateResponse, TenantTimelineDescribeResponse,
 };
 use pageserver_api::models::{
     self, DetachBehavior, LocationConfig, LocationConfigListResponse, LocationConfigMode, LsnLease,
@@ -5487,6 +5486,92 @@ impl Service {
         .ok_or_else(|| ApiError::NotFound(anyhow::anyhow!("Tenant {tenant_id} not found").into()))
     }
 
+    /* BEGIN_HADRON */
+    pub(crate) async fn tenant_timeline_describe(
+        &self,
+        tenant_id: TenantId,
+        timeline_id: TimelineId,
+    ) -> Result<TenantTimelineDescribeResponse, ApiError> {
+        self.tenant_remote_mutation(tenant_id, |locations| async move {
+            if locations.0.is_empty() {
+                return Err(ApiError::NotFound(
+                    anyhow::anyhow!("Tenant not found").into(),
+                ));
+            };
+
+            let locations: Vec<(TenantShardId, Node)> = locations
+                .0
+                .iter()
+                .map(|t| (*t.0, t.1.latest.node.clone()))
+                .collect();
+            let mut futs = FuturesUnordered::new();
+
+            for (shard_id, node) in locations {
+                futs.push({
+                    async move {
+                        let result = node
+                            .with_client_retries(
+                                |client| async move {
+                                    client
+                                        .tenant_timeline_describe(&shard_id, &timeline_id)
+                                        .await
+                                },
+                                &self.http_client,
+                                &self.config.pageserver_jwt_token,
+                                3,
+                                3,
+                                Duration::from_secs(30),
+                                &self.cancel,
+                            )
+                            .await;
+                        (result, shard_id, node.get_id())
+                    }
+                });
+            }
+
+            let mut results: Vec<TimelineInfo> = Vec::new();
+            while let Some((result, tenant_shard_id, node_id)) = futs.next().await {
+                match result {
+                    Some(Ok(timeline_info)) => results.push(timeline_info),
+                    Some(Err(e)) => {
+                        tracing::warn!(
+                            "Failed to describe tenant {} timeline {} for pageserver {}: {e}",
+                            tenant_shard_id,
+                            timeline_id,
+                            node_id,
+                        );
+                        return Err(ApiError::ResourceUnavailable(format!("{e}").into()));
+                    }
+                    None => return Err(ApiError::Cancelled),
+                }
+            }
+            let mut image_consistent_lsn: Option<Lsn> = Some(Lsn::MAX);
+            for timeline_info in &results {
+                if let Some(tline_image_consistent_lsn) = timeline_info.image_consistent_lsn {
+                    image_consistent_lsn = Some(std::cmp::min(
+                        image_consistent_lsn.unwrap(),
+                        tline_image_consistent_lsn,
+                    ));
+                } else {
+                    tracing::warn!(
+                        "Timeline {} on shard {} does not have image consistent lsn",
+                        timeline_info.timeline_id,
+                        timeline_info.tenant_id
+                    );
+                    image_consistent_lsn = None;
+                    break;
+                }
+            }
+
+            Ok(TenantTimelineDescribeResponse {
+                shards: results,
+                image_consistent_lsn,
+            })
+        })
+        .await?
+    }
+    /* END_HADRON */
+
     /// limit & offset are pagination parameters. Since we are walking an in-memory HashMap, `offset` does not
     /// avoid traversing data, it just avoid returning it. This is suitable for our purposes, since our in memory
     /// maps are small enough to traverse fast, our pagination is just to avoid serializing huge JSON responses
@@ -6027,7 +6112,7 @@ impl Service {
         tenant_id: TenantId,
         split_req: TenantShardSplitRequest,
     ) -> Result<ShardSplitAction, ApiError> {
-        fail::fail_point_sync!("shard-split-validation", |_| Err(ApiError::BadRequest(
+        fail::fail_point!("shard-split-validation", |_| Err(ApiError::BadRequest(
             anyhow::anyhow!("failpoint")
         )));
 

storage_controller/src/service/safekeeper_service.rs

@@ -25,7 +25,8 @@ use pageserver_api::models::{SafekeeperInfo, SafekeepersInfo, TimelineInfo};
 use safekeeper_api::PgVersionId;
 use safekeeper_api::membership::{self, MemberSet, SafekeeperGeneration};
 use safekeeper_api::models::{
-    PullTimelineRequest, TimelineMembershipSwitchRequest, TimelineMembershipSwitchResponse,
+    PullTimelineRequest, TimelineLocateResponse, TimelineMembershipSwitchRequest,
+    TimelineMembershipSwitchResponse,
 };
 use safekeeper_api::{INITIAL_TERM, Term};
 use safekeeper_client::mgmt_api;
@@ -37,13 +38,6 @@ use utils::lsn::Lsn;
 
 use super::Service;
 
-#[derive(serde::Serialize, serde::Deserialize, Clone)]
-pub struct TimelineLocateResponse {
-    pub generation: SafekeeperGeneration,
-    pub sk_set: Vec<NodeId>,
-    pub new_sk_set: Option<Vec<NodeId>>,
-}
-
 impl Service {
     fn make_member_set(safekeepers: &[Safekeeper]) -> Result<MemberSet, ApiError> {
         let members = safekeepers

test_runner/fixtures/endpoint/http.py

@@ -159,59 +159,16 @@ class EndpointHttpClient(requests.Session):
         res.raise_for_status()
         return res.json()
 
-    def configure_failpoints(
-        self, *args: tuple[str, str] | list[dict[str, str | dict[str, str]]]
-    ) -> None:
-        """Configure failpoints for testing purposes.
+    def configure_failpoints(self, *args: tuple[str, str]) -> None:
+        body: list[dict[str, str]] = []
 
-        Args:
-            *args: Can be one of:
-                - Variable number of (name, actions) tuples
-                - Single list of dicts with keys: name, actions, and optionally context_matchers
-
-        Examples:
-            # Basic failpoints
-            client.configure_failpoints(("test_fp", "return(error)"))
-            client.configure_failpoints(("fp1", "return"), ("fp2", "sleep(1000)"))
-
-            # Probability-based failpoint
-            client.configure_failpoints(("test_fp", "50%return(error)"))
-
-            # Context-based failpoint
-            client.configure_failpoints([{
-                "name": "test_fp",
-                "actions": "return(error)",
-                "context_matchers": {"tenant_id": ".*test.*"}
-            }])
-        """
-
-        request_body: list[dict[str, Any]] = []
-
-        if (
-            len(args) == 1
-            and isinstance(args[0], list)
-            and args[0]
-            and isinstance(args[0][0], dict)
-        ):
-            # Handle list of dicts (context-based failpoints)
-            failpoint_configs = args[0]
-            for config in failpoint_configs:
-                server_config: dict[str, Any] = {
-                    "name": config["name"],
-                    "actions": config["actions"],
+        for fp in args:
+            body.append(
+                {
+                    "name": fp[0],
+                    "action": fp[1],
                 }
-                if "context_matchers" in config:
-                    server_config["context_matchers"] = config["context_matchers"]
-                request_body.append(server_config)
-        else:
-            # Handle tuples (basic failpoints)
-            for fp in args:
-                request_body.append(
-                    {
-                        "name": fp[0],
-                        "actions": fp[1],
-                    }
-                )
+            )
 
-        res = self.post(f"http://localhost:{self.internal_port}/failpoints", json=request_body)
+        res = self.post(f"http://localhost:{self.internal_port}/failpoints", json=body)
         res.raise_for_status()

test_runner/fixtures/neon_fixtures.py

@@ -2342,6 +2342,20 @@ class NeonStorageController(MetricsGetter, LogUtils):
         response.raise_for_status()
         return response.json()
 
+    # HADRON
+    def tenant_timeline_describe(
+        self,
+        tenant_id: TenantId,
+        timeline_id: TimelineId,
+    ):
+        response = self.request(
+            "GET",
+            f"{self.api}/control/v1/tenant/{tenant_id}/timeline/{timeline_id}",
+            headers=self.headers(TokenScope.ADMIN),
+        )
+        response.raise_for_status()
+        return response.json()
+
     def nodes(self):
         """
         :return: list of {"id": ""}
@@ -2614,68 +2628,22 @@ class NeonStorageController(MetricsGetter, LogUtils):
         )
         return res.json()
 
-    def configure_failpoints(
-        self,
-        config_strings: tuple[str, str]
-        | list[tuple[str, str]]
-        | list[dict[str, str | dict[str, str]]],
-    ):
-        """
-        Configure failpoints for testing purposes.
-
-        Args:
-            config_strings: Can be one of:
-                - Single tuple of (name, actions)
-                - List of tuples [(name, actions), ...]
-                - List of dicts with keys: name, actions, and optionally context_matchers
-
-        Examples:
-            # Basic failpoint
-            client.configure_failpoints(("test_fp", "return(error)"))
-
-            # Multiple basic failpoints
-            client.configure_failpoints([("fp1", "return"), ("fp2", "sleep(1000)")])
-
-            # Probability-based failpoint
-            client.configure_failpoints(("test_fp", "50%return(error)"))
-
-            # Context-based failpoint
-            client.configure_failpoints([{
-                "name": "test_fp",
-                "actions": "return(error)",
-                "context_matchers": {"tenant_id": ".*test.*"}
-            }])
-        """
-        # Handle single tuple case
+    def configure_failpoints(self, config_strings: tuple[str, str] | list[tuple[str, str]]):
         if isinstance(config_strings, tuple):
-            config_strings = [config_strings]
+            pairs = [config_strings]
+        else:
+            pairs = config_strings
 
-        # Convert to server format
-        body: list[dict[str, str | dict[str, str]]] = []
-        for config in config_strings:
-            if isinstance(config, tuple):
-                # Simple (name, actions) tuple
-                body.append({"name": config[0], "actions": config[1]})
-            elif isinstance(config, dict):
-                # Dict with name, actions, and optional context_matchers
-                server_config: dict[str, str | dict[str, str]] = {
-                    "name": config["name"],
-                    "actions": config["actions"],
-                }
-                if "context_matchers" in config:
-                    server_config["context_matchers"] = config["context_matchers"]
-                body.append(server_config)
-            else:
-                raise ValueError(f"Invalid config format: {config}")
+        log.info(f"Requesting config failpoints: {repr(pairs)}")
 
         res = self.request(
             "PUT",
-            f"{self.api_root()}/debug/v1/failpoints",
-            json=body,
+            f"{self.api}/debug/v1/failpoints",
+            json=[{"name": name, "actions": actions} for name, actions in pairs],
             headers=self.headers(TokenScope.ADMIN),
         )
-        if res.status_code != 200:
-            self.raise_api_exception(res)
+        log.info(f"Got failpoints request response code {res.status_code}")
+        res.raise_for_status()
 
     def get_tenants_placement(self) -> defaultdict[str, dict[str, Any]]:
         """

test_runner/fixtures/pageserver/allowed_errors.py

@@ -115,8 +115,7 @@ DEFAULT_PAGESERVER_ALLOWED_ERRORS = (
     ".*Local data loss suspected.*",
     # Too many frozen layers error is normal during intensive benchmarks
     ".*too many frozen layers.*",
-    # Transient errors when resolving tenant shards by page service
-    ".*Fail to resolve tenant shard in attempt.*",
+    ".*Failed to resolve tenant shard after.*",
     # Expected warnings when pageserver has not refreshed GC info yet
     ".*pitr LSN/interval not found, skipping force image creation LSN calculation.*",
     ".*No broker updates received for a while.*",

test_runner/fixtures/pageserver/http.py

@@ -309,64 +309,25 @@ class PageserverHttpClient(requests.Session, MetricsGetter):
     def check_status(self):
         self.get(f"http://localhost:{self.port}/v1/status").raise_for_status()
 
-    def configure_failpoints(
-        self,
-        config_strings: tuple[str, str]
-        | list[tuple[str, str]]
-        | list[dict[str, str | dict[str, str]]],
-    ):
-        """
-        Configure failpoints for testing purposes.
-
-        Args:
-            config_strings: Can be one of:
-                - Single tuple of (name, actions)
-                - List of tuples [(name, actions), ...]
-                - List of dicts with keys: name, actions, and optionally context_matchers
-
-        Examples:
-            # Basic failpoint
-            client.configure_failpoints(("test_fp", "return(error)"))
-
-            # Multiple basic failpoints
-            client.configure_failpoints([("fp1", "return"), ("fp2", "sleep(1000)")])
-
-            # Probability-based failpoint
-            client.configure_failpoints(("test_fp", "50%return(error)"))
-
-            # Context-based failpoint
-            client.configure_failpoints([{
-                "name": "test_fp",
-                "actions": "return(error)",
-                "context_matchers": {"tenant_id": ".*test.*"}
-            }])
-        """
+    def configure_failpoints(self, config_strings: tuple[str, str] | list[tuple[str, str]]):
         self.is_testing_enabled_or_skip()
 
-        # Handle single tuple case
         if isinstance(config_strings, tuple):
-            config_strings = [config_strings]
+            pairs = [config_strings]
+        else:
+            pairs = config_strings
 
-        # Convert to server format
-        body: list[dict[str, str | dict[str, str]]] = []
-        for config in config_strings:
-            if isinstance(config, tuple):
-                # Simple (name, actions) tuple
-                body.append({"name": config[0], "actions": config[1]})
-            elif isinstance(config, dict):
-                # Dict with name, actions, and optional context_matchers
-                server_config = {"name": config["name"], "actions": config["actions"]}
-                if "context_matchers" in config:
-                    server_config["context_matchers"] = config["context_matchers"]
-                body.append(server_config)
-            else:
-                raise ValueError(f"Invalid config format: {config}")
+        log.info(f"Requesting config failpoints: {repr(pairs)}")
 
-        res = self.post(f"{self.base_url}/v1/failpoints", json=body)
-        if res.status_code != 200:
-            raise PageserverApiException(
-                f"Failed to configure failpoints: {res.text}", res.status_code
-            )
+        res = self.put(
+            f"http://localhost:{self.port}/v1/failpoints",
+            json=[{"name": name, "actions": actions} for name, actions in pairs],
+        )
+        log.info(f"Got failpoints request response code {res.status_code}")
+        self.verbose_error(res)
+        res_json = res.json()
+        assert res_json is None
+        return res_json
 
     def reload_auth_validation_keys(self):
         res = self.post(f"http://localhost:{self.port}/v1/reload_auth_validation_keys")

test_runner/fixtures/safekeeper/http.py

@@ -8,6 +8,7 @@ import pytest
 import requests
 
 from fixtures.common_types import Lsn, TenantId, TenantTimelineId, TimelineId
+from fixtures.log_helper import log
 from fixtures.metrics import Metrics, MetricsGetter, parse_metrics
 from fixtures.utils import EnhancedJSONEncoder, wait_until
 
@@ -154,62 +155,25 @@ class SafekeeperHttpClient(requests.Session, MetricsGetter):
         if not self.is_testing_enabled:
             pytest.skip("safekeeper was built without 'testing' feature")
 
-    def configure_failpoints(
-        self,
-        config_strings: tuple[str, str]
-        | list[tuple[str, str]]
-        | list[dict[str, str | dict[str, str]]],
-    ):
-        """
-        Configure failpoints for testing purposes.
-
-        Args:
-            config_strings: Can be one of:
-                - Single tuple of (name, actions)
-                - List of tuples [(name, actions), ...]
-                - List of dicts with keys: name, actions, and optionally context_matchers
-
-        Examples:
-            # Basic failpoint
-            client.configure_failpoints(("test_fp", "return(error)"))
-
-            # Multiple basic failpoints
-            client.configure_failpoints([("fp1", "return"), ("fp2", "sleep(1000)")])
-
-            # Probability-based failpoint
-            client.configure_failpoints(("test_fp", "50%return(error)"))
-
-            # Context-based failpoint
-            client.configure_failpoints([{
-                "name": "test_fp",
-                "actions": "return(error)",
-                "context_matchers": {"tenant_id": ".*test.*"}
-            }])
-        """
+    def configure_failpoints(self, config_strings: tuple[str, str] | list[tuple[str, str]]):
         self.is_testing_enabled_or_skip()
 
-        # Handle single tuple case
         if isinstance(config_strings, tuple):
-            config_strings = [config_strings]
+            pairs = [config_strings]
+        else:
+            pairs = config_strings
 
-        # Convert to server format
-        body: list[dict[str, str | dict[str, str]]] = []
-        for config in config_strings:
-            if isinstance(config, tuple):
-                # Simple (name, actions) tuple
-                body.append({"name": config[0], "actions": config[1]})
-            elif isinstance(config, dict):
-                # Dict with name, actions, and optional context_matchers
-                server_config = {"name": config["name"], "actions": config["actions"]}
-                if "context_matchers" in config:
-                    server_config["context_matchers"] = config["context_matchers"]
-                body.append(server_config)
-            else:
-                raise ValueError(f"Invalid config format: {config}")
+        log.info(f"Requesting config failpoints: {repr(pairs)}")
 
-        res = self.post(f"http://localhost:{self.port}/v1/failpoints", json=body)
-        if res.status_code != 200:
-            raise RuntimeError(f"Failed to configure failpoints: {res.text}")
+        res = self.put(
+            f"http://localhost:{self.port}/v1/failpoints",
+            json=[{"name": name, "actions": actions} for name, actions in pairs],
+        )
+        log.info(f"Got failpoints request response code {res.status_code}")
+        res.raise_for_status()
+        res_json = res.json()
+        assert res_json is None
+        return res_json
 
     def tenant_delete_force(self, tenant_id: TenantId) -> dict[Any, Any]:
         res = self.delete(f"http://localhost:{self.port}/v1/tenant/{tenant_id}")

test_runner/regress/test_compaction.py

@@ -960,9 +960,9 @@ def get_layer_map(env, tenant_shard_id, timeline_id, ps_id):
     return image_layer_count, delta_layer_count
 
 
-def test_image_creation_timeout(neon_env_builder: NeonEnvBuilder):
+def test_image_layer_force_creation_period(neon_env_builder: NeonEnvBuilder):
     """
-    Tests that page server can force creating new images if image creation timeout is enabled
+    Tests that page server can force creating new images if image_layer_force_creation_period is enabled
     """
     # use large knobs to disable L0 compaction/image creation except for the force image creation
     tenant_conf = {
@@ -972,10 +972,10 @@ def test_image_creation_timeout(neon_env_builder: NeonEnvBuilder):
         "checkpoint_distance": 10 * 1024,
         "checkpoint_timeout": "1s",
         "image_layer_force_creation_period": "1s",
-        # The lsn for forced image layer creations is calculated once every 10 minutes.
-        # Hence, drive compaction manually such that the test doesn't compute it at the
-        # wrong time.
-        "compaction_period": "0s",
+        "pitr_interval": "10s",
+        "gc_period": "1s",
+        "compaction_period": "1s",
+        "lsn_lease_length": "1s",
     }
 
     # consider every tenant large to run the image layer generation check more eagerly
@@ -1018,4 +1018,69 @@ def test_image_creation_timeout(neon_env_builder: NeonEnvBuilder):
     )
 
 
+def test_image_consistent_lsn(neon_env_builder: NeonEnvBuilder):
+    """
+    Test the /v1/tenant/<tenant_id>/timeline/<timeline_id> endpoint and the computation of image_consistent_lsn
+    """
+    # use large knobs to disable L0 compaction/image creation except for the force image creation
+    tenant_conf = {
+        "compaction_threshold": "100",
+        "image_creation_threshold": "100",
+        "image_layer_creation_check_threshold": "1",
+        "checkpoint_distance": 10 * 1024,
+        "checkpoint_timeout": "1s",
+        "image_layer_force_creation_period": "1s",
+        "pitr_interval": "10s",
+        "gc_period": "1s",
+        "compaction_period": "1s",
+        "lsn_lease_length": "1s",
+    }
+
+    neon_env_builder.num_pageservers = 2
+    neon_env_builder.num_safekeepers = 1
+    env = neon_env_builder.init_start(
+        initial_tenant_conf=tenant_conf,
+        initial_tenant_shard_count=4,
+        initial_tenant_shard_stripe_size=1,
+    )
+
+    tenant_id = env.initial_tenant
+    timeline_id = env.initial_timeline
+
+    endpoint = env.endpoints.create_start("main")
+    endpoint.safe_psql("CREATE TABLE foo (id INTEGER, val text)")
+    for v in range(10):
+        endpoint.safe_psql(
+            f"INSERT INTO foo (id, val) VALUES ({v}, repeat('abcde{v:0>3}', 500))", log_query=False
+        )
+
+    response = env.storage_controller.tenant_timeline_describe(tenant_id, timeline_id)
+    shards = response["shards"]
+    for shard in shards:
+        assert shard["image_consistent_lsn"] is not None
+    image_consistent_lsn = response["image_consistent_lsn"]
+    assert image_consistent_lsn is not None
+
+    # do more writes and wait for image_consistent_lsn to advance
+    for v in range(100):
+        endpoint.safe_psql(
+            f"INSERT INTO foo (id, val) VALUES ({v}, repeat('abcde{v:0>3}', 500))", log_query=False
+        )
+
+    def check_image_consistent_lsn_advanced():
+        response = env.storage_controller.tenant_timeline_describe(tenant_id, timeline_id)
+        new_image_consistent_lsn = response["image_consistent_lsn"]
+        shards = response["shards"]
+        for shard in shards:
+            print(f"shard {shard['tenant_id']} image_consistent_lsn{shard['image_consistent_lsn']}")
+        assert new_image_consistent_lsn != image_consistent_lsn
+
+    wait_until(check_image_consistent_lsn_advanced)
+
+    endpoint.stop_and_destroy()
+
+    for ps in env.pageservers:
+        ps.allowed_errors.append(".*created delta file of size.*larger than double of target.*")
+
+
 # END_HADRON

test_runner/regress/test_compatibility.py

@@ -200,6 +200,8 @@ def test_create_snapshot(
         ignore=shutil.ignore_patterns("pg_dynshmem"),
     )
 
+    log.info(f"Copied new compatibility snapshot dir to: {compatibility_snapshot_dir}")
+
 
 # check_neon_works does recovery from WAL => the compatibility snapshot's WAL is old => will log this warning
 ingest_lag_log_line = ".*ingesting record with timestamp lagging more than wait_lsn_timeout.*"
@@ -218,6 +220,10 @@ def test_backward_compatibility(
     """
     Test that the new binaries can read old data
     """
+
+    snapshot_dir_stat = os.stat(compatibility_snapshot_dir)
+    log.info(f"Snapshot dir at {compatibility_snapshot_dir} stat: {snapshot_dir_stat}")
+
     neon_env_builder.num_safekeepers = 3
     env = neon_env_builder.from_repo_dir(compatibility_snapshot_dir / "repo")
     env.pageserver.allowed_errors.append(ingest_lag_log_line)
@@ -249,6 +255,9 @@ def test_forward_compatibility(
     Test that the old binaries can read new data
     """
 
+    snapshot_dir_stat = os.stat(compatibility_snapshot_dir)
+    log.info(f"Snapshot dir at {compatibility_snapshot_dir} stat: {snapshot_dir_stat}")
+
     neon_env_builder.control_plane_hooks_api = compute_reconfigure_listener.control_plane_hooks_api
     neon_env_builder.test_may_use_compatibility_snapshot_binaries = True