feat(app): wire priority + infosync into the node

crates/app/Cargo.toml

@@ -22,6 +22,8 @@ pluto-eth2util.workspace = true
 pluto-p2p.workspace = true
 pluto-parsigex.workspace = true
 pluto-peerinfo.workspace = true
+pluto-priority.workspace = true
+pluto-infosync.workspace = true
 tokio.workspace = true
 tokio-util.workspace = true
 vise.workspace = true

crates/app/src/node/behaviour.rs

@@ -54,6 +54,8 @@ pub(crate) struct CoreBehaviour {
     pub consensus: qbft::p2p::Behaviour,
     /// Peer metadata exchange.
     pub peerinfo: peerinfo::Behaviour,
+    /// Priority protocol request/response transport (backs infosync).
+    pub priority: pluto_priority::p2p::Behaviour,
 }
 
 /// Async handles for driving the composed behaviour from the core workflow.
@@ -65,28 +67,56 @@ pub struct CoreHandles {
     /// Shared P2P runtime context (known peers + live connections), used by the
     /// monitoring API's readiness checker to compute quorum connectivity.
     pub p2p_context: P2PContext,
+    /// Priority protocol component; started by the caller.
+    pub priority: Arc<pluto_priority::Component>,
+    /// Expired-duty receiver for `priority`; move-only, consumed once by
+    /// `Component::start`.
+    pub priority_expired_rx: tokio::sync::mpsc::Receiver<pluto_core::types::Duty>,
+    /// Infosync component driving the per-epoch priority exchange.
+    pub infosync: Arc<pluto_infosync::Component>,
+}
+
+/// Inputs to [`wire_p2p`], grouped to keep the aggregating call site readable.
+pub(crate) struct WireP2PParams {
+    pub key: k256::SecretKey,
+    pub p2p_config: pluto_p2p::config::P2PConfig,
+    pub peers: Vec<Peer>,
+    pub consensus: Arc<qbft::Consensus>,
+    pub min_required: i64,
+    pub deadline_calc: Arc<dyn pluto_core::deadline::DeadlineCalculator>,
+    pub feature_set: Arc<pluto_featureset::FeatureSet>,
+    pub duty_gater: DutyGaterFn,
+    pub eth2_cl: EthBeaconNodeApiClient,
+    pub pub_shares_by_key: HashMap<PubKey, HashMap<u64, PublicKey>>,
+    pub lock_hash: Vec<u8>,
+    pub builder_enabled: bool,
+    pub nickname: String,
+    pub cancellation: CancellationToken,
 }
 
 /// Composes the core behaviours and builds the libp2p [`Node`].
 // TODO(#402 part B): QUIC transport (featureset-gated off at v1.7.1) and
 // bandwidth metrics.
-#[allow(
-    clippy::too_many_arguments,
-    reason = "wireP2P aggregates independent inputs; a config struct is deferred to part B when priority inputs are added"
-)]
 pub(crate) async fn wire_p2p(
-    key: k256::SecretKey,
-    p2p_config: pluto_p2p::config::P2PConfig,
-    peers: Vec<Peer>,
-    consensus: Arc<qbft::Consensus>,
-    duty_gater: DutyGaterFn,
-    eth2_cl: EthBeaconNodeApiClient,
-    pub_shares_by_key: HashMap<PubKey, HashMap<u64, PublicKey>>,
-    lock_hash: Vec<u8>,
-    builder_enabled: bool,
-    nickname: String,
-    cancellation: CancellationToken,
+    params: WireP2PParams,
 ) -> Result<(Node<CoreBehaviour>, CoreHandles), AppError> {
+    let WireP2PParams {
+        key,
+        p2p_config,
+        peers,
+        consensus,
+        min_required,
+        deadline_calc,
+        feature_set,
+        duty_gater,
+        eth2_cl,
+        pub_shares_by_key,
+        lock_hash,
+        builder_enabled,
+        nickname,
+        cancellation,
+    } = params;
+
     let peer_ids = peers.iter().map(|peer| peer.id).collect::<Vec<_>>();
     let local_peer_id = peer::peer_id_from_key(key.public_key())?;
 
@@ -123,6 +153,11 @@ pub(crate) async fn wire_p2p(
     );
     let (parsigex_comp, parsigex_handle) = parsigex::Behaviour::new(parsigex_config);
 
+    // Priority rides the same QBFT consensus; clone before it moves into the
+    // QBFT behaviour below.
+    let priority_consensus: Arc<dyn pluto_priority::Consensus> = consensus.clone();
+    let priority_cancellation = cancellation.clone();
+
     // QBFT consensus transport. `Behaviour::new` errors if the local peer id is
     // not present in the configured cluster peer list.
     let (consensus_comp, consensus_handle) = qbft::p2p::Behaviour::new(qbft::p2p::Config {
@@ -146,6 +181,30 @@ pub(crate) async fn wire_p2p(
     .with_peers(peer_ids.clone());
     let peerinfo_comp = peerinfo::Behaviour::new(local_peer_id, peerinfo_config);
 
+    // Priority + infosync: per-epoch negotiation of supported
+    // versions/protocols/proposal types. The 6s exchange timeout (half a slot)
+    // matches Charon; `new_component` fails fast on a peer missing from the
+    // shared `p2p_context`.
+    let (priority_comp, priority_behaviour, priority_expired_rx) = pluto_priority::new_component(
+        peer_ids.clone(),
+        min_required,
+        priority_consensus,
+        std::time::Duration::from_secs(6),
+        key.clone(),
+        deadline_calc,
+        p2p_context.clone(),
+        priority_cancellation,
+    )?;
+    let priority_comp = Arc::new(priority_comp);
+
+    let infosync = Arc::new(pluto_infosync::Component::new(
+        Arc::clone(&priority_comp),
+        pluto_core::version::SUPPORTED.to_vec(),
+        local_protocols(),
+        local_proposal_types(builder_enabled),
+        &feature_set,
+    ));
+
     // Clone the context before it is moved into the node so the readiness
     // checker observes the same shared peer/connection state the swarm updates.
     let p2p_context_for_handle = p2p_context.clone();
@@ -164,6 +223,7 @@ pub(crate) async fn wire_p2p(
                 parsigex: parsigex_comp,
                 consensus: consensus_comp,
                 peerinfo: peerinfo_comp,
+                priority: priority_behaviour,
             })
         },
     )?;
@@ -172,7 +232,72 @@ pub(crate) async fn wire_p2p(
         parsigex: parsigex_handle,
         consensus: consensus_handle,
         p2p_context: p2p_context_for_handle,
+        priority: priority_comp,
+        priority_expired_rx,
+        infosync,
     };
 
     Ok((node, handles))
 }
+
+/// Advertised proposal types in precedence order: builder first when enabled,
+/// full always last as the fallback.
+fn local_proposal_types(builder_enabled: bool) -> Vec<pluto_core::types::ProposalType> {
+    let mut proposal_types = Vec::new();
+    if builder_enabled {
+        proposal_types.push(pluto_core::types::ProposalType::Builder);
+    }
+    proposal_types.push(pluto_core::types::ProposalType::Full);
+    proposal_types
+}
+
+/// Advertised protocols in precedence order: consensus, parsigex, peerinfo,
+/// priority.
+// TODO(#402 part B): reorder by the cluster-preferred / CLI consensus protocol
+// once those inputs exist, matching Go's `PrioritizeProtocolsByName`.
+fn local_protocols() -> Vec<String> {
+    pluto_consensus::protocols::protocols()
+        .iter()
+        .map(|p| p.to_string())
+        .chain(pluto_parsigex::protocols().iter().map(|p| p.to_string()))
+        .chain(pluto_peerinfo::protocols().iter().map(|p| p.to_string()))
+        .chain(pluto_priority::protocols().iter().map(|p| p.to_string()))
+        .collect()
+}
+
+#[cfg(test)]
+mod tests {
+    use pluto_core::types::ProposalType;
+
+    use super::{local_proposal_types, local_protocols};
+
+    #[test]
+    fn proposal_types_put_builder_first_only_when_enabled() {
+        assert_eq!(local_proposal_types(false), vec![ProposalType::Full]);
+        assert_eq!(
+            local_proposal_types(true),
+            vec![ProposalType::Builder, ProposalType::Full],
+        );
+    }
+
+    #[test]
+    fn protocols_are_advertised_in_component_precedence_order() {
+        // This ordering is what keeps pluto's info_sync result byte-identical
+        // to Charon's; priority's absence is the bug this wiring fixes.
+        let got = local_protocols();
+
+        let index_of = |head: String| {
+            got.iter()
+                .position(|p| *p == head)
+                .expect("protocol advertised")
+        };
+        let consensus = index_of(pluto_consensus::protocols::protocols()[0].to_string());
+        let parsigex = index_of(pluto_parsigex::protocols()[0].to_string());
+        let peerinfo = index_of(pluto_peerinfo::protocols()[0].to_string());
+        let priority = index_of(pluto_priority::protocols()[0].to_string());
+
+        assert!(consensus < parsigex);
+        assert!(parsigex < peerinfo);
+        assert!(peerinfo < priority);
+    }
+}

crates/app/src/node/mod.rs

@@ -123,6 +123,10 @@ pub enum AppError {
     #[error("consensus p2p: {0}")]
     ConsensusP2P(#[from] qbft::p2p::Error),
 
+    /// Priority protocol component construction failed.
+    #[error("priority: {0}")]
+    Priority(#[from] pluto_priority::Error),
+
     /// A beacon node API request failed.
     #[error("beacon node api: {0}")]
     BeaconApi(#[from] pluto_eth2api::EthBeaconNodeApiClientError),
@@ -377,8 +381,8 @@ async fn run(config: AppConfig, ct: CancellationToken) -> Result<(), AppError> {
         .map_err(AppError::Gater)?
         .into_fn();
 
-    // Per-component deadline calculator, shared as an `Arc<dyn ...>` so a single
-    // beacon-derived instance backs every component's deadliner.
+    // Shared `Arc<dyn ...>` so one beacon-derived instance backs every
+    // component's deadliner (priority included).
     let deadline_calc: Arc<dyn pluto_core::deadline::DeadlineCalculator> = Arc::new(
         pluto_core::deadline::DutyDeadlineCalculator::from_client(&eth2_cl)
             .await
@@ -460,19 +464,24 @@ async fn run(config: AppConfig, ct: CancellationToken) -> Result<(), AppError> {
     let pub_shares_by_key = build_pub_shares_by_key(&lock)?;
 
     // ---- P2P behaviours (relay + parsigex + qbft + peerinfo) ----
-    let (node, handles) = behaviour::wire_p2p(
-        key.clone(),
-        config.p2p.clone(),
+    let (node, handles) = behaviour::wire_p2p(behaviour::WireP2PParams {
+        key: key.clone(),
+        p2p_config: config.p2p.clone(),
         peers,
-        Arc::clone(&consensus),
-        Arc::clone(&duty_gater),
-        eth2_cl.clone(),
+        consensus: Arc::clone(&consensus),
+        // Priority quorum = cluster signing threshold (Charon's
+        // `int(cluster.GetThreshold())`).
+        min_required: i64::try_from(threshold).unwrap_or(i64::MAX),
+        deadline_calc: Arc::clone(&deadline_calc),
+        feature_set: Arc::clone(&feature_set),
+        duty_gater: Arc::clone(&duty_gater),
+        eth2_cl: eth2_cl.clone(),
         pub_shares_by_key,
-        lock.lock_hash.clone(),
-        config.builder_api,
-        config.nickname.clone(),
-        ct.clone(),
-    )
+        lock_hash: lock.lock_hash.clone(),
+        builder_enabled: config.builder_api,
+        nickname: config.nickname.clone(),
+        cancellation: ct.clone(),
+    })
     .await?;
     // Complete the broadcaster<->behaviour cycle.
     handle_slot
@@ -547,6 +556,7 @@ async fn run(config: AppConfig, ct: CancellationToken) -> Result<(), AppError> {
             fetch_only_comm_idx0,
             seen_pubkeys: Some(seen_pubkeys_observer),
             slot_tick: vmock.clone().map(|v| simnet_slot_tick(v, ct.clone())),
+            infosync: Some(Arc::clone(&handles.infosync)),
         },
         ct.clone(),
     )
@@ -689,6 +699,17 @@ async fn run_lifecycle(
     // Self-spawning actor: consensus expired-duty pruner.
     let _consensus_task = consensus.start(ct.clone());
 
+    // Priority state-cleanup loop; the per-epoch infosync trigger is registered
+    // as a slot subscriber in `wire_core_workflow`.
+    handles
+        .priority
+        .start(handles.priority_expired_rx, ct.clone());
+
+    // TODO(#402 part B): consume the decided infosync result. Charon's
+    // `wirePrioritise` swaps the duty-consensus implementation via
+    // `ConsensusController::set_current_consensus_for_protocol`; deferred because
+    // it is a no-op while QBFTv2 is the only consensus protocol.
+
     let mut tasks: JoinSet<Result<(), AppError>> = JoinSet::new();
 
     // Supervise the scheduler actor alongside the other long-lived tasks so

crates/app/src/node/wire.rs

@@ -153,6 +153,9 @@ pub struct WireInputs {
     /// Optional per-slot subscriber; simnet wires the in-process validator
     /// mock here. `None` in production and tests.
     pub slot_tick: Option<SlotTickFn>,
+    /// Infosync component, triggered on each epoch's last slot to run the
+    /// cluster-wide priority exchange. `None` in tests.
+    pub infosync: Option<Arc<pluto_infosync::Component>>,
 }
 
 /// The wired components and long-lived handles produced by
@@ -281,6 +284,7 @@ pub async fn wire_core_workflow(
         fetch_only_comm_idx0,
         seen_pubkeys,
         slot_tick,
+        infosync,
     } = inputs;
 
     // ---- Derived validator maps ----
@@ -608,6 +612,25 @@ pub async fn wire_core_workflow(
     if let Some(slot_tick) = slot_tick {
         sched_builder.subscribe_slot(move |slot: &Slot| slot_tick(slot), "simnet.vmock");
     }
+    // Per-epoch infosync trigger, fired on each epoch's last slot. A failure is
+    // logged by `subscribe_slot` and does not fail the node.
+    if let Some(infosync) = infosync {
+        let ct = ct.clone();
+        sched_builder.subscribe_slot(
+            move |slot: &Slot| {
+                let infosync = Arc::clone(&infosync);
+                let ct = ct.clone();
+                let slot = slot.clone();
+                async move {
+                    if slot.last_in_epoch() {
+                        infosync.trigger(ct.child_token(), slot.slot).await?;
+                    }
+                    Ok::<(), AppError>(())
+                }
+            },
+            "infosync",
+        );
+    }
     // Slot subscriber: per-epoch validator cache trim + refresh (Charon's
     // `wireCoreWorkflow`).
     {

crates/app/tests/wiring.rs

@@ -269,6 +269,7 @@ fn wire_inputs_with(
         fetch_only_comm_idx0: false,
         seen_pubkeys: None,
         slot_tick: None,
+        infosync: None,
     }
 }