diff --git a/pkg/topology/kademlia/export_test.go b/pkg/topology/kademlia/export_test.go index bca662c9a7b..1a56255462c 100644 --- a/pkg/topology/kademlia/export_test.go +++ b/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 diff --git a/pkg/topology/kademlia/kademlia.go b/pkg/topology/kademlia/kademlia.go index d66b669b786..a8233e91749 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,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) } } }) diff --git a/pkg/topology/kademlia/kademlia_test.go b/pkg/topology/kademlia/kademlia_test.go index cb684685bf1..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,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()