From 009c2a4b15a204696bdf4725194490d0d6a5b56c Mon Sep 17 00:00:00 2001 From: Ljubisa Gacevic Date: Thu, 3 Sep 2026 11:51:05 +0200 Subject: [PATCH 1/2] fix(kademlia): neighborhood rebroadcast sends wrong peer lists --- pkg/topology/kademlia/export_test.go | 8 ++++ pkg/topology/kademlia/kademlia.go | 35 +++++++++------ pkg/topology/kademlia/kademlia_test.go | 61 ++++++++++++++++++++++++++ 3 files changed, 91 insertions(+), 13 deletions(-) diff --git a/pkg/topology/kademlia/export_test.go b/pkg/topology/kademlia/export_test.go index bca662c9a7b..1e0dd7a618b 100644 --- a/pkg/topology/kademlia/export_test.go +++ b/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 diff --git a/pkg/topology/kademlia/kademlia.go b/pkg/topology/kademlia/kademlia.go index d66b669b786..586cfe51975 100644 --- a/pkg/topology/kademlia/kademlia.go +++ b/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) } } }) diff --git a/pkg/topology/kademlia/kademlia_test.go b/pkg/topology/kademlia/kademlia_test.go index cb684685bf1..e0456f5a578 100644 --- a/pkg/topology/kademlia/kademlia_test.go +++ b/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() From c97300da2acdda30a27840d67c4141991812c201 Mon Sep 17 00:00:00 2001 From: Ljubisa Gacevic Date: Fri, 4 Sep 2026 08:22:03 +0200 Subject: [PATCH 2/2] test(kademlia): unit-test neighborhood broadcast sets directly --- pkg/topology/kademlia/export_test.go | 9 +-- pkg/topology/kademlia/kademlia.go | 16 ++++-- pkg/topology/kademlia/kademlia_test.go | 79 ++++++++++---------------- 3 files changed, 43 insertions(+), 61 deletions(-) diff --git a/pkg/topology/kademlia/export_test.go b/pkg/topology/kademlia/export_test.go index 1e0dd7a618b..1a56255462c 100644 --- a/pkg/topology/kademlia/export_test.go +++ b/pkg/topology/kademlia/export_test.go @@ -5,8 +5,6 @@ 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" @@ -18,6 +16,7 @@ var ( return k.pruneOversaturatedBins } GenerateCommonBinPrefixes = generateCommonBinPrefixes + NeighborhoodBroadcasts = neighborhoodBroadcasts ) // MarkConnectedPeersSeen runs the sweep the manage loop performs on every @@ -26,12 +25,6 @@ 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 diff --git a/pkg/topology/kademlia/kademlia.go b/pkg/topology/kademlia/kademlia.go index 586cfe51975..a8233e91749 100644 --- a/pkg/topology/kademlia/kademlia.go +++ b/pkg/topology/kademlia/kademlia.go @@ -538,6 +538,16 @@ 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 @@ -548,11 +558,9 @@ func (k *Kad) rebroadcastNeighborhood(ctx context.Context) { neighbors = append(neighbors, addr) return false, false, nil }) + broadcasts := neighborhoodBroadcasts(neighbors) 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 { + if err := k.discovery.BroadcastPeers(ctx, peer, broadcasts[i]...); err != nil { k.logger.Debug("broadcast neighborhood failure", "peer_address", peer, "error", err) } } diff --git a/pkg/topology/kademlia/kademlia_test.go b/pkg/topology/kademlia/kademlia_test.go index e0456f5a578..008a0611cf8 100644 --- a/pkg/topology/kademlia/kademlia_test.go +++ b/pkg/topology/kademlia/kademlia_test.go @@ -11,6 +11,7 @@ import ( "math" "math/rand" "reflect" + "slices" "strconv" "sync" "sync/atomic" @@ -1779,64 +1780,44 @@ 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) { +// 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() - 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) + neighbors := make([]swarm.Address, 4) for i := range neighbors { - neighbors[i] = swarm.RandAddressAt(t, base, i) - connectOne(t, signer, kad, ab, neighbors[i], nil) + neighbors[i] = swarm.RandAddress(t) } - waitPeers(t, kad, n) + input := slices.Clone(neighbors) + + broadcasts := kademlia.NeighborhoodBroadcasts(neighbors) - // 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 len(broadcasts) != len(neighbors) { + t.Fatalf("got %d broadcasts, want %d", len(broadcasts), len(neighbors)) } - 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) + 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) + } } - disc.Reset() + if !slices.EqualFunc(neighbors, input, swarm.Address.Equal) { + t.Fatalf("input was modified: got %v, want %v", neighbors, input) + } +} - kad.RebroadcastNeighborhood(context.Background()) +func TestNeighborhoodBroadcastsSmall(t *testing.T) { + t.Parallel() - 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) - } - } + 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) } }