142 lines
3.7 KiB
Go
142 lines
3.7 KiB
Go
package cluster
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"log/slog"
|
|
"math"
|
|
"math/rand/v2"
|
|
"slices"
|
|
"time"
|
|
|
|
"github.com/labstack/gommon/log"
|
|
"github.com/libp2p/go-libp2p/core/peer"
|
|
"github.com/libp2p/go-libp2p/p2p/net/swarm"
|
|
)
|
|
|
|
func (c *Cluster) selectRandomPeer(excluded []peer.ID) (peer.ID, error) {
|
|
// get a list of discovered peers
|
|
peers := c.dht.RoutingTable().ListPeers()
|
|
|
|
// add ourselves to the excluded list
|
|
excluded = append(excluded, c.host.ID())
|
|
|
|
// remove excluded peers from the list
|
|
peers = slices.DeleteFunc(peers, func(p peer.ID) bool {
|
|
return slices.Contains(excluded, p)
|
|
})
|
|
|
|
if len(peers) == 0 {
|
|
// no peers available
|
|
return "", ErrNoPeersAvailable
|
|
}
|
|
|
|
// select one random peer
|
|
return peers[rand.IntN(len(peers))], nil //nolint:gosec
|
|
}
|
|
|
|
// exchangeStateOnJoin performs state exchange with min(sqrt(N), 4) random connected peers on joining the cluster.
|
|
func (c *Cluster) exchangeStateOnJoin(ctx context.Context) {
|
|
// wait a few seconds to let the DHT populate
|
|
delay := max(c.cfg.Cluster.PushPullInterval/2, time.Second*10)
|
|
|
|
slog.Info("waiting for DHT to populate", slog.Duration("delay", delay))
|
|
time.Sleep(delay)
|
|
|
|
// get a list of discovered peers
|
|
peers := c.dht.RoutingTable().ListPeers()
|
|
|
|
// remove ourselves from the list
|
|
peers = slices.DeleteFunc(peers, func(p peer.ID) bool {
|
|
return p == c.host.ID()
|
|
})
|
|
|
|
// bail out early if there are no peers
|
|
if len(peers) == 0 {
|
|
// todo retry for a bit
|
|
slog.Warn("no peers discovered in DHT, skipping join state exchange")
|
|
return
|
|
}
|
|
|
|
// calculate how many peers to sync with: min(sqrt(N), 4)
|
|
n := min(min(max(int(math.Sqrt(float64(len(peers)))), 1), 4), len(peers))
|
|
|
|
slog.Info("performing state exchange with peers on join", slog.Int("count", n))
|
|
|
|
// shuffle and take the first n
|
|
rand.Shuffle(len(peers), func(i, j int) {
|
|
peers[i], peers[j] = peers[j], peers[i]
|
|
})
|
|
|
|
for _, p := range peers[:n] {
|
|
if err := c.stateExchanger.ExchangeWithPeer(ctx, p, c.cfg.Cluster.PushPullInterval); err != nil {
|
|
slog.Warn(
|
|
"failed to exchange state on join",
|
|
slog.String("peer", p.String()),
|
|
slog.String("err", err.Error()),
|
|
)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (c *Cluster) exchangeState(ctx context.Context, retries int) error {
|
|
var (
|
|
retryCount int
|
|
excludedPeers []peer.ID
|
|
)
|
|
|
|
for retryCount < retries {
|
|
// choose a random peer
|
|
peerID, err := c.selectRandomPeer(excludedPeers)
|
|
if errors.Is(err, ErrNoPeersAvailable) {
|
|
slog.Debug("no peers discovered in DHT, skipping state exchange")
|
|
|
|
break
|
|
}
|
|
|
|
// attempt to exchange state with the peer
|
|
err = c.stateExchanger.ExchangeWithPeer(ctx, peerID, c.cfg.Cluster.PushPullInterval)
|
|
if err == nil {
|
|
// state exchange was successful
|
|
return nil
|
|
}
|
|
|
|
// check for unexpected errors
|
|
var dialErr *swarm.DialError
|
|
if !errors.As(err, &dialErr) {
|
|
log.Warn(
|
|
"unexpected error during state exchange",
|
|
slog.String("peer", peerID.String()),
|
|
slog.String("err", err.Error()),
|
|
)
|
|
|
|
// exclude the peer from retry
|
|
excludedPeers = append(excludedPeers, peerID)
|
|
|
|
retryCount++
|
|
|
|
continue
|
|
}
|
|
|
|
// check for specific dial errors
|
|
// we ignore ErrDialBackoff as that can be transient
|
|
if errors.Is(dialErr.Cause, swarm.ErrAllDialsFailed) ||
|
|
errors.Is(dialErr.Cause, swarm.ErrNoAddresses) ||
|
|
errors.Is(dialErr.Cause, swarm.ErrNoGoodAddresses) {
|
|
// peer is unreachable, proactively remove it from the routing table
|
|
log.Warn("peer unreachable", slog.String("err", dialErr.Error()))
|
|
|
|
// exclude the peer from retry
|
|
excludedPeers = append(excludedPeers, peerID)
|
|
|
|
// increase the retry count and try with another peer
|
|
retryCount++
|
|
|
|
continue
|
|
}
|
|
}
|
|
|
|
return fmt.Errorf("failed to exchange state with peers after %d retries", retries)
|
|
}
|