fix(kademlia): neighborhood rebroadcast sends wrong peer lists

pkg/topology/kademlia/export_test.go

@@ -16,6 +16,7 @@ var (
 		return k.pruneOversaturatedBins
 	}
 	GenerateCommonBinPrefixes = generateCommonBinPrefixes
+	NeighborhoodBroadcasts    = neighborhoodBroadcasts
 )
 
 // MarkConnectedPeersSeen runs the sweep the manage loop performs on every

pkg/topology/kademlia/kademlia.go

@@ -13,6 +13,7 @@ import (
 	"math/big"
 	"math/rand"
 	"path/filepath"
+	"slices"
 	"sync"
 	"time"
 
@@ -537,6 +538,34 @@ func (k *Kad) markConnectedPeersSeen() error {
 	return k.addressBook.Seen(peers...)
 }
 
+// neighborhoodBroadcasts returns, for every neighbor, the other neighbors it
+// should be told about.
+func neighborhoodBroadcasts(neighbors []swarm.Address) [][]swarm.Address {
+	broadcasts := make([][]swarm.Address, len(neighbors))
+	for i := range neighbors {
+		broadcasts[i] = slices.Concat(neighbors[:i], neighbors[i+1:])
+	}
+	return broadcasts
+}
+
+// rebroadcastNeighborhood tells each neighbor about the other neighbors.
+func (k *Kad) rebroadcastNeighborhood(ctx context.Context) {
+	var neighbors []swarm.Address
+	_ = k.connectedPeers.EachBin(func(addr swarm.Address, bin uint8) (stop bool, jumpToNext bool, err error) {
+		if bin < k.neighborhoodDepth() {
+			return true, false, nil
+		}
+		neighbors = append(neighbors, addr)
+		return false, false, nil
+	})
+	broadcasts := neighborhoodBroadcasts(neighbors)
+	for i, peer := range neighbors {
+		if err := k.discovery.BroadcastPeers(ctx, peer, broadcasts[i]...); err != nil {
+			k.logger.Debug("broadcast neighborhood failure", "peer_address", peer, "error", err)
+		}
+	}
+}
+
 // manage is a forever loop that manages the connection to new peers
 // once they get added or once others leave.
 func (k *Kad) manage() {
@@ -617,19 +646,7 @@ func (k *Kad) manage() {
 			case <-k.quit:
 				return
 			case <-time.After(15 * time.Minute):
-				var neighbors []swarm.Address
-				_ = k.connectedPeers.EachBin(func(addr swarm.Address, bin uint8) (stop bool, jumpToNext bool, err error) {
-					if bin < k.neighborhoodDepth() {
-						return true, false, nil
-					}
-					neighbors = append(neighbors, addr)
-					return false, false, nil
-				})
-				for i, peer := range neighbors {
-					if err := k.discovery.BroadcastPeers(ctx, peer, append(neighbors[:i], neighbors[i+1:]...)...); err != nil {
-						k.logger.Debug("broadcast neighborhood failure", "peer_address", peer, "error", err)
-					}
-				}
+				k.rebroadcastNeighborhood(ctx)
 			}
 		}
 	})

pkg/topology/kademlia/kademlia_test.go

@@ -11,6 +11,7 @@ import (
 	"math"
 	"math/rand"
 	"reflect"
+	"slices"
 	"strconv"
 	"sync"
 	"sync/atomic"
@@ -1779,6 +1780,47 @@ func TestAnnounceNeighborhoodToNeighbor(t *testing.T) {
 	}
 }
 
+// TestNeighborhoodBroadcasts checks the sets the periodic neighborhood gossip
+// sends out: every neighbor is told about all the other neighbors and never
+// about itself, and the input is not modified along the way.
+func TestNeighborhoodBroadcasts(t *testing.T) {
+	t.Parallel()
+
+	neighbors := make([]swarm.Address, 4)
+	for i := range neighbors {
+		neighbors[i] = swarm.RandAddress(t)
+	}
+	input := slices.Clone(neighbors)
+
+	broadcasts := kademlia.NeighborhoodBroadcasts(neighbors)
+
+	if len(broadcasts) != len(neighbors) {
+		t.Fatalf("got %d broadcasts, want %d", len(broadcasts), len(neighbors))
+	}
+	for i, others := range broadcasts {
+		want := slices.Concat(input[:i], input[i+1:])
+		if !slices.EqualFunc(others, want, swarm.Address.Equal) {
+			t.Fatalf("neighbor %d: got %v, want %v", i, others, want)
+		}
+	}
+	if !slices.EqualFunc(neighbors, input, swarm.Address.Equal) {
+		t.Fatalf("input was modified: got %v, want %v", neighbors, input)
+	}
+}
+
+func TestNeighborhoodBroadcastsSmall(t *testing.T) {
+	t.Parallel()
+
+	if got := kademlia.NeighborhoodBroadcasts(nil); len(got) != 0 {
+		t.Fatalf("no neighbors: got %v, want none", got)
+	}
+
+	got := kademlia.NeighborhoodBroadcasts([]swarm.Address{swarm.RandAddress(t)})
+	if len(got) != 1 || len(got[0]) != 0 {
+		t.Fatalf("single neighbor: got %v, want one empty set", got)
+	}
+}
+
 func TestIteratorOpts(t *testing.T) {
 	t.Parallel()