smite-scenarios: skip ping_pong for parked connections

smite-scenarios/src/executor.rs

@@ -205,6 +205,10 @@ pub enum ExecuteError {
     #[error("unexpected message: expected type {expected}, got {got}")]
     UnexpectedMessage { expected: u16, got: u16 },
 
+    /// The target sent a BOLT `error`.
+    #[error("peer error on {:?}: {}", .0.channel_id, .0.message().unwrap_or("<non-utf8>"))]
+    PeerError(smite::bolt::Error),
+
     /// Wallet UTXOs could not cover the funding amount and fees.
     #[error("funding: {0}")]
     InsufficientFunds(#[from] smite::channel_tx::InsufficientFunds),
@@ -1220,6 +1224,8 @@ fn recv_non_ping(conn: &mut impl Connection, timeout: Duration) -> Result<Messag
             | Message::GossipTimestampFilter(_) => {
                 log::debug!("skipping gossip message type {}", msg.msg_type());
             }
+            // Surface the received error message.
+            Message::Error(e) => return Err(ExecuteError::PeerError(e)),
             other => return Ok(other),
         }
     })();
@@ -2461,6 +2467,33 @@ mod tests {
         ));
     }
 
+    #[test]
+    fn execute_recv_peer_error() {
+        let peer_error = smite::bolt::Error::all_channels("Wrong channel id in channel_ready");
+        let error_bytes = Message::Error(peer_error.clone()).encode();
+
+        let mut instrs = send_open_channel_instructions();
+        let sent_open_channel = instrs.len() - 1;
+        instrs.push(Instruction {
+            operation: Operation::RecvAcceptChannel,
+            inputs: vec![sent_open_channel],
+        });
+
+        let program = Program {
+            instructions: instrs,
+        };
+        let mut executor = Executor::new(
+            MockConnection::new(),
+            MockBitcoinCli::default(),
+            sample_context(),
+        );
+        executor.conn.queue_recv(error_bytes);
+        let err = executor
+            .execute(&program, std::time::Instant::now())
+            .unwrap_err();
+        assert!(matches!(err, ExecuteError::PeerError(e) if e == peer_error));
+    }
+
     #[test]
     #[allow(clippy::similar_names)] // ping and pong are the canonical names
     fn execute_recv_auto_pong() {

smite-scenarios/src/scenarios.rs

@@ -16,11 +16,38 @@ use smite::scenarios::ScenarioError;
 use std::time::Duration;
 
 use bitcoin::secp256k1::SecretKey;
-use smite::bolt::{Init, Message, Ping};
+use smite::bolt::{Error, Init, Message, Ping};
 use smite::noise::NoiseConnection;
 
 use crate::targets::Target;
 
+/// Peer `error` messages after which a target is known to "park" the
+/// connection, keeping it open but no longer servicing any messages on it.
+/// When this happens the target will not respond to our pings.
+///
+/// Every entry must describe the behavior justifying it, and be removed once
+/// the upstream fix lands.
+///
+/// - `"Wrong channel id"`: CLN permanently fails the channel without
+///   disconnecting. If a follow-on message arrives while the channel's
+///   subdaemon is still dying, lightningd fails to handle the message properly,
+///   and connectd is left waiting for lightningd's response indefinitely. The
+///   node stays alive and keeps serving other connections, but does not read
+///   this one again until the connection is closed due to inactivity 80s later.
+///   See <https://github.com/ElementsProject/lightning/issues/9369>.
+const KNOWN_PARKED_CONNECTION_ERRORS: &[&str] = &["Wrong channel id"];
+
+/// Returns true if `err` is one after which the target is known to leave the
+/// connection open but unserviced. See [`KNOWN_PARKED_CONNECTION_ERRORS`].
+#[must_use]
+pub fn is_known_parked_error(err: &Error) -> bool {
+    err.message().is_some_and(|msg| {
+        KNOWN_PARKED_CONNECTION_ERRORS
+            .iter()
+            .any(|known| msg.contains(known))
+    })
+}
+
 /// Static keys for Noise handshake. Using fixed keys ensures reproducibility
 /// of fuzz failures across runs.
 const STATIC_KEY: [u8; 32] = [
@@ -67,6 +94,15 @@ pub fn handshake_with_target<T: Target>(
     Ok((conn, init))
 }
 
+/// Outcome of a [`ping_pong_checked`] synchronization.
+pub enum PingOutcome {
+    /// The target responded with a pong.
+    Pong,
+    /// The target sent an error after which it is known to leave the connection
+    /// unserviced, so no pong is coming. Carries the error message.
+    ParkedConnection(String),
+}
+
 /// Send ping and wait for pong (for synchronization).
 ///
 /// This ensures the target has done initial processing of any previously sent
@@ -76,14 +112,72 @@ pub fn handshake_with_target<T: Target>(
 ///
 /// Returns an error if the connection is closed or times out.
 pub fn ping_pong(conn: &mut NoiseConnection) -> Result<(), ScenarioError> {
+    match ping_pong_inner(conn, false)? {
+        PingOutcome::Pong => Ok(()),
+        PingOutcome::ParkedConnection(_) => {
+            unreachable!("stopping on known errors is disabled")
+        }
+    }
+}
+
+/// Like [`ping_pong`], but stops waiting when the target sends an error after
+/// which it is known not to service the connection.
+///
+/// # Errors
+///
+/// Returns an error if the connection is closed or times out.
+pub fn ping_pong_checked(conn: &mut NoiseConnection) -> Result<PingOutcome, ScenarioError> {
+    ping_pong_inner(conn, true)
+}
+
+fn ping_pong_inner(
+    conn: &mut NoiseConnection,
+    stop_on_known_error: bool,
+) -> Result<PingOutcome, ScenarioError> {
     conn.send_message(&Message::Ping(Ping::new(0)).encode())?;
 
     // Read messages until we get a pong
     loop {
         let msg_bytes = conn.recv_message()?;
-        if matches!(Message::decode(&msg_bytes)?, Message::Pong(_)) {
-            return Ok(());
+        match Message::decode(&msg_bytes)? {
+            Message::Pong(_) => return Ok(PingOutcome::Pong),
+            Message::Error(e) if stop_on_known_error && is_known_parked_error(&e) => {
+                let msg = e.message().unwrap_or("<non-utf8>").to_string();
+                return Ok(PingOutcome::ParkedConnection(msg));
+            }
+            // Ignore other messages (warnings, errors, etc.)
+            _ => {}
         }
-        // Ignore other messages (warnings, errors, etc.)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{Error, is_known_parked_error};
+    use smite::bolt::ChannelId;
+
+    #[test]
+    fn known_parked_error_matches_substring() {
+        // CLN embeds the phrase in a longer, formatted message.
+        let err = Error::for_channel(
+            ChannelId::new([0x11; 32]),
+            "channeld: sent ERROR Wrong channel id in channel_ready (expected 1111)",
+        );
+        assert!(is_known_parked_error(&err));
+    }
+
+    #[test]
+    fn unrelated_error_is_not_parked() {
+        let err = Error::all_channels("bad funding_signed signature");
+        assert!(!is_known_parked_error(&err));
+    }
+
+    #[test]
+    fn non_utf8_error_is_not_parked() {
+        let err = Error {
+            channel_id: ChannelId::ALL,
+            data: vec![0xff, 0xfe],
+        };
+        assert!(!is_known_parked_error(&err));
     }
 }

smite-scenarios/src/scenarios/ir.rs

@@ -9,7 +9,7 @@ use smite::scenarios::{Scenario, ScenarioError, ScenarioResult};
 use smite::violation::Violation;
 use smite_ir::Program;
 
-use super::{SnapshotSetup, ping_pong};
+use super::{PingOutcome, SnapshotSetup, is_known_parked_error, ping_pong_checked};
 use crate::executor::{ExecuteError, Executor};
 use crate::targets::Target;
 
@@ -57,6 +57,10 @@ impl<T: Target, S: SnapshotSetup<T>> Scenario for IrScenario<T, S> {
             input.len(),
         );
 
+        // Set when the target has sent an error after which it is known to not
+        // service this connection any further. See `is_known_parked_error`.
+        let mut parked = false;
+
         match self.executor.execute(&program, start) {
             Ok(()) => {
                 log::debug!("[{:?}] Program executed successfully", start.elapsed());
@@ -76,6 +80,15 @@ impl<T: Target, S: SnapshotSetup<T>> Scenario for IrScenario<T, S> {
                     start.elapsed(),
                 );
             }
+            Err(ExecuteError::PeerError(e)) => {
+                // Normal protocol behavior: the target rejected our input.
+                parked = is_known_parked_error(&e);
+                log::debug!(
+                    "[{:?}] peer error (parked: {parked}): {}",
+                    start.elapsed(),
+                    e.message().unwrap_or("<non-utf8>"),
+                );
+            }
             Err(ExecuteError::Decode(e)) => {
                 // Either our decoder is incomplete or the target sent something
                 // the spec doesn't allow.
@@ -99,14 +112,29 @@ impl<T: Target, S: SnapshotSetup<T>> Scenario for IrScenario<T, S> {
         }
 
         // Ping-pong sync to ensure the target has at least done the initial
-        // processing of all previous messages. Timeouts here signal a hang.
-        if let Err(e) = ping_pong(self.executor.conn_mut()) {
-            log::debug!("[{:?}] ping_pong: {e}", start.elapsed());
-            if e.is_timeout() {
-                return ScenarioResult::Fail(Violation::Hung.to_string());
-            }
+        // processing of all previous messages. Timeouts here signal a hang,
+        // unless we know the connection has been parked by the target, in which
+        // case we continue without requiring a pong.
+        if parked {
+            log::info!(
+                "[{:?}] connection parked by target, skipping ping-pong",
+                start.elapsed()
+            );
         } else {
-            log::debug!("[{:?}] Target responded with pong", start.elapsed());
+            match ping_pong_checked(self.executor.conn_mut()) {
+                Ok(PingOutcome::Pong) => {
+                    log::debug!("[{:?}] Target responded with pong", start.elapsed());
+                }
+                Ok(PingOutcome::ParkedConnection(msg)) => {
+                    log::info!("[{:?}] connection parked by target: {msg}", start.elapsed());
+                }
+                Err(e) => {
+                    log::debug!("[{:?}] ping_pong: {e}", start.elapsed());
+                    if e.is_timeout() {
+                        return ScenarioResult::Fail(Violation::Hung.to_string());
+                    }
+                }
+            }
         }
 
         if let Err(e) = self.target.check_alive() {