feat(identify): allow updating agent_version at runtime

protocols/identify/CHANGELOG.md

@@ -1,5 +1,10 @@
 ## 0.48.0
 
+- Add `Behaviour::set_agent_version` to update the advertised agent version at runtime.
+  Established connections are updated in place; combine it with `Behaviour::push` to propagate
+  the new value immediately.
+  See [PR 6576](https://github.com/libp2p/rust-libp2p/pull/6576).
+
 - Use `futures-timer` instead of tokio's timer for stream timeouts so the bounded `Delay` works on
   `wasm32`; tokio's timer has no driver in the browser and panics at runtime.
   See [PR 6488](https://github.com/libp2p/rust-libp2p/pull/6488).

protocols/identify/src/behaviour.rs

@@ -295,6 +295,41 @@ impl Behaviour {
         }
     }
 
+    /// Updates the agent version advertised to remote peers.
+    ///
+    /// Existing connections are updated in place, i.e. the new value is used for the next
+    /// identify exchange on every established connection. Connections established from now
+    /// on pick it up as well.
+    ///
+    /// This does not send anything to remote peers on its own. Combine it with
+    /// [`Behaviour::push`] to propagate the change immediately, otherwise peers learn about
+    /// it with the next periodic identify exchange.
+    ///
+    /// Setting the currently advertised agent version is a no-op.
+    pub fn set_agent_version(&mut self, agent_version: String) {
+        if self.config.agent_version == agent_version {
+            return;
+        }
+
+        self.config.agent_version = agent_version;
+
+        // Notify all connected handlers about the changed agent version. This has to be
+        // `NotifyHandler::One` per connection: a peer may be reachable over several
+        // connections at once and each of them holds its own copy of the agent version.
+        let change_events = self
+            .connected
+            .iter()
+            .flat_map(|(peer, map)| map.keys().map(|id| (*peer, id)))
+            .map(|(peer_id, connection_id)| ToSwarm::NotifyHandler {
+                peer_id,
+                handler: NotifyHandler::One(*connection_id),
+                event: InEvent::AgentVersionChanged(self.config.agent_version.clone()),
+            })
+            .collect::<Vec<_>>();
+
+        self.events.extend(change_events);
+    }
+
     fn on_connection_established(
         &mut self,
         ConnectionEstablished {
@@ -754,4 +789,51 @@ mod tests {
         ));
         assert!(multiaddr_matches_peer_id(&addr_without_peer_id, &peer_id));
     }
+
+    #[test]
+    fn set_agent_version_does_not_notify_handlers_on_unchanged_value() {
+        let mut behaviour = Behaviour::new(
+            Config::new(
+                "/test/1.0.0".to_owned(),
+                Keypair::generate_ed25519().public(),
+            )
+            .with_agent_version("agent/1.0.0".to_owned()),
+        );
+
+        let endpoint = ConnectedPoint::Dialer {
+            address: "/ip4/127.0.0.1/tcp/4001".parse().unwrap(),
+            role_override: Endpoint::Dialer,
+            port_use: PortUse::Reuse,
+        };
+        behaviour.on_swarm_event(FromSwarm::ConnectionEstablished(ConnectionEstablished {
+            peer_id: PeerId::random(),
+            connection_id: ConnectionId::new_unchecked(0),
+            endpoint: &endpoint,
+            failed_addresses: &[],
+            other_established: 0,
+        }));
+        assert_eq!(notify_handler_events(&behaviour), 0);
+
+        behaviour.set_agent_version("agent/1.0.0".to_owned());
+        assert_eq!(
+            notify_handler_events(&behaviour),
+            0,
+            "setting the current agent version should not notify any handler"
+        );
+
+        behaviour.set_agent_version("agent/2.0.0".to_owned());
+        assert_eq!(
+            notify_handler_events(&behaviour),
+            1,
+            "setting a new agent version should notify every connection once"
+        );
+    }
+
+    fn notify_handler_events(behaviour: &Behaviour) -> usize {
+        behaviour
+            .events
+            .iter()
+            .filter(|event| matches!(event, ToSwarm::NotifyHandler { .. }))
+            .count()
+    }
 }

protocols/identify/src/handler.rs

@@ -107,6 +107,7 @@ pub struct Handler {
 #[derive(Debug)]
 pub enum InEvent {
     AddressesChanged(HashSet<Multiaddr>),
+    AgentVersionChanged(String),
     Push,
 }
 
@@ -323,6 +324,9 @@ impl ConnectionHandler for Handler {
             InEvent::AddressesChanged(addresses) => {
                 self.external_addresses = addresses;
             }
+            InEvent::AgentVersionChanged(agent_version) => {
+                self.agent_version = agent_version;
+            }
             InEvent::Push => {
                 self.events
                     .push(ConnectionHandlerEvent::OutboundSubstreamRequest {

protocols/identify/tests/smoke.rs

@@ -349,6 +349,61 @@ async fn identify_push() {
     assert!(swarm1_received_info.listen_addrs.is_empty());
 }
 
+#[tokio::test]
+async fn runtime_agent_version_update() {
+    let _ = tracing_subscriber::fmt()
+        .with_env_filter(EnvFilter::from_default_env())
+        .try_init();
+
+    let mut swarm1 = Swarm::new_ephemeral_tokio(|identity| {
+        identify::Behaviour::new(identify::Config::new("a".to_string(), identity.public()))
+    });
+    let mut swarm2 = Swarm::new_ephemeral_tokio(|identity| {
+        identify::Behaviour::new(
+            identify::Config::new("a".to_string(), identity.public())
+                .with_agent_version("b".to_string()),
+        )
+    });
+
+    swarm1.listen().with_memory_addr_external().await;
+    swarm2.connect(&mut swarm1).await;
+
+    // First, let the periodic identify do its thing so that both sides have exchanged the
+    // initial agent version.
+    let ([e1, e2], [e3, e4]) = libp2p_swarm_test::drive(&mut swarm1, &mut swarm2).await;
+
+    {
+        use identify::Event::{Received, Sent};
+
+        // These can be received in any order, hence assert them here.
+        assert!(matches!(e1, Received { .. } | Sent { .. }));
+        assert!(matches!(e2, Received { .. } | Sent { .. }));
+        assert!(matches!(e3, Received { .. } | Sent { .. }));
+        assert!(matches!(e4, Received { .. } | Sent { .. }));
+    }
+
+    // Second, change the agent version on the already established connection and push it.
+    swarm2.behaviour_mut().set_agent_version("c".to_string());
+    swarm2
+        .behaviour_mut()
+        .push(iter::once(*swarm1.local_peer_id()));
+
+    let swarm1_received_info = match libp2p_swarm_test::drive(&mut swarm1, &mut swarm2).await {
+        ([identify::Event::Received { info, .. }], [identify::Event::Pushed { .. }]) => info,
+        other => panic!("Unexpected events: {other:?}"),
+    };
+
+    assert_eq!(
+        swarm1_received_info.public_key.to_peer_id(),
+        *swarm2.local_peer_id()
+    );
+    assert_eq!(
+        swarm1_received_info.agent_version, "c",
+        "the updated agent version should be advertised on the existing connection"
+    );
+    assert_eq!(swarm1_received_info.protocol_version, "a");
+}
+
 #[tokio::test]
 async fn discover_peer_after_disconnect() {
     let _ = tracing_subscriber::fmt()