fix(bitswap): see peers connected before startup

CHANGELOG.md

@@ -22,6 +22,8 @@ The following emojis are used to highlight certain changes:
 
 ### Fixed
 
+- `bitswap/network/bsnet`: peers already connected when Bitswap starts are now recognised. libp2p only reports connections opened after a notifier is registered, so a peer connected during node startup stayed invisible to Bitswap for the life of that connection, and no want was ever sent to it. Nodes with another way to find content usually masked this; nodes relying on an already-connected peer could wait forever.
+
 ### Security
 
 ## [v0.42.1]

bitswap/network/bsnet/ipfs_impl.go

@@ -445,6 +445,20 @@ func (bsnet *impl) Start(r ...iface.Receiver) {
 		bsnet.host.SetStreamHandler(proto, bsnet.handleNewStream)
 	}
 	bsnet.host.Network().Notify((*netNotifiee)(bsnet))
+
+	// Take stock of the connections that are already open. Notify only reports
+	// what happens from here on, so without this pass a peer connected before
+	// bitswap started stays invisible for the life of that connection: we never
+	// send it a want, and nothing later corrects the record.
+	// Registering the notifiee first means a connection opening right now is
+	// counted twice at worst, which Connected handles.
+	for _, conn := range bsnet.host.Network().Conns() {
+		if conn.Stat().Limited {
+			continue
+		}
+		bsnet.connectEvtMgr.Connected(conn.RemotePeer())
+	}
+
 	bsnet.connectEvtMgr.Start()
 }
 

bitswap/network/bsnet/ipfs_impl_test.go

@@ -999,3 +999,41 @@ func TestSendMessagePanicInCloseIsRecovered(t *testing.T) {
 		return false
 	}, 2*time.Second, 50*time.Millisecond, "close goroutine did not mark stream closed")
 }
+
+// A peer can already be connected by the time bitswap starts, for example when
+// mDNS or an inbound dial lands while the node is still being built. Those
+// connections have to be picked up on start, because libp2p only reports
+// changes from the moment a notifiee is registered.
+func TestStartNoticesExistingConnections(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	mn := mocknet.New()
+	defer mn.Close()
+	streamNet, err := tn.StreamNet(ctx, mn)
+	require.NoError(t, err)
+
+	p1 := tnet.RandIdentityOrFatal(t)
+	p2 := tnet.RandIdentityOrFatal(t)
+	bsnet1 := streamNet.Adapter(p1)
+	streamNet.Adapter(p2)
+
+	require.NoError(t, mn.LinkAll())
+	_, err = mn.ConnectPeers(p1.ID(), p2.ID())
+	require.NoError(t, err)
+
+	r1 := newReceiver()
+	bsnet1.Start(r1)
+	t.Cleanup(bsnet1.Stop)
+
+	select {
+	case connected := <-r1.connectionEvent:
+		require.True(t, connected, "expected a connect event, got a disconnect")
+	case <-ctx.Done():
+		t.Fatal("bitswap never noticed the peer that was connected before Start")
+	}
+
+	r1.mu.Lock()
+	defer r1.mu.Unlock()
+	require.Contains(t, r1.peers, p2.ID())
+}