relay-pool: add TransportSink for WebSocket

Add the `TransportSink` struct as a replacement for the `sink_map_err` method that was causing panics in case of connection interruption (i.e., broken pipe error).

Fixes #984

Pull-Request: #1007
Signed-off-by: Yuki Kishimoto <[email protected]>

Author: yukibtc

crates/nostr-relay-pool/CHANGELOG.md

@@ -44,6 +44,10 @@
 - An option to ban relays that send events which don't match the subscription filter (https://github.com/rust-nostr/nostr/pull/981)
 - Add `RelayOptions::verify_subscriptions` option (https://github.com/rust-nostr/nostr/pull/997)
 
+### Fixed
+
+- Fix panic after a broken pipe error in the relay connection (https://github.com/rust-nostr/nostr/pull/1007)
+
 ## v0.42.0 - 2025/05/20
 
 ### Breaking changes

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

@@ -5,9 +5,12 @@
 //! WebSocket transport
 
 use std::fmt;
+use std::pin::Pin;
 use std::sync::Arc;
+use std::task::{Context, Poll};
 use std::time::Duration;
 
+use async_utility::futures_util::stream::SplitSink;
 use async_wsocket::futures_util::{Sink, SinkExt, Stream, StreamExt, TryStreamExt};
 use async_wsocket::{ConnectionMode, Message, WebSocket};
 use nostr::util::BoxedFuture;
@@ -94,9 +97,43 @@ 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;
+
+            // NOTE: don't use sink_map_err here, as it may cause panics!
+            // Issue: https://github.com/rust-nostr/nostr/issues/984
+            let sink: BoxSink = Box::new(TransportSink(tx)) as BoxSink;
             let stream: BoxStream = Box::new(rx.map_err(TransportError::backend)) as BoxStream;
+
             Ok((sink, stream))
         })
     }
 }
+
+struct TransportSink(SplitSink<WebSocket, Message>);
+
+impl Sink<Message> for TransportSink {
+    type Error = TransportError;
+
+    fn poll_ready(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        Pin::new(&mut self.0)
+            .poll_ready_unpin(cx)
+            .map_err(TransportError::backend)
+    }
+
+    fn start_send(mut self: Pin<&mut Self>, item: Message) -> Result<(), Self::Error> {
+        Pin::new(&mut self.0)
+            .start_send_unpin(item)
+            .map_err(TransportError::backend)
+    }
+
+    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        Pin::new(&mut self.0)
+            .poll_flush_unpin(cx)
+            .map_err(TransportError::backend)
+    }
+
+    fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        Pin::new(&mut self.0)
+            .poll_close_unpin(cx)
+            .map_err(TransportError::backend)
+    }
+}