Chat example report Publish error: Publish error: InsufficientPeers with peers already discovered

[EMUNES]

Hi all. I'm trying to learn about libp2p by the chat example. I have running the example in two terminals, they can find each other but they will report Publish error like below:

mDNS discovered a new peer: 12D3KooWCyV65nPqHkRGEG3j4eTbbCvJM5hdVADMZ3Mwavu2RgCk
mDNS discovered a new peer: 12D3KooWCyV65nPqHkRGEG3j4eTbbCvJM5hdVADMZ3Mwavu2RgCk
mDNS discovered a new peer: 12D3KooWCyV65nPqHkRGEG3j4eTbbCvJM5hdVADMZ3Mwavu2RgCk
mDNS discovered a new peer: 12D3KooWCyV65nPqHkRGEG3j4eTbbCvJM5hdVADMZ3Mwavu2RgCk
mDNS discovered a new peer: 12D3KooWCyV65nPqHkRGEG3j4eTbbCvJM5hdVADMZ3Mwavu2RgCk
mDNS discovered a new peer: 12D3KooWCyV65nPqHkRGEG3j4eTbbCvJM5hdVADMZ3Mwavu2RgCk
df
Publish error: InsufficientPeers
sd
Publish error: InsufficientPeers
asd
Publish error: InsufficientPeers

I'm confused about why they can discover each other but still shows InsufficientPeers. Anyone has clues?

I'm running on Windows 11, libp2p 0.53.

[mxinden]

Do all of these peers join the same topic? Are they added to each others mesh?

If you want to dig deeper:

rust-libp2p/protocols/gossipsub/src/behaviour.rs

Lines 603 to 763 in 7f4ba69

	/// Publishes a message with multiple topics to the network.
	pub fn publish(
	&mut self,
	topic: impl Into<TopicHash>,
	data: impl Into<Vec<u8>>,
	) -> Result<MessageId, PublishError> {
	let data = data.into();
	let topic = topic.into();
	// Transform the data before building a raw_message.
	let transformed_data = self
	.data_transform
	.outbound_transform(&topic, data.clone())?;
	let raw_message = self.build_raw_message(topic, transformed_data)?;
	// calculate the message id from the un-transformed data
	let msg_id = self.config.message_id(&Message {
	source: raw_message.source,
	data, // the uncompressed form
	sequence_number: raw_message.sequence_number,
	topic: raw_message.topic.clone(),
	});
	let event = Rpc {
	subscriptions: Vec::new(),
	messages: vec![raw_message.clone()],
	control_msgs: Vec::new(),
	}
	.into_protobuf();
	// check that the size doesn't exceed the max transmission size
	if event.get_size() > self.config.max_transmit_size() {
	return Err(PublishError::MessageTooLarge);
	}
	// Check the if the message has been published before
	if self.duplicate_cache.contains(&msg_id) {
	// This message has already been seen. We don't re-publish messages that have already
	// been published on the network.
	tracing::warn!(
	message=%msg_id,
	"Not publishing a message that has already been published"
	);
	return Err(PublishError::Duplicate);
	}
	tracing::trace!(message=%msg_id, "Publishing message");
	let topic_hash = raw_message.topic.clone();
	// If we are not flood publishing forward the message to mesh peers.
	let mesh_peers_sent = !self.config.flood_publish()
	&& self.forward_msg(&msg_id, raw_message.clone(), None, HashSet::new())?;
	let mut recipient_peers = HashSet::new();
	if let Some(set) = self.topic_peers.get(&topic_hash) {
	if self.config.flood_publish() {
	// Forward to all peers above score and all explicit peers
	recipient_peers.extend(
	set.iter()
	.filter(|p| {
	self.explicit_peers.contains(*p)
	|| !self.score_below_threshold(p, |ts| ts.publish_threshold).0
	})
	.cloned(),
	);
	} else {
	// Explicit peers
	for peer in &self.explicit_peers {
	if set.contains(peer) {
	recipient_peers.insert(*peer);
	}
	}
	// Floodsub peers
	for (peer, connections) in &self.connected_peers {
	if connections.kind == PeerKind::Floodsub
	&& !self
	.score_below_threshold(peer, |ts| ts.publish_threshold)
	.0
	{
	recipient_peers.insert(*peer);
	}
	}
	// Gossipsub peers
	if self.mesh.get(&topic_hash).is_none() {
	tracing::debug!(topic=%topic_hash, "Topic not in the mesh");
	// If we have fanout peers add them to the map.
	if self.fanout.contains_key(&topic_hash) {
	for peer in self.fanout.get(&topic_hash).expect("Topic must exist") {
	recipient_peers.insert(*peer);
	}
	} else {
	// We have no fanout peers, select mesh_n of them and add them to the fanout
	let mesh_n = self.config.mesh_n();
	let new_peers = get_random_peers(
	&self.topic_peers,
	&self.connected_peers,
	&topic_hash,
	mesh_n,
	{
	|p| {
	!self.explicit_peers.contains(p)
	&& !self
	.score_below_threshold(p, |pst| pst.publish_threshold)
	.0
	}
	},
	);
	// Add the new peers to the fanout and recipient peers
	self.fanout.insert(topic_hash.clone(), new_peers.clone());
	for peer in new_peers {
	tracing::debug!(%peer, "Peer added to fanout");
	recipient_peers.insert(peer);
	}
	}
	// We are publishing to fanout peers - update the time we published
	self.fanout_last_pub
	.insert(topic_hash.clone(), Instant::now());
	}
	}
	}
	if recipient_peers.is_empty() && !mesh_peers_sent {
	return Err(PublishError::InsufficientPeers);
	}
	// If the message isn't a duplicate and we have sent it to some peers add it to the
	// duplicate cache and memcache.
	self.duplicate_cache.insert(msg_id.clone());
	self.mcache.put(&msg_id, raw_message);
	// If the message is anonymous or has a random author add it to the published message ids
	// cache.
	if let PublishConfig::RandomAuthor | PublishConfig::Anonymous = self.publish_config {
	if !self.config.allow_self_origin() {
	self.published_message_ids.insert(msg_id.clone());
	}
	}
	// Send to peers we know are subscribed to the topic.
	let msg_bytes = event.get_size();
	for peer_id in recipient_peers.iter() {
	tracing::trace!(peer=%peer_id, "Sending message to peer");
	self.send_message(*peer_id, event.clone())?;
	if let Some(m) = self.metrics.as_mut() {
	m.msg_sent(&topic_hash, msg_bytes);
	}
	}
	tracing::debug!(message=%msg_id, "Published message");
	if let Some(metrics) = self.metrics.as_mut() {
	metrics.register_published_message(&topic_hash);
	}
	Ok(msg_id)
	}

[EMUNES]

Thank you, after making sure the topic and mesh are working in the example (it is mostly copied from the examples folder). I try to play with the code and find the reason it won't work: I have removed the tracing code:

    let _ = tracing_subscriber::fmt()
        .with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
        .try_init();

If I add those code, the chat example works as expected.

I'm puzzled why would tracing could actually affect how gossipsub works. I assume it should works fine even if I don't invoke tracing but it seems critical in this case. It is designed to must invoke tracing with libp2p?

[cyrus937]

I have the same problem.
I haven't changed the chat example, but sending messages doesn't work. Also, there's a latency in discovering peers.
On the first run, everything worked fine, but since then I've had this InsufficientPeers error

[thomaseizinger]

Confirmed that the example is currently broken. Fix in #4840. Sorry about that folks!