fix(gossipsub): gracefully disable handler on stream errors

[vnermolaev]

Description

Previously, we closed the entire connection upon receiving too many upgrade errors. This is unnecessarily aggressive. For example, an upgrade error may be caused by the remote dropping a stream during the initial handshake which is completely isolated from other protocols running on the same connection.

Instead of closing the connection, set KeepAlive::No.

Related: #3591.
Resolves: #3690.

Notes & open questions

Few items I'd like to raise.

• Pieces of code such as

  return Poll::Ready(ConnectionHandlerEvent::Close(...));

  are replaced with self.keep_alive = KeepAlive::No and the error report. Unfortunately, I cannot meaningfully return from the poll function because it returns ConnectionHamdlerEvent and not an Option thereof.
• Pieces of code such as

  loop {
      ...
      return Poll::Ready(ConnectionHandlerEvent::Close(...));
      ...
  }

  are replaced with self.keep_alive = KeepAlive::No and the error report. Again, it is impossible to return meaningfully; I just break; such an approach is also observed in the poll function.
  • sometimes errors are reported on the warn! level which is inconsistent with the proposed info! level

Change checklist

• I have performed a self-review of my own code
• I have made corresponding changes to the documentation
• I have added tests that prove my fix is effective or that my feature works
• A changelog entry has been made in the appropriate crates

[vnermolaev]

@thomaseizinger, Will you please comment on the open questions in the first message?

[thomaseizinger]

Meta-note: Please don't force push, it makes review unnecessarily difficult.

[thomaseizinger]

Nice, we are making progress in the right direction here.

[thomaseizinger]

Thanks!

Some more thoughts.

[thomaseizinger]

Thanks for pushing this forward!

[vnermolaev]

I did not expect to make harmonization changes. So in the first commits, I kept the surface as minimal as possible.

[thomaseizinger]

I did not expect to make harmonization changes. So in the first commits, I kept the surface as minimal as possible.

I figured while we are at it, might as well tidy up the code a bit :)

[thomaseizinger]

This is good from my end. I would like @mxinden to also have a look.

[mxinden]

Overall change looks good to me. Thanks for the work.

//CC @divagant-martian and @AgeManning to make sure this does not break any assumptions in lighthouse.

[mxinden]

@p-shahi and @2color this pull request should fix the issue of libp2p-gossipsub closing the connection due to an outbound stream upgrade timing out. Note that it does not fix the root cause, namely the outbound stream upgrade timing out. It will only prevent the closing of the connection.

libp2p/universal-connectivity#3

[thomaseizinger]

Thank you Max!

[thomaseizinger]

f999f3e splits the Handler into an EnabledHandler and a DisabledHandler making many invalid states impossible at compile time. In my eyes this makes it significantly easier to reason about the state of the handler and removes multiple edge cases. E.g. without this patch send_queue could have been growing unbounded in case the handler reached MAX_SUBSTREAM_CREATION but the gossipsub NetworkBehaviour continued pushing messages to the handler.

I know this is yet another large change to this already large pull request. To value everyone's time, @thomaseizinger would you mind taking a look first? Once this looks good to you we can ask for another larger round of reviews from other folks.

Yes, thank you for doing this. I think this makes a lot of sense. I am liking the direction this is going. Eventually, we might be able to add first-class support for "disabling" a handler to the swarm.

[mxinden]

@thomaseizinger can you give this pull request another review?

[thomaseizinger]

LGTM. Should we extract the changes around void and ConnectionEvent into separate PRs?

[thomaseizinger]

Suggested change

[thomaseizinger]

Could flatten this match into the outer one.

[thomaseizinger]

@vnermolaev I am sorry that we hijacked your PR like this 🙈

I hope you don't mind. We discovered several issues around handler errors with gossipsub and this PR helped us uncover those. Thanks for kicking it off and I hope it doesn't scare you back from contributing in the future :)

[divagant-martian]

I think this way it's possible we don't send the PeerKind::UnsupportedProtocol "on time" unless there is some kind of guarantee that poll will be called first (?). This event is important since we might want to get rid of (ban, or otherwise prevent connections to) this peer if it doesn't support what we need

[thomaseizinger]

A handler is always completed drained of its events first before we even ask whether or not it should be shut down.

[thomaseizinger]

See

rust-libp2p/swarm/src/connection.rs

Lines 216 to 303 in 06ea8ce

	loop {
	match requested_substreams.poll_next_unpin(cx) {
	Poll::Ready(Some(Ok(()))) => continue,
	Poll::Ready(Some(Err(info))) => {
	handler.on_connection_event(ConnectionEvent::DialUpgradeError(
	DialUpgradeError {
	info,
	error: ConnectionHandlerUpgrErr::Timeout,
	},
	));
	continue;
	}
	Poll::Ready(None) | Poll::Pending => {}
	}
	// Poll the [`ConnectionHandler`].
	match handler.poll(cx) {
	Poll::Pending => {}
	Poll::Ready(ConnectionHandlerEvent::OutboundSubstreamRequest { protocol }) => {
	let timeout = *protocol.timeout();
	let (upgrade, user_data) = protocol.into_upgrade();
	requested_substreams.push(SubstreamRequested::new(user_data, timeout, upgrade));
	continue; // Poll handler until exhausted.
	}
	Poll::Ready(ConnectionHandlerEvent::Custom(event)) => {
	return Poll::Ready(Ok(Event::Handler(event)));
	}
	Poll::Ready(ConnectionHandlerEvent::Close(err)) => {
	return Poll::Ready(Err(ConnectionError::Handler(err)));
	}
	}
	// In case the [`ConnectionHandler`] can not make any more progress, poll the negotiating outbound streams.
	match negotiating_out.poll_next_unpin(cx) {
	Poll::Pending | Poll::Ready(None) => {}
	Poll::Ready(Some((info, Ok(protocol)))) => {
	handler.on_connection_event(ConnectionEvent::FullyNegotiatedOutbound(
	FullyNegotiatedOutbound { protocol, info },
	));
	continue;
	}
	Poll::Ready(Some((info, Err(error)))) => {
	handler.on_connection_event(ConnectionEvent::DialUpgradeError(
	DialUpgradeError { info, error },
	));
	continue;
	}
	}
	// In case both the [`ConnectionHandler`] and the negotiating outbound streams can not
	// make any more progress, poll the negotiating inbound streams.
	match negotiating_in.poll_next_unpin(cx) {
	Poll::Pending | Poll::Ready(None) => {}
	Poll::Ready(Some((info, Ok(protocol)))) => {
	handler.on_connection_event(ConnectionEvent::FullyNegotiatedInbound(
	FullyNegotiatedInbound { protocol, info },
	));
	continue;
	}
	Poll::Ready(Some((info, Err(error)))) => {
	handler.on_connection_event(ConnectionEvent::ListenUpgradeError(
	ListenUpgradeError { info, error },
	));
	continue;
	}
	}
	// Ask the handler whether it wants the connection (and the handler itself)
	// to be kept alive, which determines the planned shutdown, if any.
	let keep_alive = handler.connection_keep_alive();
	match (&mut *shutdown, keep_alive) {
	(Shutdown::Later(timer, deadline), KeepAlive::Until(t)) => {
	if *deadline != t {
	*deadline = t;
	if let Some(dur) = deadline.checked_duration_since(Instant::now()) {
	timer.reset(dur)
	}
	}
	}
	(_, KeepAlive::Until(t)) => {
	if let Some(dur) = t.checked_duration_since(Instant::now()) {
	*shutdown = Shutdown::Later(Delay::new(dur), t)
	}
	}
	(_, KeepAlive::No) => *shutdown = Shutdown::Asap,
	(_, KeepAlive::Yes) => *shutdown = Shutdown::None,
	};

.

[mxinden]

Adding to the above, we first "drain" (i.e. poll) the ConnectionHandler till it returns Poll::Pending:

rust-libp2p/swarm/src/connection.rs

Lines 231 to 247 in 06ea8ce

	// Poll the [`ConnectionHandler`].
	match handler.poll(cx) {
	Poll::Pending => {}
	Poll::Ready(ConnectionHandlerEvent::OutboundSubstreamRequest { protocol }) => {
	let timeout = *protocol.timeout();
	let (upgrade, user_data) = protocol.into_upgrade();
	requested_substreams.push(SubstreamRequested::new(user_data, timeout, upgrade));
	continue; // Poll handler until exhausted.
	}
	Poll::Ready(ConnectionHandlerEvent::Custom(event)) => {
	return Poll::Ready(Ok(Event::Handler(event)));
	}
	Poll::Ready(ConnectionHandlerEvent::Close(err)) => {
	return Poll::Ready(Err(ConnectionError::Handler(err)));
	}
	}

Only then do we check the connection_keep_alive statsu:

rust-libp2p/swarm/src/connection.rs

Lines 284 to 303 in 06ea8ce

	// Ask the handler whether it wants the connection (and the handler itself)
	// to be kept alive, which determines the planned shutdown, if any.
	let keep_alive = handler.connection_keep_alive();
	match (&mut *shutdown, keep_alive) {
	(Shutdown::Later(timer, deadline), KeepAlive::Until(t)) => {
	if *deadline != t {
	*deadline = t;
	if let Some(dur) = deadline.checked_duration_since(Instant::now()) {
	timer.reset(dur)
	}
	}
	}
	(_, KeepAlive::Until(t)) => {
	if let Some(dur) = t.checked_duration_since(Instant::now()) {
	*shutdown = Shutdown::Later(Delay::new(dur), t)
	}
	}
	(_, KeepAlive::No) => *shutdown = Shutdown::Asap,
	(_, KeepAlive::Yes) => *shutdown = Shutdown::None,
	};

[mxinden]

@divagant-martian @AgeManning does either of you mind giving this another review?

[mxinden]

@p-shahi with #3765 and #3767 merged into master and master merged into this pull request, this pull request again should fix all the issues that you are facing on https://github.com/libp2p/universal-connectivity/ with rust-libp2p. Mind updating your fork of rust-libp2p to the latest commit on this branch?

[divagant-martian]

LGTM, management of handler state as an enum seems like a great code quality improvement on top, ty @mxinden

addressed with latest commits

[mxinden]

@Mergifyio refresh

[mergify]

refresh

✅ Pull request refreshed

Approvals have been dismissed because the PR was updated after the send-it label was applied.

[dariusc93]

@mxinden The changelog from this PR may have to get updated since libp2p-gossipsub 0.44.3 didnt include this change before 0.51.3 patch release.