chore(deps): replace parity-ipc-server with interprocess (#7862)

Author: DaniPopes

Cargo.lock

@@ -1,8 +1,8 @@
-//! Utility functions for Ethereum adapted from https://github.dev/rust-blockchain/ethereum/blob/755dffaa4903fbec1269f50cde9863cf86269a14/src/util.rs
-use std::collections::BTreeMap;
+//! Utility functions for Ethereum adapted from <https://github.com/rust-blockchain/ethereum/blob/755dffaa4903fbec1269f50cde9863cf86269a14/src/util.rs>
 
 use alloy_primitives::{fixed_bytes, B256};
 use alloy_trie::{HashBuilder, Nibbles};
+use std::collections::BTreeMap;
 
 /// The KECCAK of the RLP encoding of empty data.
 pub const KECCAK_NULL_RLP: B256 =

crates/anvil/core/src/eth/trie.rs

@@ -25,7 +25,7 @@ parking_lot = "0.12"
 futures = "0.3"
 
 # ipc
-parity-tokio-ipc = { version = "0.9", optional = true }
+interprocess = { version = "2", optional = true, features = ["tokio"] }
 bytes = { version = "1.4", optional = true }
 tokio-util = { version = "0.7", features = ["codec"], optional = true }
 
@@ -40,4 +40,4 @@ pin-project = "1"
 
 [features]
 default = ["ipc"]
-ipc = ["parity-tokio-ipc", "bytes", "tokio-util"]
+ipc = ["dep:interprocess", "dep:bytes", "dep:tokio-util"]

crates/anvil/server/Cargo.toml

@@ -4,7 +4,7 @@ use crate::{error::RequestError, pubsub::PubSubConnection, PubSubRpcHandler};
 use anvil_rpc::request::Request;
 use bytes::BytesMut;
 use futures::{ready, Sink, Stream, StreamExt};
-use parity_tokio_ipc::Endpoint;
+use interprocess::local_socket::{self as ls, tokio::prelude::*};
 use std::{
     future::Future,
     io,
@@ -18,14 +18,14 @@ use std::{
 pub struct IpcEndpoint<Handler> {
     /// the handler for the websocket connection
     handler: Handler,
-    /// The endpoint we listen for incoming transactions
-    endpoint: Endpoint,
+    /// The path to the socket
+    path: String,
 }
 
 impl<Handler: PubSubRpcHandler> IpcEndpoint<Handler> {
     /// Creates a new endpoint with the given handler
-    pub fn new(handler: Handler, endpoint: String) -> Self {
-        Self { handler, endpoint: Endpoint::new(endpoint) }
+    pub fn new(handler: Handler, path: String) -> Self {
+        Self { handler, path }
     }
 
     /// Returns a stream of incoming connection handlers
@@ -34,39 +34,43 @@ impl<Handler: PubSubRpcHandler> IpcEndpoint<Handler> {
     /// connections, See [`PubSubConnection`] that should be spawned
     #[instrument(target = "ipc", skip_all)]
     pub fn incoming(self) -> io::Result<impl Stream<Item = impl Future<Output = ()>>> {
-        let IpcEndpoint { handler, endpoint } = self;
-        trace!(endpoint=?endpoint.path(), "starting IPC server" );
+        let IpcEndpoint { handler, path } = self;
+
+        trace!(%path, "starting IPC server");
 
         if cfg!(unix) {
             // ensure the file does not exist
-            if std::fs::remove_file(endpoint.path()).is_ok() {
-                warn!(endpoint=?endpoint.path(), "removed existing file");
+            if std::fs::remove_file(&path).is_ok() {
+                warn!(%path, "removed existing file");
             }
         }
 
-        let connections = match endpoint.incoming() {
-            Ok(connections) => connections,
-            Err(err) => {
-                error!(%err, "Failed to create IPC listener");
-                return Err(err)
-            }
-        };
+        let name = to_name(path.as_ref())?;
+        let listener = ls::ListenerOptions::new().name(name).create_tokio()?;
+        // TODO: https://github.com/kotauskas/interprocess/issues/64
+        let connections = futures::stream::unfold(listener, |listener| async move {
+            let conn = listener.accept().await;
+            Some((conn, listener))
+        });
 
         trace!("established connection listener");
 
-        let connections = connections.filter_map(move |stream| {
+        Ok(connections.filter_map(move |stream| {
             let handler = handler.clone();
-            Box::pin(async move {
-                if let Ok(stream) = stream {
-                    trace!("successful incoming IPC connection");
-                    let framed = tokio_util::codec::Decoder::framed(JsonRpcCodec, stream);
-                    Some(PubSubConnection::new(IpcConn(framed), handler))
-                } else {
-                    None
+            async move {
+                match stream {
+                    Ok(stream) => {
+                        trace!("successful incoming IPC connection");
+                        let framed = tokio_util::codec::Decoder::framed(JsonRpcCodec, stream);
+                        Some(PubSubConnection::new(IpcConn(framed), handler))
+                    }
+                    Err(err) => {
+                        trace!(%err, "unsuccessful incoming IPC connection");
+                        None
+                    }
                 }
-            })
-        });
-        Ok(connections)
+            }
+        }))
     }
 }
 
@@ -118,7 +122,7 @@ where
 
 struct JsonRpcCodec;
 
-// Adapted from <https://github.dev/paritytech/jsonrpc/blob/38af3c9439aa75481805edf6c05c6622a5ab1e70/server-utils/src/stream_codec.rs#L47-L105>
+// Adapted from <https://github.com/paritytech/jsonrpc/blob/38af3c9439aa75481805edf6c05c6622a5ab1e70/server-utils/src/stream_codec.rs#L47-L105>
 impl tokio_util::codec::Decoder for JsonRpcCodec {
     type Item = String;
     type Error = io::Error;
@@ -171,3 +175,11 @@ impl tokio_util::codec::Encoder<String> for JsonRpcCodec {
         Ok(())
     }
 }
+
+fn to_name(path: &std::ffi::OsStr) -> io::Result<ls::Name<'_>> {
+    if cfg!(windows) && !path.as_encoded_bytes().starts_with(br"\\.\pipe\") {
+        ls::ToNsName::to_ns_name::<ls::GenericNamespaced>(path)
+    } else {
+        ls::ToFsName::to_fs_name::<ls::GenericFilePath>(path)
+    }
+}

crates/anvil/server/src/ipc.rs

@@ -59,12 +59,8 @@ pub const CHAIN_ID: u64 = 31337;
 pub const DEFAULT_MNEMONIC: &str = "test test test test test test test test test test test junk";
 
 /// The default IPC endpoint
-#[cfg(windows)]
-pub const DEFAULT_IPC_ENDPOINT: &str = r"\\.\pipe\anvil.ipc";
-
-/// The default IPC endpoint
-#[cfg(not(windows))]
-pub const DEFAULT_IPC_ENDPOINT: &str = "/tmp/anvil.ipc";
+pub const DEFAULT_IPC_ENDPOINT: &str =
+    if cfg!(unix) { "/tmp/anvil.ipc" } else { r"\\.\pipe\anvil.ipc" };
 
 /// `anvil 0.1.0 (f01b232bc 2022-04-13T23:28:39.493201+00:00)`
 pub const VERSION_MESSAGE: &str = concat!(
@@ -801,7 +797,7 @@ impl NodeConfig {
 
     /// Returns the ipc path for the ipc endpoint if any
     pub fn get_ipc_path(&self) -> Option<String> {
-        match self.ipc_path.as_ref() {
+        match &self.ipc_path {
             Some(path) => path.clone().or_else(|| Some(DEFAULT_IPC_ENDPOINT.to_string())),
             None => None,
         }

crates/anvil/src/config.rs

@@ -12,6 +12,7 @@ use crate::{
     },
     filter::Filters,
     logging::{LoggingManager, NodeLogLayer},
+    server::error::{NodeError, NodeResult},
     service::NodeService,
     shutdown::Signal,
     tasks::TaskManager,
@@ -23,6 +24,7 @@ use foundry_common::provider::{ProviderBuilder, RetryProvider};
 use foundry_evm::revm;
 use futures::{FutureExt, TryFutureExt};
 use parking_lot::Mutex;
+use server::try_spawn_ipc;
 use std::{
     future::Future,
     io,
@@ -41,11 +43,8 @@ mod service;
 
 mod config;
 pub use config::{AccountGenerator, NodeConfig, CHAIN_ID, VERSION_MESSAGE};
+
 mod hardfork;
-use crate::server::{
-    error::{NodeError, NodeResult},
-    spawn_ipc,
-};
 pub use hardfork::Hardfork;
 
 /// ethereum related implementations
@@ -223,7 +222,8 @@ pub async fn try_spawn(mut config: NodeConfig) -> io::Result<(EthApi, NodeHandle
     let (signal, on_shutdown) = shutdown::signal();
     let task_manager = TaskManager::new(tokio_handle, on_shutdown);
 
-    let ipc_task = config.get_ipc_path().map(|path| spawn_ipc(api.clone(), path));
+    let ipc_task =
+        config.get_ipc_path().map(|path| try_spawn_ipc(api.clone(), path)).transpose()?;
 
     let handle = NodeHandle {
         config,
@@ -310,8 +310,10 @@ impl NodeHandle {
 
     /// Constructs a [`RetryProvider`] for this handle's HTTP endpoint.
     pub fn http_provider(&self) -> RetryProvider {
-        ProviderBuilder::new(&self.http_endpoint()).build().expect("failed to build HTTP provider")
-        // .interval(Duration::from_millis(500))
+        ProviderBuilder::new(&self.http_endpoint())
+            .aggressive()
+            .build()
+            .expect("failed to build HTTP provider")
     }
 
     /// Constructs a [`RetryProvider`] for this handle's WS endpoint.

crates/anvil/src/lib.rs

@@ -1,28 +1,34 @@
 //! IPC tests
 
-use crate::utils::connect_pubsub;
+use crate::{init_tracing, utils::connect_pubsub};
 use alloy_primitives::U256;
 use alloy_provider::Provider;
 use anvil::{spawn, NodeConfig};
 use futures::StreamExt;
-
-pub fn rand_ipc_endpoint() -> String {
-    let num: u64 = rand::Rng::gen(&mut rand::thread_rng());
-    if cfg!(windows) {
-        format!(r"\\.\pipe\anvil-ipc-{num}")
+use tempfile::TempDir;
+
+fn ipc_config() -> (Option<TempDir>, NodeConfig) {
+    let path;
+    let dir;
+    if cfg!(unix) {
+        let tmp = tempfile::tempdir().unwrap();
+        path = tmp.path().join("anvil.ipc").to_string_lossy().into_owned();
+        dir = Some(tmp);
     } else {
-        format!(r"/tmp/anvil-ipc-{num}")
+        dir = None;
+        path = format!(r"\\.\pipe\anvil_test_{}.ipc", rand::random::<u64>());
     }
-}
-
-fn ipc_config() -> NodeConfig {
-    NodeConfig::test().with_ipc(Some(Some(rand_ipc_endpoint())))
+    let config = NodeConfig::test().with_ipc(Some(Some(path)));
+    (dir, config)
 }
 
 #[tokio::test(flavor = "multi_thread")]
-#[cfg_attr(target_os = "windows", ignore)]
+#[cfg_attr(windows, ignore = "TODO")]
 async fn can_get_block_number_ipc() {
-    let (api, handle) = spawn(ipc_config()).await;
+    init_tracing();
+
+    let (_dir, config) = ipc_config();
+    let (api, handle) = spawn(config).await;
 
     let block_num = api.block_number().unwrap();
     assert_eq!(block_num, U256::ZERO);
@@ -34,17 +40,19 @@ async fn can_get_block_number_ipc() {
 }
 
 #[tokio::test(flavor = "multi_thread")]
-#[cfg_attr(target_os = "windows", ignore)]
+#[cfg_attr(windows, ignore = "TODO")]
 async fn test_sub_new_heads_ipc() {
-    let (api, handle) = spawn(ipc_config()).await;
-
-    let provider = connect_pubsub(handle.ipc_path().unwrap().as_str()).await;
+    init_tracing();
 
-    let blocks = provider.subscribe_blocks().await.unwrap().into_stream();
+    let (_dir, config) = ipc_config();
+    let (api, handle) = spawn(config).await;
 
+    let provider = connect_pubsub(handle.ipc_path().unwrap().as_str()).await;
     // mine a block every 1 seconds
     api.anvil_set_interval_mining(1).unwrap();
 
+    let blocks = provider.subscribe_blocks().await.unwrap().into_stream();
+
     let blocks = blocks.take(3).collect::<Vec<_>>().await;
     let block_numbers = blocks.into_iter().map(|b| b.header.number.unwrap()).collect::<Vec<_>>();
 

crates/anvil/tests/it/ipc.rs

@@ -195,7 +195,7 @@ impl RuntimeTransport {
     async fn connect_ipc(&self) -> Result<InnerTransport, RuntimeTransportError> {
         let path = url_to_file_path(&self.url)
             .map_err(|_| RuntimeTransportError::BadPath(self.url.to_string()))?;
-        let ipc_connector: IpcConnect<PathBuf> = path.clone().into();
+        let ipc_connector = IpcConnect::new(path.clone());
         let ipc = ipc_connector.into_service().await.map_err(|e| {
             RuntimeTransportError::TransportError(e, path.clone().display().to_string())
         })?;