feat(node,config,network-libp2p): per-worker network swarms (#555, PR 2) #1315
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change | |||||||||
|---|---|---|---|---|---|---|---|---|---|---|---|
|
|
@@ -90,8 +90,9 @@ pub(crate) struct EpochManager<P, DB> { | ||||||||||
| tn_datadir: P, | |||||||||||
| /// Primary network handle. | |||||||||||
| primary_network_handle: Option<PrimaryNetworkHandle>, | |||||||||||
| /// Worker network handles, indexed by [`WorkerId`](tn_types::WorkerId). Empty until | |||||||||||
| /// [`spawn_node_networks`](Self::spawn_node_networks) runs. | |||||||||||
| worker_network_handles: Vec<WorkerNetworkHandle>, | |||||||||||
| /// Key config - loaded once for application lifetime. | |||||||||||
| key_config: KeyConfig, | |||||||||||
| /// The epoch manager's [ShutdownNotifier] to shutdown all node processes. | |||||||||||
|
|
@@ -124,9 +125,10 @@ pub(crate) struct EpochManager<P, DB> { | ||||||||||
| /// Application-scoped consensus bus. Survives epoch boundaries and is reset between epochs via | |||||||||||
| /// `reset_for_epoch`; carries `recent_blocks`, node mode, and other cross-component state. | |||||||||||
| consensus_bus: ConsensusBusApp, | |||||||||||
| /// Persistent event streams for the long-running worker networks, one per configured worker | |||||||||||
| /// and indexed by [`WorkerId`](tn_types::WorkerId). Outlive any single epoch so the worker | |||||||||||
| /// swarms do not have to be rebuilt on each transition. | |||||||||||
| worker_event_streams: Vec<QueChannel<NetworkEvent<WorkerRequest, WorkerResponse>>>, | |||||||||||
|
|
|||||||||||
| /// Final consensus header of the epoch that just closed, carried into the next epoch so it can | |||||||||||
| /// be used as the starting point for the new epoch's chain. | |||||||||||
|
|
@@ -610,7 +612,10 @@ where | ||||||||||
| // Don't risk keeping the default CVV active mode... | |||||||||||
| consensus_bus.node_mode().send_replace(NodeMode::Observer); | |||||||||||
| } | |||||||||||
| // one event stream per configured worker, indexed by worker id | |||||||||||
| let worker_event_streams = (0..builder.tn_config.node_info.p2p_info.num_workers()) | |||||||||||
| .map(|_| QueChannel::new()) | |||||||||||
| .collect(); | |||||||||||
| let bootstrap_servers = if let Ok(committee_zero) = | |||||||||||
| Config::load_from_path_or_default::<Committee>( | |||||||||||
| tn_datadir.committee_path(), | |||||||||||
|
|
@@ -639,15 +644,15 @@ where | ||||||||||
| builder, | |||||||||||
| tn_datadir, | |||||||||||
| primary_network_handle: None, | |||||||||||
| worker_network_handles: Vec::new(), | |||||||||||
| key_config, | |||||||||||
| node_shutdown, | |||||||||||
| epoch_boundary: Default::default(), | |||||||||||
| network_initialized: false, | |||||||||||
| reth_db, | |||||||||||
| consensus_db, | |||||||||||
| consensus_bus, | |||||||||||
| worker_event_streams, | |||||||||||
| last_consensus_header: None, | |||||||||||
| last_forwarded_consensus_number: 0, | |||||||||||
| consensus_chain, | |||||||||||
|
|
@@ -994,8 +999,8 @@ where | ||||||||||
| /// Spawn the process-lifetime primary and worker [`ConsensusNetwork`] swarms. | |||||||||||
| /// | |||||||||||
| /// Each swarm runs as a critical task until node shutdown. The resulting network handles are | |||||||||||
| /// stored on the manager for use by every epoch; the worker handles are seeded with the | |||||||||||
| /// starting `epoch` and their task spawners are refreshed on each epoch transition. | |||||||||||
| async fn spawn_node_networks( | |||||||||||
| &mut self, | |||||||||||
| node_task_spawner: TaskSpawner, | |||||||||||
|
|
@@ -1046,59 +1051,75 @@ where | ||||||||||
| self.primary_network_handle = | |||||||||||
| Some(PrimaryNetworkHandle::new(primary_network_handle, network_config.chain_id())); | |||||||||||
|
|
|||||||||||
| // | |||||||||||
| //=== WORKERS | |||||||||||
| // | |||||||||||
|
|
|||||||||||
| // create one long-running swarm per configured worker | |||||||||||
| // the per-epoch code still drives worker 0 only (#557 loops over worker components) | |||||||||||
|
Contributor
There was a problem hiding this comment. Workers 1 and up get full swarms that never listen or dial, which costs a failed startup publish and a permanent heartbeat log line per extra swarm. This comment is accurate and the staging is clearly deliberate — I traced it and it holds. Two things I checked that are fine, recorded so nobody re-derives them. The real cost per extra swarm is one failing startup The sequencing question is the one that actually needs an answer, and it's yours to make:
One note on a claim you may hear elsewhere: it is not the case that the kad store is keyed by
Contributor
Author
There was a problem hiding this comment. |
|||||||||||
| let workers = self.builder.tn_config.node_info.p2p_info.workers.clone(); | |||||||||||
|
Contributor
There was a problem hiding this comment. The number of worker swarms comes straight from local node config, with no cross-check against the committee's worker count or the multi-workers fork gate. This line decides how many swarms get spawned, and nothing between here and The gap is wider than "validation happens late". A multi-worker local config is reachable without any code change. So: duplicate the entry under The chain count is already available before this point: let configured = self.builder.tn_config.node_info.p2p_info.num_workers();
eyre::ensure!(
configured == 1 || tn_types::forks::multi_workers_fork_active(epoch),
"node config `node_info.p2p_info.workers` lists {configured} workers but the \
multi-workers fork is not active at epoch {epoch}: configure exactly one worker"
);
let on_chain = gas_accumulator.num_workers();
eyre::ensure!(
configured == on_chain,
"node config `node_info.p2p_info.workers` lists {configured} workers but the \
chain-derived count for epoch {epoch} is {on_chain}: every validator must run the \
worker count the committee carries (see `Committee::number_of_workers`)"
);Worth also adding a third One deployment-visible consequence: a node whose local config disagrees with the chain would now fail to start rather than start degraded. That's the intent, but it's worth a release note. |
|||||||||||
| self.worker_network_handles = workers | |||||||||||
|
Contributor
There was a problem hiding this comment. No test covers the multi-swarm spawn path this PR introduces. This assignment is the new behaviour — one swarm per configured worker — and nothing exercises it. The gap that matters most is downstream of that: no test anywhere in the repo constructs
Contributor
Author
There was a problem hiding this comment. Added coverage in 1748aba for the production worker-preparation helper: per-worker IDs/keys/addresses/streams, invalid RPC descriptors, stream/count mismatches, and WorkerId bounds. The storage tests now exercise Worker(0)/Worker(1) on one DB, cross-worker signature rejection, and sibling preservation during network construction. This covers preparation and construction; a full live two-swarm harness remains outside this PR. |
|||||||||||
| .into_iter() | |||||||||||
| .zip(self.worker_event_streams.iter()) | |||||||||||
| .enumerate() | |||||||||||
| .map(|(idx, (worker_p2p, worker_event_stream))| { | |||||||||||
| let worker_id = u16::try_from(idx) | |||||||||||
| .map_err(|_| eyre!("worker index {idx} exceeds the WorkerId range"))?; | |||||||||||
|
|
|||||||||||
| // pass through the worker's RPC descriptor so peers can discover this | |||||||||||
| // validator's JSON-RPC endpoint via kademlia. validators that did not | |||||||||||
| // configure RPC leave the descriptor `None`. fail fast on a misconfigured | |||||||||||
| // endpoint rather than advertising something peers will reject. | |||||||||||
| let worker_rpc = worker_p2p.rpc; | |||||||||||
| if let Some(rpc) = &worker_rpc { | |||||||||||
| rpc.validate().wrap_err_with(|| { | |||||||||||
| format!( | |||||||||||
| "invalid `node_info.p2p_info.workers[{worker_id}].rpc` endpoint in \ | |||||||||||
| node config" | |||||||||||
| ) | |||||||||||
| })?; | |||||||||||
| } | |||||||||||
|
|
|||||||||||
| // create long-running network task for this worker | |||||||||||
| let worker_network = ConsensusNetwork::new_for_worker( | |||||||||||
| worker_id, | |||||||||||
|
Contributor
There was a problem hiding this comment. Kad record accounting is per-store but the underlying table is shared, so one worker's eviction desynchronizes another worker's counter and can wedge it permanently. Each
The interesting failure isn't the cap bypass, it's the wedge. Worker 1's One correction to flag, since it changes the scope: The ownership filter from my
Contributor
Author
There was a problem hiding this comment. Addressed in 1748aba. Startup counts and eviction now operate only on the store's owned rows, with consistent accounting for corrupt-row repair/removal. Added a regression that fills worker 0 to capacity, runs eviction on worker 1, and verifies worker 0 can still insert records and providers. |
|||||||||||
| network_config, | |||||||||||
| worker_event_stream.clone(), | |||||||||||
| self.key_config.clone(), | |||||||||||
| self.consensus_db.clone(), | |||||||||||
|
Contributor
There was a problem hiding this comment. All worker swarms share a single kad record table, so sibling swarms overwrite and purge each other's node records. This is the blocker. The code I'm describing is
Every swarm then keys its own record on the same bytes. What makes the collision destructive rather than merely redundant is that the values are not interchangeable. Two consequences, both worse than a lost row: Honest peers ban this node. A remote worker-0 peer that receives a It recurs, and it self-purges on every restart. One imprecision worth noting so nobody re-derives it: Fix, part 1 — namespace the row keyfn key_to_hash(&self, key: &RecordKey) -> BlockHash {
// Namespace the row by (role, worker id). `NetworkType::Worker(_)` selects one shared pair
// of tables for EVERY worker, and every ConsensusNetwork keys its own record on the same
// primary BLS key (consensus.rs:618). Without this, two worker swarms land on one row:
// they publish NodeRecords signed for different `RecordDomain`s, so each overwrites the
// other's value and each purges the other's rows in the verify-and-remove loop in
// `ConsensusNetwork::new`. Discriminants match `RecordDomain::role_parts`.
let (role, worker_id): (u8, tn_types::WorkerId) = match self.kad_type {
NetworkType::Primary => (0, 0),
NetworkType::Worker(id) => (1, id),
};
let mut h = DefaultHashFunction::new();
h.update(&[role]);
h.update(&worker_id.to_le_bytes());
h.update(encode(key).as_ref());
BlockHash::from_slice(h.finalize().as_bytes())
}Fix, part 2 — ownership filter on whole-table scansNamespacing the key alone doesn't stop a scan from touching a sibling's rows. fn owns(&self, key: &RecordKey, hash: &BlockHash) -> bool { self.key_to_hash(key) == *hash }Apply it in Migration — your call
I'd go with A. Kad rows are TTL'd discovery data re-learned on first connect via TestsAll three fail today, and no test anywhere currently constructs
Contributor
Author
There was a problem hiding this comment. Addressed in 1748aba: Kademlia keys now include the role and worker ID, and scans, eviction, cleanup, and initial counters only consider owned rows. Used option A for migration with fresh |
|||||||||||
| node_task_spawner.clone(), | |||||||||||
| worker_p2p.network_address, | |||||||||||
| worker_rpc, | |||||||||||
| )?; | |||||||||||
| let worker_network_handle = worker_network.network_handle(); | |||||||||||
| let node_shutdown = self.node_shutdown.subscribe(); | |||||||||||
|
|
|||||||||||
| // spawn long-running worker network task | |||||||||||
| node_task_spawner.spawn_critical_task( | |||||||||||
| format!("Worker Network {worker_id}"), | |||||||||||
|
Contributor
There was a problem hiding this comment. The network metrics label collapses every Worker(id) to the string "worker", so per-worker swarms silently overwrite each other's gauges in the process-global registry. The task name here already carries the worker id, which is the right instinct — the metrics don't. The registry is process-global: So every worker swarm writes the same series. Gauges that collide (last writer wins): The rest of the codebase already does this correctly — Fix is to change Collateral worth calling out in the PR description: dashboards and alerts querying
Contributor
Author
There was a problem hiding this comment. Implemented in 1748aba: |
|||||||||||
| async move { | |||||||||||
| tokio::select!( | |||||||||||
| _ = &node_shutdown => { | |||||||||||
| Ok(()) | |||||||||||
| } | |||||||||||
| res = worker_network.run() => { | |||||||||||
| warn!(target: "epoch-manager", ?res, "worker network stopped"); | |||||||||||
| Ok(res?) | |||||||||||
| } | |||||||||||
| ) | |||||||||||
| }, | |||||||||||
| ); | |||||||||||
|
|
|||||||||||
| // set temporary task spawner - this is updated with each epoch | |||||||||||
| Ok(WorkerNetworkHandle::new( | |||||||||||
| worker_network_handle, | |||||||||||
| node_task_spawner.clone(), | |||||||||||
| worker_id, | |||||||||||
| epoch, | |||||||||||
| network_config.chain_id(), | |||||||||||
| )) | |||||||||||
| }) | |||||||||||
| .collect::<eyre::Result<Vec<_>>>()?; | |||||||||||
|
|
|||||||||||
| Ok(()) | |||||||||||
| } | |||||||||||
|
|
|||||||||||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
The derived Debug on KeyConfigInner renders the worker network seed, and this PR dropped the positive assertion that kept the worker side of the leak test honest.
KeyConfigInneris#[derive(Debug)](keys.rs:211) and now holdsworker_network_seed: Stringon this line.KeyConfigis#[derive(Debug, Clone)](:234) overArc<KeyConfigInner>, andArc'sDebugforwards to the inner value — soformat!("{:?}", key_config)does render the seed.This is informational rather than urgent, for two reasons. The seed is a domain-separation label, not key material: it's operator-configurable, read from
node_keys_path().join(WORKER_NETWORK_SEED_FILE)atkeys.rs:356-358. And no production log path debug-prints aKeyConfig— I grepped every tracing macro plusformat!,dbg!andprintln!and found zero sites.Still worth closing, since the type is one
?key_configaway from leaking, and the swap to a stored seed removed the only thing that would have caught a regression. A manualDebugforKeyConfigInnerrendering the seed as"<redacted>"handles the first part.For the second: the diff replaces the worker positive anchor in
debug_does_not_leak_key_materialwith a comment explaining that the worker side has no anchor now. That's true as written, but it leaves the worker half of the test with no assertion at all. A negative assertion would restore coverage and would fail if the seed ever starts rendering:That pins the redaction directly, rather than relying on the absence of a field that could come back.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Added the manual
Debugimplementation in 1748aba, rendering the stored seed as[REDACTED]. The leak test now explicitly rejects the seed and checks positive anchors for redaction and public fields.