Auto-reconnect after MMP peer removal, directed outbound configs, sim improvements

Auto-reconnect: - Add per-peer auto_reconnect config (default true) to PeerConfig - schedule_reconnect() feeds removed peers back into retry system with unlimited retries and exponential backoff after MMP dead timeout - RetryState gains reconnect flag to distinguish startup retries (max_retries-limited) from auto-reconnect (unlimited) Retry re-fire fix: - process_pending_retries() now pushes retry_after_ms past the handshake timeout window after successful initiate_peer_connection(), preventing retries from firing every tick with no backoff Chaos sim improvements: - Directed outbound configs: BFS spanning tree + lower-ID-first assignment eliminates dual-connect race conditions in simulation - Save runner log (runner.log) alongside per-node logs for event correlation - Increase churn-20 traffic aggressiveness and node churn (max_down_nodes 3→5, traffic interval min 0s, duration max 90s, concurrent flows 5→10)
2026-07-30 19:46:15 +00:00 · 2026-02-21 13:00:09 +00:00
parent 66c268a564
commit 78a73e1749
9 changed files with 190 additions and 34 deletions
@@ -51,6 +51,10 @@ fn default_priority() -> u8 {
    100
 }

+fn default_auto_reconnect() -> bool {
+    true
+}
+
 impl PeerAddress {
    /// Create a new peer address.
    pub fn new(transport: impl Into<String>, addr: impl Into<String>) -> Self {
@@ -75,7 +79,7 @@ impl PeerAddress {
 ///
 /// Peers are identified by their Nostr public key (npub) and can have
 /// multiple transport addresses for reaching them.
-#[derive(Debug, Clone, Default, Serialize, Deserialize)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
 #[serde(deny_unknown_fields)]
 pub struct PeerConfig {
    /// The peer's Nostr public key in npub (bech32) or hex format.
@@ -92,6 +96,24 @@ pub struct PeerConfig {
    /// Connection policy for this peer.
    #[serde(default)]
    pub connect_policy: ConnectPolicy,
+
+    /// Whether to automatically reconnect after link-dead removal.
+    /// When true (default), the node will retry connecting with exponential
+    /// backoff after MMP removes this peer due to liveness timeout.
+    #[serde(default = "default_auto_reconnect")]
+    pub auto_reconnect: bool,
+}
+
+impl Default for PeerConfig {
+    fn default() -> Self {
+        Self {
+            npub: String::new(),
+            alias: None,
+            addresses: Vec::new(),
+            connect_policy: ConnectPolicy::default(),
+            auto_reconnect: default_auto_reconnect(),
+        }
+    }
 }

 impl PeerConfig {
@@ -102,6 +124,7 @@ impl PeerConfig {
            alias: None,
            addresses: vec![PeerAddress::new(transport, addr)],
            connect_policy: ConnectPolicy::default(),
+            auto_reconnect: default_auto_reconnect(),
        }
    }

@@ -418,7 +418,12 @@ impl Node {
            }
        }

-        // Remove dead peers
+        // Remove dead peers and schedule auto-reconnect
+        let now_ms = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .map(|d| d.as_millis() as u64)
+            .unwrap_or(0);
+
        for addr in &dead_peers {
            warn!(
                peer = %self.peer_display_name(addr),
@@ -426,6 +431,7 @@ impl Node {
                "Removing peer: link dead timeout"
            );
            self.remove_active_peer(addr);
+            self.schedule_reconnect(*addr, now_ms);
        }

        // Send heartbeats (skip peers we just removed)
@@ -51,7 +51,7 @@ impl Node {
                if conn.is_outbound()
                    && let Some(identity) = conn.expected_identity()
                {
-                    self.schedule_retry(*identity.node_addr(), now_ms);
+                    self.schedule_retry(*identity.node_addr(), now_ms, false);
                }
            }
            self.cleanup_stale_connection(link_id, now_ms);
@@ -22,6 +22,9 @@ pub struct RetryState {

    /// Timestamp (Unix ms) when the next retry should be attempted.
    pub retry_after_ms: u64,
+
+    /// Whether this is an auto-reconnect (unlimited retries, ignores max_retries).
+    pub reconnect: bool,
 }

 impl RetryState {
@@ -31,6 +34,7 @@ impl RetryState {
            peer_config,
            retry_count: 0,
            retry_after_ms: 0,
+            reconnect: false,
        }
    }

@@ -48,12 +52,18 @@ impl Node {
    /// Schedule a retry for a failed outbound connection, if applicable.
    ///
    /// Only schedules if the peer is an auto-connect peer and max retries
-    /// have not been exhausted. Does nothing if the peer is already connected
-    /// or has a connection in progress.
-    pub(super) fn schedule_retry(&mut self, node_addr: NodeAddr, now_ms: u64) {
+    /// have not been exhausted (unless `reconnect` is true, which retries
+    /// indefinitely). Does nothing if the peer is already connected or has
+    /// a connection in progress.
+    pub(super) fn schedule_retry(
+        &mut self,
+        node_addr: NodeAddr,
+        now_ms: u64,
+        reconnect: bool,
+    ) {
        let retry_cfg = &self.config.node.retry;
        let max_retries = retry_cfg.max_retries;
-        if max_retries == 0 {
+        if max_retries == 0 && !reconnect {
            return;
        }

@@ -69,7 +79,7 @@ impl Node {
        if let Some(state) = self.retry_pending.get_mut(&node_addr) {
            // Already tracking — increment
            state.retry_count += 1;
-            if state.retry_count > max_retries {
+            if !state.reconnect && state.retry_count > max_retries {
                info!(
                    peer = %peer_name,
                    attempts = state.retry_count,
@@ -83,6 +93,7 @@ impl Node {
            debug!(
                peer = %peer_name,
                retry = state.retry_count,
+                reconnect = state.reconnect,
                delay_secs = delay / 1000,
                "Scheduling connection retry"
            );
@@ -101,10 +112,12 @@ impl Node {
            if let Some(pc) = peer_config {
                let mut state = RetryState::new(pc);
                state.retry_count = 1;
+                state.reconnect = reconnect;
                let delay = state.backoff_ms(base_interval_ms, max_backoff_ms);
                state.retry_after_ms = now_ms + delay;
                debug!(
                    peer = %self.peer_display_name(&node_addr),
+                    reconnect = reconnect,
                    delay_secs = delay / 1000,
                    "First connection attempt failed, scheduling retry"
                );
@@ -114,6 +127,51 @@ impl Node {
        }
    }

+    /// Schedule auto-reconnect for a peer removed by MMP dead timeout.
+    ///
+    /// Looks up the peer in auto-connect config and checks `auto_reconnect`.
+    /// If enabled, feeds the peer into the retry system with unlimited retries.
+    pub(super) fn schedule_reconnect(&mut self, node_addr: NodeAddr, now_ms: u64) {
+        // Find peer in auto-connect config
+        let peer_config = self
+            .config
+            .auto_connect_peers()
+            .find(|pc| {
+                PeerIdentity::from_npub(&pc.npub)
+                    .map(|id| *id.node_addr() == node_addr)
+                    .unwrap_or(false)
+            })
+            .cloned();
+
+        let Some(pc) = peer_config else {
+            return; // Not an auto-connect peer, no reconnect
+        };
+
+        if !pc.auto_reconnect {
+            debug!(
+                peer = %self.peer_display_name(&node_addr),
+                "Auto-reconnect disabled for peer, skipping"
+            );
+            return;
+        }
+
+        let base_interval_ms = self.config.node.retry.base_interval_secs * 1000;
+        let max_backoff_ms = self.config.node.retry.max_backoff_secs * 1000;
+
+        let mut state = RetryState::new(pc);
+        state.reconnect = true;
+        let delay = state.backoff_ms(base_interval_ms, max_backoff_ms);
+        state.retry_after_ms = now_ms + delay;
+
+        info!(
+            peer = %self.peer_display_name(&node_addr),
+            delay_secs = delay / 1000,
+            "Scheduling auto-reconnect after link-dead removal"
+        );
+
+        self.retry_pending.insert(node_addr, state);
+    }
+
    /// Process pending retries whose time has arrived.
    ///
    /// For each due retry, initiates a fresh connection attempt. The retry
@@ -155,12 +213,21 @@ impl Node {

            match self.initiate_peer_connection(&peer_config).await {
                Ok(()) => {
+                    // Push retry_after_ms past the handshake timeout window so
+                    // we don't re-fire on the next tick. If the handshake
+                    // succeeds, promote_connection() clears retry_pending. If
+                    // it times out, check_timeouts() calls schedule_retry()
+                    // which bumps the counter and applies proper backoff.
+                    let hs_timeout_ms =
+                        self.config.node.rate_limit.handshake_timeout_secs * 1000;
+                    if let Some(state) = self.retry_pending.get_mut(&node_addr) {
+                        state.retry_after_ms = now_ms + hs_timeout_ms;
+                    }
                    debug!(
                        peer = %self.peer_display_name(&node_addr),
-                        "Retry connection initiated"
+                        "Retry connection initiated, suppressing re-fire for {}s",
+                        self.config.node.rate_limit.handshake_timeout_secs,
                    );
-                    // Don't remove from retry_pending — wait for promotion
-                    // (success) or next timeout (failure triggers schedule_retry again)
                }
                Err(e) => {
                    warn!(
@@ -169,7 +236,8 @@ impl Node {
                        "Retry connection initiation failed"
                    );
                    // Immediate failure counts as an attempt — schedule next retry
-                    self.schedule_retry(node_addr, now_ms);
+                    // (reconnect flag is preserved on existing retry_pending entry)
+                    self.schedule_retry(node_addr, now_ms, false);
                }
            }
        }
@@ -189,6 +257,7 @@ mod tests {
            peer_config: PeerConfig::default(),
            retry_count: 0,
            retry_after_ms: 0,
+            reconnect: false,
        };
        // base = 5000ms
        assert_eq!(state.backoff_ms(5000, TEST_MAX_BACKOFF_MS), 5000); // 5s * 2^0
@@ -224,6 +293,7 @@ mod tests {
            peer_config: PeerConfig::default(),
            retry_count: 20, // 2^20 * 5000 would be huge
            retry_after_ms: 0,
+            reconnect: false,
        };
        assert_eq!(state.backoff_ms(5000, TEST_MAX_BACKOFF_MS), TEST_MAX_BACKOFF_MS);
    }
@@ -234,6 +304,7 @@ mod tests {
            peer_config: PeerConfig::default(),
            retry_count: 3,
            retry_after_ms: 0,
+            reconnect: false,
        };
        assert_eq!(state.backoff_ms(0, TEST_MAX_BACKOFF_MS), 0);
    }
@@ -566,7 +566,7 @@ fn test_schedule_retry_creates_entry() {

    assert!(node.retry_pending.is_empty());

-    node.schedule_retry(peer_node_addr, 1000);
+    node.schedule_retry(peer_node_addr, 1000, false);

    assert_eq!(node.retry_pending.len(), 1);
    let state = node.retry_pending.get(&peer_node_addr).unwrap();
@@ -593,11 +593,11 @@ fn test_schedule_retry_increments() {
    let mut node = Node::new(config).unwrap();

    // First failure
-    node.schedule_retry(peer_node_addr, 1000);
+    node.schedule_retry(peer_node_addr, 1000, false);
    assert_eq!(node.retry_pending.get(&peer_node_addr).unwrap().retry_count, 1);

    // Second failure
-    node.schedule_retry(peer_node_addr, 11_000);
+    node.schedule_retry(peer_node_addr, 11_000, false);
    let state = node.retry_pending.get(&peer_node_addr).unwrap();
    assert_eq!(state.retry_count, 2);
    // backoff_ms(5000) with retry_count=2 = 5000 * 4 = 20000
@@ -622,14 +622,14 @@ fn test_schedule_retry_max_retries_exhausted() {
    let mut node = Node::new(config).unwrap();

    // Attempts 1 and 2 should schedule retries
-    node.schedule_retry(peer_node_addr, 1000);
+    node.schedule_retry(peer_node_addr, 1000, false);
    assert!(node.retry_pending.contains_key(&peer_node_addr));

-    node.schedule_retry(peer_node_addr, 2000);
+    node.schedule_retry(peer_node_addr, 2000, false);
    assert!(node.retry_pending.contains_key(&peer_node_addr));

    // Attempt 3 exceeds max_retries=2, should remove entry
-    node.schedule_retry(peer_node_addr, 3000);
+    node.schedule_retry(peer_node_addr, 3000, false);
    assert!(
        !node.retry_pending.contains_key(&peer_node_addr),
        "Should be removed after max retries exhausted"
@@ -653,7 +653,7 @@ fn test_schedule_retry_disabled() {

    let mut node = Node::new(config).unwrap();

-    node.schedule_retry(peer_node_addr, 1000);
+    node.schedule_retry(peer_node_addr, 1000, false);
    assert!(
        node.retry_pending.is_empty(),
        "No retry should be scheduled when max_retries=0"
@@ -669,7 +669,7 @@ fn test_schedule_retry_ignores_non_autoconnect() {
    // No peers configured at all
    let mut node = make_node();

-    node.schedule_retry(peer_node_addr, 1000);
+    node.schedule_retry(peer_node_addr, 1000, false);
    assert!(
        node.retry_pending.is_empty(),
        "No retry for unconfigured peer"
@@ -691,7 +691,7 @@ fn test_schedule_retry_skips_connected_peer() {
    assert_eq!(node.peer_count(), 1);

    // Scheduling a retry for an already-connected peer should be a no-op
-    node.schedule_retry(node_addr, 3000);
+    node.schedule_retry(node_addr, 3000, false);
    assert!(
        node.retry_pending.is_empty(),
        "No retry for already-connected peer"
@@ -39,15 +39,15 @@ link_flaps:

 traffic:
  enabled: true
-  max_concurrent: 5
-  interval_secs: { min: 10, max: 30 }
-  duration_secs: { min: 5, max: 15 }
+  max_concurrent: 10
+  interval_secs: { min: 0, max: 30 }
+  duration_secs: { min: 5, max: 90 }
  parallel_streams: 4

 node_churn:
  enabled: true
  interval_secs: { min: 60, max: 90 }
-  max_down_nodes: 3
+  max_down_nodes: 5
  down_duration_secs: { min: 30, max: 90 }
  protect_connectivity: false

@@ -17,14 +17,19 @@ def _load_template() -> str:
        return f.read()


-def generate_peers_block(topology: SimTopology, node_id: str) -> str:
-    """Generate the YAML peers block for a node."""
-    peers = topology.nodes[node_id].peers
-    if not peers:
+def generate_peers_block(
+    topology: SimTopology, node_id: str, outbound_peers: list[str]
+) -> str:
+    """Generate the YAML peers block for a node.
+
+    Only includes peers that this node is responsible for connecting to
+    (outbound direction). The link is still bidirectional once established.
+    """
+    if not outbound_peers:
        return "  []"

    lines = []
-    for peer_id in sorted(peers):
+    for peer_id in sorted(outbound_peers):
        peer = topology.nodes[peer_id]
        lines.append(f'  - npub: "{peer.npub}"')
        lines.append(f'    alias: "{peer_id}"')
@@ -35,11 +40,13 @@ def generate_peers_block(topology: SimTopology, node_id: str) -> str:
    return "\n".join(lines)


-def generate_node_config(topology: SimTopology, node_id: str) -> str:
+def generate_node_config(
+    topology: SimTopology, node_id: str, outbound_peers: list[str]
+) -> str:
    """Generate a complete FIPS config YAML for one node."""
    template = _load_template()
    node = topology.nodes[node_id]
-    peers_yaml = generate_peers_block(topology, node_id)
+    peers_yaml = generate_peers_block(topology, node_id, outbound_peers)

    config = template
    config = config.replace("{{NODE_NAME}}", node_id.upper())
@@ -64,8 +71,9 @@ def write_configs(topology: SimTopology, output_dir: str):
    """Write all node configs and npubs.env to the output directory."""
    os.makedirs(output_dir, exist_ok=True)

+    outbound = topology.directed_outbound()
    for node_id in topology.nodes:
-        config = generate_node_config(topology, node_id)
+        config = generate_node_config(topology, node_id, outbound[node_id])
        path = os.path.join(output_dir, f"{node_id}.yaml")
        with open(path, "w") as f:
            f.write(config)
@@ -71,6 +71,19 @@ class SimRunner:
        s = self.scenario
        mesh_name = f"sim-{s.name}-{s.seed}"

+        # Set up runner log file so all sim orchestration output is captured
+        # alongside the per-node FIPS logs for post-run analysis.
+        os.makedirs(self.output_dir, exist_ok=True)
+        runner_log_path = os.path.join(self.output_dir, "runner.log")
+        fh = logging.FileHandler(runner_log_path, mode="w")
+        fh.setLevel(logging.DEBUG)
+        fh.setFormatter(logging.Formatter(
+            "%(asctime)s %(levelname)-5s %(name)s: %(message)s",
+            datefmt="%H:%M:%S",
+        ))
+        logging.getLogger().addHandler(fh)
+        log.info("Runner log: %s", runner_log_path)
+
        # 1. Generate topology
        log.info(
            "Generating %d-node %s topology (seed=%d)...",
@@ -226,7 +239,6 @@ class SimRunner:
                self.node_mgr.restore_all()

            # Collect logs before stopping containers
-            os.makedirs(self.output_dir, exist_ok=True)
            container_names = [
                self.topology.container_name(nid) for nid in sorted(self.topology.nodes)
            ]
@@ -60,6 +60,42 @@ class SimTopology:
    def container_name(self, node_id: str) -> str:
        return f"fips-node-{node_id}"

+    def directed_outbound(self) -> dict[str, list[str]]:
+        """Assign each edge to exactly one node for outbound connection.
+
+        Returns a mapping from node_id to the list of peers that node
+        should connect to (outbound only). Every edge appears in exactly
+        one direction, ensuring auto-reconnect is testable — if B goes
+        down, only A (the outbound owner) will attempt to reconnect.
+
+        Strategy: BFS spanning tree edges go parent→child. Non-tree
+        edges go from the lower node ID to the higher. This guarantees
+        every node is reachable via at least one inbound connection.
+        """
+        outbound: dict[str, list[str]] = {nid: [] for nid in self.nodes}
+
+        # BFS spanning tree from first node
+        root = min(self.nodes)
+        visited: set[str] = set()
+        tree_edges: set[tuple[str, str]] = set()
+        queue = deque([root])
+        visited.add(root)
+        while queue:
+            node = queue.popleft()
+            for peer in self.nodes[node].peers:
+                if peer not in visited:
+                    visited.add(peer)
+                    queue.append(peer)
+                    tree_edges.add((node, peer))  # parent → child
+                    outbound[node].append(peer)
+
+        # Non-tree edges: lower ID → higher ID
+        for a, b in self.edges:
+            if (a, b) not in tree_edges and (b, a) not in tree_edges:
+                outbound[a].append(b)  # a < b by _make_edge convention
+
+        return outbound
+

 def generate_topology(
    config: TopologyConfig,