fix(kademlia): neighborhood rebroadcast sends wrong peer lists

Author: gacevicljubisa

pkg/topology/kademlia/export_test.go

@@ -5,6 +5,8 @@
 package kademlia
 
 import (
+	"context"
+
 	"github.com/ethersphere/bee/v2/pkg/swarm"
 	"github.com/ethersphere/bee/v2/pkg/topology"
 	im "github.com/ethersphere/bee/v2/pkg/topology/kademlia/internal/metrics"
@@ -24,6 +26,12 @@ func (k *Kad) MarkConnectedPeersSeen() error {
 	return k.markConnectedPeersSeen()
 }
 
+// RebroadcastNeighborhood runs the neighborhood gossip the manage loop
+// performs every fifteen minutes.
+func (k *Kad) RebroadcastNeighborhood(ctx context.Context) {
+	k.rebroadcastNeighborhood(ctx)
+}
+
 const (
 	DefaultBitSuffixLength     = defaultBitSuffixLength
 	DefaultSaturationPeers     = defaultSaturationPeers

pkg/topology/kademlia/kademlia.go

@@ -13,6 +13,7 @@ import (
 	"math/big"
 	"math/rand"
 	"path/filepath"
+	"slices"
 	"sync"
 	"time"
 
@@ -537,6 +538,26 @@ func (k *Kad) markConnectedPeersSeen() error {
 	return k.addressBook.Seen(peers...)
 }
 
+// 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
+	})
+	for i, peer := range neighbors {
+		// Concat allocates a new slice; appending to neighbors[:i] in place
+		// would overwrite the entries the loop has yet to visit.
+		others := slices.Concat(neighbors[:i], neighbors[i+1:])
+		if err := k.discovery.BroadcastPeers(ctx, peer, others...); 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 +638,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

@@ -1779,6 +1779,67 @@ func TestAnnounceNeighborhoodToNeighbor(t *testing.T) {
 	}
 }
 
+// TestRebroadcastNeighborhood checks that the periodic neighborhood gossip
+// sends every neighbor exactly one message listing all the other neighbors.
+func TestRebroadcastNeighborhood(t *testing.T) {
+	t.Parallel()
+
+	var (
+		conns                       int32
+		base, kad, ab, disc, signer = newTestKademlia(t, &conns, nil, kademlia.Options{})
+	)
+
+	// With a storage radius of zero every connected peer is a neighbor.
+	kad.SetStorageRadius(0)
+
+	if err := kad.Start(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+	testutil.CleanupCloser(t, kad)
+
+	const n = 4
+	neighbors := make([]swarm.Address, n)
+	for i := range neighbors {
+		neighbors[i] = swarm.RandAddressAt(t, base, i)
+		connectOne(t, signer, kad, ab, neighbors[i], nil)
+	}
+	waitPeers(t, kad, n)
+
+	// Connecting peer j announces it to the j peers connected before it and
+	// introduces those j peers to it. Wait for all of that to land before
+	// clearing the recorder, so only the rebroadcast is measured.
+	wantAnnounces := 0
+	for j := 1; j < n; j++ {
+		wantAnnounces += j + 1
+	}
+	if err := spinlock.Wait(spinLockWaitTime, func() bool {
+		return disc.Broadcasts() == wantAnnounces
+	}); err != nil {
+		t.Fatalf("waiting for announce broadcasts: got %d, want %d", disc.Broadcasts(), wantAnnounces)
+	}
+	disc.Reset()
+
+	kad.RebroadcastNeighborhood(context.Background())
+
+	for _, addressee := range neighbors {
+		got, ok := disc.AddresseeRecords(addressee)
+		if !ok {
+			t.Fatalf("neighbor %s received no rebroadcast", addressee)
+		}
+		if len(got) != n-1 {
+			t.Fatalf("neighbor %s received %d peers, want %d", addressee, len(got), n-1)
+		}
+		if swarm.ContainsAddress(got, addressee) {
+			t.Fatalf("neighbor %s was told about itself", addressee)
+		}
+		for _, other := range neighbors {
+			if !other.Equal(addressee) && !swarm.ContainsAddress(got, other) {
+				t.Fatalf("neighbor %s was not told about %s", addressee, other)
+			}
+		}
+	}
+}
+
 func TestIteratorOpts(t *testing.T) {
 	t.Parallel()