From 0bf031dd3224ee2cc75e0a944767821b5293c7bc Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Mon, 13 Jul 2026 12:05:17 +0000 Subject: [PATCH] node: drive net-new inbound establish through the per-peer machine Cut the net-new inbound handshake (a fresh msg1 that promotes to a new peer, plus the at-capacity reject) over from the inline handle_msg1 logic to the per-peer control machine. handle_msg1 still classifies via establish_inbound and still owns the Noise wire step, the late ACL check, and the promote_connection registry surgery; for the net-new path it now builds the machine, steps it, and executes the returned actions. Authorization is interposed between two machine steps so the session index is allocated only after the ACL check passes: a rejected or unauthorized msg1 consumes no index, matching the prior order exactly. The msg2 wire bytes, the index-allocation sequence, and the reject metrics are all byte-neutral. Restart, resend, rekey-respond, and the other reject arms stay inline unchanged; they move to the machine once outbound establish is cut over and every promoted peer has a machine. Also fills in the executor's send-failure and promote-failure cleanup so a mid-establish error tears the leg down and frees its index exactly as before. --- src/node/dataplane/mod.rs | 2 + src/node/dataplane/peer_actions.rs | 43 +++++- src/node/handlers/handshake.rs | 190 ++++++++++++++++++++++- src/peer/machine.rs | 234 +++++++++++++++++++++++------ 4 files changed, 412 insertions(+), 57 deletions(-) diff --git a/src/node/dataplane/mod.rs b/src/node/dataplane/mod.rs index e0643a1..bcf5d61 100644 --- a/src/node/dataplane/mod.rs +++ b/src/node/dataplane/mod.rs @@ -14,3 +14,5 @@ mod encrypted; mod forwarding; mod peer_actions; mod rx_loop; + +pub(in crate::node) use peer_actions::PeerActionCtx; diff --git a/src/node/dataplane/peer_actions.rs b/src/node/dataplane/peer_actions.rs index 88f00c3..40d1891 100644 --- a/src/node/dataplane/peer_actions.rs +++ b/src/node/dataplane/peer_actions.rs @@ -20,6 +20,7 @@ //! `RegisterDecryptSession` is a deliberate no-op — see its arm for the C3-2 note. use crate::node::Node; +use crate::node::reject::{HandshakeReject, RejectReason}; use crate::peer::machine::{PeerAction, PeerEvent}; use crate::proto::fmp::wire::build_msg2; use crate::transport::{LinkId, TransportAddr, TransportId}; @@ -96,7 +97,6 @@ impl Node { ambient: &PeerActionCtx, actions: Vec, ) { - let _ = link; let mut queue: VecDeque = actions.into(); while let Some(action) = queue.pop_front() { match action { @@ -115,8 +115,30 @@ impl Node { (ambient.our_index, ambient.their_index) { let frame = build_msg2(sender_idx, receiver_idx, &bytes); - if let Some(transport) = self.transports.get(&ambient.transport_id) { - let _ = transport.send(&ambient.remote_addr, &frame).await; + // GAP-5: surface the send Result. A missing transport skips + // the send and continues (mirrors `handle_msg1`'s + // `if let Some(transport)` guard); a send *error* runs the + // pre-refactor msg2-send-failure cleanup (`handle_msg1` + // L494-503) and ABORTS the remaining queue so the queued + // `PromoteToActive` never runs. + let send_err = match self.transports.get(&ambient.transport_id) { + Some(transport) => { + transport.send(&ambient.remote_addr, &frame).await.is_err() + } + None => false, + }; + if send_err { + self.connections.remove(&link); + self.links.remove(&link); + self.addr_to_link + .remove(&(ambient.transport_id, ambient.remote_addr.clone())); + if let Some(idx) = ambient.our_index { + let _ = self.index_allocator.free(idx); + } + self.peer_machines.remove(&link); + self.stats_mut() + .record_reject(RejectReason::Handshake(HandshakeReject::BadState)); + return; } } } @@ -151,8 +173,19 @@ impl Node { queue.extend(follow); } Err(_e) => { - // C3-2 realizes the promotion-failure cleanup tail - // (`handle_msg1:587` / `handle_msg2:1005`). + // GAP-4: promotion failed. `promote_connection` already + // removed `connections[link]`; mirror the pre-refactor + // cleanup (`handle_msg1` L587-591): drop the link + + // reverse map, free our index, discard the machine, and + // record the reject. The queue is drained (PromoteToActive + // is the last establish action), so no explicit abort. + self.remove_link(&promote_link); + if let Some(idx) = ambient.our_index { + let _ = self.index_allocator.free(idx); + } + self.peer_machines.remove(&promote_link); + self.stats_mut() + .record_reject(RejectReason::Handshake(HandshakeReject::BadState)); } } } diff --git a/src/node/handlers/handshake.rs b/src/node/handlers/handshake.rs index 8b6875f..9aff408 100644 --- a/src/node/handlers/handshake.rs +++ b/src/node/handlers/handshake.rs @@ -3,8 +3,10 @@ use crate::NodeAddr; use crate::PeerIdentity; use crate::node::acl::PeerAclContext; +use crate::node::dataplane::PeerActionCtx; use crate::node::reject::{HandshakeReject, RejectReason}; use crate::node::{Node, NodeError}; +use crate::peer::machine::{FailReason, HandshakePhase, PeerEvent, PeerMachine, PeerState}; use crate::peer::{ActivePeer, PeerConnection}; use crate::proto::fmp::wire::{Msg1Header, Msg2Header, build_msg2}; use crate::proto::fmp::{ @@ -272,13 +274,47 @@ impl Node { // shared authorize → allocate → send-msg2 → promote tail; the other // variants complete the rate-limiter and return here. match self.fmp.establish_inbound(&est, &wire) { - InboundDecision::Reject { reason } => { + InboundDecision::Reject { + reason: InboundReject::AtMaxPeers, + } => { + // C3-2a net-new arm: drive the reject through the machine to + // prove realization A — a net-new msg1 at the max-peers cap + // reaches `Failed{Rejected}` with the index allocator untouched + // (no allocate before the reject). The transient machine is + // discarded (never inserted into `peer_machines`); `conn`/ + // `link_id` were never inserted into the registry either. + debug!( + peer = %self.peer_display_name(&peer_node_addr), + max = self.max_peers(), + "Silent-dropping Msg1 at max_peers cap (early gate; no Msg2 sent)" + ); + let mut machine = PeerMachine::new_inbound(link_id, packet.timestamp_ms); + let _ = machine.step( + PeerEvent::InboundMsg1 { + link: link_id, + wire, + est, + }, + packet.timestamp_ms, + &mut self.index_allocator, + ); + debug_assert!(matches!( + machine.state(), + PeerState::Failed { + reason: FailReason::Rejected + } + )); + self.msg1_rate_limiter.complete_handshake(); + self.stats_mut() + .record_reject(RejectReason::Handshake(HandshakeReject::BadState)); + return; + } + InboundDecision::Reject { + reason: reason @ (InboundReject::PendingSession | InboundReject::DualRekeyWon), + } => { + // Existing-peer rekey rejects — still inline (C4). Byte-unchanged + // from the pre-refactor shared reject tail. match reason { - InboundReject::AtMaxPeers => debug!( - peer = %self.peer_display_name(&peer_node_addr), - max = self.max_peers(), - "Silent-dropping Msg1 at max_peers cap (early gate; no Msg2 sent)" - ), InboundReject::PendingSession => debug!( peer = %self.peer_display_name(&peer_node_addr), "Rekey msg1 received but already have pending session, dropping" @@ -287,6 +323,7 @@ impl Node { peer = %self.peer_display_name(&peer_node_addr), "Dual rekey initiation: we win (smaller addr), dropping their msg1" ), + InboundReject::AtMaxPeers => unreachable!(), } // `conn`/`link_id` were never inserted into the registry, so the // local drop suffices — no cleanup needed. @@ -419,7 +456,146 @@ impl Node { .unwrap_or(0); self.note_link_dead(peer, now_ms); } - InboundDecision::Promote => {} + InboundDecision::Promote => { + // === C3-2a: net-new inbound establish, driven by the machine. === + // Realization A (two-phase authorize): Phase 1 classifies with no + // allocation; the shell interposes the late-ACL gate here; Phase 2 + // allocates the single index and emits [SendHandshake, + // PromoteToActive]. A rejected/unauthorized msg1 therefore + // allocates NO index — matching the pre-refactor + // authorize-before-allocate ordering exactly. + + // Keep the shell's own copies of the msg2 framing inputs before + // the machine event consumes `wire` (WireOutcome is not Clone). + let msg2_payload = wire.msg2_payload.clone(); + let their_index = wire.their_index; + + let mut machine = PeerMachine::new_inbound(link_id, packet.timestamp_ms); + + // Phase 1: classify (no allocation, no actions for a net-new leg). + let phase1 = machine.step( + PeerEvent::InboundMsg1 { + link: link_id, + wire, + est, + }, + packet.timestamp_ms, + &mut self.index_allocator, + ); + debug_assert!(phase1.is_empty()); + debug_assert!(matches!( + machine.state(), + PeerState::Handshaking { + phase: HandshakePhase::ReceivedMsg1, + .. + } + )); + + // Shell interposition: late-ACL authorize BEFORE any allocation. + if self + .authorize_peer( + &peer_identity, + PeerAclContext::InboundHandshake, + packet.transport_id, + &packet.remote_addr, + ) + .is_err() + { + let _ = machine.step( + PeerEvent::Rejected, + packet.timestamp_ms, + &mut self.index_allocator, + ); + self.msg1_rate_limiter.complete_handshake(); + self.stats_mut() + .record_reject(RejectReason::Handshake(HandshakeReject::BadState)); + return; + } + + // Phase 2: allocate our index + emit [SendHandshake, PromoteToActive]. + let promote_actions = machine.step( + PeerEvent::Authorized, + packet.timestamp_ms, + &mut self.index_allocator, + ); + let our_index = match machine.our_index() { + Some(idx) => idx, + None => { + // Allocation exhausted in Phase 2 (mirrors the pre-refactor + // allocate-failure path): no msg2, no promote. + self.msg1_rate_limiter.complete_handshake(); + self.stats_mut() + .record_reject(RejectReason::Handshake(HandshakeReject::BadState)); + return; + } + }; + + // Shell registry surgery (Option A1), in the pre-refactor order: + // set indices on the shell connection, insert link / reverse map / + // connection, then build + store the framed msg2. + conn.set_our_index(our_index); + conn.set_their_index(their_index); + let link = Link::connectionless( + link_id, + packet.transport_id, + packet.remote_addr.clone(), + LinkDirection::Inbound, + Duration::from_millis(self.config().node.base_rtt_ms), + ); + self.links.insert(link_id, link); + self.addr_to_link.insert(addr_key, link_id); + self.connections.insert(link_id, conn); + let wire_msg2 = build_msg2(our_index, their_index, &msg2_payload); + if let Some(conn) = self.connections.get_mut(&link_id) { + conn.set_handshake_msg2(wire_msg2.clone()); + } + + // Register the machine (Promote tail only — discarded on every + // reject/resend/rekey arm per the insertion discipline). + self.peer_machines.insert(link_id, machine); + + // Execute [SendHandshake, PromoteToActive]. The executor frames + + // sends msg2 (bytes identical to `wire_msg2`), promotes via + // `promote_connection`, feeds PromotionResolved back, and runs the + // inert RegisterDecryptSession (R2 — register stays in + // `promote_connection`). Its send-failure / promote-failure arms + // run the pre-refactor cleanup and remove the machine, leaving it + // absent (not Established). + let ambient = PeerActionCtx { + verified_identity: peer_identity, + transport_id: packet.transport_id, + remote_addr: packet.remote_addr.clone(), + our_index: Some(our_index), + their_index: Some(their_index), + now_ms: packet.timestamp_ms, + }; + self.execute_peer_actions(link_id, &ambient, promote_actions) + .await; + + // Post-`Promoted` shell tail (byte-identical to the pre-refactor + // Promoted arm), reached only when promotion succeeded (the machine + // is now Established); a send/promote failure removed the machine + // and already cleaned up. + if matches!( + self.peer_machines.get(&link_id).map(|m| m.state()), + Some(PeerState::Established { .. }) + ) { + // Store msg2 on peer for resend on duplicate msg1 + if let Some(peer) = self.peers.get_mut(&peer_node_addr) { + peer.set_handshake_msg2(wire_msg2.clone()); + } + // Send initial tree announce to new peer + if let Err(e) = self.send_tree_announce_to_peer(&peer_node_addr).await { + debug!(peer = %self.peer_display_name(&peer_node_addr), error = %e, "Failed to send initial TreeAnnounce"); + } + // Schedule filter announce (sent on next tick via debounce) + self.bloom_state.mark_update_needed(peer_node_addr); + self.reset_lookup_backoff(); + } + + self.msg1_rate_limiter.complete_handshake(); + return; + } } if self diff --git a/src/peer/machine.rs b/src/peer/machine.rs index 179b4a4..2b5e885 100644 --- a/src/peer/machine.rs +++ b/src/peer/machine.rs @@ -305,6 +305,10 @@ pub(crate) struct PeerMachine { conn: ConnectionState, /// Remote startup epoch (establish-path-only; NOT in send-state). remote_epoch: Option<[u8; 8]>, + /// Inbound two-phase authorize (realization A): the opaque Noise msg2 + /// payload stashed in Phase 1 (`InboundMsg1`) and emitted in Phase 2 + /// (`on_authorized`), so a rejected/unauthorized msg1 allocates no index. + pending_msg2_payload: Option>, // --- rekey negotiation sub-state (control tier; NOT the pending send slot) --- rekey_in_progress: bool, @@ -343,6 +347,7 @@ impl PeerMachine { identity: Some(identity), conn: ConnectionState::outbound(link, identity, now), remote_epoch: None, + pending_msg2_payload: None, rekey_in_progress: false, rekey_our_index: None, rekey_msg1: None, @@ -368,6 +373,7 @@ impl PeerMachine { identity: None, conn: ConnectionState::inbound(link, now), remote_epoch: None, + pending_msg2_payload: None, rekey_in_progress: false, rekey_our_index: None, rekey_msg1: None, @@ -387,6 +393,14 @@ impl PeerMachine { self.state } + /// The index we allocated for this peer's inbound session, once Phase 2 + /// (`on_authorized`) has run. `None` before allocation (and after a + /// rejected/unauthorized msg1). The inbound cutover reads this to perform + /// the shell registry surgery with the machine-owned index (Option A1). + pub(crate) fn our_index(&self) -> Option { + self.our_index + } + /// The crystallized node address, if identity is known. fn addr(&self) -> Option { self.identity.map(|id| *id.node_addr()) @@ -421,7 +435,7 @@ impl PeerMachine { PeerEvent::Msg2 { their_index, out } => { self.on_msg2(their_index, out, now, index_allocator) } - PeerEvent::Authorized => Vec::new(), + PeerEvent::Authorized => self.on_authorized(now, index_allocator), PeerEvent::Rejected => self.fail(FailReason::AclRejected), PeerEvent::PromotionResolved { result } => self.on_promotion_resolved(result, now), PeerEvent::RekeyMsg1 { wire, est } => { @@ -608,40 +622,67 @@ impl PeerMachine { actions.push(PeerAction::UnregisterDecryptSession { index: idx }); } actions.push(PeerAction::ReportLost { peer }); - actions.extend(self.inbound_promote(link, &wire, now, alloc)); + actions.extend(self.inbound_classify(link, &wire)); actions } - InboundDecision::Promote => self.inbound_promote(link, &wire, now, alloc), + InboundDecision::Promote => self.inbound_classify(link, &wire), } } - /// The inbound Promote tail: allocate our index, record indices/epoch/msg2, - /// emit msg2 + drive promotion. `RegisterDecryptSession` follows on the - /// `PromotionResolved{Promoted}` feedback (§3.2 "then on PromotionResult"). - fn inbound_promote( - &mut self, - link: LinkId, - wire: &WireOutcome, - _now: u64, - alloc: &mut IndexAllocator, - ) -> Vec { + /// Inbound **Phase 1** (realization A): classify the fresh leg *without* + /// allocating an index. Records identity/epoch/their-index and stashes the + /// opaque msg2 payload, parking at `Handshaking{ReceivedMsg1}` — the + /// "awaiting Authorized" marker. The index allocation and the msg2/promote + /// emission happen in Phase 2 ([`Self::on_authorized`]) only after the + /// shell's late-ACL gate passes, so a rejected/unauthorized msg1 allocates + /// nothing (preserving the pre-refactor global index-allocation sequence). + fn inbound_classify(&mut self, link: LinkId, wire: &WireOutcome) -> Vec { self.identity = Some(wire.peer_identity); self.remote_epoch = wire.remote_epoch; self.conn.set_their_index(wire.their_index); - let our_index = alloc.allocate().ok(); - if let Some(idx) = our_index { - self.conn.set_our_index(idx); - self.our_index = Some(idx); - } - self.conn.set_handshake_msg2(wire.msg2_payload.clone()); + self.pending_msg2_payload = Some(wire.msg2_payload.clone()); self.state = PeerState::Handshaking { link, phase: HandshakePhase::ReceivedMsg1, }; + Vec::new() + } + + /// Inbound **Phase 2** (realization A): the late-ACL gate passed shell-side. + /// Allocate our index NOW — the single inbound allocation point — record it + /// on `conn`, and emit the msg2 send + promotion. `RegisterDecryptSession` + /// follows on the `PromotionResolved{Promoted}` feedback (§3.2). Guarded to + /// the inbound `ReceivedMsg1` phase so the benign outbound `Authorized` + /// confirmation stays a no-op (state `Handshaking{SentMsg1}` and every other + /// state fall through to `Vec::new()`). + fn on_authorized(&mut self, _now: u64, alloc: &mut IndexAllocator) -> Vec { + if !matches!( + self.state, + PeerState::Handshaking { + phase: HandshakePhase::ReceivedMsg1, + .. + } + ) { + return Vec::new(); + } + let our_index = match alloc.allocate() { + Ok(idx) => idx, + Err(_) => { + // Allocation exhausted: no index, no msg2, no promote. The shell + // records the reject + completes the rate-limiter bracket + // (mirrors the pre-refactor `handle_msg1` allocate-failure path). + self.state = PeerState::Failed { + reason: FailReason::Rejected, + }; + return Vec::new(); + } + }; + self.conn.set_our_index(our_index); + self.our_index = Some(our_index); + let bytes = self.pending_msg2_payload.take().unwrap_or_default(); + let link = self.link; vec![ - PeerAction::SendHandshake { - bytes: wire.msg2_payload.clone(), - }, + PeerAction::SendHandshake { bytes }, PeerAction::PromoteToActive { link }, ] } @@ -1352,22 +1393,37 @@ mod tests { &mut alloc, ); - // Restart tail: invalidate, unregister old, report lost, then Promote. - assert_eq!(actions[0], PeerAction::InvalidateSendState); + // Phase 1: restart tail only (invalidate, unregister old, report lost), + // then park at ReceivedMsg1 — no index allocated yet (realization A). assert_eq!( - actions[1], - PeerAction::UnregisterDecryptSession { - index: SessionIndex::new(0xDEAD) - } + actions, + vec![ + PeerAction::InvalidateSendState, + PeerAction::UnregisterDecryptSession { + index: SessionIndex::new(0xDEAD) + }, + PeerAction::ReportLost { peer: peer_addr }, + ] ); - assert_eq!(actions[2], PeerAction::ReportLost { peer: peer_addr }); + assert!(matches!( + m.state(), + PeerState::Handshaking { + phase: HandshakePhase::ReceivedMsg1, + .. + } + )); + assert_eq!(m.our_index(), None); + assert_eq!(alloc.count(), 0); + + // Phase 2: late-ACL gate passed -> allocate + Promote tail. + let promote = m.step(PeerEvent::Authorized, 1_000, &mut alloc); assert!( - actions + promote .iter() .any(|a| matches!(a, PeerAction::SendHandshake { .. })) ); assert!( - actions + promote .iter() .any(|a| matches!(a, PeerAction::PromoteToActive { .. })) ); @@ -1400,7 +1456,7 @@ mod tests { let our = *peer_identity().node_addr(); let est_w = est_new_peer(our); let wire_w = wire_outcome(peer, Some([3u8; 8]), 0x77); - let wa = winner.step( + let wp1 = winner.step( PeerEvent::InboundMsg1 { link: LinkId::new(1), wire: wire_w, @@ -1409,6 +1465,8 @@ mod tests { 100, &mut alloc, ); + assert!(wp1.is_empty()); // Phase 1 classifies without emitting. + let wa = winner.step(PeerEvent::Authorized, 100, &mut alloc); assert!( wa.iter().any( |a| matches!(a, PeerAction::PromoteToActive { link } if *link == LinkId::new(1)) @@ -1432,7 +1490,7 @@ mod tests { let mut loser = PeerMachine::new_inbound(LinkId::new(2), 0); let est_l = est_new_peer(our); let wire_l = wire_outcome(peer, Some([3u8; 8]), 0x88); - let la = loser.step( + let lp1 = loser.step( PeerEvent::InboundMsg1 { link: LinkId::new(2), wire: wire_l, @@ -1441,6 +1499,8 @@ mod tests { 100, &mut alloc, ); + assert!(lp1.is_empty()); // Phase 1 classifies without emitting. + let la = loser.step(PeerEvent::Authorized, 100, &mut alloc); assert!( la.iter() .any(|a| matches!(a, PeerAction::PromoteToActive { .. })) @@ -1490,7 +1550,9 @@ mod tests { let est = est_new_peer(our); let wire = wire_outcome(peer, Some([4u8; 8]), 0x77); - let mut actions = m.step( + // Phase 1 (InboundMsg1): classify only — no actions, no allocation, + // parked at ReceivedMsg1 (realization A). + let phase1 = m.step( PeerEvent::InboundMsg1 { link: LinkId::new(1), wire, @@ -1499,29 +1561,111 @@ mod tests { 200, &mut alloc, ); - actions.extend(m.step( + assert!(phase1.is_empty()); + assert_eq!( + m.state(), + PeerState::Handshaking { + link: LinkId::new(1), + phase: HandshakePhase::ReceivedMsg1 + } + ); + assert_eq!(m.our_index(), None); + assert_eq!(alloc.count(), 0); // allocator untouched pre-authorize + + // Phase 2 (Authorized): allocate + [SendHandshake, PromoteToActive]. + let phase2 = m.step(PeerEvent::Authorized, 200, &mut alloc); + assert!(matches!(phase2[0], PeerAction::SendHandshake { .. })); + assert_eq!( + phase2[1], + PeerAction::PromoteToActive { + link: LinkId::new(1) + } + ); + assert!(m.our_index().is_some()); + assert_eq!(alloc.count(), 1); // exactly one index allocated + + // Phase 3 (PromotionResolved{Promoted}): register + Established. + let phase3 = m.step( PeerEvent::PromotionResolved { result: PromotionResult::Promoted(peer_addr), }, 200, &mut alloc, - )); - - // Combined promote sequence (§3.2 "then on PromotionResult ..."). - assert!(matches!(actions[0], PeerAction::SendHandshake { .. })); - assert_eq!( - actions[1], - PeerAction::PromoteToActive { - link: LinkId::new(1) - } ); assert!(matches!( - actions[2], + phase3[0], PeerAction::RegisterDecryptSession { .. } )); assert_eq!(m.state(), PeerState::Established { addr: peer_addr }); } + // ---- Test 6b: inbound late-ACL rejected -> no allocation -------------- + #[test] + fn inbound_authorize_rejected_no_alloc() { + let mut alloc = IndexAllocator::new(); + let peer = peer_identity(); + let mut m = PeerMachine::new_inbound(LinkId::new(1), 0); + let our = *peer_identity().node_addr(); + let est = est_new_peer(our); + let wire = wire_outcome(peer, Some([4u8; 8]), 0x77); + + // Phase 1 classifies (no alloc). + let phase1 = m.step( + PeerEvent::InboundMsg1 { + link: LinkId::new(1), + wire, + est, + }, + 200, + &mut alloc, + ); + assert!(phase1.is_empty()); + assert_eq!(alloc.count(), 0); + + // Late-ACL rejects -> Failed{AclRejected}, still no allocation. + let rej = m.step(PeerEvent::Rejected, 200, &mut alloc); + assert!(rej.is_empty()); + assert_eq!( + m.state(), + PeerState::Failed { + reason: FailReason::AclRejected + } + ); + assert_eq!(alloc.count(), 0); + assert_eq!(m.our_index(), None); + } + + // ---- Test 6c: inbound reject at max_peers -> no allocation ------------ + #[test] + fn inbound_at_max_peers_reject_no_alloc() { + let mut alloc = IndexAllocator::new(); + let peer = peer_identity(); + let mut m = PeerMachine::new_inbound(LinkId::new(1), 0); + let our = *peer_identity().node_addr(); + let mut est = est_new_peer(our); + est.at_max_peers = true; + let wire = wire_outcome(peer, Some([4u8; 8]), 0x77); + + let actions = m.step( + PeerEvent::InboundMsg1 { + link: LinkId::new(1), + wire, + est, + }, + 200, + &mut alloc, + ); + assert!(actions.is_empty()); + assert_eq!( + m.state(), + PeerState::Failed { + reason: FailReason::Rejected + } + ); + assert_eq!(alloc.count(), 0); + assert_eq!(m.our_index(), None); + } + // ---- Test 7: outbound establish (+ cross-connection) ------------------ #[test] fn outbound_establish() {