tsarevvv/rust-ipfs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge #499

Browse Source 
499: Update libp2p to v0.43.0 r=koivunej a=rand0m-cloud

This PR updates the dependency on libp2p. This is needed because currently building `ipfs-http` fails with an ambiguous type error inside of `libp2p`.

This PR passes all tests!!

### Checklist (can be deleted from PR description once items are checked)

- [x] **New** code is “linted” i.e. code formatting via rustfmt and language idioms via clippy
- [ ] There are no extraneous changes like formatting, line reordering, etc. Keep the patch sizes small!
- [ ] There are functional and/or unit tests written, and they are passing
- [ ] There is suitable documentation. In our case, this means:
    - [ ] Each command has a usage example and API specification
    - [ ] Rustdoc tests are passing on all code-level comments
    - [ ] Differences between Rust’s IPFS implementation and the Go or JS implementations are explained
- [x] Additions to CHANGELOG.md files



Co-authored-by: Addy Bryant <rand0m-cloud@outlook.com>
Co-authored-by: Joonas Koivunen <joonas.koivunen@gmail.com>
This commit is contained in:
bors[bot] 2022-04-01 13:22:58 +00:00 committed by GitHub
commit e8f6d66a60
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
16 changed files with 374 additions and 185 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
.github/workflows/ci.yml vendored
View File

@ -33,7 +33,7 @@ jobs:
- target: x86_64-pc-windows-msvc
name: windows
host: windows-latest
host: windows-2019
cross: false
# Mobile platforms disabled until we get a good estimate of them being
@ -86,7 +86,7 @@ jobs:
- name: Install and cache vcpkg (windows)
uses: lukka/run-vcpkg@v7.4
id: windows-runvcpkg
if: matrix.platform.host == 'windows-latest'
if: matrix.platform.host == 'windows-2019'
with:
vcpkgDirectory: '${{ runner.workspace }}/vcpkg'
vcpkgTriplet: 'x64-windows'
@ -94,7 +94,7 @@ jobs:
setupOnly: true # required for caching
- name: Install depedencies (windows)
if: matrix.platform.host == 'windows-latest'
if: matrix.platform.host == 'windows-2019'
run: "$VCPKG_ROOT/vcpkg install openssl:x64-windows"
shell: bash
env:

2
CHANGELOG.md
View File

@ -13,6 +13,7 @@
* chore: upgrade to libp2p 0.39.1, update most of the other deps with the notable exception of cid and multihash [#472]
* refactor(swarm): swarm cleanup following libp2p upgrade to v0.39.1 [#473]
* fix: strict ordering for DAG-CBOR-encoded map keys [#493]
* feat: upgrade libp2p to v0.43.0 [#499]
[#429]: https://github.com/rs-ipfs/rust-ipfs/pull/429
[#428]: https://github.com/rs-ipfs/rust-ipfs/pull/428
@ -29,6 +30,7 @@
[#472]: https://github.com/rs-ipfs/rust-ipfs/pull/472
[#473]: https://github.com/rs-ipfs/rust-ipfs/pull/473
[#493]: https://github.com/rs-ipfs/rust-ipfs/pull/493
[#499]: https://github.com/rs-ipfs/rust-ipfs/pull/499
# 0.2.1

4
Cargo.toml
View File

@ -31,10 +31,10 @@ either = { default-features = false, version = "1.5" }
futures = { default-features = false, version = "0.3.9", features = ["alloc", "std"] }
hash_hasher = "2.0.3"
ipfs-unixfs = { version = "0.2", path = "unixfs" }
libp2p = { default-features = false, features = ["floodsub", "identify", "kad", "tcp-tokio", "mplex", "noise", "ping", "yamux", "dns-tokio"], version = "0.39.1" }
libp2p = { default-features = false, features = ["floodsub", "identify", "kad", "tcp-tokio", "mplex", "noise", "ping", "yamux", "dns-tokio"], version = "0.43.0" }
multibase = { default-features = false, version = "0.9" }
multihash = { default-features = false, version = "0.11" }
prost = { default-features = false, version = "0.8" }
prost = { default-features = false, version = "0.9" }
serde = { default-features = false, features = ["derive"], version = "1.0" }
serde_json = { default-features = false, features = ["std"], version = "1.0" }
thiserror = { default-features = false, version = "1.0" }

6
bitswap/Cargo.toml
View File

@ -15,10 +15,10 @@ cid = { default-features = false, version = "0.5" }
fnv = { default-features = false, version = "1.0" }
futures = { default-features = false, version = "0.3" }
hash_hasher = "2.0.3"
libp2p-core = { default-features = false, version = "0.29" }
libp2p-swarm = { default-features = false, version = "0.30" }
libp2p-core = { default-features = false, version = "0.32" }
libp2p-swarm = { default-features = false, version = "0.34" }
multihash = { default-features = false, version = "0.11" }
prost = { default-features = false, version = "0.8" }
prost = { default-features = false, version = "0.9" }
thiserror = { default-features = false, version = "1.0" }
tokio = { default-features = false, version = "1", features = ["rt"] }
tracing = { default-features = false, version = "0.1" }

56
bitswap/src/behaviour.rs
View File

@ -12,11 +12,10 @@ use cid::Cid;
use fnv::FnvHashSet;
use futures::channel::mpsc::{unbounded, UnboundedReceiver, UnboundedSender};
use hash_hasher::HashedMap;
use libp2p_core::{connection::ConnectionId, Multiaddr, PeerId};
use libp2p_swarm::protocols_handler::{IntoProtocolsHandler, OneShotHandler, ProtocolsHandler};
use libp2p_swarm::{
DialPeerCondition, NetworkBehaviour, NetworkBehaviourAction, NotifyHandler, PollParameters,
};
use libp2p_core::{connection::ConnectionId, ConnectedPoint, Multiaddr, PeerId};
use libp2p_swarm::dial_opts::{DialOpts, PeerCondition};
use libp2p_swarm::handler::OneShotHandler;
use libp2p_swarm::{NetworkBehaviour, NetworkBehaviourAction, NotifyHandler, PollParameters};
use std::task::{Context, Poll};
use std::{
collections::{HashMap, VecDeque},
@ -88,7 +87,13 @@ impl Stats {
/// Network behaviour that handles sending and receiving IPFS blocks.
pub struct Bitswap {
/// Queue of events to report to the user.
events: VecDeque<NetworkBehaviourAction<Message, BitswapEvent>>,
events: VecDeque<
NetworkBehaviourAction<
BitswapEvent,
<Bitswap as NetworkBehaviour>::ConnectionHandler,
Message,
>,
>,
/// List of prospect peers to connect to.
target_peers: FnvHashSet<PeerId>,
/// Ledger
@ -150,9 +155,12 @@ impl Bitswap {
/// Called from Kademlia behaviour.
pub fn connect(&mut self, peer_id: PeerId) {
if self.target_peers.insert(peer_id) {
self.events.push_back(NetworkBehaviourAction::DialPeer {
peer_id,
condition: DialPeerCondition::Disconnected,
let handler = self.new_handler();
self.events.push_back(NetworkBehaviourAction::Dial {
opts: DialOpts::peer_id(peer_id)
.condition(PeerCondition::Disconnected)
.build(),
handler,
});
}
}
@ -209,10 +217,10 @@ impl Bitswap {
}
impl NetworkBehaviour for Bitswap {
type ProtocolsHandler = OneShotHandler<BitswapConfig, Message, MessageWrapper>;
type ConnectionHandler = OneShotHandler<BitswapConfig, Message, MessageWrapper>;
type OutEvent = BitswapEvent;
fn new_handler(&mut self) -> Self::ProtocolsHandler {
fn new_handler(&mut self) -> Self::ConnectionHandler {
debug!("bitswap: new_handler");
Default::default()
}
@ -222,7 +230,14 @@ impl NetworkBehaviour for Bitswap {
Vec::new()
}
fn inject_connected(&mut self, peer_id: &PeerId) {
fn inject_connection_established(
&mut self,
peer_id: &PeerId,
_connection_id: &ConnectionId,
_endpoint: &ConnectedPoint,
_failed_addresses: Option<&Vec<Multiaddr>>,
_other_established: usize,
) {
debug!("bitswap: inject_connected {}", peer_id);
let ledger = Ledger::new();
self.stats.entry(*peer_id).or_default();
@ -230,7 +245,14 @@ impl NetworkBehaviour for Bitswap {
self.send_want_list(*peer_id);
}
fn inject_disconnected(&mut self, peer_id: &PeerId) {
fn inject_connection_closed(
&mut self,
peer_id: &PeerId,
_connection_id: &ConnectionId,
_endpoint: &ConnectedPoint,
_handler: Self::ConnectionHandler,
_remaining_established: usize,
) {
debug!("bitswap: inject_disconnected {:?}", peer_id);
self.connected_peers.remove(peer_id);
// the related stats are not dropped, so that they
@ -289,9 +311,11 @@ impl NetworkBehaviour for Bitswap {
}
#[allow(clippy::type_complexity)]
fn poll(&mut self, ctx: &mut Context, _: &mut impl PollParameters)
-> Poll<NetworkBehaviourAction<<<Self::ProtocolsHandler as IntoProtocolsHandler>::Handler as ProtocolsHandler>::InEvent, Self::OutEvent>>
{
fn poll(
&mut self,
ctx: &mut Context,
_: &mut impl PollParameters,
) -> Poll<NetworkBehaviourAction<Self::OutEvent, Self::ConnectionHandler>> {
use futures::stream::StreamExt;
while let Poll::Ready(Some((peer_id, block))) = self.ready_blocks.poll_next_unpin(ctx) {

2
conformance/setup.sh
View File

@ -28,7 +28,7 @@ if [ -d "patches" ]; then
fi
if ! [ -f "../target/debug/ipfs-http" ]; then
echo "Please build a debug version of Rust IPFS first via `cargo build --workspace` in the project root first." >&2
echo 'Please build a debug version of Rust IPFS first via `cargo build --workspace` in the project root first.' >&2
exit 1
fi

13
conformance/test/index.js
View File

@ -52,8 +52,17 @@ tests.miscellaneous(factory, { skip: [
// Phase 1.1
// these are a bit flaky
tests.pubsub(factory)
if (process.platform !== "win32") {
// the following tests started failing with the libp2p 0.43 upgrade for yet unknown reasons:
//
// 1) .pubsub.subscribe > multiple connected nodes > should send/receive 100 messages
// 2) .pubsub.peers > should not return extra peers
// 3) .pubsub.peers > should return peers for a topic - one peer
// 4) .pubsub.peers > should return peers for a topic - multiple peers
//
// also, these are known to be a bit flaky
tests.pubsub(factory)
}
// these are rarely flaky
tests.swarm(factory)

2
http/Cargo.toml
View File

@ -24,7 +24,7 @@ multihash = { default-features = false, version = "0.11" }
# openssl is required for rsa keygen but not used by the rust-ipfs or its dependencies
openssl = { default-features = false, version = "0.10" }
percent-encoding = { default-features = false, version = "2.1" }
prost = { default-features = false, version = "0.8" }
prost = { default-features = false, version = "0.9" }
serde = { default-features = false, features = ["derive"], version = "1.0" }
serde_json = { default-features = false, version = "1.0" }
structopt = { default-features = false, version = "0.3" }

6
http/src/config.rs
View File

@ -91,7 +91,7 @@ pub fn init(
let kp = ipfs::Keypair::rsa_from_pkcs8(&mut pkcs8)
.expect("Failed to turn pkcs#8 into libp2p::identity::Keypair");
let peer_id = kp.public().into_peer_id().to_string();
let peer_id = kp.public().to_peer_id().to_string();
// TODO: this part could be PR'd to rust-libp2p as they already have some public key
// import/export but probably not if ring does not support these required conversions.
@ -193,7 +193,7 @@ pub fn load(config: File) -> Result<Config, LoadingError> {
let kp = config_file.identity.load_keypair()?;
let peer_id = kp.public().into_peer_id().to_string();
let peer_id = kp.public().to_peer_id().to_string();
if peer_id != config_file.identity.peer_id {
return Err(LoadingError::PeerIdMismatch {
@ -370,7 +370,7 @@ aGVsbG8gd29ybGQ=
.load_keypair()
.unwrap()
.public()
.into_peer_id()
.to_peer_id()
.to_string();
assert_eq!(peer_id, input.peer_id);

4
http/src/v0/id.rs
View File

@ -39,9 +39,9 @@ async fn identity_query<T: IpfsTypes>(
match ipfs.identity().await {
Ok((public_key, addresses)) => {
let peer_id = public_key.clone().into_peer_id();
let peer_id = public_key.to_peer_id();
let id = peer_id.to_string();
let public_key = Base64Pad.encode(public_key.into_protobuf_encoding());
let public_key = Base64Pad.encode(public_key.to_protobuf_encoding());
let addresses = addresses.into_iter().map(|addr| addr.to_string()).collect();

15
src/lib.rs
View File

@ -208,7 +208,6 @@ impl<I: Borrow<Keypair>> fmt::Debug for DebuggableKeypair<I> {
let kind = match self.get_ref() {
Keypair::Ed25519(_) => "Ed25519",
Keypair::Rsa(_) => "Rsa",
Keypair::Secp256k1(_) => "Secp256k1",
};
write!(fmt, "Keypair::{}", kind)
@ -743,7 +742,7 @@ impl<Types: IpfsTypes> Ipfs<Types> {
.await?;
let mut addresses = rx.await?;
let public_key = self.keys.get_ref().public();
let peer_id = public_key.clone().into_peer_id();
let peer_id = public_key.to_peer_id();
for addr in &mut addresses {
addr.push(Protocol::P2p(peer_id.into()))
@ -1476,12 +1475,14 @@ impl<TRepoTypes: RepoTypes> Future for IpfsFuture<TRepoTypes> {
IpfsEvent::RemoveListeningAddress(addr, ret) => {
let removed = if let Some((id, _)) = self.listening_addresses.remove(&addr)
{
self.swarm.remove_listener(id).map_err(|_: ()| {
format_err!(
if !self.swarm.remove_listener(id) {
Err(format_err!(
"Failed to remove previously added listening address: {}",
addr
)
})
))
} else {
Ok(())
}
} else {
Err(format_err!("Address was not listened to before: {}", addr))
};
@ -1689,7 +1690,7 @@ mod node {
/// Returns a new `Node` based on `IpfsOptions`.
pub async fn with_options(opts: IpfsOptions) -> Self {
let id = opts.keypair.public().into_peer_id();
let id = opts.keypair.public().to_peer_id();
// for future: assume UninitializedIpfs handles instrumenting any futures with the
// given span

28
src/p2p/behaviour.rs
View File

@ -15,6 +15,7 @@ use libp2p::kad::{Kademlia, KademliaConfig, KademliaEvent, Quorum};
// use libp2p::mdns::{MdnsEvent, TokioMdns};
use libp2p::ping::{Ping, PingEvent};
// use libp2p::swarm::toggle::Toggle;
use libp2p::floodsub::FloodsubEvent;
use libp2p::swarm::{NetworkBehaviour, NetworkBehaviourEventProcess};
use multibase::Base;
use std::{convert::TryInto, sync::Arc};
@ -22,18 +23,19 @@ use tokio::task;
/// Behaviour type.
#[derive(libp2p::NetworkBehaviour)]
#[behaviour(event_process = true)]
pub struct Behaviour<Types: IpfsTypes> {
#[behaviour(ignore)]
repo: Arc<Repo<Types>>,
// mdns: Toggle<TokioMdns>,
kademlia: Kademlia<MemoryStore>,
#[behaviour(ignore)]
kad_subscriptions: SubscriptionRegistry<KadResult, String>,
bitswap: Bitswap,
ping: Ping,
identify: Identify,
pubsub: Pubsub,
pub swarm: SwarmApi,
#[behaviour(ignore)]
repo: Arc<Repo<Types>>,
#[behaviour(ignore)]
kad_subscriptions: SubscriptionRegistry<KadResult, String>,
}
/// Represents the result of a Kademlia query.
@ -84,7 +86,7 @@ impl<Types: IpfsTypes> NetworkBehaviourEventProcess<KademliaEvent> for Behaviour
};
match event {
InboundRequestServed { request } => {
InboundRequest { request } => {
trace!("kad: inbound {:?} request handled", request);
}
OutboundQueryCompleted { result, id, .. } => {
@ -377,7 +379,7 @@ impl<Types: IpfsTypes> NetworkBehaviourEventProcess<BitswapEvent> for Behaviour<
impl<Types: IpfsTypes> NetworkBehaviourEventProcess<PingEvent> for Behaviour<Types> {
fn inject_event(&mut self, event: PingEvent) {
use libp2p::ping::handler::{PingFailure, PingSuccess};
use libp2p::ping::{PingFailure, PingSuccess};
match event {
PingEvent {
peer,
@ -409,6 +411,12 @@ impl<Types: IpfsTypes> NetworkBehaviourEventProcess<PingEvent> for Behaviour<Typ
} => {
error!("ping: failure with {}: {}", peer.to_base58(), error);
}
PingEvent {
peer,
result: Result::Err(PingFailure::Unsupported),
} => {
error!("ping: failure with {}: unsupported", peer.to_base58());
}
}
}
}
@ -419,6 +427,12 @@ impl<Types: IpfsTypes> NetworkBehaviourEventProcess<IdentifyEvent> for Behaviour
}
}
impl<Types: IpfsTypes> NetworkBehaviourEventProcess<FloodsubEvent> for Behaviour<Types> {
fn inject_event(&mut self, event: FloodsubEvent) {
trace!("floodsub: {:?}", event);
}
}
impl<Types: IpfsTypes> Behaviour<Types> {
/// Create a Kademlia behaviour with the IPFS bootstrap nodes.
pub async fn new(options: SwarmOptions, repo: Arc<Repo<Types>>) -> Self {
@ -580,7 +594,7 @@ impl<Types: IpfsTypes> Behaviour<Types> {
pub fn dht_get(&mut self, key: Key, quorum: Quorum) -> SubscriptionFuture<KadResult, String> {
self.kad_subscriptions
.create_subscription(self.kademlia.get_record(&key, quorum).into(), None)
.create_subscription(self.kademlia.get_record(key, quorum).into(), None)
}
pub fn dht_put(

2
src/p2p/mod.rs
View File

@ -37,7 +37,7 @@ pub struct SwarmOptions {
impl From<&IpfsOptions> for SwarmOptions {
fn from(options: &IpfsOptions) -> Self {
let keypair = options.keypair.clone();
let peer_id = keypair.public().into_peer_id();
let peer_id = keypair.public().to_peer_id();
let bootstrap = options.bootstrap.clone();
let mdns = options.mdns;
let kad_protocol = options.kad_protocol.clone();

76
src/p2p/pubsub.rs
View File

@ -12,7 +12,9 @@ use libp2p::core::{
Multiaddr, PeerId,
};
use libp2p::floodsub::{Floodsub, FloodsubConfig, FloodsubEvent, FloodsubMessage, Topic};
use libp2p::swarm::{NetworkBehaviour, NetworkBehaviourAction, PollParameters, ProtocolsHandler};
use libp2p::swarm::{
ConnectionHandler, DialError, NetworkBehaviour, NetworkBehaviourAction, PollParameters,
};
/// Currently a thin wrapper around Floodsub, perhaps supporting both Gossipsub and Floodsub later.
/// Allows single subscription to a topic with only unbounded senders. Tracks the peers subscribed
@ -233,15 +235,16 @@ impl Pubsub {
}
type PubsubNetworkBehaviourAction = NetworkBehaviourAction<
<<Pubsub as NetworkBehaviour>::ProtocolsHandler as ProtocolsHandler>::InEvent,
<Pubsub as NetworkBehaviour>::OutEvent,
<Floodsub as NetworkBehaviour>::OutEvent,
<Pubsub as NetworkBehaviour>::ConnectionHandler,
<<Pubsub as NetworkBehaviour>::ConnectionHandler as ConnectionHandler>::InEvent,
>;
impl NetworkBehaviour for Pubsub {
type ProtocolsHandler = <Floodsub as NetworkBehaviour>::ProtocolsHandler;
type OutEvent = void::Void;
type ConnectionHandler = <Floodsub as NetworkBehaviour>::ConnectionHandler;
type OutEvent = FloodsubEvent;
fn new_handler(&mut self) -> Self::ProtocolsHandler {
fn new_handler(&mut self) -> Self::ConnectionHandler {
self.floodsub.new_handler()
}
@ -249,55 +252,56 @@ impl NetworkBehaviour for Pubsub {
self.floodsub.addresses_of_peer(peer_id)
}
fn inject_connected(&mut self, peer_id: &PeerId) {
self.floodsub.inject_connected(peer_id)
}
fn inject_disconnected(&mut self, peer_id: &PeerId) {
self.floodsub.inject_disconnected(peer_id)
}
fn inject_connection_established(
&mut self,
peer_id: &PeerId,
connection_id: &ConnectionId,
connected_point: &ConnectedPoint,
endpoint: &ConnectedPoint,
failed_addresses: Option<&Vec<Multiaddr>>,
other_established: usize,
) {
self.floodsub
.inject_connection_established(peer_id, connection_id, connected_point)
self.floodsub.inject_connection_established(
peer_id,
connection_id,
endpoint,
failed_addresses,
other_established,
)
}
fn inject_connection_closed(
&mut self,
peer_id: &PeerId,
connection_id: &ConnectionId,
connected_point: &ConnectedPoint,
endpoint: &ConnectedPoint,
handler: Self::ConnectionHandler,
remaining_established: usize,
) {
self.floodsub
.inject_connection_closed(peer_id, connection_id, connected_point)
self.floodsub.inject_connection_closed(
peer_id,
connection_id,
endpoint,
handler,
remaining_established,
)
}
fn inject_event(
&mut self,
peer_id: PeerId,
connection: ConnectionId,
event: <Self::ProtocolsHandler as ProtocolsHandler>::OutEvent,
event: <Self::ConnectionHandler as ConnectionHandler>::OutEvent,
) {
self.floodsub.inject_event(peer_id, connection, event)
}
fn inject_addr_reach_failure(
fn inject_dial_failure(
&mut self,
peer_id: Option<&PeerId>,
addr: &Multiaddr,
error: &dyn std::error::Error,
peer_id: Option<PeerId>,
handler: Self::ConnectionHandler,
error: &DialError,
) {
self.floodsub
.inject_addr_reach_failure(peer_id, addr, error)
}
fn inject_dial_failure(&mut self, peer_id: &PeerId) {
self.floodsub.inject_dial_failure(peer_id)
self.floodsub.inject_dial_failure(peer_id, handler, error)
}
fn inject_new_listen_addr(&mut self, id: ListenerId, addr: &Multiaddr) {
@ -378,7 +382,9 @@ impl NetworkBehaviour for Pubsub {
peer_id,
topic,
}) => {
self.add_node_to_partial_view(peer_id);
let topics = self.peers.entry(peer_id).or_insert_with(Vec::new);
let appeared = topics.is_empty();
if !topics.contains(&topic) {
@ -403,16 +409,14 @@ impl NetworkBehaviour for Pubsub {
if topics.is_empty() {
debug!("peer disappeared as pubsub subscriber: {}", peer_id);
oe.remove();
self.remove_node_from_partial_view(&peer_id);
}
}
continue;
}
NetworkBehaviourAction::DialAddress { address } => {
return Poll::Ready(NetworkBehaviourAction::DialAddress { address });
}
NetworkBehaviourAction::DialPeer { peer_id, condition } => {
return Poll::Ready(NetworkBehaviourAction::DialPeer { peer_id, condition });
action @ NetworkBehaviourAction::Dial { .. } => {
return Poll::Ready(action);
}
NetworkBehaviourAction::NotifyHandler {
peer_id,

174
src/p2p/swarm.rs
View File

@ -2,10 +2,12 @@ use crate::p2p::{MultiaddrWithPeerId, MultiaddrWithoutPeerId};
use crate::subscription::{SubscriptionFuture, SubscriptionRegistry};
use core::task::{Context, Poll};
use libp2p::core::{connection::ConnectionId, ConnectedPoint, Multiaddr, PeerId};
use libp2p::swarm::protocols_handler::{
DummyProtocolsHandler, IntoProtocolsHandler, ProtocolsHandler,
use libp2p::swarm::handler::DummyConnectionHandler;
use libp2p::swarm::{
self,
dial_opts::{DialOpts, PeerCondition},
ConnectionHandler, DialError, NetworkBehaviour, PollParameters, Swarm,
};
use libp2p::swarm::{self, DialPeerCondition, NetworkBehaviour, PollParameters, Swarm};
use std::collections::{hash_map::Entry, HashMap, HashSet, VecDeque};
use std::convert::{TryFrom, TryInto};
use std::time::Duration;
@ -33,7 +35,10 @@ impl Disconnector {
}
// Currently this is swarm::NetworkBehaviourAction<Void, Void>
type NetworkBehaviourAction = swarm::NetworkBehaviourAction<<<<SwarmApi as NetworkBehaviour>::ProtocolsHandler as IntoProtocolsHandler>::Handler as ProtocolsHandler>::InEvent, <SwarmApi as NetworkBehaviour>::OutEvent>;
type NetworkBehaviourAction = swarm::NetworkBehaviourAction<
<<SwarmApi as NetworkBehaviour>::ConnectionHandler as ConnectionHandler>::OutEvent,
<SwarmApi as NetworkBehaviour>::ConnectionHandler,
>;
#[derive(Debug, Default)]
pub struct SwarmApi {
@ -106,11 +111,14 @@ impl SwarmApi {
.connect_registry
.create_subscription(addr.clone().into(), None);
self.events.push_back(NetworkBehaviourAction::DialPeer {
peer_id: addr.peer_id,
let handler = self.new_handler();
self.events.push_back(NetworkBehaviourAction::Dial {
// rationale: this is sort of explicit command, perhaps the old address is no longer
// valid. Always would be even better but it's bugged at the moment.
condition: DialPeerCondition::NotDialing,
opts: DialOpts::peer_id(addr.peer_id)
.condition(PeerCondition::NotDialing)
.build(),
handler,
});
self.pending_addresses
@ -140,10 +148,10 @@ impl SwarmApi {
}
impl NetworkBehaviour for SwarmApi {
type ProtocolsHandler = DummyProtocolsHandler;
type ConnectionHandler = DummyConnectionHandler;
type OutEvent = void::Void;
fn new_handler(&mut self) -> Self::ProtocolsHandler {
fn new_handler(&mut self) -> Self::ConnectionHandler {
Default::default()
}
@ -165,12 +173,14 @@ impl NetworkBehaviour for SwarmApi {
fn inject_connection_established(
&mut self,
peer_id: &PeerId,
_id: &ConnectionId,
cp: &ConnectedPoint,
_connection_id: &ConnectionId,
endpoint: &ConnectedPoint,
_failed_addresses: Option<&Vec<Multiaddr>>,
_other_established: usize,
) {
// TODO: could be that the connection is not yet fully established at this point
trace!("inject_connection_established {} {:?}", peer_id, cp);
let addr = connection_point_addr(cp);
trace!("inject_connection_established {} {:?}", peer_id, endpoint);
let addr = connection_point_addr(endpoint);
self.peers.insert(*peer_id);
let connections = self.connected_peers.entry(*peer_id).or_default();
@ -185,7 +195,11 @@ impl NetworkBehaviour for SwarmApi {
);
}
if let ConnectedPoint::Dialer { address } = cp {
if let ConnectedPoint::Dialer {
address,
role_override: _,
} = endpoint
{
// we dialed to the `address`
match self.pending_connections.entry(*peer_id) {
Entry::Occupied(mut oe) => {
@ -211,9 +225,7 @@ impl NetworkBehaviour for SwarmApi {
}
}
}
}
fn inject_connected(&mut self, peer_id: &PeerId) {
// we have at least one fully open connection and handler is running
//
// just finish all of the subscriptions that remain.
@ -247,10 +259,12 @@ impl NetworkBehaviour for SwarmApi {
&mut self,
peer_id: &PeerId,
_id: &ConnectionId,
cp: &ConnectedPoint,
endpoint: &ConnectedPoint,
_handler: Self::ConnectionHandler,
_remaining_established: usize,
) {
trace!("inject_connection_closed {} {:?}", peer_id, cp);
let closed_addr = connection_point_addr(cp);
trace!("inject_connection_closed {} {:?}", peer_id, endpoint);
let closed_addr = connection_point_addr(endpoint);
match self.connected_peers.entry(*peer_id) {
Entry::Occupied(mut oe) => {
@ -277,7 +291,7 @@ impl NetworkBehaviour for SwarmApi {
closed_addr
);
if let ConnectedPoint::Dialer { .. } = cp {
if let ConnectedPoint::Dialer { .. } = endpoint {
let addr = MultiaddrWithPeerId::from((closed_addr, peer_id.to_owned()));
match self.pending_connections.entry(*peer_id) {
@ -290,7 +304,7 @@ impl NetworkBehaviour for SwarmApi {
// this needs to be guarded, so that the connect test case doesn't cause a
// panic following inject_connection_established, inject_connection_closed
// if there's only the DummyProtocolsHandler, which doesn't open a
// if there's only the DummyConnectionHandler, which doesn't open a
// substream and closes up immediatedly.
self.connect_registry.finish_subscription(
addr.into(),
@ -310,76 +324,61 @@ impl NetworkBehaviour for SwarmApi {
}
}
fn inject_disconnected(&mut self, peer_id: &PeerId) {
trace!("inject_disconnected: {}", peer_id);
assert!(!self.connected_peers.contains_key(peer_id));
self.roundtrip_times.remove(peer_id);
let failed = self
.pending_addresses
.remove(peer_id)
.unwrap_or_default()
.into_iter()
.chain(
self.pending_connections
.remove(peer_id)
.unwrap_or_default()
.into_iter(),
);
for addr in failed {
self.connect_registry
.finish_subscription(addr.into(), Err("disconnected".into()));
}
}
fn inject_event(&mut self, _peer_id: PeerId, _connection: ConnectionId, _event: void::Void) {}
fn inject_dial_failure(&mut self, peer_id: &PeerId) {
trace!("inject_dial_failure: {}", peer_id);
if self.pending_addresses.contains_key(peer_id) {
// it is possible that these addresses have not been tried yet; they will be asked
// for soon.
self.events
.push_back(swarm::NetworkBehaviourAction::DialPeer {
peer_id: *peer_id,
condition: DialPeerCondition::NotDialing,
});
}
// this should not be executed once, but probably will be in case unsupported addresses or something
// surprising happens.
for failed in self.pending_connections.remove(peer_id).unwrap_or_default() {
self.connect_registry
.finish_subscription(failed.into(), Err("addresses exhausted".into()));
}
}
fn inject_addr_reach_failure(
fn inject_dial_failure(
&mut self,
peer_id: Option<&PeerId>,
addr: &Multiaddr,
error: &dyn std::error::Error,
peer_id: Option<PeerId>,
_handler: Self::ConnectionHandler,
error: &DialError,
) {
trace!("inject_addr_reach_failure {} {}", addr, error);
// TODO: there might be additional connections we should attempt
// (i.e) a new MultiAddr was found after sending the existing ones
// off to dial
if let Some(peer_id) = peer_id {
match self.pending_connections.entry(*peer_id) {
match self.pending_connections.entry(peer_id) {
Entry::Occupied(mut oe) => {
let addresses = oe.get_mut();
let addr = MultiaddrWithPeerId::try_from(addr.clone())
.expect("dialed address contains peerid in libp2p 0.38");
let pos = addresses.iter().position(|a| *a == addr);
if let Some(pos) = pos {
addresses.swap_remove(pos);
self.connect_registry
.finish_subscription(addr.into(), Err(error.to_string()));
match error {
DialError::Transport(multiaddrs) => {
for (addr, error) in multiaddrs {
let addr = MultiaddrWithPeerId::try_from(addr.clone())
.expect("to recieve an MultiAddrWithPeerId from DialError");
self.connect_registry.finish_subscription(
addr.clone().into(),
Err(error.to_string()),
);
if let Some(pos) = addresses.iter().position(|a| *a == addr) {
addresses.swap_remove(pos);
}
}
}
DialError::WrongPeerId { .. } => {
for addr in addresses.iter() {
self.connect_registry.finish_subscription(
addr.clone().into(),
Err(error.to_string()),
);
}
addresses.clear();
}
error => {
warn!(
?error,
"unexpected DialError; some futures might never complete"
);
}
}
if addresses.is_empty() {
oe.remove();
}
// FIXME from libp2p-0.43 upgrade: unclear if there could be a need for new
// dial attempt if new entries to self.pending_addresses arrived.
}
Entry::Vacant(_) => {}
}
@ -401,7 +400,10 @@ impl NetworkBehaviour for SwarmApi {
fn connection_point_addr(cp: &ConnectedPoint) -> MultiaddrWithoutPeerId {
match cp {
ConnectedPoint::Dialer { address } => MultiaddrWithPeerId::try_from(address.to_owned())
ConnectedPoint::Dialer {
address,
role_override: _,
} => MultiaddrWithPeerId::try_from(address.to_owned())
.expect("dialed address contains peerid in libp2p 0.38")
.into(),
ConnectedPoint::Listener { send_back_addr, .. } => send_back_addr
@ -451,11 +453,13 @@ mod tests {
loop {
tokio::select! {
biased;
_ = (&mut swarm1).next() => {},
_ = (&mut swarm2).next() => {},
res = (&mut sub) => {
// this is currently a success even though the connection is never really
// established, the DummyProtocolsHandler doesn't do anything nor want the
// established, the DummyConnectionHandler doesn't do anything nor want the
// connection to be kept alive and thats it.
//
// it could be argued that this should be `Err("keepalive disconnected")`
@ -482,10 +486,10 @@ mod tests {
#[tokio::test]
async fn wrong_peerid() {
let (_, mut swarm1) = build_swarm();
let (swarm1_peerid, mut swarm1) = build_swarm();
let (_, mut swarm2) = build_swarm();
let peer3_id = Keypair::generate_ed25519().public().into_peer_id();
let peer3_id = Keypair::generate_ed25519().public().to_peer_id();
Swarm::listen_on(&mut swarm1, "/ip4/127.0.0.1/tcp/0".parse().unwrap()).unwrap();
@ -515,7 +519,9 @@ mod tests {
_ = swarm1.next() => {},
_ = swarm2.next() => {},
res = &mut fut => {
assert_eq!(res.unwrap_err(), Some("Pending connection: Invalid peer ID.".into()));
let err = res.unwrap_err().unwrap();
let expected_start = format!("Dial error: Unexpected peer ID {}", swarm1_peerid);
assert_eq!(&err[0..expected_start.len()], expected_start);
return;
}
}
@ -581,7 +587,7 @@ mod tests {
fn build_swarm() -> (PeerId, libp2p::swarm::Swarm<SwarmApi>) {
let key = Keypair::generate_ed25519();
let peer_id = key.public().into_peer_id();
let peer_id = key.public().to_peer_id();
let transport = build_transport(key).unwrap();
let swarm = SwarmBuilder::new(transport, SwarmApi::default(), peer_id)

163
tests/pubsub.rs
View File

@ -48,10 +48,8 @@ async fn can_publish_without_subscribing() {
}
#[tokio::test]
#[allow(clippy::mutable_key_type)] // clippy doesn't like Vec inside HashSet
async fn publish_between_two_nodes() {
async fn publish_between_two_nodes_single_topic() {
use futures::stream::StreamExt;
use std::collections::HashSet;
let nodes = spawn_nodes(2, Topology::Line).await;
@ -98,26 +96,50 @@ async fn publish_between_two_nodes() {
// the order is not defined, but both should see the other's message and the message they sent
let expected = [
(&[topic.clone()], &nodes[0].id, b"foobar"),
(&[topic.clone()], &nodes[1].id, b"barfoo"),
// first node should witness it's the message it sent
(&[topic.clone()], nodes[0].id, b"foobar", nodes[0].id),
// second node should witness first nodes message, and so on.
(&[topic.clone()], nodes[0].id, b"foobar", nodes[1].id),
(&[topic.clone()], nodes[1].id, b"barfoo", nodes[0].id),
(&[topic.clone()], nodes[1].id, b"barfoo", nodes[1].id),
]
.iter()
.cloned()
.map(|(topics, id, data)| (topics.to_vec(), *id, data.to_vec()))
.collect::<HashSet<_>>();
.map(|(topics, sender, data, witness)| (topics.to_vec(), sender, data.to_vec(), witness))
.collect::<Vec<_>>();
for st in &mut [b_msgs.by_ref(), a_msgs.by_ref()] {
let actual = st
.take(2)
// Arc::try_unwrap will fail sometimes here as the sender side in src/p2p/pubsub.rs:305
// can still be looping
.map(|msg| (*msg).clone())
.map(|msg| (msg.topics, msg.source, msg.data))
.collect::<HashSet<_>>()
.await;
assert_eq!(expected, actual);
let mut actual = Vec::new();
for (st, own_peer_id) in &mut [
(b_msgs.by_ref(), nodes[1].id),
(a_msgs.by_ref(), nodes[0].id),
] {
let received = timeout(
Duration::from_secs(2),
st.take(2)
// Arc::try_unwrap will fail sometimes here as the sender side in src/p2p/pubsub.rs:305
// can still be looping
.map(|msg| (*msg).clone())
.map(|msg| (msg.topics, msg.source, msg.data, *own_peer_id))
.collect::<Vec<_>>(),
)
.await
.unwrap();
actual.extend(received);
}
// sort the received messages both in expected and actual to make sure they are comparable;
// order of receiving is not part of the tuple and shouldn't matter.
let mut expected = expected;
expected.sort_unstable();
actual.sort_unstable();
assert_eq!(
actual, expected,
"sent and received messages must be present on both nodes' streams"
);
drop(b_msgs);
let mut disappeared = false;
@ -139,6 +161,113 @@ async fn publish_between_two_nodes() {
assert!(disappeared, "timed out before a saw b's unsubscription");
}
#[tokio::test]
async fn publish_between_two_nodes_different_topics() {
use futures::stream::StreamExt;
let nodes = spawn_nodes(2, Topology::Line).await;
let node_a = &nodes[0];
let node_b = &nodes[1];
let topic_a = "shared-a".to_owned();
let topic_b = "shared-b".to_owned();
// Node A subscribes to Topic B
// Node B subscribes to Topic A
let mut a_msgs = node_a.pubsub_subscribe(topic_b.clone()).await.unwrap();
let mut b_msgs = node_b.pubsub_subscribe(topic_a.clone()).await.unwrap();
// need to wait to see both sides so that the messages will get through
let mut appeared = false;
for _ in 0..100usize {
if node_a
.pubsub_peers(Some(topic_a.clone()))
.await
.unwrap()
.contains(&node_b.id)
&& node_b
.pubsub_peers(Some(topic_b.clone()))
.await
.unwrap()
.contains(&node_a.id)
{
appeared = true;
break;
}
timeout(Duration::from_millis(100), pending::<()>())
.await
.unwrap_err();
}
assert!(
appeared,
"timed out before both nodes appeared as pubsub peers"
);
// Each node publishes to their own topic
node_a
.pubsub_publish(topic_a.clone(), b"foobar".to_vec())
.await
.unwrap();
node_b
.pubsub_publish(topic_b.clone(), b"barfoo".to_vec())
.await
.unwrap();
// the order between messages is not defined, but both should see the other's message. since we
// receive messages first from node_b's stream we expect this order.
//
// in this test case the nodes are not expected to see their own message because nodes are not
// subscribing to the streams they are sending to.
let expected = [
(&[topic_a.clone()], node_a.id, b"foobar", node_b.id),
(&[topic_b.clone()], node_b.id, b"barfoo", node_a.id),
]
.iter()
.cloned()
.map(|(topics, sender, data, witness)| (topics.to_vec(), sender, data.to_vec(), witness))
.collect::<Vec<_>>();
let mut actual = Vec::new();
for (st, own_peer_id) in &mut [(b_msgs.by_ref(), node_b.id), (a_msgs.by_ref(), node_a.id)] {
let received = timeout(
Duration::from_secs(2),
st.take(1)
.map(|msg| (*msg).clone())
.map(|msg| (msg.topics, msg.source, msg.data, *own_peer_id))
.next(),
)
.await
.unwrap()
.unwrap();
actual.push(received);
}
// ordering is defined for expected and actual by the order of the looping above and the
// initial expected creation.
assert_eq!(expected, actual);
drop(b_msgs);
let mut disappeared = false;
for _ in 0..100usize {
if !node_a
.pubsub_peers(Some(topic_a.clone()))
.await
.unwrap()
.contains(&node_b.id)
{
disappeared = true;
break;
}
timeout(Duration::from_millis(100), pending::<()>())
.await
.unwrap_err();
}
assert!(disappeared, "timed out before a saw b's unsubscription");
}
#[cfg(any(feature = "test_go_interop", feature = "test_js_interop"))]
#[tokio::test]
#[ignore = "doesn't work yet"]