From 8d18541bc7d15bd87bb756cc2417f8f050b2c09d Mon Sep 17 00:00:00 2001 From: Federico Carrone Date: Mon, 9 Nov 2026 13:15:49 -0300 Subject: [PATCH 1/9] chore: upgrade spawned to v0.7 Point spawned-concurrency and spawned-rt at the unreleased v0.7 candidate (lambdaclass/spawned fd28d2e9: main 32796a88 plus the spawned-test crate), and declare spawned-test at the same revision so every spawned crate shares one set of types. Mechanical only: ethlambda's 0.5 protocol and handler signatures compile unchanged against v0.7, both crates keep their default features (which now include `async`), and every actor still starts the way it did. Nothing is placed in an ActorSystem yet. --- Cargo.lock | 88 +++++++++++++++++++++++++++++++++++++++++++++++++----- Cargo.toml | 7 +++-- 2 files changed, 85 insertions(+), 10 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 0bec6c9e0..97701e95a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2040,7 +2040,7 @@ dependencies = [ "serde", "serde_json", "snap", - "spawned-concurrency", + "spawned-concurrency 0.6.0", "thiserror 2.0.18", "tokio", "tokio-util", @@ -2100,7 +2100,7 @@ name = "ethlambda-network-api" version = "1.0.0" dependencies = [ "ethlambda-types", - "spawned-concurrency", + "spawned-concurrency 0.6.0", ] [[package]] @@ -2128,7 +2128,7 @@ dependencies = [ "secp256k1 0.30.0", "sha2", "snap", - "spawned-concurrency", + "spawned-concurrency 0.6.0", "thiserror 2.0.18", "tokio", "tokio-stream", @@ -2159,7 +2159,7 @@ dependencies = [ "libssz-types 0.3.0 (git+https://github.com/lambdaclass/libssz?rev=5cb14378ffdadcb51b387c003567e2630d0c7073)", "serde", "serde_json", - "spawned-concurrency", + "spawned-concurrency 0.6.0", "tokio", "tokio-util", "tower", @@ -2482,8 +2482,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 +2867,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 +4449,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" @@ -6180,6 +6208,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" @@ -6554,8 +6588,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.6.0" +source = "git+https://github.com/lambdaclass/spawned?rev=fd28d2e93ceae3657073b030b0c64d81ec2c11d9#fd28d2e93ceae3657073b030b0c64d81ec2c11d9" +dependencies = [ + "futures", + "pin-project-lite", + "spawned-macros 0.6.0", + "spawned-rt 0.6.0", "thiserror 2.0.18", "tracing", ] @@ -6571,6 +6618,16 @@ dependencies = [ "syn 2.0.117", ] +[[package]] +name = "spawned-macros" +version = "0.6.0" +source = "git+https://github.com/lambdaclass/spawned?rev=fd28d2e93ceae3657073b030b0c64d81ec2c11d9#fd28d2e93ceae3657073b030b0c64d81ec2c11d9" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.117", +] + [[package]] name = "spawned-rt" version = "0.5.0" @@ -6585,6 +6642,21 @@ dependencies = [ "tracing-subscriber", ] +[[package]] +name = "spawned-rt" +version = "0.6.0" +source = "git+https://github.com/lambdaclass/spawned?rev=fd28d2e93ceae3657073b030b0c64d81ec2c11d9#fd28d2e93ceae3657073b030b0c64d81ec2c11d9" +dependencies = [ + "ctrlc", + "loom", + "nix 0.30.1", + "tokio", + "tokio-stream", + "tokio-util", + "tracing", + "tracing-subscriber", +] + [[package]] name = "sphincs" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index e13475448..5308cec18 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -90,8 +90,11 @@ hex = "0.4" hmac = "0.12" base64 = "0.22" -spawned-concurrency = "0.5.0" -spawned-rt = "0.5.0" +# Unreleased spawned v0.7 (main 32796a88 plus the spawned-test crate). Every +# spawned crate must name the same revision so they share one set of types. +spawned-concurrency = { git = "https://github.com/lambdaclass/spawned", rev = "fd28d2e93ceae3657073b030b0c64d81ec2c11d9" } +spawned-rt = { git = "https://github.com/lambdaclass/spawned", rev = "fd28d2e93ceae3657073b030b0c64d81ec2c11d9" } +spawned-test = { git = "https://github.com/lambdaclass/spawned", rev = "fd28d2e93ceae3657073b030b0c64d81ec2c11d9" } tokio = "1.0" tokio-util = "0.7" From c891553c4ffb33df16d0cec2d6d87853d60e6a40 Mon Sep 17 00:00:00 2001 From: Federico Carrone Date: Mon, 9 Nov 2026 13:25:15 -0300 Subject: [PATCH 2/9] refactor: own actors in an ActorSystem with a shutdown deadline Startup and shutdown were wired by hand: the chain actor on its own `Backend::Thread`, P2P on the shared runtime, and on ctrl-c both were stopped and joined with no deadline. A clean shutdown during catch-up took 14.3 s, past Docker's 10 s grace period (1d7902da), and only SIGINT was handled, so `docker stop` (SIGTERM) killed the node outright unless it was run with `--stop-signal SIGINT`. Now one ActorSystem owns both actors and their threads: - `chain`: a one-worker `thread_group`, the same isolation the chain actor had on `Backend::Thread`: one dedicated native thread for its long synchronous handlers, plus a one-worker Tokio runtime for its timers and the aggregation worker's `spawn_blocking`. - `network`: an `async_group` for P2P, sized like the default Tokio runtime it used to share (one worker under Shadow). The chain actor is admitted before P2P, so the system's reverse-admission shutdown stops P2P, the ingress, before the actor it feeds. discv5 now binds its socket in `run_node` before any actor starts, so a busy port still aborts startup with nothing running; `P2P::spawn` takes the handle and an execution group, `BlockChain::spawn{,_beacon}` take a group. Shutdown answers SIGTERM as well as SIGINT and is bounded by an 8 s deadline: the HTTP server gets 2 s to close (an open SSE stream could otherwise hold it), then the system stops the actors with what is left. A shutdown that misses the deadline logs the unfinished groups and actors by name and exits 1 instead of waiting. Repeated ctrl-c still forces an exit. --- Cargo.lock | 1 + bin/ethlambda/Cargo.toml | 1 + bin/ethlambda/src/main.rs | 300 ++++++++++++++++++++++++++--------- crates/blockchain/src/lib.rs | 40 +++-- crates/net/p2p/src/lib.rs | 39 +++-- 5 files changed, 276 insertions(+), 105 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 97701e95a..7d9e40fb9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2010,6 +2010,7 @@ dependencies = [ "serde_yaml_ng", "sha2", "snap", + "spawned-concurrency 0.6.0", "tempfile", "thiserror 2.0.18", "tikv-jemallocator", 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/main.rs b/bin/ethlambda/src/main.rs index db3f62368..164ab6377 100644 --- a/bin/ethlambda/src/main.rs +++ b/bin/ethlambda/src/main.rs @@ -31,12 +31,14 @@ 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::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 +55,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 +233,29 @@ 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: P2P, the ingress, before the + /// chain actor it feeds. 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` @@ -769,8 +791,37 @@ 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, on a multi-thread pool sized like the + // default Tokio runtime it used to share with everything else. + // + // Admission order is shutdown order reversed, so the chain actor is + // admitted first and P2P, which feeds it, is stopped 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,31 +834,77 @@ 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(); + // What the HTTP server and P2P take from values the chain actor's spawn + // below moves. + let rpc_store = setup.store.clone(); + let rpc_events = events.clone(); + 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(), + }; + // Block production builds its payloads with the same execution client the + // chain actor validates them with. + let rpc_engine = match &setup.chain { + ChainActor::Beacon { engine, .. } => engine.clone(), + ChainActor::Lean(..) => None, + }; + + // Which HTTP surface this node serves follows from the store's own chain + // tag rather than from the sub-command, so the two can never disagree. + // A beacon node served `/lean/v0` until now, off a store those handlers + // cannot read: they reach for lean state variants and metadata keys a + // beacon directory never carries, so calling one panicked that request. + let serves_beacon_api = setup.store.chain() == ethlambda_storage::Chain::Beacon; + + // 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, &chain_group) + } + ChainActor::Beacon { + custody_columns, + engine, + safe_slots_to_import_optimistically, + anchor_envelope_urls, + } => { + check_custody_set(&custody_columns)?; + anchor_envelope_fetch = missing_anchor_envelope_root(&setup.store) + .filter(|_| !anchor_envelope_urls.is_empty()) + .map(|root| (anchor_envelope_urls, root)); + BlockChain::spawn_beacon( + setup.store, + sync_status, + events, + custody_columns, + engine, + safe_slots_to_import_optimistically, + &chain_group, + ) + } + } + .wrap_err("failed to start the chain actor")?; + let p2p = P2P::spawn( built, - setup.store.clone(), + p2p_store, setup.node_names, discovery, payload_attestation_pool.clone(), builder_market.clone(), sync_committee_pool.clone(), + &network_group, ) - .await - .wrap_err("failed to start discv5 discovery")?; + .wrap_err("failed to start the P2P actor")?; let shutdown = CancellationToken::new(); let rpc_shutdown = shutdown.clone(); - 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_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( @@ -818,20 +915,6 @@ async fn run_node(options: Options) -> eyre::Result<()> { &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 { - ChainActor::Beacon { engine, .. } => engine.clone(), - ChainActor::Lean(..) => None, - }; - - // Which HTTP surface this node serves follows from the store's own chain - // tag rather than from the sub-command, so the two can never disagree. - // A beacon node served `/lean/v0` until now, off a store those handlers - // cannot read: they reach for lean state variants and metadata keys a - // 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( @@ -868,32 +951,6 @@ async fn run_node(options: Options) -> eyre::Result<()> { let _ = served.inspect_err(|err| error!(%err, "RPC server failed")); }); - 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) - } - ChainActor::Beacon { - custody_columns, - engine, - safe_slots_to_import_optimistically, - anchor_envelope_urls, - } => { - check_custody_set(&custody_columns)?; - anchor_envelope_fetch = missing_anchor_envelope_root(&setup.store) - .filter(|_| !anchor_envelope_urls.is_empty()) - .map(|root| (anchor_envelope_urls, root)); - BlockChain::spawn_beacon( - setup.store, - sync_status, - events, - custody_columns, - engine, - safe_slots_to_import_optimistically, - ) - } - }; - let p2p_ref = p2p.actor_ref(); let p2p_to_block_chain = p2p_ref.to_block_chain_to_p2p_ref(); @@ -931,13 +988,24 @@ 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 +1041,32 @@ 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()`. +/// Ingress first: the HTTP server is cancelled and given +/// [`HTTP_SHUTDOWN_TIMEOUT`] to close, then the actor system stops the actors +/// in reverse admission order (P2P, then the chain actor) with whatever is left +/// of the deadline. 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 +1083,91 @@ 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(); + let RunningNode { + system, + p2p, + blockchain, + mut http, + shutdown, + } = node; + // Read now: the report names unfinished actors by id only. + let actors = [ + (p2p.actor_ref().id(), "p2p"), + (blockchain.actor_ref().id(), "blockchain"), + ]; + + shutdown.cancel(); + if tokio::time::timeout(HTTP_SHUTDOWN_TIMEOUT, &mut http) + .await + .is_err() + { + warn!( + timeout_secs = HTTP_SHUTDOWN_TIMEOUT.as_secs(), + "HTTP server did not close in time; aborting it" + ); + http.abort(); + } - blockchain_ref.context().stop(); - p2p_ref.context().stop(); - node.shutdown.cancel(); + let report = system + .shutdown(SHUTDOWN_DEADLINE.saturating_sub(started.elapsed())) + .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/src/lib.rs b/crates/net/p2p/src/lib.rs index cc9ca2bd1..bb87542d2 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 { From b33738d97e0ca2074d745758b874b3c75cb2c61e Mon Sep 17 00:00:00 2001 From: Federico Carrone Date: Mon, 9 Nov 2026 13:28:50 -0300 Subject: [PATCH 3/9] feat: supervise restartable actors Put the HTTP server (API, metrics and debug routes) under a OneForOne supervisor in the actor system. It was a bare `tokio::spawn`: a failure logged one error and the node ran on with no API and no metrics. It is now an `HttpServer` actor, admitted on the `network` group, that serves on an owned helper and stops when the server returns, so the supervisor builds a fresh one. Each run builds its router afresh from cloned handles, which is what makes a restart safe; what one loses is per connection (open event streams, Beacon API fee recipients until the validator client re-registers them). Budget: Permanent, at most 3 restarts a minute, backoff from 1 s to 10 s. A server that keeps failing (its port taken by another process) exhausts it, the supervisor gives up with an error log, and the node runs on without HTTP, as it did before. Restarts and the give-up are logged by a watcher, since spawned logs neither. The tree is admitted last, so shutdown stops it first, with 2 s of the 8 s deadline. The chain and P2P actors stay unsupervised, with the reasons in `run_node`: the chain actor's in-memory state, including the validator keys' prepared signing state, is not rebuilt from the store, and P2P owns the swarm and discv5 sockets built once at startup; both are referenced by the other actors through refs a replacement would not rebind. For test nodes only, `ETHLAMBDA_FAULT_INJECTION=1` makes SIGUSR1 crash the current HTTP server with a real handler panic, to exercise recovery. --- bin/ethlambda/src/http_server.rs | 268 +++++++++++++++++++++++++++++++ bin/ethlambda/src/main.rs | 141 +++++++++------- 2 files changed, 352 insertions(+), 57 deletions(-) create mode 100644 bin/ethlambda/src/http_server.rs 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 164ab6377..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; @@ -32,6 +33,7 @@ static ALLOC: tikv_jemallocator::Jemalloc = tikv_jemallocator::Jemalloc; 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}, @@ -253,8 +255,9 @@ const HTTP_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(2); /// 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: P2P, the ingress, before the - /// chain actor it feeds. See [`run_node`] for why that is the order. + /// 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 @@ -263,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 @@ -809,11 +810,29 @@ async fn run_node(options: Options) -> eyre::Result<()> { // 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, on a multi-thread pool sized like the - // default Tokio runtime it used to share with everything else. + // - `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 P2P, which feeds it, is stopped first. + // 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) @@ -901,8 +920,6 @@ async fn run_node(options: Options) -> eyre::Result<()> { ) .wrap_err("failed to start the P2P actor")?; - let shutdown = CancellationToken::new(); - let rpc_shutdown = shutdown.clone(); let rpc_aggregator = aggregator.clone(); let rpc_p2p = p2p.actor_ref().to_rpc_to_p2p_ref(); let rpc_identity = ethlambda_rpc::BeaconIdentity { @@ -915,41 +932,65 @@ async fn run_node(options: Options) -> eyre::Result<()> { &local_peer_id, ) }; - let http = tokio::spawn(async move { - let served = if serves_beacon_api { - ethlambda_rpc::start_beacon_rpc_server( + // 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, - 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, - }, + handles, local_peer_id, - rpc_shutdown, - ) - .await + shutdown, + )) } else { - ethlambda_rpc::start_rpc_server( + Box::pin(ethlambda_rpc::start_rpc_server( rpc_config, rpc_store, - rpc_aggregator, + rpc_aggregator.clone(), rpc_sync_status, local_peer_id, rpc_events, - rpc_shutdown, - ) - .await - }; - let _ = served.inspect_err(|err| error!(%err, "RPC server failed")); + 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(); @@ -992,7 +1033,6 @@ async fn run_node(options: Options) -> eyre::Result<()> { p2p, blockchain, http, - shutdown, }) .await } @@ -1044,10 +1084,9 @@ fn check_custody_set(custody_columns: &[u64]) -> eyre::Result<()> { /// Wait for SIGINT or SIGTERM, then stop everything within /// [`SHUTDOWN_DEADLINE`]. /// -/// Ingress first: the HTTP server is cancelled and given -/// [`HTTP_SHUTDOWN_TIMEOUT`] to close, then the actor system stops the actors -/// in reverse admission order (P2P, then the chain actor) with whatever is left -/// of the deadline. A shutdown that misses it names what is still running and +/// 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. /// @@ -1087,30 +1126,18 @@ async fn wait_for_shutdown(node: RunningNode) -> eyre::Result<()> { system, p2p, blockchain, - mut http, - shutdown, + http, } = node; // Read now: the report names unfinished actors by id only. - let actors = [ + let mut actors = vec![ (p2p.actor_ref().id(), "p2p"), (blockchain.actor_ref().id(), "blockchain"), ]; - - shutdown.cancel(); - if tokio::time::timeout(HTTP_SHUTDOWN_TIMEOUT, &mut http) - .await - .is_err() - { - warn!( - timeout_secs = HTTP_SHUTDOWN_TIMEOUT.as_secs(), - "HTTP server did not close in time; aborting it" - ); - http.abort(); + if let Ok(server) = http.current() { + actors.push((server.id(), "http")); } - let report = system - .shutdown(SHUTDOWN_DEADLINE.saturating_sub(started.elapsed())) - .await; + let report = system.shutdown(SHUTDOWN_DEADLINE).await; let elapsed_ms = started.elapsed().as_millis() as u64; if report.complete { info!(elapsed_ms, "Shutdown complete"); From edcddbf1fe1cd5b41cca2d5306ac2dc4eed3c558 Mon Sep 17 00:00:00 2001 From: Federico Carrone Date: Mon, 9 Nov 2026 13:30:50 -0300 Subject: [PATCH 4/9] test: replace hand-written protocol stubs with spawned-test mocks The p2p forwarding tests stood in for the chain actor with four hand-written `P2PToBlockChain` impls: `RecordingChain` in column_checks.rs and in test_support, `Recorder` in envelope_checks.rs (also used by envelope_client.rs) and `GloasRecorder` in verdict.rs. Each implemented all eight protocol methods to record one or two of them. They are now `spawned_test::tasks::Mock` actors converted with `to_p2p_to_block_chain_ref()`, so the code under test still holds the protocol ref it holds in production: - `Mock::new().allow_sends()` logs every send; tests read what they care about with `expect` / `take_all` and the arrival order with `received()`. - Tests asserting that nothing reaches the chain actor use a strict `Mock::new()`, which fails on any message at all rather than only the one method the old stub recorded. The stubs were synchronous; a mock is an actor, so tests settle it (or expect the last send) before reading the log. Where the old tests closed a channel to prove nothing followed, the new ones rely on the code's own ordering (the parked batch is the task's last send; the range release fires after the last envelope) and assert the complete log. --- Cargo.lock | 27 +++- crates/net/p2p/Cargo.toml | 5 + crates/net/p2p/src/beacon/column_checks.rs | 119 +++----------- crates/net/p2p/src/beacon/envelope_checks.rs | 107 +++---------- crates/net/p2p/src/beacon/verdict.rs | 149 ++++++------------ crates/net/p2p/src/gossipsub/handler.rs | 18 ++- crates/net/p2p/src/lib.rs | 79 ---------- .../net/p2p/src/req_resp/envelope_client.rs | 54 +++++-- 8 files changed, 173 insertions(+), 385 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 7d9e40fb9..bd8ad4729 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]] @@ -2130,6 +2130,7 @@ dependencies = [ "sha2", "snap", "spawned-concurrency 0.6.0", + "spawned-test", "thiserror 2.0.18", "tokio", "tokio-stream", @@ -4796,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]] @@ -6086,7 +6087,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys 0.12.1", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -6578,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]] @@ -6658,6 +6659,16 @@ dependencies = [ "tracing-subscriber", ] +[[package]] +name = "spawned-test" +version = "0.6.0" +source = "git+https://github.com/lambdaclass/spawned?rev=fd28d2e93ceae3657073b030b0c64d81ec2c11d9#fd28d2e93ceae3657073b030b0c64d81ec2c11d9" +dependencies = [ + "spawned-concurrency 0.6.0", + "spawned-rt 0.6.0", + "tracing", +] + [[package]] name = "sphincs" version = "0.1.0" @@ -6835,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]] @@ -7622,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/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 bb87542d2..94a90f20a 100644 --- a/crates/net/p2p/src/lib.rs +++ b/crates/net/p2p/src/lib.rs @@ -2825,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)]); } } From 611bbe53fafce2b34f165b7b28b61ba4cb6d4bff Mon Sep 17 00:00:00 2001 From: Federico Carrone Date: Mon, 9 Nov 2026 13:32:01 -0300 Subject: [PATCH 5/9] feat: export spawned actor metrics on /metrics Enable spawned-concurrency's `metrics` feature and append its Prometheus text to the existing `/metrics` response: mailbox depth, queue wait, handler duration, request outcomes and actor lifecycle (spawned, exited by reason), all prefixed `spawned_` and summed over every actor in the process. They are what to watch on a canary of the v0.7 runtime: a restarted HTTP server shows as an exit and a new spawn, and a stalled chain actor as queue wait and mailbox depth. Collection is the library's fixed set of atomic counters, timing every message (the default sampling); no exporter or labels are added. --- Cargo.toml | 4 +++- crates/net/rpc/Cargo.toml | 3 ++- crates/net/rpc/src/metrics.rs | 9 ++++++--- crates/net/rpc/tests/http_servers.rs | 8 +++++++- 4 files changed, 18 insertions(+), 6 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 5308cec18..2da5020f9 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -92,7 +92,9 @@ base64 = "0.22" # Unreleased spawned v0.7 (main 32796a88 plus the spawned-test crate). Every # spawned crate must name the same revision so they share one set of types. -spawned-concurrency = { git = "https://github.com/lambdaclass/spawned", rev = "fd28d2e93ceae3657073b030b0c64d81ec2c11d9" } +# `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 = "fd28d2e93ceae3657073b030b0c64d81ec2c11d9", features = ["metrics"] } spawned-rt = { git = "https://github.com/lambdaclass/spawned", rev = "fd28d2e93ceae3657073b030b0c64d81ec2c11d9" } spawned-test = { git = "https://github.com/lambdaclass/spawned", rev = "fd28d2e93ceae3657073b030b0c64d81ec2c11d9" } tokio = "1.0" 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!( From 047860fac6f92949ddff1036a6ccf28ca5314c05 Mon Sep 17 00:00:00 2001 From: Federico Carrone Date: Mon, 9 Nov 2026 13:52:30 -0300 Subject: [PATCH 6/9] chore: pin spawned to the v0.7.0 release plus spawned-test --- Cargo.lock | 34 +++++++++++++++++----------------- Cargo.toml | 8 ++++---- 2 files changed, 21 insertions(+), 21 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index bd8ad4729..4ee3c9c6d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2010,7 +2010,7 @@ dependencies = [ "serde_yaml_ng", "sha2", "snap", - "spawned-concurrency 0.6.0", + "spawned-concurrency 0.7.0", "tempfile", "thiserror 2.0.18", "tikv-jemallocator", @@ -2041,7 +2041,7 @@ dependencies = [ "serde", "serde_json", "snap", - "spawned-concurrency 0.6.0", + "spawned-concurrency 0.7.0", "thiserror 2.0.18", "tokio", "tokio-util", @@ -2101,7 +2101,7 @@ name = "ethlambda-network-api" version = "1.0.0" dependencies = [ "ethlambda-types", - "spawned-concurrency 0.6.0", + "spawned-concurrency 0.7.0", ] [[package]] @@ -2129,7 +2129,7 @@ dependencies = [ "secp256k1 0.30.0", "sha2", "snap", - "spawned-concurrency 0.6.0", + "spawned-concurrency 0.7.0", "spawned-test", "thiserror 2.0.18", "tokio", @@ -2161,7 +2161,7 @@ dependencies = [ "libssz-types 0.3.0 (git+https://github.com/lambdaclass/libssz?rev=5cb14378ffdadcb51b387c003567e2630d0c7073)", "serde", "serde_json", - "spawned-concurrency 0.6.0", + "spawned-concurrency 0.7.0", "tokio", "tokio-util", "tower", @@ -6598,13 +6598,13 @@ dependencies = [ [[package]] name = "spawned-concurrency" -version = "0.6.0" -source = "git+https://github.com/lambdaclass/spawned?rev=fd28d2e93ceae3657073b030b0c64d81ec2c11d9#fd28d2e93ceae3657073b030b0c64d81ec2c11d9" +version = "0.7.0" +source = "git+https://github.com/lambdaclass/spawned?rev=39be3f3b62fa87ade2b62dbf45f0437f95024480#39be3f3b62fa87ade2b62dbf45f0437f95024480" dependencies = [ "futures", "pin-project-lite", - "spawned-macros 0.6.0", - "spawned-rt 0.6.0", + "spawned-macros 0.7.0", + "spawned-rt 0.7.0", "thiserror 2.0.18", "tracing", ] @@ -6622,8 +6622,8 @@ dependencies = [ [[package]] name = "spawned-macros" -version = "0.6.0" -source = "git+https://github.com/lambdaclass/spawned?rev=fd28d2e93ceae3657073b030b0c64d81ec2c11d9#fd28d2e93ceae3657073b030b0c64d81ec2c11d9" +version = "0.7.0" +source = "git+https://github.com/lambdaclass/spawned?rev=39be3f3b62fa87ade2b62dbf45f0437f95024480#39be3f3b62fa87ade2b62dbf45f0437f95024480" dependencies = [ "proc-macro2", "quote", @@ -6646,8 +6646,8 @@ dependencies = [ [[package]] name = "spawned-rt" -version = "0.6.0" -source = "git+https://github.com/lambdaclass/spawned?rev=fd28d2e93ceae3657073b030b0c64d81ec2c11d9#fd28d2e93ceae3657073b030b0c64d81ec2c11d9" +version = "0.7.0" +source = "git+https://github.com/lambdaclass/spawned?rev=39be3f3b62fa87ade2b62dbf45f0437f95024480#39be3f3b62fa87ade2b62dbf45f0437f95024480" dependencies = [ "ctrlc", "loom", @@ -6661,11 +6661,11 @@ dependencies = [ [[package]] name = "spawned-test" -version = "0.6.0" -source = "git+https://github.com/lambdaclass/spawned?rev=fd28d2e93ceae3657073b030b0c64d81ec2c11d9#fd28d2e93ceae3657073b030b0c64d81ec2c11d9" +version = "0.7.0" +source = "git+https://github.com/lambdaclass/spawned?rev=39be3f3b62fa87ade2b62dbf45f0437f95024480#39be3f3b62fa87ade2b62dbf45f0437f95024480" dependencies = [ - "spawned-concurrency 0.6.0", - "spawned-rt 0.6.0", + "spawned-concurrency 0.7.0", + "spawned-rt 0.7.0", "tracing", ] diff --git a/Cargo.toml b/Cargo.toml index 2da5020f9..95656c01d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -90,13 +90,13 @@ hex = "0.4" hmac = "0.12" base64 = "0.22" -# Unreleased spawned v0.7 (main 32796a88 plus the spawned-test crate). Every +# spawned v0.7.0 (release 02b9400b) 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 = "fd28d2e93ceae3657073b030b0c64d81ec2c11d9", features = ["metrics"] } -spawned-rt = { git = "https://github.com/lambdaclass/spawned", rev = "fd28d2e93ceae3657073b030b0c64d81ec2c11d9" } -spawned-test = { git = "https://github.com/lambdaclass/spawned", rev = "fd28d2e93ceae3657073b030b0c64d81ec2c11d9" } +spawned-concurrency = { git = "https://github.com/lambdaclass/spawned", rev = "39be3f3b62fa87ade2b62dbf45f0437f95024480", features = ["metrics"] } +spawned-rt = { git = "https://github.com/lambdaclass/spawned", rev = "39be3f3b62fa87ade2b62dbf45f0437f95024480" } +spawned-test = { git = "https://github.com/lambdaclass/spawned", rev = "39be3f3b62fa87ade2b62dbf45f0437f95024480" } tokio = "1.0" tokio-util = "0.7" From 77db8d24191f0772b4ab18d6a9d78a342a1ce1f8 Mon Sep 17 00:00:00 2001 From: Federico Carrone Date: Mon, 9 Nov 2026 14:06:25 -0300 Subject: [PATCH 7/9] docs: add a testing and verification roadmap --- README.md | 2 + docs/SUMMARY.md | 1 + docs/testing_roadmap.md | 293 ++++++++++++++++++++++++++++++++++++++++ 3 files changed, 296 insertions(+) create mode 100644 docs/testing_roadmap.md 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/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 From 0660f21f92c93575b1f6a0ee8f28fb370a2185a7 Mon Sep 17 00:00:00 2001 From: Federico Carrone Date: Mon, 9 Nov 2026 14:52:40 -0300 Subject: [PATCH 8/9] chore: pin spawned to main 342b707b (canary fixes) plus spawned-test --- Cargo.lock | 8 ++++---- Cargo.toml | 8 ++++---- 2 files changed, 8 insertions(+), 8 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 4ee3c9c6d..56844e501 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6599,7 +6599,7 @@ dependencies = [ [[package]] name = "spawned-concurrency" version = "0.7.0" -source = "git+https://github.com/lambdaclass/spawned?rev=39be3f3b62fa87ade2b62dbf45f0437f95024480#39be3f3b62fa87ade2b62dbf45f0437f95024480" +source = "git+https://github.com/lambdaclass/spawned?rev=646cc0a4887b323e7e5231fa42d45509aa205ca5#646cc0a4887b323e7e5231fa42d45509aa205ca5" dependencies = [ "futures", "pin-project-lite", @@ -6623,7 +6623,7 @@ dependencies = [ [[package]] name = "spawned-macros" version = "0.7.0" -source = "git+https://github.com/lambdaclass/spawned?rev=39be3f3b62fa87ade2b62dbf45f0437f95024480#39be3f3b62fa87ade2b62dbf45f0437f95024480" +source = "git+https://github.com/lambdaclass/spawned?rev=646cc0a4887b323e7e5231fa42d45509aa205ca5#646cc0a4887b323e7e5231fa42d45509aa205ca5" dependencies = [ "proc-macro2", "quote", @@ -6647,7 +6647,7 @@ dependencies = [ [[package]] name = "spawned-rt" version = "0.7.0" -source = "git+https://github.com/lambdaclass/spawned?rev=39be3f3b62fa87ade2b62dbf45f0437f95024480#39be3f3b62fa87ade2b62dbf45f0437f95024480" +source = "git+https://github.com/lambdaclass/spawned?rev=646cc0a4887b323e7e5231fa42d45509aa205ca5#646cc0a4887b323e7e5231fa42d45509aa205ca5" dependencies = [ "ctrlc", "loom", @@ -6662,7 +6662,7 @@ dependencies = [ [[package]] name = "spawned-test" version = "0.7.0" -source = "git+https://github.com/lambdaclass/spawned?rev=39be3f3b62fa87ade2b62dbf45f0437f95024480#39be3f3b62fa87ade2b62dbf45f0437f95024480" +source = "git+https://github.com/lambdaclass/spawned?rev=646cc0a4887b323e7e5231fa42d45509aa205ca5#646cc0a4887b323e7e5231fa42d45509aa205ca5" dependencies = [ "spawned-concurrency 0.7.0", "spawned-rt 0.7.0", diff --git a/Cargo.toml b/Cargo.toml index 95656c01d..a06f4f7f9 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -90,13 +90,13 @@ hex = "0.4" hmac = "0.12" base64 = "0.22" -# spawned v0.7.0 (release 02b9400b) plus the unreleased spawned-test crate (lambdaclass/spawned#198). Every +# spawned main 342b707b (v0.7.0 plus canary 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 = "39be3f3b62fa87ade2b62dbf45f0437f95024480", features = ["metrics"] } -spawned-rt = { git = "https://github.com/lambdaclass/spawned", rev = "39be3f3b62fa87ade2b62dbf45f0437f95024480" } -spawned-test = { git = "https://github.com/lambdaclass/spawned", rev = "39be3f3b62fa87ade2b62dbf45f0437f95024480" } +spawned-concurrency = { git = "https://github.com/lambdaclass/spawned", rev = "646cc0a4887b323e7e5231fa42d45509aa205ca5", features = ["metrics"] } +spawned-rt = { git = "https://github.com/lambdaclass/spawned", rev = "646cc0a4887b323e7e5231fa42d45509aa205ca5" } +spawned-test = { git = "https://github.com/lambdaclass/spawned", rev = "646cc0a4887b323e7e5231fa42d45509aa205ca5" } tokio = "1.0" tokio-util = "0.7" From 57a6e13bf31463946838ed9677c29c7eb801a6d3 Mon Sep 17 00:00:00 2001 From: Federico Carrone Date: Mon, 9 Nov 2026 15:42:35 -0300 Subject: [PATCH 9/9] chore: pin spawned to main b1a17e6c plus spawned-test --- Cargo.lock | 8 ++++---- Cargo.toml | 8 ++++---- 2 files changed, 8 insertions(+), 8 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 56844e501..1afda0a4c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6599,7 +6599,7 @@ dependencies = [ [[package]] name = "spawned-concurrency" version = "0.7.0" -source = "git+https://github.com/lambdaclass/spawned?rev=646cc0a4887b323e7e5231fa42d45509aa205ca5#646cc0a4887b323e7e5231fa42d45509aa205ca5" +source = "git+https://github.com/lambdaclass/spawned?rev=fc35abafe7a850aa27f4b511b60e28dc49b4a393#fc35abafe7a850aa27f4b511b60e28dc49b4a393" dependencies = [ "futures", "pin-project-lite", @@ -6623,7 +6623,7 @@ dependencies = [ [[package]] name = "spawned-macros" version = "0.7.0" -source = "git+https://github.com/lambdaclass/spawned?rev=646cc0a4887b323e7e5231fa42d45509aa205ca5#646cc0a4887b323e7e5231fa42d45509aa205ca5" +source = "git+https://github.com/lambdaclass/spawned?rev=fc35abafe7a850aa27f4b511b60e28dc49b4a393#fc35abafe7a850aa27f4b511b60e28dc49b4a393" dependencies = [ "proc-macro2", "quote", @@ -6647,7 +6647,7 @@ dependencies = [ [[package]] name = "spawned-rt" version = "0.7.0" -source = "git+https://github.com/lambdaclass/spawned?rev=646cc0a4887b323e7e5231fa42d45509aa205ca5#646cc0a4887b323e7e5231fa42d45509aa205ca5" +source = "git+https://github.com/lambdaclass/spawned?rev=fc35abafe7a850aa27f4b511b60e28dc49b4a393#fc35abafe7a850aa27f4b511b60e28dc49b4a393" dependencies = [ "ctrlc", "loom", @@ -6662,7 +6662,7 @@ dependencies = [ [[package]] name = "spawned-test" version = "0.7.0" -source = "git+https://github.com/lambdaclass/spawned?rev=646cc0a4887b323e7e5231fa42d45509aa205ca5#646cc0a4887b323e7e5231fa42d45509aa205ca5" +source = "git+https://github.com/lambdaclass/spawned?rev=fc35abafe7a850aa27f4b511b60e28dc49b4a393#fc35abafe7a850aa27f4b511b60e28dc49b4a393" dependencies = [ "spawned-concurrency 0.7.0", "spawned-rt 0.7.0", diff --git a/Cargo.toml b/Cargo.toml index a06f4f7f9..bc69fda7d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -90,13 +90,13 @@ hex = "0.4" hmac = "0.12" base64 = "0.22" -# spawned main 342b707b (v0.7.0 plus canary fixes) plus the unreleased spawned-test crate (lambdaclass/spawned#198). Every +# 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 = "646cc0a4887b323e7e5231fa42d45509aa205ca5", features = ["metrics"] } -spawned-rt = { git = "https://github.com/lambdaclass/spawned", rev = "646cc0a4887b323e7e5231fa42d45509aa205ca5" } -spawned-test = { git = "https://github.com/lambdaclass/spawned", rev = "646cc0a4887b323e7e5231fa42d45509aa205ca5" } +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"