Handling Duplicate Messages in libp2p with gossipsub

[vinay10949]

Description:

I'm using libp2p with gossipsub for message publication and subscription in my application. I have configured gossipsub with a duplicate cache time of 1 second, intending to allow duplicate messages after this interval. However, I'm encountering a "Duplicate" error when publishing messages even after the specified time interval has passed.

Code Snippet:

let gossipsub_config = gossipsub::ConfigBuilder::default()
    .heartbeat_interval(Duration::from_secs(10))
    .validation_mode(gossipsub::ValidationMode::Strict)
    .duplicate_cache_time(Duration::from_secs(1))
    .published_message_ids_cache_time(Duration::from_secs(1))
    .message_id_fn(message_id_fn)
    .build()
    .map_err(|msg| io::Error::new(io::ErrorKind::Other, msg))?;

Questions:

1. What settings or configurations do I need to adjust to properly handle duplicate messages after the specified time interval?
2. Are there any additional considerations or configurations related to duplicate message handling in libp2p with gossipsub that I might be missing?
3. Also How do I need to make any specific changes for gossip sub or anything in Kademlia Behaviour to allow public ipaddress for communication.

[jxs]

Hi, have you seen #2587 (comment) ?
But basically, when publishing via gossipsub, it checks on DuplicateCache if contains the entry:

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

Lines 607 to 616 in ab4f664

	// 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);
	}

but DuplicateCache only removes stale entries when you insert a new message:

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

Lines 172 to 179 in ab4f664

	pub(crate) fn insert(&mut self, key: Key) -> bool {
	if let Entry::Vacant(entry) = self.0.entry(key) {
	entry.insert(());
	true
	} else {
	false
	}
	}

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

Lines 133 to 144 in ab4f664

	pub(crate) fn entry(&mut self, key: Key) -> Entry<Key, Value> {
	let now = Instant::now();
	self.remove_expired_keys(now);
	match self.map.entry(key) {
	Occupied(entry) => Entry::Occupied(OccupiedEntry { entry }),
	Vacant(entry) => Entry::Vacant(VacantEntry {
	expiration: now + self.ttl,
	entry,
	list: &mut self.list,
	}),
	}
	}

not when contains() is called:

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

Lines 181 to 184 in ab4f664

	pub(crate) fn contains(&self, key: &Key) -> bool {
	self.0.contains_key(key)
	}
	}

I'd say you are probably trying to send the same message sequentially

[AgeManning]

What settings or configurations do I need to adjust to properly handle duplicate messages after the specified time interval?

Duplicates on gossipsub should be expressly forbidden. In an ideal world, gossipsub should never permit duplicates, however we are limited in resources (i.e RAM) such that we can't remember all messages for an infinite amount of time. Given these constraints, sometimes a duplicate can get through, in Rust, we loosely time-bound the message. The time-bound ensures that no duplicates will happen within that time (i.e 1 second) but may prevent duplicates longer. With unlimited resources, we'd set this time limit to infinity.

This begs the question, why do you want duplicates? Duplicates are a source of amplification and excess bandwidth use in gossipsub. We don't want old messages permanently being slowly bouncing around the network and re-transmitted forever, so having a large enough cache prevents this.

A message is determined by its message-id. In your code-snippet you have changed the default message to something else but have not said what it is. The default message-id has a sequence number associated with it. So if you had a chat app, and wanted to send the message "hey" and then send "hey" again immediately after, this would be fine. These messages are not duplicates because the message-id will be different. If you have a message whose contents need to be re-transmitted regularly, it sounds like your message-id is not correctly built for it.

Are there any additional considerations or configurations related to duplicate message handling in libp2p with gossipsub that I might be missing?

Yep, sounds like you need to change your message-id so that it is unique per message you want to transmit.

Also How do I need to make any specific changes for gossip sub or anything in Kademlia Behaviour to allow public ipaddress for communication.

I don't really follow this. Gossipsub works fine with public addresses. This is part of libp2p, you can dial any public multiaddr. I think there is minimal difference in libp2p between public/private addresses.

[dariusc93]

To add to above, the default message_id_fn does generate a unique message id

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

Lines 389 to 402 in b890b29

	message_id_fn: Arc::new(|message| {
	// default message id is: source + sequence number
	// NOTE: If either the peer_id or source is not provided, we set to 0;
	let mut source_string = if let Some(peer_id) = message.source.as_ref() {
	peer_id.to_base58()
	} else {
	PeerId::from_bytes(&[0, 1, 0])
	.expect("Valid peer id")
	.to_base58()
	};
	source_string
	.push_str(&message.sequence_number.unwrap_or_default().to_string());
	MessageId::from(source_string)
	}),

However, if youre going based on the chat example the message id would likely be the same if the message is the same because it generates it from the hash

rust-libp2p/examples/chat/src/main.rs

Lines 55 to 59 in b890b29

	let message_id_fn = |message: &gossipsub::Message| {
	let mut s = DefaultHasher::new();
	message.data.hash(&mut s);
	gossipsub::MessageId::from(s.finish().to_string())
	};

I would suggest omitting that segment of code as well as the message_id_fn in the chain and see if that helps (though this is all assuming you are using the example. If you are using your own code to generate the id, you might want to factor in Message::sequence_number or some timestamp, though the default should be more than enough unless you have a use-case)