relay-pool: don't send ws close msg if the stream has been terminated

Don't send the WebSocket close message if the stream has been terminated.

Fixes #984

Signed-off-by: Yuki Kishimoto <yukikishimoto@protonmail.com>

Author: yukibtc

crates/nostr-relay-pool/src/relay/inner.rs

@@ -37,6 +37,7 @@ use crate::policy::AdmitStatus;
 use crate::pool::RelayPoolNotification;
 use crate::relay::status::AtomicRelayStatus;
 use crate::shared::SharedState;
+use crate::transport::error::TransportError;
 use crate::transport::websocket::{BoxSink, BoxStream};
 
 type ClientMessageJson = String;
@@ -742,7 +743,15 @@ impl InnerRelay {
             // Message sender handler
             res = self.sender_message_handler(&mut ws_tx, rx_nostr, &ping) => match res {
                 Ok(()) => tracing::trace!(url = %self.url, "Relay sender exited."),
-                Err(e) => tracing::error!(url = %self.url, error = %e, "Relay sender exited with error.")
+                // Sink error!
+                Err(Error::Transport(TransportError::Sink(e))) => {
+                    tracing::error!(url = %self.url, error = %e, "Relay sender exited with sink error.");
+
+                    // The sink returned an error, so an additional call on the sink (like closing the sink) will cause a panic.
+                    // Issue: https://github.com/rust-nostr/nostr/issues/984
+                    return;
+                },
+                Err(e) => tracing::error!(url = %self.url, error = %e, "Relay sender exited with error."),
             },
             // Message receiver handler
             res = self.receiver_message_handler(ws_rx, &ping, ingester_tx) => match res {

crates/nostr-relay-pool/src/transport/error.rs

@@ -11,6 +11,10 @@ use core::fmt;
 pub enum TransportError {
     /// An error happened in the underlying backend.
     Backend(Box<dyn std::error::Error + Send + Sync>),
+    /// Sink error
+    Sink(Box<dyn std::error::Error + Send + Sync>),
+    /// Stream error
+    Stream(Box<dyn std::error::Error + Send + Sync>),
 }
 
 impl std::error::Error for TransportError {}
@@ -19,6 +23,8 @@ impl fmt::Display for TransportError {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
             Self::Backend(e) => write!(f, "{e}"),
+            Self::Sink(e) => write!(f, "{e}"),
+            Self::Stream(e) => write!(f, "{e}"),
         }
     }
 }
@@ -34,4 +40,26 @@ impl TransportError {
     {
         Self::Backend(Box::new(error))
     }
+
+    /// Create a new sink error
+    ///
+    /// Shorthand for `Error::Sink(Box::new(error))`.
+    #[inline]
+    pub fn sink<E>(error: E) -> Self
+    where
+        E: std::error::Error + Send + Sync + 'static,
+    {
+        Self::Sink(Box::new(error))
+    }
+
+    /// Create a new stream error
+    ///
+    /// Shorthand for `Error::Sink(Box::new(error))`.
+    #[inline]
+    pub fn stream<E>(error: E) -> Self
+    where
+        E: std::error::Error + Send + Sync + 'static,
+    {
+        Self::Stream(Box::new(error))
+    }
 }

crates/nostr-relay-pool/src/transport/websocket.rs

@@ -94,8 +94,8 @@ impl WebSocketTransport for DefaultWebsocketTransport {
 
             // Split sink and stream
             let (tx, rx) = socket.split();
-            let sink: BoxSink = Box::new(tx.sink_map_err(TransportError::backend)) as BoxSink;
-            let stream: BoxStream = Box::new(rx.map_err(TransportError::backend)) as BoxStream;
+            let sink: BoxSink = Box::new(tx.sink_map_err(TransportError::sink)) as BoxSink;
+            let stream: BoxStream = Box::new(rx.map_err(TransportError::stream)) as BoxStream;
             Ok((sink, stream))
         })
     }