diff --git a/Cargo.lock b/Cargo.lock index 0bec6c9e0..1afda0a4c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -172,7 +172,7 @@ version = "1.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -183,7 +183,7 @@ checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" dependencies = [ "anstyle", "once_cell_polyfill", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -1916,7 +1916,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -2010,6 +2010,7 @@ dependencies = [ "serde_yaml_ng", "sha2", "snap", + "spawned-concurrency 0.7.0", "tempfile", "thiserror 2.0.18", "tikv-jemallocator", @@ -2040,7 +2041,7 @@ dependencies = [ "serde", "serde_json", "snap", - "spawned-concurrency", + "spawned-concurrency 0.7.0", "thiserror 2.0.18", "tokio", "tokio-util", @@ -2100,7 +2101,7 @@ name = "ethlambda-network-api" version = "1.0.0" dependencies = [ "ethlambda-types", - "spawned-concurrency", + "spawned-concurrency 0.7.0", ] [[package]] @@ -2128,7 +2129,8 @@ dependencies = [ "secp256k1 0.30.0", "sha2", "snap", - "spawned-concurrency", + "spawned-concurrency 0.7.0", + "spawned-test", "thiserror 2.0.18", "tokio", "tokio-stream", @@ -2159,7 +2161,7 @@ dependencies = [ "libssz-types 0.3.0 (git+https://github.com/lambdaclass/libssz?rev=5cb14378ffdadcb51b387c003567e2630d0c7073)", "serde", "serde_json", - "spawned-concurrency", + "spawned-concurrency 0.7.0", "tokio", "tokio-util", "tower", @@ -2482,8 +2484,8 @@ dependencies = [ "serde", "sha2", "snap", - "spawned-concurrency", - "spawned-rt", + "spawned-concurrency 0.5.0", + "spawned-rt 0.5.0", "thiserror 2.0.18", "tokio", "tokio-stream", @@ -2867,6 +2869,21 @@ dependencies = [ "slab", ] +[[package]] +name = "generator" +version = "0.8.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1be81b31c9ce4caf7f3178eeb90662bb1864d3ad363ec6bdada3201adff19c37" +dependencies = [ + "cc", + "cfg-if", + "libc", + "log", + "rustversion", + "windows-link", + "windows-result 0.4.1", +] + [[package]] name = "generic-array" version = "0.14.7" @@ -4434,6 +4451,19 @@ version = "0.4.29" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5e5032e24019045c762d3c0f28f5b6b8bbf38563a65908389bf7978758920897" +[[package]] +name = "loom" +version = "0.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "419e0dc8046cb947daa77eb95ae174acfbddb7673b4151f56d1eed8e93fbfaca" +dependencies = [ + "cfg-if", + "generator", + "scoped-tls", + "tracing", + "tracing-subscriber", +] + [[package]] name = "lru" version = "0.16.4" @@ -4767,7 +4797,7 @@ version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -6057,7 +6087,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys 0.12.1", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -6180,6 +6210,12 @@ dependencies = [ "serde_json", ] +[[package]] +name = "scoped-tls" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e1cf6437eb19a8f4a6cc0f7dca544973b0b78843adbfeb3683d1a94a0024a294" + [[package]] name = "scopeguard" version = "1.2.0" @@ -6543,7 +6579,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3a766e1110788c36f4fa1c2b71b387a7815aa65f88ce0229841826633d93723e" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -6554,8 +6590,21 @@ checksum = "bc21166874e8cd7584ea795c223303160461f0bb1b571bc23e92ca2abb7c5149" dependencies = [ "futures", "pin-project-lite", - "spawned-macros", - "spawned-rt", + "spawned-macros 0.5.0", + "spawned-rt 0.5.0", + "thiserror 2.0.18", + "tracing", +] + +[[package]] +name = "spawned-concurrency" +version = "0.7.0" +source = "git+https://github.com/lambdaclass/spawned?rev=fc35abafe7a850aa27f4b511b60e28dc49b4a393#fc35abafe7a850aa27f4b511b60e28dc49b4a393" +dependencies = [ + "futures", + "pin-project-lite", + "spawned-macros 0.7.0", + "spawned-rt 0.7.0", "thiserror 2.0.18", "tracing", ] @@ -6571,6 +6620,16 @@ dependencies = [ "syn 2.0.117", ] +[[package]] +name = "spawned-macros" +version = "0.7.0" +source = "git+https://github.com/lambdaclass/spawned?rev=fc35abafe7a850aa27f4b511b60e28dc49b4a393#fc35abafe7a850aa27f4b511b60e28dc49b4a393" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.117", +] + [[package]] name = "spawned-rt" version = "0.5.0" @@ -6585,6 +6644,31 @@ dependencies = [ "tracing-subscriber", ] +[[package]] +name = "spawned-rt" +version = "0.7.0" +source = "git+https://github.com/lambdaclass/spawned?rev=fc35abafe7a850aa27f4b511b60e28dc49b4a393#fc35abafe7a850aa27f4b511b60e28dc49b4a393" +dependencies = [ + "ctrlc", + "loom", + "nix 0.30.1", + "tokio", + "tokio-stream", + "tokio-util", + "tracing", + "tracing-subscriber", +] + +[[package]] +name = "spawned-test" +version = "0.7.0" +source = "git+https://github.com/lambdaclass/spawned?rev=fc35abafe7a850aa27f4b511b60e28dc49b4a393#fc35abafe7a850aa27f4b511b60e28dc49b4a393" +dependencies = [ + "spawned-concurrency 0.7.0", + "spawned-rt 0.7.0", + "tracing", +] + [[package]] name = "sphincs" version = "0.1.0" @@ -6762,7 +6846,7 @@ dependencies = [ "getrandom 0.4.2", "once_cell", "rustix 1.1.4", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -7549,7 +7633,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index e13475448..bc69fda7d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -90,8 +90,13 @@ hex = "0.4" hmac = "0.12" base64 = "0.22" -spawned-concurrency = "0.5.0" -spawned-rt = "0.5.0" +# spawned main b1a17e6c (v0.7.0 plus canary and drop-panic fixes) plus the unreleased spawned-test crate (lambdaclass/spawned#198). Every +# spawned crate must name the same revision so they share one set of types. +# `metrics`: the actor runtime's own counters (mailbox depth, queue wait, +# handler time, actor exits), appended to `/metrics` by ethlambda-rpc. +spawned-concurrency = { git = "https://github.com/lambdaclass/spawned", rev = "fc35abafe7a850aa27f4b511b60e28dc49b4a393", features = ["metrics"] } +spawned-rt = { git = "https://github.com/lambdaclass/spawned", rev = "fc35abafe7a850aa27f4b511b60e28dc49b4a393" } +spawned-test = { git = "https://github.com/lambdaclass/spawned", rev = "fc35abafe7a850aa27f4b511b60e28dc49b4a393" } tokio = "1.0" tokio-util = "0.7" diff --git a/README.md b/README.md index bbe23f7cc..0e024b3d7 100644 --- a/README.md +++ b/README.md @@ -494,6 +494,8 @@ Some features we are looking to implement in the near future, in order of priori - [Add a guest program and ZK proving of the STF](https://github.com/lambdaclass/ethlambda/issues/156) - Rewrite the STF in the concrete programming language to enable formal verification +The [testing and verification roadmap](docs/testing_roadmap.md) proposes how we plan to test the client: a seeded chain-logic simulator first, then model checking, fault injection and long-running soak tests. + ### Experimental features We have a proof-of-concept formalization of a part of the state transition function in Lean4 in PR [#269](https://github.com/lambdaclass/ethlambda/pull/269). diff --git a/bin/ethlambda/Cargo.toml b/bin/ethlambda/Cargo.toml index 330940e28..f0b6f7bb5 100644 --- a/bin/ethlambda/Cargo.toml +++ b/bin/ethlambda/Cargo.toml @@ -36,6 +36,7 @@ rayon.workspace = true libssz.workspace = true libssz-types.workspace = true +spawned-concurrency.workspace = true tokio.workspace = true tokio-util.workspace = true tracing.workspace = true diff --git a/bin/ethlambda/src/http_server.rs b/bin/ethlambda/src/http_server.rs new file mode 100644 index 000000000..47d425c6a --- /dev/null +++ b/bin/ethlambda/src/http_server.rs @@ -0,0 +1,268 @@ +//! The HTTP server (API, metrics and debug routes) as a supervised actor. +//! +//! The one piece of the node that restarts safely, so the one piece under a +//! supervisor. It holds no state of its own beyond what each run builds afresh +//! from cloned handles (the store, the P2P actor's protocol ref, the shared +//! pools and controllers), so a fresh instance after a failure serves what the +//! old one did. What a restart loses is per connection: open event streams end +//! and their clients reconnect, and the Beacon API's fee recipients, registered +//! through `prepare_beacon_proposer`, are forgotten until the validator client +//! sends them again, which it does every epoch. +//! +//! The chain and P2P actors are deliberately not supervised; see `run_node`. +//! +//! Before this, a server that failed logged one error and the node ran on with +//! no API and no metrics until someone restarted the process. + +use std::{future::Future, io, pin::Pin, sync::Arc, time::Duration}; + +use spawned_concurrency::actor; +use spawned_concurrency::message::Message; +use spawned_concurrency::supervision::{ + Backoff, SupervisedChild, Supervisor, SupervisorConfig, SupervisorStrategy, +}; +use spawned_concurrency::tasks::{Actor, Context, Handler}; +use tokio_util::sync::CancellationToken; +use tracing::{error, info, warn}; + +/// One run of the HTTP server: bind its listeners, then serve until the token +/// is cancelled. Called once per actor generation. +pub(crate) type Serve = Arc< + dyn Fn(CancellationToken) -> Pin> + Send>> + Send + Sync, +>; + +/// The restart budget: at most three restarts a minute, the first after a +/// second and backing off to ten. A server that keeps failing faster than that +/// (its port taken by another process, say) exhausts the budget, the +/// supervisor gives up, and the node runs on without HTTP as it did before +/// supervision, with an error saying so. +pub(crate) fn supervisor_config() -> SupervisorConfig { + SupervisorConfig { + strategy: SupervisorStrategy::OneForOne, + max_restarts: 3, + restart_window: Duration::from_secs(60), + backoff: Backoff { + initial: Duration::from_secs(1), + maximum: Duration::from_secs(10), + stable_after: Duration::from_secs(60), + ..Backoff::default() + }, + ..SupervisorConfig::default() + } +} + +pub(crate) struct HttpServer { + serve: Serve, + /// Cancelled when this generation stops, so axum closes its listeners + /// gracefully. + shutdown: CancellationToken, +} + +impl HttpServer { + pub(crate) fn new(serve: Serve) -> Self { + Self { + serve, + shutdown: CancellationToken::new(), + } + } +} + +#[actor] +impl HttpServer { + /// Serve on an owned helper, which the actor system cancels and joins when + /// this generation stops, and stop the actor when the server returns: the + /// exit is what tells the supervisor to build a fresh one. + #[started] + async fn start_serving(&mut self, ctx: &Context) { + let server = (self.serve)(self.shutdown.clone()); + let owner = ctx.clone(); + let served = ctx.spawn_helper(async move { + match server.await { + Ok(()) => info!("HTTP server stopped serving"), + Err(err) => error!(%err, "HTTP server failed"), + } + // A stopped Permanent child is restarted while its supervisor is + // open. During node shutdown this generation is already stopping + // and nothing restarts it. + owner.stop(); + }); + if let Err(err) = served { + warn!(%err, "Could not start the HTTP server"); + ctx.stop(); + } + } + + #[stopped] + async fn stop_serving(&mut self, _ctx: &Context) { + self.shutdown.cancel(); + } +} + +/// Fault injection only: crash the HTTP server as a panicking handler would. +/// Sent by [`install_fault_injection`], never by the node itself. +pub(crate) struct InjectCrash; + +impl Message for InjectCrash { + type Result = (); +} + +impl Handler for HttpServer { + async fn handle(&mut self, _msg: InjectCrash, _ctx: &Context) { + panic!("fault injection: crashing the HTTP server"); + } +} + +/// Log every restart of the supervised HTTP server, and the supervisor giving +/// up if it ever exhausts its budget. Spawned does not log either itself. +pub(crate) fn watch(http: SupervisedChild, supervisor: Supervisor) { + tokio::spawn(async move { + let mut seen = http.snapshot().generation; + while http.wait_for_generation(seen).await.is_ok() { + let snapshot = http.snapshot(); + warn!( + child = %snapshot.name, + generation = snapshot.generation, + last_exit = ?snapshot.last_exit, + last_error = ?snapshot.last_error, + "Supervisor restarted the HTTP server" + ); + seen = snapshot.generation; + } + supervisor.join().await; + if let Some(failure) = supervisor.failure() { + error!( + %failure, + "HTTP server supervisor gave up; the node runs on without its HTTP server" + ); + } + }); +} + +/// The environment variable that arms [`install_fault_injection`]. +const FAULT_INJECTION_ENV: &str = "ETHLAMBDA_FAULT_INJECTION"; + +/// With `ETHLAMBDA_FAULT_INJECTION=1`, crash the current HTTP server on each +/// SIGUSR1, to exercise supervised recovery on a test node: +/// `kill -USR1 `. Off by default and out of every network-facing path. +pub(crate) fn install_fault_injection(http: SupervisedChild) { + if std::env::var(FAULT_INJECTION_ENV).as_deref() != Ok("1") { + return; + } + #[cfg(unix)] + { + use tokio::signal::unix::{SignalKind, signal}; + let mut usr1 = match signal(SignalKind::user_defined1()) { + Ok(usr1) => usr1, + Err(err) => { + warn!(%err, "Fault injection requested but SIGUSR1 cannot be handled"); + return; + } + }; + warn!("Fault injection armed: SIGUSR1 crashes the HTTP server"); + tokio::spawn(async move { + while usr1.recv().await.is_some() { + match http.current() { + Ok(server) => { + warn!(actor = %server.id(), "Fault injection: crashing the HTTP server"); + let _ = server.send(InjectCrash); + } + Err(unavailable) => { + warn!(%unavailable, "Fault injection: no HTTP server to crash") + } + } + } + }); + } +} + +#[cfg(test)] +mod tests { + use std::sync::atomic::{AtomicUsize, Ordering}; + + use spawned_concurrency::supervision::ChildSpec; + use spawned_concurrency::system::ActorSystem; + + use super::*; + + const RECOVERY: Duration = Duration::from_secs(15); + + /// A server that counts its runs; the first `failures` runs fail at once, + /// the rest serve until cancelled. + fn counting_server(failures: usize) -> (Serve, Arc) { + let runs = Arc::new(AtomicUsize::new(0)); + let counter = runs.clone(); + let serve: Serve = Arc::new(move |shutdown: CancellationToken| { + let run = counter.fetch_add(1, Ordering::SeqCst); + Box::pin(async move { + if run < failures { + return Err(io::Error::other("address in use")); + } + shutdown.cancelled().await; + Ok(()) + }) + }); + (serve, runs) + } + + async fn supervised( + system: &ActorSystem, + serve: Serve, + ) -> (SupervisedChild, Supervisor) { + let group = system.async_group("network", 1).unwrap(); + let mut tree = system.supervisor(supervisor_config()).unwrap(); + let http = tree + .add(ChildSpec::new("http", &group, move |_| { + let serve = serve.clone(); + async move { Ok::<_, std::convert::Infallible>(HttpServer::new(serve)) } + })) + .unwrap(); + let supervisor = tree.start().await.unwrap(); + (http, supervisor) + } + + /// The fault-injection crash is a real handler panic: the supervisor + /// builds a fresh actor, which serves again. + #[tokio::test] + async fn a_crashed_server_is_replaced_and_serves_again() { + let system = ActorSystem::new().unwrap(); + let (serve, runs) = counting_server(0); + let (http, _supervisor) = supervised(&system, serve).await; + let first = http.wait_ready().await.unwrap(); + let generation = http.snapshot().generation; + + first.send(InjectCrash).unwrap(); + let second = tokio::time::timeout(RECOVERY, http.wait_for_generation(generation)) + .await + .expect("the supervisor restarts the server") + .unwrap(); + + assert_ne!(first.id(), second.id()); + assert!(matches!( + http.snapshot().last_exit, + Some(spawned_concurrency::ExitReason::Panic(_)) + )); + assert_eq!(runs.load(Ordering::SeqCst), 2); + assert!(system.shutdown(RECOVERY).await.complete); + } + + /// A server that fails to bind is retried, and one that keeps failing + /// exhausts the budget and leaves the supervisor failed, not looping. + #[tokio::test] + async fn a_server_that_keeps_failing_exhausts_the_budget() { + let system = ActorSystem::new().unwrap(); + let (serve, runs) = counting_server(usize::MAX); + let (_http, supervisor) = supervised(&system, serve).await; + + tokio::time::timeout(Duration::from_secs(60), supervisor.join()) + .await + .expect("the supervisor gives up within its budget"); + + assert!(supervisor.failure().is_some()); + // The first run and one per restart the budget allows. + assert_eq!( + runs.load(Ordering::SeqCst), + supervisor_config().max_restarts + 1 + ); + assert!(system.shutdown(RECOVERY).await.complete); + } +} diff --git a/bin/ethlambda/src/main.rs b/bin/ethlambda/src/main.rs index db3f62368..ac02d1042 100644 --- a/bin/ethlambda/src/main.rs +++ b/bin/ethlambda/src/main.rs @@ -6,6 +6,7 @@ mod cli; mod command; mod era; mod fd_limit; +mod http_server; mod keygen; mod network; mod slashing_protection; @@ -31,12 +32,15 @@ static ALLOC: tikv_jemallocator::Jemalloc = tikv_jemallocator::Jemalloc; #[unsafe(export_name = "malloc_conf")] static malloc_conf: &[u8] = b"prof:true,prof_active:true,lg_prof_sample:19\0"; +use spawned_concurrency::ActorId; +use spawned_concurrency::supervision::{ChildSpec, SupervisedChild}; +use spawned_concurrency::system::{ActorSystem, ShutdownReport}; use std::{ collections::{BTreeMap, HashMap, HashSet}, net::{IpAddr, SocketAddr}, path::{Path, PathBuf}, sync::Arc, - time::SystemTime, + time::{Duration, Instant, SystemTime}, }; use tokio_util::sync::CancellationToken; @@ -53,7 +57,9 @@ use ethlambda_network_api::{ }; use ethlambda_p2p::{ LeanWireConfig, P2P, PeerId, SwarmConfig, WireConfig, attestation_subscription_subnets, - build_swarm, discovery::DiscoverySpawnConfig, parse_enrs, + build_swarm, + discovery::{DiscoverySpawnConfig, spawn_discovery}, + parse_enrs, }; use ethlambda_state_transition::beacon::fork_choice; use ethlambda_types::primitives::{H256, HashTreeRoot as _}; @@ -229,11 +235,30 @@ fn init_info_stderr_logging() -> eyre::Result<()> { .wrap_err("failed to set global tracing subscriber") } +/// How long a graceful shutdown may take, from the signal to the last actor +/// joined. +/// +/// Comfortably under Docker's default 10 s `docker stop` grace period, so the +/// node reports what it could not stop instead of being SIGKILLed with no +/// record of why. Before this deadline existed a clean shutdown during +/// catch-up took 14.3 s and was cut off by that SIGKILL (1d7902da). +const SHUTDOWN_DEADLINE: Duration = Duration::from_secs(8); + +/// The part of [`SHUTDOWN_DEADLINE`] the HTTP server gets to close before the +/// actors are stopped. An open `/events` stream would otherwise hold axum's +/// graceful shutdown, and the actors' share of the deadline with it. +const HTTP_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(2); + /// A node that has started, whichever chain it follows. /// /// What [`wait_for_shutdown`] needs to stop, and all [`run_node`] has left to /// assemble once the chain-specific half is done. struct RunningNode { + /// Owns every actor below and the threads they run on. Shutting it down + /// stops them in reverse admission order: the HTTP server, then P2P, the + /// ingresses, before the chain actor they feed. See [`run_node`] for why + /// that is the order. + system: ActorSystem, p2p: P2P, /// The chain actor, on both lean and mainnet: the beacon follower runs one /// too, so there is always exactly one to stop and join. `run_node` @@ -241,10 +266,8 @@ struct RunningNode { /// this a plain `BlockChain` rather than an `Option` describing a state /// the node never reaches. blockchain: BlockChain, - /// The HTTP server task. Returns once `shutdown` is cancelled. - http: tokio::task::JoinHandle<()>, - /// Cancelled by [`wait_for_shutdown`] to stop `http`. - shutdown: CancellationToken, + /// The supervised HTTP server; whichever generation is current. + http: SupervisedChild, } /// What one chain claims to serve, in the entries its ENR publishes and its @@ -769,8 +792,55 @@ async fn run_node(options: Options) -> eyre::Result<()> { // `/lean/v0/node/identity` endpoint reports it. let local_peer_id = built.local_peer_id.to_string(); - // `P2P::spawn` starts the discv5 server from this and owns the resulting - // handle. + // discv5 binds its UDP socket here, before any actor runs, so a busy port + // aborts startup with nothing to stop. `P2P::spawn` owns the handle. + let discovery = match discovery { + Some(config) => Some( + spawn_discovery(config) + .await + .wrap_err("failed to start discv5 discovery")?, + ), + None => None, + }; + + // The owner of every actor this node runs, and of the threads they run + // on. Two groups: + // + // - `chain`: one native thread for the chain actor, whose handlers are + // long synchronous CPU (block import, state transition, signing) that + // would starve a shared runtime. One worker with a one-worker helper + // runtime is what `Backend::Thread` gave it before. + // - `network`: the P2P actor and the HTTP server, on a multi-thread pool + // sized like the default Tokio runtime they used to share. + // + // Admission order is shutdown order reversed, so the chain actor is + // admitted first and the ingresses that feed it, P2P and then HTTP, are + // stopped before it. + // + // Only the HTTP server is supervised (see `http_server`). The other two + // are not, on purpose: + // + // - The chain actor is the consensus-critical singleton. Its in-memory + // state (pending blocks and columns, the sync gate, the aggregation + // session, the validator keys' prepared XMSS signing state) is not + // rebuilt from the store, and a fresh instance would sign again from a + // key manager that does not know what the old one signed. P2P also holds + // its `ActorRef` from `InitBlockChain`, which a replacement would not + // rebind. + // - P2P owns the libp2p swarm and the discv5 server, built once here with + // their sockets bound; a fresh instance would need both rebuilt, and the + // chain actor and the HTTP server hold refs to the old one. + // + // A crash in either still just stops that actor, as before; supervising + // them needs that state recovery and ref rebinding first. + let system = ActorSystem::new().wrap_err("failed to create the actor system")?; + let chain_group = system + .thread_group("chain", 1) + .wrap_err("failed to create the chain execution group")?; + let network_group = system + .async_group("network", network_workers()) + .wrap_err("failed to create the network execution group")?; + // Filled by gossip and the Beacon API's payload attestation endpoint, read // by block production and `GET .../pool/payload_attestations`. let payload_attestation_pool = @@ -783,41 +853,18 @@ async fn run_node(options: Options) -> eyre::Result<()> { // block production and the contribution endpoint. let sync_committee_pool = ethlambda_state_transition::beacon::sync_committee_pool::SharedSyncCommitteePool::default(); - let p2p = P2P::spawn( - built, - setup.store.clone(), - setup.node_names, - discovery, - payload_attestation_pool.clone(), - builder_market.clone(), - sync_committee_pool.clone(), - ) - .await - .wrap_err("failed to start discv5 discovery")?; - - let shutdown = CancellationToken::new(); - let rpc_shutdown = shutdown.clone(); + // What the HTTP server and P2P take from values the chain actor's spawn + // below moves. let rpc_store = setup.store.clone(); - let rpc_aggregator = aggregator.clone(); - let rpc_sync_status = sync_status.clone(); let rpc_events = events.clone(); - let rpc_p2p = p2p.actor_ref().to_rpc_to_p2p_ref(); + let rpc_sync_status = sync_status.clone(); + let p2p_store = setup.store.clone(); let rpc_custody_columns = match &setup.chain { ChainActor::Beacon { custody_columns, .. } => ethlambda_rpc::CustodyColumns(custody_columns.clone()), ChainActor::Lean(..) => ethlambda_rpc::CustodyColumns::default(), }; - let rpc_identity = ethlambda_rpc::BeaconIdentity { - metadata: p2p.local_metadata(), - ..beacon_identity( - p2p.local_enr().unwrap_or_default(), - common.discovery.advertise_ip, - common.gossipsub_port, - common.discovery.port, - &local_peer_id, - ) - }; // Block production builds its payloads with the same execution client the // chain actor validates them with. let rpc_engine = match &setup.chain { @@ -832,46 +879,11 @@ async fn run_node(options: Options) -> eyre::Result<()> { // beacon directory never carries, so calling one panicked that request. let serves_beacon_api = setup.store.chain() == ethlambda_storage::Chain::Beacon; - let http = tokio::spawn(async move { - let served = if serves_beacon_api { - ethlambda_rpc::start_beacon_rpc_server( - rpc_config, - rpc_store, - rpc_sync_status, - ethlambda_rpc::BeaconApiHandles { - p2p: rpc_p2p, - payload_attestation_pool: payload_attestation_pool.clone(), - builder_market: builder_market.clone(), - sync_committee_pool: sync_committee_pool.clone(), - custody_columns: rpc_custody_columns, - engine: rpc_engine, - events: rpc_events, - client_version: version::engine_client_version(), - identity: rpc_identity, - }, - local_peer_id, - rpc_shutdown, - ) - .await - } else { - ethlambda_rpc::start_rpc_server( - rpc_config, - rpc_store, - rpc_aggregator, - rpc_sync_status, - local_peer_id, - rpc_events, - rpc_shutdown, - ) - .await - }; - let _ = served.inspect_err(|err| error!(%err, "RPC server failed")); - }); - + // The chain actor before P2P: see the `system` comment above. let mut anchor_envelope_fetch: Option<(Vec, H256)> = None; let blockchain = match setup.chain { ChainActor::Lean(validator_keys, config) => { - BlockChain::spawn(setup.store, validator_keys, config, events) + BlockChain::spawn(setup.store, validator_keys, config, events, &chain_group) } ChainActor::Beacon { custody_columns, @@ -890,9 +902,95 @@ async fn run_node(options: Options) -> eyre::Result<()> { custody_columns, engine, safe_slots_to_import_optimistically, + &chain_group, ) } + } + .wrap_err("failed to start the chain actor")?; + + let p2p = P2P::spawn( + built, + p2p_store, + setup.node_names, + discovery, + payload_attestation_pool.clone(), + builder_market.clone(), + sync_committee_pool.clone(), + &network_group, + ) + .wrap_err("failed to start the P2P actor")?; + + let rpc_aggregator = aggregator.clone(); + let rpc_p2p = p2p.actor_ref().to_rpc_to_p2p_ref(); + let rpc_identity = ethlambda_rpc::BeaconIdentity { + metadata: p2p.local_metadata(), + ..beacon_identity( + p2p.local_enr().unwrap_or_default(), + common.discovery.advertise_ip, + common.gossipsub_port, + common.discovery.port, + &local_peer_id, + ) }; + // One run of the server per actor generation, from clones of everything + // it serves: see `http_server` for why that makes a restart safe. + let serve: http_server::Serve = Arc::new(move |shutdown| { + let rpc_config = rpc_config.clone(); + let rpc_store = rpc_store.clone(); + let rpc_sync_status = rpc_sync_status.clone(); + let rpc_events = rpc_events.clone(); + let local_peer_id = local_peer_id.clone(); + if serves_beacon_api { + let handles = ethlambda_rpc::BeaconApiHandles { + p2p: rpc_p2p.clone(), + payload_attestation_pool: payload_attestation_pool.clone(), + builder_market: builder_market.clone(), + sync_committee_pool: sync_committee_pool.clone(), + custody_columns: rpc_custody_columns.clone(), + engine: rpc_engine.clone(), + events: rpc_events, + client_version: version::engine_client_version(), + identity: rpc_identity.clone(), + }; + Box::pin(ethlambda_rpc::start_beacon_rpc_server( + rpc_config, + rpc_store, + rpc_sync_status, + handles, + local_peer_id, + shutdown, + )) + } else { + Box::pin(ethlambda_rpc::start_rpc_server( + rpc_config, + rpc_store, + rpc_aggregator.clone(), + rpc_sync_status, + local_peer_id, + rpc_events, + shutdown, + )) + } + }); + // Admitted last, so stopped first. See `http_server` for the budget. + let mut http_tree = system + .supervisor(http_server::supervisor_config()) + .wrap_err("failed to create the HTTP server supervisor")?; + let http = http_tree + .add( + ChildSpec::new("http", &network_group, move |_| { + let serve = serve.clone(); + async move { Ok::<_, std::convert::Infallible>(http_server::HttpServer::new(serve)) } + }) + .shutdown_timeout(HTTP_SHUTDOWN_TIMEOUT), + ) + .wrap_err("failed to declare the HTTP server")?; + let http_supervisor = http_tree + .start() + .await + .wrap_err("failed to start the HTTP server")?; + http_server::watch(http.clone(), http_supervisor); + http_server::install_fault_injection(http.clone()); let p2p_ref = p2p.actor_ref(); let p2p_to_block_chain = p2p_ref.to_block_chain_to_p2p_ref(); @@ -931,13 +1029,23 @@ async fn run_node(options: Options) -> eyre::Result<()> { } wait_for_shutdown(RunningNode { + system, p2p, blockchain, http, - shutdown, }) - .await; - Ok(()) + .await +} + +/// Async workers for the `network` group: what a default `#[tokio::main]` +/// runtime would get, so P2P keeps the parallelism it had when it shared that +/// runtime. One under Shadow, for the reason `run_node` runs single-threaded +/// there. +fn network_workers() -> usize { + if cfg!(feature = "shadow-integration") { + return 1; + } + std::thread::available_parallelism().map_or(4, std::num::NonZero::get) } /// The root of the anchor block when it is a gloas block whose execution @@ -973,21 +1081,31 @@ fn check_custody_set(custody_columns: &[u64]) -> eyre::Result<()> { Ok(()) } -/// Wait for ctrl-c, then stop and join whatever is running. +/// Wait for SIGINT or SIGTERM, then stop everything within +/// [`SHUTDOWN_DEADLINE`]. /// -/// A 2nd, 3rd and 4th ctrl-c escalate to `std::process::exit(1)` rather than -/// leaving shutdown stuck on an actor that hangs in `stop()`/`join()`. +/// The actor system stops everything in reverse admission order: the HTTP +/// server (given [`HTTP_SHUTDOWN_TIMEOUT`] of it), then P2P, then the chain +/// actor. A shutdown that misses it names what is still running and +/// exits non-zero rather than wait on it: Docker would SIGKILL the process two +/// seconds later anyway, and then nothing would say which actor held it. +/// +/// A 2nd, 3rd and 4th ctrl-c still escalate to `std::process::exit(1)`. /// /// Shared by both chains. Mainnet used to park on `pending()` here, so a /// follower killed mid-write left recovery to do; it now tears down the same /// way the lean node does. -async fn wait_for_shutdown(node: RunningNode) { +async fn wait_for_shutdown(node: RunningNode) -> eyre::Result<()> { info!("Node initialized"); - // 1st ctrl+c: start graceful shutdown - tokio::signal::ctrl_c().await.ok(); + let signal = shutdown_signal().await; + let started = Instant::now(); - info!("Shutdown signal received, stopping actors and servers..."); + info!( + signal, + deadline_secs = SHUTDOWN_DEADLINE.as_secs(), + "Shutdown signal received, stopping actors and servers..." + ); tokio::spawn(async move { // This can be turned into a loop @@ -1004,18 +1122,79 @@ async fn wait_for_shutdown(node: RunningNode) { std::process::exit(1); }); - let blockchain_ref = node.blockchain.actor_ref().clone(); - let p2p_ref = node.p2p.actor_ref().clone(); - - blockchain_ref.context().stop(); - p2p_ref.context().stop(); - node.shutdown.cancel(); + let RunningNode { + system, + p2p, + blockchain, + http, + } = node; + // Read now: the report names unfinished actors by id only. + let mut actors = vec![ + (p2p.actor_ref().id(), "p2p"), + (blockchain.actor_ref().id(), "blockchain"), + ]; + if let Ok(server) = http.current() { + actors.push((server.id(), "http")); + } + + let report = system.shutdown(SHUTDOWN_DEADLINE).await; + let elapsed_ms = started.elapsed().as_millis() as u64; + if report.complete { + info!(elapsed_ms, "Shutdown complete"); + return Ok(()); + } + log_incomplete_shutdown(&report, &actors, elapsed_ms); + // The unfinished work keeps its threads; returning would have the runtime + // wait on them with no deadline at all. + std::process::exit(1); +} - blockchain_ref.join().await; - p2p_ref.join().await; - let _ = node.http.await; +/// Name what a shutdown that missed [`SHUTDOWN_DEADLINE`] left running. +fn log_incomplete_shutdown(report: &ShutdownReport, actors: &[(ActorId, &str)], elapsed_ms: u64) { + let name = |id: &ActorId| { + actors + .iter() + .find(|(known, _)| known == id) + .map_or_else(|| id.to_string(), |(_, name)| (*name).to_string()) + }; + let unfinished_actors: Vec = report.unfinished_actors.iter().map(name).collect(); + let exits: Vec = report + .exits + .iter() + .map(|exit| format!("{}: {:?}", name(&exit.id), exit.reason)) + .collect(); + error!( + elapsed_ms, + deadline_secs = SHUTDOWN_DEADLINE.as_secs(), + unfinished_groups = ?report.unfinished_groups, + ?unfinished_actors, + ?exits, + supervisor_failures = ?report.supervisor_failures, + group_failures = ?report.group_failures, + "Shutdown deadline exceeded; exiting with work still running" + ); +} - info!("Shutdown complete"); +/// Resolve on the first SIGINT (ctrl-c) or SIGTERM, naming which one. +/// +/// SIGTERM is what `docker stop`, systemd and Kubernetes send, so a container +/// no longer needs `--stop-signal SIGINT` to shut down gracefully. +async fn shutdown_signal() -> &'static str { + #[cfg(unix)] + { + use tokio::signal::unix::{SignalKind, signal}; + match signal(SignalKind::terminate()) { + Ok(mut terminate) => { + return tokio::select! { + _ = tokio::signal::ctrl_c() => "SIGINT", + _ = terminate.recv() => "SIGTERM", + }; + } + Err(err) => warn!(%err, "Cannot listen for SIGTERM; only ctrl-c shuts down gracefully"), + } + } + tokio::signal::ctrl_c().await.ok(); + "SIGINT" } /// The ENRs to start from when `--bootnodes` was not given. diff --git a/crates/blockchain/src/lib.rs b/crates/blockchain/src/lib.rs index 251048b7a..c927002ba 100644 --- a/crates/blockchain/src/lib.rs +++ b/crates/blockchain/src/lib.rs @@ -38,9 +38,8 @@ use crate::sync_status::SyncStatusTracker; use spawned_concurrency::actor; use spawned_concurrency::error::ActorError; use spawned_concurrency::protocol; -use spawned_concurrency::tasks::{ - Actor, ActorRef, ActorStart, Backend, Context, Handler, send_after, -}; +use spawned_concurrency::system::{ExecutionGroup, SystemError}; +use spawned_concurrency::tasks::{Actor, ActorRef, ActorStart, Context, Handler, send_after}; use tokio_util::sync::CancellationToken; use tracing::{debug, error, info, trace, warn}; @@ -234,12 +233,16 @@ impl BlockChain { /// touched it, so this is a programming error to catch here rather than /// a runtime condition to branch on. Use [`Self::spawn_beacon`] for a /// beacon store. + /// + /// The actor is admitted to `group`, which owns its thread and stops it on + /// shutdown; see [`Self::start_actor`] for what kind of group it needs. pub fn spawn( store: Store, validator_keys: HashMap, config: BlockChainConfig, events: EventBus, - ) -> BlockChain { + group: &ExecutionGroup, + ) -> Result { assert_eq!( store.chain(), Chain::Lean, @@ -311,6 +314,7 @@ impl BlockChain { Vec::new(), None, constants::SAFE_SLOTS_TO_IMPORT_OPTIMISTICALLY, + group, ) } @@ -345,7 +349,8 @@ impl BlockChain { custody_columns: Vec, engine: Option, safe_slots_to_import_optimistically: u64, - ) -> BlockChain { + group: &ExecutionGroup, + ) -> Result { assert_eq!( store.chain(), Chain::Beacon, @@ -363,6 +368,7 @@ impl BlockChain { custody_columns, engine, safe_slots_to_import_optimistically, + group, ) } @@ -380,6 +386,13 @@ impl BlockChain { /// layer, so [`BlockChain::spawn`] passes an empty vector and `None`. It /// passes the specification's own default for /// `safe_slots_to_import_optimistically`, which nothing on that arm reads. + /// + /// `group` must give the actor a thread of its own: its handlers are long + /// synchronous CPU (block import, state transition, signing) that would + /// starve a shared async runtime. `run_node` passes a one-worker + /// `thread_group`, which is what `Backend::Thread` gave it before: a + /// dedicated native thread driving the handlers, plus a one-worker Tokio + /// runtime for its timers and the aggregation worker's `spawn_blocking`. #[allow(clippy::too_many_arguments)] fn start_actor( mut store: Store, @@ -390,7 +403,8 @@ impl BlockChain { custody_columns: Vec, engine: Option, safe_slots_to_import_optimistically: u64, - ) -> BlockChain { + group: &ExecutionGroup, + ) -> Result { let genesis_time = store.config().genesis_time; // Payloads verified before a restart have no recorded verdict, since @@ -430,8 +444,9 @@ impl BlockChain { events, duties, } - // Own thread: these handlers are long synchronous CPU that starves a shared runtime. - .start_with_backend(Backend::Thread); + // Own thread: these handlers are long synchronous CPU that starves a + // shared runtime. See this function's doc for what `group` must be. + .start_on(group)?; let time_until_genesis = (SystemTime::UNIX_EPOCH + Duration::from_secs(genesis_time)) .duration_since(SystemTime::now()) .unwrap_or_default(); @@ -440,7 +455,7 @@ impl BlockChain { handle.context(), block_chain_protocol::Tick, ); - BlockChain { handle } + Ok(BlockChain { handle }) } pub fn actor_ref(&self) -> &ActorRef { @@ -4437,7 +4452,9 @@ mod tests { }, }; - let _ = BlockChain::spawn(store, HashMap::new(), config, EventBus::default()); + let system = spawned_concurrency::system::ActorSystem::new().unwrap(); + let group = system.thread_group("chain", 1).unwrap(); + let _ = BlockChain::spawn(store, HashMap::new(), config, EventBus::default(), &group); } #[test] @@ -4452,6 +4469,8 @@ mod tests { DEFAULT_MILLISECONDS_PER_SLOT, ); + let system = spawned_concurrency::system::ActorSystem::new().unwrap(); + let group = system.thread_group("chain", 1).unwrap(); let _ = BlockChain::spawn_beacon( store, SyncStatusController::default(), @@ -4459,6 +4478,7 @@ mod tests { Vec::new(), None, constants::SAFE_SLOTS_TO_IMPORT_OPTIMISTICALLY, + &group, ); } diff --git a/crates/net/p2p/Cargo.toml b/crates/net/p2p/Cargo.toml index 22154e141..e9b19a188 100644 --- a/crates/net/p2p/Cargo.toml +++ b/crates/net/p2p/Cargo.toml @@ -83,3 +83,8 @@ libssz-types.workspace = true sha2 = "0.10" hex.workspace = true + +[dev-dependencies] +# Mock chain actors for the forwarding tests: an `ActorRef` stands in for +# any protocol ref, so tests need no hand-written protocol stubs. +spawned-test.workspace = true diff --git a/crates/net/p2p/src/beacon/column_checks.rs b/crates/net/p2p/src/beacon/column_checks.rs index 91fbd1efe..cdfa906a9 100644 --- a/crates/net/p2p/src/beacon/column_checks.rs +++ b/crates/net/p2p/src/beacon/column_checks.rs @@ -107,89 +107,22 @@ fn count_drop(outcome: Outcome) { #[cfg(test)] mod tests { - use std::sync::Arc; + use std::time::Duration; - use ethlambda_network_api::{ - AggregateArrival, BlockAnnouncement, BlockArrival, BlockSource, P2PToBlockChain, + use ethlambda_network_api::ToP2PToBlockChainRef as _; + use ethlambda_network_api::p2p_to_block_chain::{ + DataColumnSidecarsAwaitingParent, NewDataColumnSidecars, }; - use ethlambda_types::attestation::{SignedAggregatedAttestation, SignedAttestation}; use ethlambda_types::beacon::config::Config; - use ethlambda_types::beacon::containers::{SignedAggregateAndProof, SignedBeaconBlock, gloas}; - use ethlambda_types::beacon::primitives::{Root, ValidatorIndex}; - use spawned_concurrency::error::ActorError; - use tokio::sync::mpsc; + use ethlambda_types::beacon::containers::gloas; + use ethlambda_types::beacon::primitives::Root; + use spawned_test::tasks::Mock; use super::*; use crate::test_support::{unconnected_beacon_server, valid_shaped_sidecar}; - /// What [`check_and_forward`] sent the chain actor. - #[derive(Debug, PartialEq)] - enum Forwarded { - Checked(Vec), - AwaitingParent(Vec), - } - - /// A chain actor stand-in that reports every sidecar batch it receives. - struct RecordingChain(mpsc::UnboundedSender); - - impl P2PToBlockChain for RecordingChain { - fn new_block( - &self, - _block: SignedBeaconBlock, - _source: BlockSource, - _arrival: BlockArrival, - _announcement: BlockAnnouncement, - ) -> Result<(), ActorError> { - Ok(()) - } - fn new_attestation(&self, _attestation: SignedAttestation) -> Result<(), ActorError> { - Ok(()) - } - fn new_aggregated_attestation( - &self, - _attestation: SignedAggregatedAttestation, - ) -> Result<(), ActorError> { - Ok(()) - } - fn new_data_column_sidecars( - &self, - sidecars: Vec, - ) -> Result<(), ActorError> { - let _ = self.0.send(Forwarded::Checked(sidecars)); - Ok(()) - } - fn data_column_sidecars_awaiting_parent( - &self, - sidecars: Vec, - ) -> Result<(), ActorError> { - let _ = self.0.send(Forwarded::AwaitingParent(sidecars)); - Ok(()) - } - fn new_beacon_aggregate( - &self, - _aggregate: Box, - _attesting_indices: Vec, - _arrival: AggregateArrival, - ) -> Result<(), ActorError> { - Ok(()) - } - fn new_execution_payload_envelope( - &self, - _envelope: Box< - ethlambda_types::beacon::containers::gloas::SignedExecutionPayloadEnvelope, - >, - _arrival: BlockArrival, - ) -> Result<(), ActorError> { - Ok(()) - } - fn new_payload_attestation_message( - &self, - _message: ethlambda_types::beacon::containers::gloas::PayloadAttestationMessage, - _arrival: BlockArrival, - ) -> Result<(), ActorError> { - Ok(()) - } - } + /// Long enough for the checks' blocking tasks on a loaded test machine. + const FORWARDED: Duration = Duration::from_secs(10); /// A sidecar the checks cannot judge yet goes back to the chain actor to /// be parked, and one they refuse goes nowhere: neither reaches the batch @@ -197,8 +130,8 @@ mod tests { #[tokio::test] async fn only_what_passes_is_forwarded_as_checked() { let mut server = unconnected_beacon_server(Config::mainnet(), 0).await; - let (sender, mut received) = mpsc::unbounded_channel(); - server.blockchain = Some(Arc::new(RecordingChain(sender))); + let chain = Mock::new().allow_sends().start(); + server.blockchain = Some(chain.to_p2p_to_block_chain_ref()); // Nothing in the store is its parent. let orphan = valid_shaped_sidecar(5, 0); @@ -213,16 +146,12 @@ mod tests { ], ); - assert_eq!( - received.recv().await, - Some(Forwarded::AwaitingParent(vec![DataColumnSidecar::Fulu( - orphan - )])) - ); - // The task sends at most one message per kind and has now finished, - // so the channel closes with nothing else in it. - drop(server); - assert_eq!(received.recv().await, None); + let parked: DataColumnSidecarsAwaitingParent = chain.expect(FORWARDED).await; + assert_eq!(parked.sidecars, vec![DataColumnSidecar::Fulu(orphan)]); + // The task sends its checked batch, if it has one, before the parked + // one and nothing after it, so the log is already complete. + assert_eq!(chain.count::(), 0); + chain.assert_no_unexpected(); } /// A gloas sidecar is judged against its own block, not a parent: one @@ -232,8 +161,8 @@ mod tests { #[tokio::test] async fn a_gloas_sidecar_for_an_unknown_block_is_parked_and_a_future_one_dropped() { let mut server = unconnected_beacon_server(Config::mainnet(), 0).await; - let (sender, mut received) = mpsc::unbounded_channel(); - server.blockchain = Some(Arc::new(RecordingChain(sender))); + let chain = Mock::new().allow_sends().start(); + server.blockchain = Some(chain.to_p2p_to_block_chain_ref()); let unknown_block = DataColumnSidecar::Gloas(gloas::DataColumnSidecar { index: 0, @@ -250,11 +179,9 @@ mod tests { check_and_forward(&server, vec![unknown_block.clone(), future]); - assert_eq!( - received.recv().await, - Some(Forwarded::AwaitingParent(vec![unknown_block])) - ); - drop(server); - assert_eq!(received.recv().await, None); + let parked: DataColumnSidecarsAwaitingParent = chain.expect(FORWARDED).await; + assert_eq!(parked.sidecars, vec![unknown_block]); + assert_eq!(chain.count::(), 0); + chain.assert_no_unexpected(); } } diff --git a/crates/net/p2p/src/beacon/envelope_checks.rs b/crates/net/p2p/src/beacon/envelope_checks.rs index 5dbbd43de..ee20a3f5f 100644 --- a/crates/net/p2p/src/beacon/envelope_checks.rs +++ b/crates/net/p2p/src/beacon/envelope_checks.rs @@ -147,97 +147,33 @@ pub(crate) fn check_and_forward_then( #[cfg(test)] pub(crate) mod tests { - use std::sync::Arc; + use std::time::Duration; - use ethlambda_network_api::{ - AggregateArrival, BlockAnnouncement, BlockSource, P2PToBlockChain, - }; + use ethlambda_network_api::ToP2PToBlockChainRef as _; + use ethlambda_network_api::p2p_to_block_chain::NewExecutionPayloadEnvelope; use ethlambda_state_transition::beacon::gossip::{IgnoreReason, QueueReason, RejectReason}; - use ethlambda_types::attestation::{SignedAggregatedAttestation, SignedAttestation}; use ethlambda_types::beacon::config::Config; - use ethlambda_types::beacon::containers::{ - DataColumnSidecar, SignedAggregateAndProof, SignedBeaconBlock, - }; - use ethlambda_types::beacon::primitives::ValidatorIndex; use ethlambda_types::primitives::H256; - use spawned_concurrency::error::ActorError; - use tokio::sync::mpsc; + use spawned_test::tasks::Mock; use super::*; use crate::test_support::unconnected_beacon_server; - /// What a [`Recorder`] saw the chain actor be sent, in order. - #[derive(Debug, PartialEq)] - pub(crate) enum Seen { - Block(u64), - Envelope(H256, u64), - } - - /// A chain actor stand-in that reports every block and envelope it gets, - /// in one stream so their relative order is observable. - pub(crate) struct Recorder(pub mpsc::UnboundedSender); - - impl P2PToBlockChain for Recorder { - fn new_block( - &self, - block: SignedBeaconBlock, - _: BlockSource, - _: BlockArrival, - _: BlockAnnouncement, - ) -> Result<(), ActorError> { - let _ = self.0.send(Seen::Block(block.slot())); - Ok(()) - } - fn new_attestation(&self, _: SignedAttestation) -> Result<(), ActorError> { - Ok(()) - } - fn new_aggregated_attestation( - &self, - _: SignedAggregatedAttestation, - ) -> Result<(), ActorError> { - Ok(()) - } - fn new_data_column_sidecars(&self, _: Vec) -> Result<(), ActorError> { - Ok(()) - } - fn data_column_sidecars_awaiting_parent( - &self, - _: Vec, - ) -> Result<(), ActorError> { - Ok(()) - } - fn new_beacon_aggregate( - &self, - _: Box, - _: Vec, - _: AggregateArrival, - ) -> Result<(), ActorError> { - Ok(()) - } - fn new_execution_payload_envelope( - &self, - envelope: Box, - _: BlockArrival, - ) -> Result<(), ActorError> { - let _ = self.0.send(Seen::Envelope( - envelope.message.beacon_block_root, - envelope.message.payload.slot_number, - )); - Ok(()) - } - fn new_payload_attestation_message( - &self, - _: ethlambda_types::beacon::containers::gloas::PayloadAttestationMessage, - _: BlockArrival, - ) -> Result<(), ActorError> { - Ok(()) - } - } + /// Long enough for the checks' blocking tasks on a loaded test machine. + pub(crate) const FORWARDED: Duration = Duration::from_secs(10); pub(crate) fn envelope(slot: u64, byte: u8) -> SignedExecutionPayloadEnvelope { crate::beacon::encoding::test_support::envelope(H256::repeat_byte(byte), slot, H256::ZERO) } + /// The `(block root, slot)` an envelope the chain actor was sent names. + pub(crate) fn sent(message: &NewExecutionPayloadEnvelope) -> (H256, u64) { + ( + message.envelope.message.beacon_block_root, + message.envelope.message.payload.slot_number, + ) + } + #[test] fn only_accept_and_queue_reach_the_chain_actor() { assert!(forwards(&Outcome::Accept)); @@ -252,8 +188,8 @@ pub(crate) mod tests { #[tokio::test] async fn envelopes_for_unknown_blocks_are_forwarded_in_order() { let mut server = unconnected_beacon_server(Config::mainnet(), 0).await; - let (sender, mut received) = mpsc::unbounded_channel(); - server.blockchain = Some(Arc::new(Recorder(sender))); + let chain = Mock::new().allow_sends().start(); + server.blockchain = Some(chain.to_p2p_to_block_chain_ref()); check_and_forward( &server, @@ -261,12 +197,11 @@ pub(crate) mod tests { ); for (slot, byte) in [(7, 1), (8, 2), (9, 3)] { - assert_eq!( - received.recv().await, - Some(Seen::Envelope(H256::repeat_byte(byte), slot)) - ); + let forwarded: NewExecutionPayloadEnvelope = chain.expect(FORWARDED).await; + assert_eq!(sent(&forwarded), (H256::repeat_byte(byte), slot)); } - drop(server); - assert_eq!(received.recv().await, None); + // The last of the batch is the task's last send. + assert_eq!(chain.received().len(), 3); + chain.assert_no_unexpected(); } } diff --git a/crates/net/p2p/src/beacon/verdict.rs b/crates/net/p2p/src/beacon/verdict.rs index ab5651c26..143d17ddf 100644 --- a/crates/net/p2p/src/beacon/verdict.rs +++ b/crates/net/p2p/src/beacon/verdict.rs @@ -691,19 +691,40 @@ pub(crate) fn guarded(checks: impl FnOnce() -> Outcome) -> Outcome { mod tests { use std::sync::Arc; - use ethlambda_network_api::P2PToBlockChain; + use ethlambda_network_api::ToP2PToBlockChainRef as _; + use ethlambda_network_api::p2p_to_block_chain::{ + NewBlock, NewExecutionPayloadEnvelope, NewPayloadAttestationMessage, + }; use ethlambda_state_transition::beacon::gossip::{QueueReason, RejectReason}; - use ethlambda_types::attestation::{SignedAggregatedAttestation, SignedAttestation}; use ethlambda_types::beacon::config::Config; use ethlambda_types::beacon::containers::{ AttestationData, Checkpoint, electra, gloas, phase0, }; - use spawned_concurrency::error::ActorError; + use spawned_test::tasks::{Mock, MockRef}; use super::*; - use crate::test_support::{ - RecordingChain, envelope, unconnected_beacon_server, valid_shaped_sidecar, - }; + use crate::test_support::{envelope, unconnected_beacon_server, valid_shaped_sidecar}; + + /// A chain actor stand-in that takes every send and logs it. + fn recording_chain(server: &mut P2PServer) -> MockRef { + let chain = Mock::new().allow_sends().start(); + server.blockchain = Some(chain.to_p2p_to_block_chain_ref()); + chain + } + + /// A chain actor stand-in that fails the test on any message at all. + fn silent_chain(server: &mut P2PServer) -> MockRef { + let chain = Mock::new().start(); + server.blockchain = Some(chain.to_p2p_to_block_chain_ref()); + chain + } + + /// Assert the chain actor was sent nothing. `forward` sends synchronously, + /// so once the mock has handled what was sent before, the log is complete. + async fn assert_nothing_sent(chain: &MockRef) { + chain.settle().await; + assert_eq!(chain.received(), []); + } /// A minimal fulu block for a given `(slot, proposer)`: `settle` and /// `record_seen` only ever read those two fields plus the root passed @@ -914,8 +935,7 @@ mod tests { #[tokio::test] async fn only_an_accepted_gossip_block_is_announced() { let mut server = unconnected_beacon_server(Config::mainnet(), 0).await; - let chain = Arc::new(RecordingChain::default()); - server.blockchain = Some(chain.clone()); + let chain = recording_chain(&mut server); let block = || Validated::Block { block: Box::new(fulu_block(5, 1)), block_root: Root::repeat_byte(1), @@ -933,21 +953,27 @@ mod tests { Outcome::Ignore(IgnoreReason::Overloaded), ); + chain.settle().await; + let announcements: Vec<_> = chain + .take_all::() + .into_iter() + .map(|sent| sent.announcement) + .collect(); assert_eq!( - *chain.announcements.lock().unwrap(), + announcements, [ BlockAnnouncement::Announce, BlockAnnouncement::Silent, BlockAnnouncement::Silent ] ); + chain.assert_no_unexpected(); } #[tokio::test] async fn an_overloaded_aggregate_is_not_forwarded() { let mut server = unconnected_beacon_server(Config::mainnet(), 0).await; - let chain = Arc::new(RecordingChain::default()); - server.blockchain = Some(chain.clone()); + let chain = silent_chain(&mut server); let object = Validated::Aggregate { aggregate: Box::new(phase0_aggregate(5, 1)), attesting_indices: Vec::new(), @@ -959,7 +985,7 @@ mod tests { Outcome::Ignore(IgnoreReason::Overloaded), ); - assert!(chain.aggregates.lock().unwrap().is_empty()); + assert_nothing_sent(&chain).await; } /// An accepted electra aggregate is what block production packs other @@ -1102,8 +1128,7 @@ mod tests { #[tokio::test] async fn a_subnet_attestation_is_never_forwarded_even_on_accept() { let mut server = unconnected_beacon_server(Config::mainnet(), 0).await; - let chain = Arc::new(RecordingChain::default()); - server.blockchain = Some(chain.clone()); + let chain = silent_chain(&mut server); let object = Validated::Attestation { attestation: Box::new(electra_attestation(5, 1)), subnet_id: 0, @@ -1111,74 +1136,7 @@ mod tests { object.forward(&server, Instant::now(), Outcome::Accept); - assert!(chain.aggregates.lock().unwrap().is_empty()); - } - - /// What `forward` handed the chain actor, by method. - #[derive(Debug, PartialEq)] - enum Sent { - Envelope, - PayloadAttestation, - } - - /// A chain actor stand-in that reports only the two gloas deliveries. - struct GloasRecorder(tokio::sync::mpsc::UnboundedSender); - - impl P2PToBlockChain for GloasRecorder { - fn new_block( - &self, - _block: SignedBeaconBlock, - _source: BlockSource, - _arrival: BlockArrival, - _announcement: BlockAnnouncement, - ) -> Result<(), ActorError> { - Ok(()) - } - fn new_attestation(&self, _attestation: SignedAttestation) -> Result<(), ActorError> { - Ok(()) - } - fn new_aggregated_attestation( - &self, - _attestation: SignedAggregatedAttestation, - ) -> Result<(), ActorError> { - Ok(()) - } - fn new_data_column_sidecars( - &self, - _sidecars: Vec, - ) -> Result<(), ActorError> { - Ok(()) - } - fn data_column_sidecars_awaiting_parent( - &self, - _sidecars: Vec, - ) -> Result<(), ActorError> { - Ok(()) - } - fn new_beacon_aggregate( - &self, - _aggregate: Box, - _attesting_indices: Vec, - _arrival: AggregateArrival, - ) -> Result<(), ActorError> { - Ok(()) - } - fn new_execution_payload_envelope( - &self, - _envelope: Box, - _arrival: BlockArrival, - ) -> Result<(), ActorError> { - let _ = self.0.send(Sent::Envelope); - Ok(()) - } - fn new_payload_attestation_message( - &self, - _message: PayloadAttestationMessage, - _arrival: BlockArrival, - ) -> Result<(), ActorError> { - let _ = self.0.send(Sent::PayloadAttestation); - Ok(()) - } + assert_nothing_sent(&chain).await; } pub(crate) fn payload_attestation(slot: u64, validator: u64) -> PayloadAttestationMessage { @@ -1240,27 +1198,25 @@ mod tests { #[tokio::test] async fn an_envelope_is_forwarded_on_accept_and_queue_only() { let mut server = unconnected_beacon_server(Config::mainnet(), 0).await; - let (sender, mut received) = tokio::sync::mpsc::unbounded_channel(); - server.blockchain = Some(Arc::new(GloasRecorder(sender))); + let chain = recording_chain(&mut server); let forward = |outcome| { Validated::Envelope(Box::new(envelope(1, 3))).forward(&server, Instant::now(), outcome) }; forward(Outcome::Accept); - assert_eq!(received.try_recv(), Ok(Sent::Envelope)); forward(Outcome::Queue(QueueReason::BlockUnknown)); - assert_eq!(received.try_recv(), Ok(Sent::Envelope)); forward(Outcome::Ignore(IgnoreReason::Overloaded)); forward(Outcome::Reject(RejectReason::BadSignature)); - assert!(received.try_recv().is_err()); + chain.settle().await; + assert_eq!(chain.count::(), 2); + assert_eq!(chain.received().len(), 2); } /// A payload attestation reaches the chain actor on `Accept` alone. #[tokio::test] async fn a_payload_attestation_is_forwarded_on_accept_only() { let mut server = unconnected_beacon_server(Config::mainnet(), 0).await; - let (sender, mut received) = tokio::sync::mpsc::unbounded_channel(); - server.blockchain = Some(Arc::new(GloasRecorder(sender))); + let chain = recording_chain(&mut server); let forward = |outcome| { Validated::PayloadAttestation(payload_attestation(5, 1)).forward( @@ -1272,9 +1228,10 @@ mod tests { forward(Outcome::Ignore(IgnoreReason::UnknownBlock)); forward(Outcome::Ignore(IgnoreReason::Overloaded)); forward(Outcome::Reject(RejectReason::NotInPtc)); - assert!(received.try_recv().is_err()); forward(Outcome::Accept); - assert_eq!(received.try_recv(), Ok(Sent::PayloadAttestation)); + chain.settle().await; + assert_eq!(chain.count::(), 1); + assert_eq!(chain.received().len(), 1); } /// An accepted payload attestation is pooled for block production, with @@ -1447,8 +1404,7 @@ mod tests { #[tokio::test] async fn bids_and_preferences_never_reach_the_chain_actor() { let mut server = unconnected_beacon_server(Config::mainnet(), 0).await; - let (sender, mut received) = tokio::sync::mpsc::unbounded_channel(); - server.blockchain = Some(Arc::new(GloasRecorder(sender))); + let chain = silent_chain(&mut server); for outcome in [ Outcome::Accept, @@ -1463,7 +1419,7 @@ mod tests { ); } - assert!(received.try_recv().is_err()); + assert_nothing_sent(&chain).await; } /// Bids and preferences draw from a pool of their own, so a burst of them @@ -1749,8 +1705,7 @@ mod tests { #[tokio::test] async fn sync_committee_gossip_never_reaches_the_chain_actor() { let mut server = unconnected_beacon_server(Config::mainnet(), 0).await; - let (sender, mut received) = tokio::sync::mpsc::unbounded_channel(); - server.blockchain = Some(Arc::new(GloasRecorder(sender))); + let chain = silent_chain(&mut server); for outcome in [ Outcome::Accept, Outcome::Queue(QueueReason::BlockUnknown), @@ -1764,7 +1719,7 @@ mod tests { outcome, ); } - assert!(received.try_recv().is_err()); + assert_nothing_sent(&chain).await; } /// The sync pool is its own: exhausting it must not starve the block and diff --git a/crates/net/p2p/src/gossipsub/handler.rs b/crates/net/p2p/src/gossipsub/handler.rs index d6b90bce0..25d5a11d6 100644 --- a/crates/net/p2p/src/gossipsub/handler.rs +++ b/crates/net/p2p/src/gossipsub/handler.rs @@ -1283,7 +1283,7 @@ mod tests { use ethlambda_types::beacon::primitives::Slot; use super::*; - use crate::test_support::{RecordingChain, unconnected_beacon_server, valid_shaped_sidecar}; + use crate::test_support::{unconnected_beacon_server, valid_shaped_sidecar}; /// An electra-shaped aggregate at `slot` with `data.index` set to /// `data_index`. Only what [`gossip::aggregate::cheap_checks`]'s very @@ -1730,15 +1730,25 @@ mod tests { unreachable!("unconnected_beacon_server builds a beacon wire"); }; beacon.topics.fork_digest = beacon.digest_for_slot(5); - let chain = std::sync::Arc::new(RecordingChain::default()); - server.blockchain = Some(chain.clone()); + use ethlambda_network_api::ToP2PToBlockChainRef as _; + use ethlambda_network_api::p2p_to_block_chain::NewBeaconAggregate; + + let chain = spawned_test::tasks::Mock::new().allow_sends().start(); + server.blockchain = Some(chain.to_p2p_to_block_chain_ref()); let aggregate = ethlambda_types::beacon::containers::SignedAggregateAndProof::Electra( electra_aggregate(5, 0), ); publish_beacon_aggregate(&mut server, aggregate.clone(), vec![3, 4]).await; - assert_eq!(*chain.aggregates.lock().unwrap(), [(aggregate, vec![3, 4])]); + chain.settle().await; + let handed: Vec<_> = chain + .take_all::() + .into_iter() + .map(|sent| (*sent.aggregate, sent.attesting_indices)) + .collect(); + assert_eq!(handed, [(aggregate, vec![3, 4])]); + assert_eq!(chain.received().len(), 1); } /// A payload that does not decode as a gloas envelope is the sender's diff --git a/crates/net/p2p/src/lib.rs b/crates/net/p2p/src/lib.rs index cc9ca2bd1..94a90f20a 100644 --- a/crates/net/p2p/src/lib.rs +++ b/crates/net/p2p/src/lib.rs @@ -84,6 +84,7 @@ use spawned_concurrency::actor; use spawned_concurrency::error::ActorError; use spawned_concurrency::message::Message; use spawned_concurrency::protocol; +use spawned_concurrency::system::{ExecutionGroup, SystemError}; use spawned_concurrency::tasks::{ Actor, ActorRef, ActorStart, Context, Handler, send_after, spawn_listener, }; @@ -91,10 +92,9 @@ use tracing::{debug, info, trace, warn}; use crate::{ discovery::{ - DIAL_INTERVAL_AT_TARGET, DIAL_INTERVAL_AT_ZERO_PEERS, DiscoveryError, DiscoverySpawnConfig, + DIAL_INTERVAL_AT_TARGET, DIAL_INTERVAL_AT_ZERO_PEERS, DiscoveryHandle, dial::{DiscoveryState, dial_interval, dial_progress, dial_tick, forget_discovered_peer}, enr::{dialable_port, read_ip, read_public_key, read_quic_port, read_tcp_port}, - spawn_discovery, }, gossipsub::{ aggregation_topic, attestation_subnet_topic, block_topic, publish_aggregated_attestation, @@ -1157,39 +1157,36 @@ pub struct P2P { } impl P2P { - /// Start discovery, start the I/O adapter, spawn the actor, and wire the - /// swarm event stream. + /// Start the I/O adapter, spawn the actor on `group`, and wire the swarm + /// event stream. /// - /// `discovery` is `Some` when discv5 runs: the server is started here, and - /// its handle seeds the dial loop's state and schedules its first tick. It - /// is started before the swarm adapter so a fatal discovery failure (a busy - /// UDP port, say) surfaces before any actor is running. `None` leaves the - /// dial loop unscheduled, so peering relies solely on the static bootnode - /// list `build_swarm` dialed. + /// `discovery` is `Some` when discv5 runs: the caller starts the server + /// with [`discovery::spawn_discovery`] before any actor, so a fatal + /// discovery failure (a busy UDP port, say) surfaces before anything is + /// running, and its handle seeds the dial loop's state and schedules its + /// first tick here. `None` leaves the dial loop unscheduled, so peering + /// relies solely on the static bootnode list `build_swarm` dialed. /// /// Always `Some` on beacon, which has no other way to find a peer: /// published mainnet bootnode ENRs carry no `quic` entry, so none of them /// is statically dialable. Opt-in on lean, where nothing else speaks discv5 /// and co-located devnet nodes would otherwise all claim one UDP port. #[allow(clippy::too_many_arguments)] - pub async fn spawn( + pub fn spawn( built: BuiltSwarm, store: Store, node_names: HashMap, - discovery: Option, + discovery: Option, payload_attestation_pool: SharedPayloadAttestationPool, builder_market: SharedBuilderMarket, sync_committee_pool: SharedSyncCommitteePool, - ) -> Result { - let (discovery, local_enr) = match discovery { - Some(config) => { - let discovery = spawn_discovery(config).await?; - let local_enr = discovery.local_enr.clone(); - (Some(discovery), Some(local_enr)) - } + group: &ExecutionGroup, + ) -> Result { + let local_enr = match &discovery { + Some(discovery) => Some(discovery.local_enr.clone()), None => { info!("discv5 discovery disabled; peering from the static bootnode list only"); - (None, None) + None } }; let (swarm_stream, swarm_handle) = @@ -1268,7 +1265,7 @@ impl P2P { }; let scores_peers = server.peer_scoring.is_some(); let discovery_enabled = server.discovery.is_some(); - let handle = server.start(); + let handle = server.start_on(group)?; // At once rather than a slot in: the swarm starts on placeholder // topic scores, and the head's are already known. if scores_peers { @@ -2828,85 +2825,6 @@ pub(crate) mod test_support { use crate::beacon::swarm::BeaconWireConfig; use crate::{P2PServer, SwarmConfig, WireConfig, build_swarm}; - /// A chain actor stand-in recording what reaches it: each block's - /// announcement and each aggregate with its attesting indices. The rest - /// of the protocol is accepted and dropped. - #[derive(Default)] - pub(crate) struct RecordingChain { - pub(crate) announcements: std::sync::Mutex>, - pub(crate) aggregates: std::sync::Mutex< - Vec<( - ethlambda_types::beacon::containers::SignedAggregateAndProof, - Vec, - )>, - >, - } - - impl ethlambda_network_api::P2PToBlockChain for RecordingChain { - fn new_block( - &self, - _block: ethlambda_types::beacon::containers::SignedBeaconBlock, - _source: ethlambda_network_api::BlockSource, - _arrival: ethlambda_network_api::BlockArrival, - announcement: ethlambda_network_api::BlockAnnouncement, - ) -> Result<(), spawned_concurrency::error::ActorError> { - self.announcements.lock().unwrap().push(announcement); - Ok(()) - } - fn new_attestation( - &self, - _attestation: ethlambda_types::attestation::SignedAttestation, - ) -> Result<(), spawned_concurrency::error::ActorError> { - Ok(()) - } - fn new_aggregated_attestation( - &self, - _attestation: ethlambda_types::attestation::SignedAggregatedAttestation, - ) -> Result<(), spawned_concurrency::error::ActorError> { - Ok(()) - } - fn new_data_column_sidecars( - &self, - _sidecars: Vec, - ) -> Result<(), spawned_concurrency::error::ActorError> { - Ok(()) - } - fn data_column_sidecars_awaiting_parent( - &self, - _sidecars: Vec, - ) -> Result<(), spawned_concurrency::error::ActorError> { - Ok(()) - } - fn new_beacon_aggregate( - &self, - aggregate: Box, - attesting_indices: Vec, - _arrival: ethlambda_network_api::AggregateArrival, - ) -> Result<(), spawned_concurrency::error::ActorError> { - self.aggregates - .lock() - .unwrap() - .push((*aggregate, attesting_indices)); - Ok(()) - } - fn new_execution_payload_envelope( - &self, - _envelope: Box< - ethlambda_types::beacon::containers::gloas::SignedExecutionPayloadEnvelope, - >, - _arrival: ethlambda_network_api::BlockArrival, - ) -> Result<(), spawned_concurrency::error::ActorError> { - Ok(()) - } - fn new_payload_attestation_message( - &self, - _message: ethlambda_types::beacon::containers::gloas::PayloadAttestationMessage, - _arrival: ethlambda_network_api::BlockArrival, - ) -> Result<(), spawned_concurrency::error::ActorError> { - Ok(()) - } - } - /// A gloas envelope for `(root_byte, builder)` at slot `slot_number` /// zero. Only what the cheap gossip checks and the seen cache read is /// meaningful; nothing here is signature-valid. diff --git a/crates/net/p2p/src/req_resp/envelope_client.rs b/crates/net/p2p/src/req_resp/envelope_client.rs index 1714ce3dc..e806e13f5 100644 --- a/crates/net/p2p/src/req_resp/envelope_client.rs +++ b/crates/net/p2p/src/req_resp/envelope_client.rs @@ -372,7 +372,6 @@ pub(super) async fn handle_envelopes_by_range_response( #[cfg(test)] mod tests { use std::collections::HashMap; - use std::sync::Arc; use std::time::Duration; use ethlambda_network_api::{BlockArrival, BlockSource}; @@ -594,7 +593,7 @@ mod tests { /// or no chain to forward to, so range sync cannot wait forever. #[tokio::test] async fn the_range_release_fires_when_there_is_nothing_to_forward() { - use crate::beacon::envelope_checks::tests::Recorder; + use ethlambda_network_api::ToP2PToBlockChainRef as _; let mut server = crate::test_support::unconnected_beacon_server(Config::mainnet(), 0).await; let (done_tx, mut done_rx) = tokio::sync::mpsc::unbounded_channel(); @@ -613,10 +612,12 @@ mod tests { assert_eq!(done_rx.recv().await, Some(())); // An empty answer. - let (sender, _received) = tokio::sync::mpsc::unbounded_channel(); - server.blockchain = Some(Arc::new(Recorder(sender))); + let chain = spawned_test::tasks::Mock::new().start(); + server.blockchain = Some(chain.to_p2p_to_block_chain_ref()); envelope_checks::check_and_forward_then(&server, Vec::new(), Some(callback(done_tx))); assert_eq!(done_rx.recv().await, Some(())); + chain.settle().await; + assert!(chain.received().is_empty()); } fn pending_lookup(server: &mut P2PServer, root: H256, attempts: u32) { @@ -691,19 +692,21 @@ mod tests { /// handed it, and in slot order, whichever order the peer listed them. #[tokio::test] async fn range_envelopes_are_forwarded_after_the_batch_blocks_in_slot_order() { - use crate::beacon::envelope_checks::tests::{Recorder, Seen}; + use ethlambda_network_api::P2PToBlockChain as _; + use ethlambda_network_api::ToP2PToBlockChainRef as _; + use ethlambda_network_api::p2p_to_block_chain::{NewBlock, NewExecutionPayloadEnvelope}; + + use crate::beacon::envelope_checks::tests::sent; let mut server = crate::test_support::unconnected_beacon_server(Config::mainnet(), 0).await; - let (sender, mut received) = tokio::sync::mpsc::unbounded_channel(); - let chain = Arc::new(Recorder(sender)); - server.blockchain = Some(chain.clone()); + let chain = spawned_test::tasks::Mock::new().allow_sends().start(); + server.blockchain = Some(chain.to_p2p_to_block_chain_ref()); let root = H256::repeat_byte(3); // What `handle_beacon_blocks_by_range_response` does before it asks // for the span's envelopes. for slot in [20, 21] { - let _ = ethlambda_network_api::P2PToBlockChain::new_block( - chain.as_ref(), + let _ = chain.new_block( fulu_block_at(slot), BlockSource::Sync, BlockArrival::now(), @@ -724,11 +727,32 @@ mod tests { })), ); - assert_eq!(received.recv().await, Some(Seen::Block(20))); - assert_eq!(received.recv().await, Some(Seen::Block(21))); - assert_eq!(received.recv().await, Some(Seen::Envelope(root, 20))); - assert_eq!(received.recv().await, Some(Seen::Envelope(root, 21))); - // The release comes only after the last envelope was sent. + // The release comes only after the last envelope was sent, so once it + // fires and the mock has handled what was sent before, the log is + // complete. assert_eq!(done_rx.recv().await, Some(())); + chain.settle().await; + let order: Vec<_> = chain.received().iter().map(|r| r.type_name).collect(); + assert_eq!( + order, + [ + std::any::type_name::(), + std::any::type_name::(), + std::any::type_name::(), + std::any::type_name::(), + ] + ); + let blocks: Vec = chain + .take_all::() + .iter() + .map(|sent| sent.block.slot()) + .collect(); + assert_eq!(blocks, [20, 21]); + let envelopes: Vec<_> = chain + .take_all::() + .iter() + .map(sent) + .collect(); + assert_eq!(envelopes, [(root, 20), (root, 21)]); } } diff --git a/crates/net/rpc/Cargo.toml b/crates/net/rpc/Cargo.toml index 547639dbf..9949c6717 100644 --- a/crates/net/rpc/Cargo.toml +++ b/crates/net/rpc/Cargo.toml @@ -31,10 +31,11 @@ hex.workspace = true tracing.workspace = true jemalloc_pprof.workspace = true futures-util = "0.3" +# `/metrics` appends the actor runtime's series (the `metrics` feature). +spawned-concurrency.workspace = true [dev-dependencies] ethlambda-validator = { workspace = true, features = ["test-utils"] } -spawned-concurrency.workspace = true ethlambda-state-transition = { workspace = true, features = ["test-utils"] } ethlambda-types.workspace = true libssz-types.workspace = true diff --git a/crates/net/rpc/src/metrics.rs b/crates/net/rpc/src/metrics.rs index b07459e7e..ae94aebb7 100644 --- a/crates/net/rpc/src/metrics.rs +++ b/crates/net/rpc/src/metrics.rs @@ -34,12 +34,15 @@ pub(crate) async fn get_health() -> impl IntoResponse { } pub(crate) async fn get_metrics() -> impl IntoResponse { - let mut response = gather_default_metrics() + let mut body = gather_default_metrics() .inspect_err(|err| { warn!(%err, "Failed to gather Prometheus metrics"); }) - .unwrap_or_default() - .into_response(); + .unwrap_or_default(); + // The actor runtime's own series, all prefixed `spawned_`: mailbox depth, + // queue wait, handler duration and actor lifecycle, summed over every actor. + body.push_str(&spawned_concurrency::metrics::render_prometheus()); + let mut response = body.into_response(); let content_type = HeaderValue::from_static("text/plain; version=0.0.4; charset=utf-8"); response .headers_mut() diff --git a/crates/net/rpc/tests/http_servers.rs b/crates/net/rpc/tests/http_servers.rs index 6171b1730..b6ef5c282 100644 --- a/crates/net/rpc/tests/http_servers.rs +++ b/crates/net/rpc/tests/http_servers.rs @@ -82,10 +82,16 @@ async fn with_no_api_router_metrics_and_debug_are_served_on_the_metrics_port() { let served = tokio::spawn(start_http_servers(config.clone(), None, shutdown.clone())); // Metrics: the whole reason `beacon` serves anything at all. + let metrics = get(port, "/metrics").await; assert!( - status_line(&get(port, "/metrics").await).contains("200"), + status_line(&metrics).contains("200"), "/metrics must be served" ); + // The actor runtime's series ride along with the node's own. + assert!( + metrics.contains("spawned_mailbox_depth"), + "/metrics must carry the actor runtime's series" + ); // Debug: `beacon` gains these by going through the shared entry point. // Heap profiling was previously node-only. assert!( diff --git a/docs/SUMMARY.md b/docs/SUMMARY.md index 311798283..2a8c425f6 100644 --- a/docs/SUMMARY.md +++ b/docs/SUMMARY.md @@ -34,3 +34,4 @@ - [Benchmarking](./benchmarking.md) - [Era Replay](./era_replay.md) - [Spec Deviations](./spec_deviations.md) +- [Testing Roadmap](./testing_roadmap.md) diff --git a/docs/testing_roadmap.md b/docs/testing_roadmap.md new file mode 100644 index 000000000..effb06140 --- /dev/null +++ b/docs/testing_roadmap.md @@ -0,0 +1,293 @@ +# Testing and Verification Roadmap + +The goal is to test ethlambda the way FoundationDB, TigerBeetle, Dropbox, etcd +and AWS test their systems: run the real logic under simulated time and a +seeded, adversarial network, check invariants on every step, and keep every +failing seed as a regression test. The highest-ROI first step is a **seeded +in-process chain-logic simulator**. It drives N in-memory lean stores through +`on_tick` and gossip that a seeded scheduler delays, drops, reorders and +partitions. Most of the seams it needs already exist (see +[What makes it testable](#what-makes-it-testable)), and most of the incidents +under [Gaps](#gaps-tied-to-real-incidents) are timing and ordering bugs of the +kind such a simulator finds. + +Priorities and phases on this page are proposals, open for discussion. Code +references are `path:line` at `c3f5f566`. + +## What exists today + +| Suite | What it catches | Deterministic? | Where it runs | +|---|---|---|---| +| leanSpec `forkchoice_spectests`, `signature_spectests` (`crates/blockchain/Cargo.toml:54,63`), `stf_spectests` (`crates/blockchain/state_transition/Cargo.toml:116`), `ssz_spectests` (`crates/common/types/Cargo.toml:42`) | Divergence from leanSpec fork choice, STF, signature verification, SSZ | Yes | **Nowhere.** All four are `test = false` because the released fixtures still carry the old leanVM key format. CI only clippies them (`.github/workflows/ci.yml:65`) | +| Beacon consensus-spec-tests (`beacon-spec-tests` feature, `make test-beacon-{mainnet,minimal}`) | Beacon STF, fork choice and crypto divergence from consensus-specs | Yes | CI `beacon-spec-tests` job, both presets | +| Hive test driver (`crates/net/rpc/src/test_driver.rs`, `HIVE_LEAN_TEST_DRIVER=1`) | leanSpec fixtures replayed over HTTP by Hive's lean simulator | Yes | Hive, outside this repo | +| proptest: `crates/common/ssz-tree/tests/model.rs`, `crates/storage/src/beacon_state_delta.rs:177` | SSZ tree against a reference model (`libssz_types`); state-delta round trips | Seeded, with a checked-in regressions file | `make test` | +| `#[tokio::test(start_paused = true)]` (`bin/ethlambda/src/main.rs:2681`) | Checkpoint-sync retry and DB-resume windows under paused time | Yes | `make test` | +| `daily_ci.yml` | The same two test halves as CI, reported to Slack | As above | Daily, 12:00 UTC | +| `era_replay.yml` | Beacon STF divergence on real mainnet/hoodi eras ([Era Replay](./era_replay.md)) | Yes | Push to main touching STF/types paths | +| `kurtosis_assertoor.yml` | Beacon node and validator client next to lighthouse: stability, proposals, sync | No | Push to main | +| Shadow build (`shadow-integration` feature, `shadow/README.md`) | Runs the real binary under the Shadow simulator for [lean-shadow-fuzzer](https://github.com/kamilsa/lean-shadow-fuzzer) | Shadow is deterministic per seed | External; `-shadow` images built by `docker_publish.yaml`. No scheduled run here | + +Nothing exercises **several lean nodes' chain logic together under controlled +time and message order** in CI, and nothing checks process-level lifecycle +(signals, resource limits). + +## What makes it testable + +- **Fork-choice time arrives as data.** `store::on_tick(store, timestamp_ms, has_proposal)` + (`crates/blockchain/src/store.rs:379`) converts a timestamp into intervals; + nothing inside reads the clock. A simulator can feed any time sequence. +- **The swarm sits behind a channel adapter.** `SwarmHandle` sends + `SwarmCommand`s over an mpsc channel, and events come back as a stream + (`crates/net/p2p/src/swarm_adapter.rs:23,112,247`). `SwarmHandle::capturing()` + (`:241`, `#[cfg(test)]`) already lets a test read what was sent. +- **Storage is a trait with an in-memory backend.** `StorageBackend` + (`crates/storage/src/api/traits.rs:10`) has `InMemoryBackend` + (`crates/storage/src/backend/in_memory.rs:26`) next to RocksDB, and the test + driver already builds stores on it. +- **Crypto can be stubbed.** `on_block_without_verification` + (`crates/blockchain/src/store.rs:640`); `benchmark synthetic --mock-crypto` + (`bin/ethlambda/src/benchmark/mod.rs:169`); `dummy_sig()` in aggregation tests + (`crates/blockchain/src/aggregation.rs:1781`); Shadow's fake-proof backend + (`crates/common/crypto/src/shadow_cost.rs`). +- **Keys are reproducible.** `ValidatorSecretKey::generate_from_seed` + (`crates/common/crypto/src/signature.rs:154`). + +What makes it hard: + +- **No clock trait.** `unix_now_ms()` (`crates/common/types/src/time.rs:14`) reads + `SystemTime` directly, from 19 sites in `crates/blockchain/src/lib.rs` and + others in p2p (e.g. 11 in `gossipsub/handler.rs`). The actor's tick + scheduling mixes this wall clock with tokio's monotonic sleep, which is + exactly where 3210931f came from. +- **Unseeded randomness.** Peer choice uses `rand::thread_rng()` + (`crates/net/p2p/src/req_resp/handlers.rs:1000,1156`). +- **A process-global prover thread.** `prover_queue()` spawns one + `leanvm-prover` thread for the life of the process + (`crates/common/crypto/src/lib.rs:131`), so in-process nodes share it. +- **Global metrics.** Metrics are `LazyLock` statics (208 in + `crates/blockchain/src/metrics.rs` alone), so N in-process nodes write the + same series, and metric-based assertions cannot tell nodes apart. + +## Gaps tied to real incidents + +| Commit | What broke | What would have caught it | +|---|---|---| +| a2f65803 | Ticks were scheduled on the unix-epoch 800 ms grid, not genesis's, so every interval fired 600 ms late on the default config | Kani/property test on `ms_until_next_interval`; simulator with a non-aligned genesis | +| 514e21a7 | `on_tick` overran its interval (inline proving, ~1.3 s) and the next interval's duty was skipped: a proposer never attested on its own slot, so 1-2 node devnets could not justify | Simulator with injected duty latency; liveness check "every validator attests each slot" | +| 3210931f | Wall clock drifting behind the monotonic clock re-ran an interval's duties (`StateSlotIsNewer` on re-proposal) | Clock-jump tests on the tick actor | +| 67a69ab2 | devnet4: during a finalization stall, fallback pruning deleted states while blocks stayed, so the `has_state` idempotency check let blocks be re-processed forever (100% CPU, ~3.5 GB logs per node) | Simulator with a long partition (finality stall) and a "no block imported twice" monitor | +| 32e3972a | Fork-choice votes were derived from bounded proof buffers, so FIFO eviction dropped live votes from fork-choice weight | Reference-model test of vote tracking; simulator at high attestation volume | +| 65c0bf77 | Beacon: unrealized justifications lived only in memory, so after a restart `get_head` failed every tick and the head froze | Metamorphic restart-from-DB test | +| 6168a0e7 | Hoodi: the missing-ancestor walk stopped at an already-imported block, so a dropped block was never re-requested and ~900 blocks stayed parked for three hours | Reference model of the pending-block fetch set; "every pending root gets an outcome" monitor | +| 64a6bce9 | Two req/resp paths left a root in `pending_root_requests` forever, so it could never be fetched again | Same monitor; failpoints on error responses | +| d2593f88 | Checkpoint sync failed fast on one timeout and the process exited (flaky Hive starts) | Failpoint on checkpoint sync | +| afc491b2 | devnet-5 aggregators with `--prover-arena` were OOM-killed about every 40 minutes: each proving thread pinned an arena slab | Soak with an RSS budget | +| 1d7902da | Release feedback: `docker stop` force-killed after 10 s while a clean shutdown took 14.3 s, losing state; clean Ctrl+C logged a P2P `ERROR` | Process-level shutdown test with a time budget and a clean-log assertion | + +## Techniques + +Each entry: what it is, precedent, the ethlambda target, effort, and the +acceptance criterion. + +### Seeded in-process chain-logic simulator (flagship) + +- **What:** N lean `Store`s on `InMemoryBackend`, one simulated clock driving + `on_tick`, blocks and attestations delivered through a seeded scheduler that + delays, drops, duplicates, reorders and partitions. Signatures stubbed via + `on_block_without_verification`. Each run is a pure function of its seed. +- **Precedent:** [FoundationDB simulation](https://apple.github.io/foundationdb/testing.html), [TigerBeetle VOPR](https://github.com/tigerbeetle/tigerbeetle/blob/main/docs/internals/vopr.md). +- **Target:** `crates/blockchain` store functions plus validator duties, above + the swarm. No libp2p and no actor. +- **Invariants:** safety (no two conflicting finalized checkpoints across nodes; + finalized slot never decreases); liveness (finalization resumes within K slots + after a partition heals with >2/3 honest connected); no block imported twice. +- **Effort:** 2-3 weeks for a first version. Needs the duty path callable + without the actor, and a clock parameter where duties call `unix_now_ms()`. +- **Done when:** runs in CI on a fixed seed set plus a few hundred random seeds + nightly; a failing seed reproduces locally from one command; at least one + incident above (514e21a7 or 67a69ab2) is reproduced by reverting its fix. + +### Re-enable the leanSpec runners + +- **What:** regenerate or bump to fixtures with the current leanVM key format + and drop `test = false` from the four runners. +- **Precedent:** every consensus client runs its spec tests on every PR. +- **Effort:** days, gated on a leanSpec fixture release. +- **Done when:** `make test` runs all four and CI fails on a fork-choice fixture + mismatch. + +### Kani on slot and justification math + +- **What:** bounded proofs of `slot_is_justifiable_after` + (`crates/blockchain/state_transition/src/lib.rs:629`; no overflow, agrees with + a naive square/pronic definition), `ms_until_next_interval` + (`crates/blockchain/src/lib.rs:208`; result in `1..=interval`, lands on a + genesis-aligned boundary, the property a2f65803 broke), and + `on_tick`'s timestamp-to-interval conversion. +- **Precedent:** AWS runs [Kani harnesses on Firecracker](https://model-checking.github.io/kani-verifier-blog/2023/08/31/using-kani-to-validate-security-boundaries-in-aws-firecracker.html) in CI. +- **Effort:** about a week. **Done when:** a `kani` CI job proves the harnesses. + +### TLA+ / Apalache model of 3SF-mini justification and finality + +- **What:** a model of justification and finalization with justifiable-slot + rules, checked for accountable safety. Start from Konnov et al., + [arXiv:2501.07958](https://arxiv.org/abs/2501.07958), which translated 3SF's + executable Python specification to TLA+ and checked accountable safety with + Apalache (exhaustively up to 7 checkpoints and 24 votes). +- **Target:** [3SF-mini](./3sf_mini.md) as ethlambda implements it, then trace + validation against simulator runs. +- **Effort:** weeks. **Done when:** the model checks in CI at small bounds and + simulator traces replay in it. + +### Metamorphic tests + +- **What:** the same block set in any arrival order gives the same head and + finalized checkpoint; a node restarted from its DB gives the same head as one + that never stopped (the property 65c0bf77 lacked). +- **Effort:** days on the simulator's harness. **Done when:** both run per seed. + +### Reference models + +- **What:** proptest a simple model against `justified_slots_ops` + (`set_justified`, `shift_window`, + `crates/blockchain/state_transition/src/justified_slots_ops.rs:45,72`) and + against the pending-block / `pending_root_requests` bookkeeping (6168a0e7, + 64a6bce9), as `ssz-tree/tests/model.rs` already does for SSZ. +- **Effort:** days each. **Done when:** random op sequences agree with the + model after every step. + +### Failpoints + +- **What:** named fault sites (e.g. the `fail` crate) in the aggregation + worker, RocksDB writes, swarm-loop exit, and checkpoint-sync fetch. +- **Effort:** days. **Done when:** each site has a test asserting the node + either recovers or exits with a clear error, never hangs. + +### Process-level shutdown test + +- **What:** start the binary, send SIGINT and SIGTERM, assert exit within a + budget with a clean log and a DB that resumes. On `main` only SIGINT is + handled (`tokio::signal::ctrl_c()`, `bin/ethlambda/src/main.rs:988`), so + SIGTERM uses the default action and kills the node in about 0.01 s with no + graceful shutdown (measured during the `spawned-v0.7` work); the README works + around it with `--stop-signal SIGINT`. The `spawned-v0.7` branch handles + SIGTERM and bounds shutdown with an 8 s deadline. +- **Effort:** days. **Done when:** CI kills a running node both ways and checks + time, exit code, log level and a successful resume. + +### Single-CPU runs + +- **What:** run the test suite and a short devnet under `taskset -c 0` (or a + 1-CPU container) to surface starvation between the actor, the + `spawn_blocking` aggregation worker and the prover thread. +- **Effort:** a CI job. **Done when:** a nightly 1-CPU devnet finalizes. + +### Clock-jump tests + +- **What:** feed the store and the tick actor wall-clock jumps forward and + backward and monotonic/wall drift (3210931f, a2f65803). Requires a clock + parameter in place of `unix_now_ms()` on the tick path. +- **Done when:** no interval's duties run twice or are skipped under drift. + +### Soak with resource budgets + +- **What:** multi-hour devnet with aggregators and `--prover-arena`, asserting + RSS, FD and DB-size ceilings (afc491b2). +- **Done when:** a nightly soak fails on budget breach, not only on crash. + +### TSAN / LSan + +- **What:** nightly `-Zsanitizer=thread` and `leak` runs of the node test half. +- **Done when:** clean, with any suppressions listed and justified. + +### Log-based invariant monitors + +- **What:** a checker over devnet and Shadow logs: slot monotonicity, tick + lateness under a bound, every pending root reaching an outcome (64a6bce9, + 6168a0e7), no block imported twice (67a69ab2). `tooling/event-monitor` is a + natural home. +- **Done when:** every Kurtosis, Shadow and soak run ends with the monitor. + +### Nightly Shadow runs with seeds + +- **What:** a scheduled multi-node Shadow run of the `-shadow` image with a + recorded seed, partitions and latency, plus the log monitors. +- **Precedent:** lean-shadow-fuzzer already drives ethlambda under Shadow. +- **Done when:** nightly, failing seeds filed with a repro command. + +### Kurtosis/Assertoor and Attacknet-style chaos for lean devnets + +- **What:** extend `kurtosis_assertoor.yml`, today beacon-only, to a lean + devnet; add [Attacknet](https://github.com/crytic/attacknet)-style faults (partitions, clock skew, container kills). +- **Done when:** a lean devnet job runs on main with at least one fault scenario. + +### spawned-test fault injection and lifecycle monitors + +- **What:** after the spawned v0.7 upgrade (branch `spawned-v0.7`), use + `spawned-test` mocks (already replacing hand-written protocol stubs there), + supervision fault injection, and lifecycle monitors that assert every actor + stops within the shutdown deadline. +- **Done when:** a test panics each restartable actor and checks the node keeps + importing. + +### Antithesis (optional) + +- **What:** hosted deterministic hypervisor that explores fault schedules + against the real containers. Worth evaluating once the in-process simulator + and Shadow runs stop finding bugs. + +## Phased plan + +**Phase 1: now, on `main`.** + +| Item | Acceptance | +|---|---| +| Re-enable the leanSpec runners | All four run in `make test` | +| Chain-logic simulator | Fixed seeds in CI, random seeds nightly, one incident reproduced | +| Kani slot math | CI job proves the three harnesses | +| Shutdown test | SIGINT and SIGTERM both covered; SIGTERM expected to fail until Phase 2 | +| Nightly Shadow | Scheduled seeded run with log monitors | +| Single-CPU | Nightly 1-CPU devnet finalizes | +| Soak | Nightly run with RSS/FD/DB budgets | + +**Phase 2: with spawned v0.7** (`spawned-v0.7` branch, canary on fede256). + +| Item | Acceptance | +|---|---| +| `ActorSystem` shutdown deadline | SIGTERM test passes within 8 s | +| Supervision fault injection | Restartable actors panicked in tests; import continues | +| spawned-test mocks | Done on the branch | +| Lifecycle monitors | Every actor stop observed within the deadline | + +**Phase 3.** + +| Item | Acceptance | +|---|---| +| Actor-level exploration once spawned ships `explore` | Low value while ethlambda runs two main actors (`BlockChain`, `P2P`); revisit if that grows | +| TLA+ 3SF-mini model | Checks in CI; simulator traces validate against it | + +## Shared with ethrex + +- **Storage pattern:** the same backend-trait-plus-in-memory approach makes + restart and failpoint tests portable between the two clients. +- **Discovery:** ethlambda runs ethrex's discv5 `DiscoveryServer` + (`crates/net/p2p/src/discovery/mod.rs:21`), so discovery tests belong upstream. +- **Harness:** both build on spawned; its testing harness and monitors serve + both. + +Related plans: [ethrex testing and verification roadmap (ethrex#7517)](https://github.com/lambdaclass/ethrex/pull/7517), +[spawned: Database-grade testing program](https://github.com/lambdaclass/spawned/blob/main/ROADMAP.md#database-grade-testing-program), +[spawned: systematic testing research](https://github.com/lambdaclass/spawned/blob/main/docs/research/systematic-testing.md). + +## Sources + +- Konnov, Kukovec, Pani, Saltini, Tran, *Exploring Automatic Model-Checking of the Ethereum specification*: +- FoundationDB, *Simulation and Testing*: +- TigerBeetle, *VOPR*: +- Kani on AWS Firecracker: +- Attacknet: +- Shadow network simulator: +- lean-shadow-fuzzer: +- ethrex PR #7517: +- spawned ROADMAP and research note: links above