From 06656db888c4f0be7aef016ca93a0200f52db659 Mon Sep 17 00:00:00 2001 From: Bronek Kozicki Date: Wed, 16 Sep 2026 17:43:25 +0100 Subject: [PATCH 1/2] Subscribe to peer events before starting producers Connections established before the application boundary first reads PeerEvent can be absent from the Beacon API for their lifetime. Update Flux to the revision tagged v0.3.2 and subscribe in on_attach. Construct the boundary runner before starting producer tiles, while keeping its worker startup position and thread configuration. Add a regression test that publishes a connection before the boundary worker starts and checks the peer endpoint. It fails with an empty peer list when early subscription is disabled. Assisted-by: Codex:gpt-6-astra --- Cargo.lock | 36 ++++++++-------- Cargo.toml | 6 +-- crates/application_boundary/src/lib.rs | 4 ++ crates/application_boundary/tests/tile.rs | 51 +++++++++++++++++++++++ crates/bin/src/main.rs | 19 ++++++--- 5 files changed, 89 insertions(+), 27 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 7f53b3bb..065a53d5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1636,8 +1636,8 @@ dependencies = [ [[package]] name = "flux" -version = "0.3.1" -source = "git+https://github.com/gattaca-com/flux?rev=5402c63afc122293f04ea69cd9ae295d7d011c8f#5402c63afc122293f04ea69cd9ae295d7d011c8f" +version = "0.3.2" +source = "git+https://github.com/gattaca-com/flux?rev=c877c5edfdedcc03d3d37af9977624793e821fe3#c877c5edfdedcc03d3d37af9977624793e821fe3" dependencies = [ "bitcode", "core_affinity", @@ -1657,8 +1657,8 @@ dependencies = [ [[package]] name = "flux-communication" -version = "0.3.1" -source = "git+https://github.com/gattaca-com/flux?rev=5402c63afc122293f04ea69cd9ae295d7d011c8f#5402c63afc122293f04ea69cd9ae295d7d011c8f" +version = "0.3.2" +source = "git+https://github.com/gattaca-com/flux?rev=c877c5edfdedcc03d3d37af9977624793e821fe3#c877c5edfdedcc03d3d37af9977624793e821fe3" dependencies = [ "directories", "flux-timing", @@ -1673,8 +1673,8 @@ dependencies = [ [[package]] name = "flux-profiler" -version = "0.3.1" -source = "git+https://github.com/gattaca-com/flux?rev=5402c63afc122293f04ea69cd9ae295d7d011c8f#5402c63afc122293f04ea69cd9ae295d7d011c8f" +version = "0.3.2" +source = "git+https://github.com/gattaca-com/flux?rev=c877c5edfdedcc03d3d37af9977624793e821fe3#c877c5edfdedcc03d3d37af9977624793e821fe3" dependencies = [ "bytesize", "clap", @@ -1692,8 +1692,8 @@ dependencies = [ [[package]] name = "flux-profiler-macros" -version = "0.3.1" -source = "git+https://github.com/gattaca-com/flux?rev=5402c63afc122293f04ea69cd9ae295d7d011c8f#5402c63afc122293f04ea69cd9ae295d7d011c8f" +version = "0.3.2" +source = "git+https://github.com/gattaca-com/flux?rev=c877c5edfdedcc03d3d37af9977624793e821fe3#c877c5edfdedcc03d3d37af9977624793e821fe3" dependencies = [ "quote", "syn 2.0.118", @@ -1701,8 +1701,8 @@ dependencies = [ [[package]] name = "flux-timing" -version = "0.3.1" -source = "git+https://github.com/gattaca-com/flux?rev=5402c63afc122293f04ea69cd9ae295d7d011c8f#5402c63afc122293f04ea69cd9ae295d7d011c8f" +version = "0.3.2" +source = "git+https://github.com/gattaca-com/flux?rev=c877c5edfdedcc03d3d37af9977624793e821fe3#c877c5edfdedcc03d3d37af9977624793e821fe3" dependencies = [ "bitcode", "chrono", @@ -1718,8 +1718,8 @@ dependencies = [ [[package]] name = "flux-utils" -version = "0.3.1" -source = "git+https://github.com/gattaca-com/flux?rev=5402c63afc122293f04ea69cd9ae295d7d011c8f#5402c63afc122293f04ea69cd9ae295d7d011c8f" +version = "0.3.2" +source = "git+https://github.com/gattaca-com/flux?rev=c877c5edfdedcc03d3d37af9977624793e821fe3#c877c5edfdedcc03d3d37af9977624793e821fe3" dependencies = [ "bytes", "core_affinity", @@ -5366,8 +5366,8 @@ dependencies = [ [[package]] name = "spine-derive" -version = "0.3.1" -source = "git+https://github.com/gattaca-com/flux?rev=5402c63afc122293f04ea69cd9ae295d7d011c8f#5402c63afc122293f04ea69cd9ae295d7d011c8f" +version = "0.3.2" +source = "git+https://github.com/gattaca-com/flux?rev=c877c5edfdedcc03d3d37af9977624793e821fe3#c877c5edfdedcc03d3d37af9977624793e821fe3" dependencies = [ "proc-macro2", "quote", @@ -5884,13 +5884,13 @@ checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" [[package]] name = "type-hash" -version = "0.3.1" -source = "git+https://github.com/gattaca-com/flux?rev=5402c63afc122293f04ea69cd9ae295d7d011c8f#5402c63afc122293f04ea69cd9ae295d7d011c8f" +version = "0.3.2" +source = "git+https://github.com/gattaca-com/flux?rev=c877c5edfdedcc03d3d37af9977624793e821fe3#c877c5edfdedcc03d3d37af9977624793e821fe3" [[package]] name = "type-hash-derive" -version = "0.3.1" -source = "git+https://github.com/gattaca-com/flux?rev=5402c63afc122293f04ea69cd9ae295d7d011c8f#5402c63afc122293f04ea69cd9ae295d7d011c8f" +version = "0.3.2" +source = "git+https://github.com/gattaca-com/flux?rev=c877c5edfdedcc03d3d37af9977624793e821fe3#c877c5edfdedcc03d3d37af9977624793e821fe3" dependencies = [ "proc-macro-crate", "proc-macro2", diff --git a/Cargo.toml b/Cargo.toml index 3ea731d2..b95c0818 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -89,9 +89,9 @@ silver_network = {path = "crates/network" } silver_peer = {path = "crates/peer" } silver_storage = { path = "crates/storage" } silver_engine_api = { path = "crates/engine_api" } -flux = { git = "https://github.com/gattaca-com/flux", rev = "5402c63afc122293f04ea69cd9ae295d7d011c8f"} -flux-utils = { git = "https://github.com/gattaca-com/flux", rev = "5402c63afc122293f04ea69cd9ae295d7d011c8f", features = ["bytes"]} -flux-profiler = { git = "https://github.com/gattaca-com/flux", rev = "5402c63afc122293f04ea69cd9ae295d7d011c8f"} +flux = { git = "https://github.com/gattaca-com/flux", rev = "c877c5edfdedcc03d3d37af9977624793e821fe3"} +flux-utils = { git = "https://github.com/gattaca-com/flux", rev = "c877c5edfdedcc03d3d37af9977624793e821fe3", features = ["bytes"]} +flux-profiler = { git = "https://github.com/gattaca-com/flux", rev = "c877c5edfdedcc03d3d37af9977624793e821fe3"} # External backtrace = "0.3" diff --git a/crates/application_boundary/src/lib.rs b/crates/application_boundary/src/lib.rs index 841f67f0..0e87a844 100644 --- a/crates/application_boundary/src/lib.rs +++ b/crates/application_boundary/src/lib.rs @@ -24,6 +24,10 @@ pub struct ApplicationBoundaryTile { } impl Tile for ApplicationBoundaryTile { + fn on_attach(&mut self, adapter: &mut SpineAdapter) { + adapter.subscribe_broadcast::(); + } + fn loop_body(&mut self, adapter: &mut SpineAdapter) { self.engine.intake(adapter); self.readiness.wait(Duration::ZERO); diff --git a/crates/application_boundary/tests/tile.rs b/crates/application_boundary/tests/tile.rs index 3781b48c..9d244634 100644 --- a/crates/application_boundary/tests/tile.rs +++ b/crates/application_boundary/tests/tile.rs @@ -131,6 +131,57 @@ fn no_el() -> EngineConfig { EngineConfig { unsafe_no_el: true, ..EngineConfig::default() } } +#[test] +fn peers_include_connections_published_before_boundary_starts() { + let base = ShmemDir::new().unwrap(); + let mut spine = Box::new(SilverSpine::new_with_base_dir(base.path(), None)); + let tile = boundary_tile(&Bind::parse("127.0.0.1:0"), no_el(), [ + "cs_startup_gossip", + "cs_startup_rpc", + "cs_startup_resp", + ]); + let [Bind::Tcp(addr)] = tile.beacon.local_addrs()[..] else { panic!("expected one tcp bind") }; + let peer_id = Keypair::from_secret(&[2; 32]).unwrap().peer_id(); + + let response = std::thread::scope(|scope| { + let mut scoped = + flux::spine::ScopedSpine { spine: &mut *spine, scope, stop_flag: Default::default() }; + let run = flux::tile::tile_runner( + tile, + &mut scoped, + flux::tile::TileConfig::background(None, None).without_metrics(), + ); + let mut inj = SpineAdapter::connect_tile_with_stop_flag( + &Injector, + scoped.spine, + scoped.stop_flag.clone(), + ); + inj.produce(PeerEvent::P2pNewConnection { + p2p_peer_id: 1, + peer_id_full: peer_id, + ip: IpBytes::V4([127, 0, 0, 1]), + port: 9000, + local_dial: false, + }); + let worker = scope.spawn(run); + let response = (|| -> Result> { + let response = ureq::get(&format!("http://{addr}/eth/v1/node/peers")) + .timeout(Duration::from_secs(10)) + .call()?; + Ok(serde_json::from_reader(response.into_reader())?) + })(); + inj.request_stop_scope(); + worker.join().unwrap(); + response.unwrap() + }); + + let peers = response["data"].as_array().unwrap(); + assert_eq!(peers.len(), 1); + let peer_addr = silver_common::Eth2Addr::PeerId(peer_id).to_string(); + assert_eq!(peers[0]["peer_id"], peer_addr.strip_prefix("/p2p/").unwrap()); + assert_eq!(peers[0]["state"], "connected"); +} + fn http_get(mut stream: impl Read + Write, path: &str) -> String { write!(stream, "GET {path} HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n").unwrap(); stream.flush().unwrap(); diff --git a/crates/bin/src/main.rs b/crates/bin/src/main.rs index 41ae1c46..706bda52 100644 --- a/crates/bin/src/main.rs +++ b/crates/bin/src/main.rs @@ -8,7 +8,7 @@ use std::{ }; use flux::{ - tile::{TileConfig, attach_tile}, + tile::{Tile, TileConfig, attach_tile, tile_runner}, utils::ThreadNiceness, }; use mimalloc::MiMalloc; @@ -437,6 +437,14 @@ fn main() -> Result<(), Box> { // Spine let spine = SilverSpine::new(None); spine.start(None, None, |scoped_spine| { + // Establish peer subscriptions before any producer starts. + let boundary_name = application_boundary_tile.name().as_str().to_owned(); + let boundary_run = tile_runner( + application_boundary_tile, + scoped_spine, + TileConfig::new(5, Some(ThreadNiceness::Highest)), + ); + // TODO core config attach_tile(control_tile, scoped_spine, TileConfig::new(1, Some(ThreadNiceness::Highest))); attach_tile(network_tile, scoped_spine, TileConfig::new(2, Some(ThreadNiceness::Highest))); @@ -446,11 +454,10 @@ fn main() -> Result<(), Box> { TileConfig::new(3, Some(ThreadNiceness::Highest)), ); attach_tile(storage_tile, scoped_spine, TileConfig::new(4, Some(ThreadNiceness::Highest))); - attach_tile( - application_boundary_tile, - scoped_spine, - TileConfig::new(5, Some(ThreadNiceness::Highest)), - ); + std::thread::Builder::new() + .name(boundary_name) + .spawn_scoped(scoped_spine.scope, boundary_run) + .expect("spawn application boundary"); attach_tile( data_columns_tile, scoped_spine, From d937ebad35dd98c1dc0a397472a007c2ad5054a7 Mon Sep 17 00:00:00 2001 From: Bronek Kozicki Date: Fri, 18 Sep 2026 11:17:56 +0100 Subject: [PATCH 2/2] Simplify and document order of attach_tile --- crates/bin/src/main.rs | 13 ++++--------- 1 file changed, 4 insertions(+), 9 deletions(-) diff --git a/crates/bin/src/main.rs b/crates/bin/src/main.rs index 706bda52..6b559637 100644 --- a/crates/bin/src/main.rs +++ b/crates/bin/src/main.rs @@ -8,7 +8,7 @@ use std::{ }; use flux::{ - tile::{Tile, TileConfig, attach_tile, tile_runner}, + tile::{TileConfig, attach_tile}, utils::ThreadNiceness, }; use mimalloc::MiMalloc; @@ -437,15 +437,14 @@ fn main() -> Result<(), Box> { // Spine let spine = SilverSpine::new(None); spine.start(None, None, |scoped_spine| { - // Establish peer subscriptions before any producer starts. - let boundary_name = application_boundary_tile.name().as_str().to_owned(); - let boundary_run = tile_runner( + // Attach application_boundary_tiles first so its `on_attach` can subscribe to + // peer events before their producers start. + attach_tile( application_boundary_tile, scoped_spine, TileConfig::new(5, Some(ThreadNiceness::Highest)), ); - // TODO core config attach_tile(control_tile, scoped_spine, TileConfig::new(1, Some(ThreadNiceness::Highest))); attach_tile(network_tile, scoped_spine, TileConfig::new(2, Some(ThreadNiceness::Highest))); attach_tile( @@ -454,10 +453,6 @@ fn main() -> Result<(), Box> { TileConfig::new(3, Some(ThreadNiceness::Highest)), ); attach_tile(storage_tile, scoped_spine, TileConfig::new(4, Some(ThreadNiceness::Highest))); - std::thread::Builder::new() - .name(boundary_name) - .spawn_scoped(scoped_spine.scope, boundary_run) - .expect("spawn application boundary"); attach_tile( data_columns_tile, scoped_spine,