Skip to content

fix: fail e2e scenarios when a required node dies #416

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
matthias-wright merged 3 commits into from
Jun 29, 2026
Merged

fix: fail e2e scenarios when a required node dies #416

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
68026df
fix: fail e2e scenarios when a required node dies
matthias-wright Jun 17, 2026
b4507b1
fix: fail e2e scenarios on unexpected node-handle exit or join failure
matthias-wright Jun 24, 2026
018839f
fix: fail e2e scenarios on remaining unguarded node-exit paths
matthias-wright Jun 26, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 13 additions & 8 deletions node/src/args.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -593,7 +593,7 @@ pub fn run_node_local(
flags: RunFlags,
checkpoint: Option<ConsensusState>,
checkpoint_parent_block: Option<Block>,
) -> Handle<()> {
) -> Handle<anyhow::Result<()>> {
context.spawn(async move |context| {
let key_store = expect_key_store(&flags.key_store_path);
run_node_local_inner(
Expand All @@ -603,7 +603,7 @@ pub fn run_node_local(
checkpoint,
checkpoint_parent_block,
)
.await;
.await
})
}

Expand All @@ -613,7 +613,7 @@ async fn run_node_local_inner(
key_store: KeyStore<PrivateKey>,
checkpoint: Option<ConsensusState>,
checkpoint_parent_block: Option<Block>,
) {
) -> anyhow::Result<()> {
let context = context.with_label("summit_cw");

let genesis = acquire_genesis(&context, &flags).await;
Expand Down Expand Up @@ -749,14 +749,19 @@ async fn run_node_local_inner(
MetricServer::new(config).serve(stop_signal).await.unwrap();
}

// Bring the node down as soon as any core task exits, then return so the runtime
// tears down cleanly and destructors run. Unlike the production path we do NOT
// `process::exit` here: this entrypoint runs inside a caller-managed runtime/thread
// bring the node down as soon as any core task exits, then return so the runtime
// tears down cleanly and destructors run. unlike the production path we do not
// process::exit here: this entrypoint runs inside a caller managed runtime/thread
// (testnet and the e2e scenario binaries), and an abrupt exit would skip the caller's
// shutdown — e.g. orphaning the child Reth processes the scenarios spawn.
if let Err(e) = supervise_node_tasks(&context, p2p, engine, rpc_handle).await {
// shutdown, e.g. orphaning the child reth processes the scenarios spawn. instead we
// propagate the supervise outcome so the caller can decide: a coordinated shutdown
// (graceful stop or committee exit) returns ok, while a genuine core task failure
// returns err so the caller can fail the scenario instead of masking a dead node.
let result = supervise_node_tasks(&context, p2p, engine, rpc_handle).await;
if let Err(e) = &result {
error!(?e, "node core task failed; shutting down node runtime");
}
result
}

/// Supervise the core node tasks (P2P, consensus engine, RPC): as soon as any of them
Expand Down
41 changes: 33 additions & 8 deletions node/src/bin/observer.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -189,8 +189,14 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
let executor = cw_tokio::Runner::new(cfg);

executor.start(|node_context| async move {
let node_handle = node_context.clone().spawn(|ctx| async move {
run_node_local(ctx, flags, None, None).await.unwrap();
let node_handle = node_context.clone().spawn(move |ctx| async move {
// a coordinated shutdown (graceful stop or committee exit) returns
// ok; a genuine core task failure returns err and must fail the
// scenario instead of being masked as a clean node exit.
if let Err(e) = run_node_local(ctx, flags, None, None).await.unwrap() {
eprintln!("node {x} core task failed: {e:?}; failing scenario");
std::process::exit(1);
}
});

let stop_fut = stop_rx.recv().fuse();
Expand All @@ -201,7 +207,11 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
node_context.stop(0, Some(Duration::from_secs(30))).await.unwrap();
}
_ = node_handle.fuse() => {
println!("Node {} handle completed", x);
// Every validator here is a required participant; its handle
// completing without a stop signal means it went down
// unexpectedly, so fail the scenario.
eprintln!("Node {} handle completed unexpectedly; failing scenario", x);
std::process::exit(1);
}
}
});
Expand Down Expand Up @@ -296,10 +306,16 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
let executor = cw_tokio::Runner::new(cfg);

executor.start(|node_context| async move {
let node_handle = node_context.clone().spawn(|ctx| async move {
run_node_local(ctx, observer_flags, None, None)
let node_handle = node_context.clone().spawn(move |ctx| async move {
// the observer is a required participant: a genuine core task failure
// (err) must fail the scenario rather than be masked as a clean exit.
if let Err(e) = run_node_local(ctx, observer_flags, None, None)
.await
.unwrap();
.unwrap()
{
eprintln!("observer core task failed: {e:?}; failing scenario");
std::process::exit(1);
}
});

let stop_fut = observer_stop_rx.recv().fuse();
Expand All @@ -310,7 +326,11 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
node_context.stop(0, Some(Duration::from_secs(30))).await.unwrap();
}
_ = node_handle.fuse() => {
println!("Observer handle completed");
// The observer is a required participant; its handle completing
// without a stop signal means it went down unexpectedly, so
// fail the scenario.
eprintln!("Observer handle completed unexpectedly; failing scenario");
std::process::exit(1);
}
}
});
Expand Down Expand Up @@ -475,7 +495,12 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
println!("Waiting for node index {} to join...", idx);
match node_runtime.thread.join() {
Ok(_) => println!("Node index {} thread joined successfully", idx),
Err(e) => println!("Node index {} thread join failed: {:?}", idx, e),
// A join failure means the node panicked or was killed: a required
// participant going down unexpectedly, so fail the scenario.
Err(e) => {
eprintln!("Node index {} thread join failed: {:?}", idx, e);
std::process::exit(1);
}
}
}

Expand Down
16 changes: 13 additions & 3 deletions node/src/bin/protocol_params.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -177,8 +177,14 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
let executor = cw_tokio::Runner::new(cfg);

executor.start(|node_context| async move {
let node_handle = node_context.clone().spawn(|ctx| async move {
run_node_local(ctx, flags, None, None).await.unwrap();
let node_handle = node_context.clone().spawn(move |ctx| async move {
// a coordinated shutdown (graceful stop or committee exit) returns
// ok; a genuine core task failure returns err and must fail the
// scenario instead of being masked as a clean node exit.
if let Err(e) = run_node_local(ctx, flags, None, None).await.unwrap() {
eprintln!("node {x} core task failed: {e:?}; failing scenario");
std::process::exit(1);
}
});

// Wait for stop signal or node completion
Expand All @@ -190,7 +196,11 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
node_context.stop(0, Some(Duration::from_secs(30))).await.unwrap();
}
_ = node_handle.fuse() => {
println!("Node {} handle completed", x);
// Every node here is a required participant; its handle
// completing without a stop signal means it went down
// unexpectedly, so fail the scenario.
eprintln!("Node {} handle completed unexpectedly; failing scenario", x);
std::process::exit(1);
}
}
});
Expand Down
56 changes: 44 additions & 12 deletions node/src/bin/stake_and_checkpoint.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -199,8 +199,14 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
let executor = cw_tokio::Runner::new(cfg);

executor.start(|node_context| async move {
let node_handle = node_context.clone().spawn(|ctx| async move {
run_node_local(ctx, flags, None, None).await.unwrap();
let node_handle = node_context.clone().spawn(move |ctx| async move {
// a coordinated shutdown (graceful stop or committee exit) returns
// ok; a genuine core task failure returns err and must fail the
// scenario instead of being masked as a clean node exit.
if let Err(e) = run_node_local(ctx, flags, None, None).await.unwrap() {
eprintln!("node {x} core task failed: {e:?}; failing scenario");
std::process::exit(1);
}
});

// Wait for stop signal or node completion
Expand All @@ -212,7 +218,11 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
node_context.stop(0, Some(Duration::from_secs(30))).await.unwrap();
}
_ = node_handle.fuse() => {
println!("Node {} handle completed", x);
// Every node here is a required participant; its handle
// completing without a stop signal means it went down
// unexpectedly, so fail the scenario.
eprintln!("Node {} handle completed unexpectedly; failing scenario", x);
std::process::exit(1);
}
}
});
Expand Down Expand Up @@ -424,7 +434,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {

// executor.start(|node_context| async move {
// let flags = get_node_flags(source_node);
// let node_handle = node_context.clone().spawn(|ctx| async move {
// let node_handle = node_context.clone().spawn(move |ctx| async move {
// run_node_with_runtime(ctx, flags, None).await.unwrap();
// });

Expand Down Expand Up @@ -527,8 +537,15 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
let executor = cw_tokio::Runner::new(cfg);

executor.start(|node_context| async move {
let node_handle = node_context.clone().spawn(|ctx| async move {
run_node_local(ctx, flags, Some(checkpoint_state), None).await.unwrap();
let node_handle = node_context.clone().spawn(move |ctx| async move {
// the checkpoint restarted node is a required participant: a genuine
// core task failure (err) must fail the scenario, not be masked.
if let Err(e) =
run_node_local(ctx, flags, Some(checkpoint_state), None).await.unwrap()
{
eprintln!("node {x} core task failed: {e:?}; failing scenario");
std::process::exit(1);
}
});

// Wait for stop signal or node completion
Expand All @@ -540,7 +557,11 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
node_context.stop(0, Some(Duration::from_secs(30))).await.unwrap();
}
_ = node_handle.fuse() => {
println!("Node {} handle completed", x);
// The checkpoint-restart node is a required participant; its
// handle completing without a stop signal means it went down
// unexpectedly, so fail the scenario.
eprintln!("Node {} handle completed unexpectedly; failing scenario", x);
std::process::exit(1);
}
}
});
Expand Down Expand Up @@ -575,8 +596,11 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
);
loop {
let mut all_ready = true;
//for idx in 0..num_nodes {
// Skip node0
// node0 is intentionally stopped for checkpoint copying, so it is
// excluded from this progress check; its expected clean shutdown is
// asserted by the thread join below (a join failure fails the
// scenario), and the checkpoint-restart participant is asserted to
// sync above. Every other required node must reach the target height.
for idx in 1..num_nodes {
let rpc_port = get_node_flags(idx as usize, &e2e_genesis_path_str).rpc_port;
match get_latest_height(rpc_port).await {
Expand Down Expand Up @@ -616,12 +640,17 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
// Join all node threads outside of async context to avoid runtime drop issues
println!("Waiting for all nodes to shut down...");

// Join the stopped node0 first if it exists
// Join the stopped node0 first if it exists. A thread join failure means the
// node panicked or was killed: that is a required participant going down
// unexpectedly, so fail the scenario instead of logging and continuing.
if let Some(node0) = node_runtimes.1 {
println!("Joining node0 thread...");
match node0.thread.join() {
Ok(_) => println!("Node0 thread joined successfully"),
Err(e) => println!("Node0 thread join failed: {:?}", e),
Err(e) => {
eprintln!("Node0 thread join failed: {:?}", e);
std::process::exit(1);
}
}
}

Expand All @@ -630,7 +659,10 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
println!("Waiting for node index {} to join...", idx);
match node_runtime.thread.join() {
Ok(_) => println!("Node index {} thread joined successfully", idx),
Err(e) => println!("Node index {} thread join failed: {:?}", idx, e),
Err(e) => {
eprintln!("Node index {} thread join failed: {:?}", idx, e);
std::process::exit(1);
}
}
}

Expand Down
Loading
Loading