From 04680fc52cf550371cd64d65cca17968b6aab3e4 Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Fri, 11 Oct 2019 12:56:14 +0000 Subject: [PATCH] Update weaveworks/mesh to v0.3 --- Gopkg.lock | 6 +- vendor/github.com/weaveworks/mesh/README.md | 9 +++ .../github.com/weaveworks/mesh/connection.go | 26 +------ .../weaveworks/mesh/connection_maker.go | 2 +- .../github.com/weaveworks/mesh/local_peer.go | 54 ++++++++++--- vendor/github.com/weaveworks/mesh/overlay.go | 6 +- vendor/github.com/weaveworks/mesh/peers.go | 30 +++++++- vendor/github.com/weaveworks/mesh/protocol.go | 2 +- vendor/github.com/weaveworks/mesh/router.go | 31 +++++--- vendor/github.com/weaveworks/mesh/routes.go | 75 ++++++++++++------- vendor/github.com/weaveworks/mesh/status.go | 2 +- .../weaveworks/mesh/surrogate_gossiper.go | 5 ++ 12 files changed, 171 insertions(+), 77 deletions(-) diff --git a/Gopkg.lock b/Gopkg.lock index 334246d1..171cecbc 100644 --- a/Gopkg.lock +++ b/Gopkg.lock @@ -686,12 +686,12 @@ revision = "6b0aa22550d9325eb8f43418185859e13dc0de1d" [[projects]] - digest = "1:277e090a58afaeb00fb9fd04e2b2f3466ff8fb327a331e0e531a3399a72697e3" + digest = "1:ef8d0a1f90a33a968984ba2159a02304c591bbe964ad13e6408bf361975b3a58" name = "github.com/weaveworks/mesh" packages = ["."] pruneopts = "UT" - revision = "38e8100dfbc7b28ae8ffde5a2bbeceec57ffa52a" - version = "v0.1" + revision = "8889a805f7f6fe19a40dd175575d7f5efb7eda79" + version = "v0.3" [[projects]] branch = "master" diff --git a/vendor/github.com/weaveworks/mesh/README.md b/vendor/github.com/weaveworks/mesh/README.md index 856768fc..4b23a3f4 100644 --- a/vendor/github.com/weaveworks/mesh/README.md +++ b/vendor/github.com/weaveworks/mesh/README.md @@ -77,3 +77,12 @@ All contributions should be made as pull requests that satisfy the guidelines, b In addition, several mechanical checks are enforced. See [the lint script](/lint) for details. +## Getting Help + +If you have any questions about, feedback for or problems with `mesh`: + +- Invite yourself to the Weave Users Slack. +- Ask a question on the [#general](https://weave-community.slack.com/messages/general/) slack channel. +- [File an issue](https://github.com/weaveworks/mesh/issues/new). + +Your feedback is always welcome! diff --git a/vendor/github.com/weaveworks/mesh/connection.go b/vendor/github.com/weaveworks/mesh/connection.go index 545f98bd..233aba9f 100644 --- a/vendor/github.com/weaveworks/mesh/connection.go +++ b/vendor/github.com/weaveworks/mesh/connection.go @@ -71,7 +71,6 @@ type LocalConnection struct { heartbeatTCP *time.Ticker router *Router uid uint64 - actionChan chan<- connectionAction errorChan chan<- error finished <-chan struct{} // closed to signal that actorLoop has finished senders *gossipSenders @@ -84,7 +83,6 @@ func startLocalConnection(connRemote *remoteConnection, tcpConn *net.TCPConn, ro if connRemote.local != router.Ourself.Peer { panic("attempt to create local connection from a peer which is not ourself") } - actionChan := make(chan connectionAction, ChannelSize) errorChan := make(chan error, 1) finished := make(chan struct{}) conn := &LocalConnection{ @@ -93,13 +91,12 @@ func startLocalConnection(connRemote *remoteConnection, tcpConn *net.TCPConn, ro tcpConn: tcpConn, trustRemote: router.trusts(connRemote), uid: randUint64(), - actionChan: actionChan, errorChan: errorChan, finished: finished, logger: logger, } conn.senders = newGossipSenders(conn, finished) - go conn.run(actionChan, errorChan, finished, acceptNewPeer) + go conn.run(errorChan, finished, acceptNewPeer) } func (conn *LocalConnection) logf(format string, args ...interface{}) { @@ -158,18 +155,9 @@ func (conn *LocalConnection) shutdown(err error) { } } -// Send an actor request to the actorLoop, but don't block if actorLoop has -// exited. See http://blog.golang.org/pipelines for pattern. -func (conn *LocalConnection) sendAction(action connectionAction) { - select { - case conn.actionChan <- action: - case <-conn.finished: - } -} - // ACTOR server -func (conn *LocalConnection) run(actionChan <-chan connectionAction, errorChan <-chan error, finished chan<- struct{}, acceptNewPeer bool) { +func (conn *LocalConnection) run(errorChan <-chan error, finished chan<- struct{}, acceptNewPeer bool) { var err error // important to use this var and not create another one with 'err :=' defer func() { conn.teardown(err) }() defer close(finished) @@ -265,7 +253,7 @@ func (conn *LocalConnection) run(actionChan <-chan connectionAction, errorChan < // running AddConnection in a separate goroutine, at least not // without some synchronisation. Which in turn requires the // launching of the receiveTCP goroutine to precede actorLoop. - err = conn.actorLoop(actionChan, errorChan) + err = conn.actorLoop(errorChan) } func (conn *LocalConnection) makeFeatures() map[string]string { @@ -354,7 +342,7 @@ func (conn *LocalConnection) registerRemote(remote *Peer, acceptNewPeer bool) er return nil } -func (conn *LocalConnection) actorLoop(actionChan <-chan connectionAction, errorChan <-chan error) (err error) { +func (conn *LocalConnection) actorLoop(errorChan <-chan error) (err error) { fwdErrorChan := conn.OverlayConn.ErrorChannel() fwdEstablishedChan := conn.OverlayConn.EstablishedChannel() @@ -364,8 +352,6 @@ func (conn *LocalConnection) actorLoop(actionChan <-chan connectionAction, error case err = <-fwdErrorChan: default: select { - case action := <-actionChan: - err = action() case <-conn.heartbeatTCP.C: err = conn.sendSimpleProtocolMsg(ProtocolHeartbeat) case <-fwdEstablishedChan: @@ -486,10 +472,6 @@ func (err *peerNameCollisionError) Error() string { return fmt.Sprintf("local %q and remote %q peer names collision", err.local, err.remote) } -// The actor closure used by LocalConnection. If an action returns an error, -// it will terminate the actor loop, which terminates the connection in turn. -type connectionAction func() error - func mustHave(features map[string]string, keys []string) error { for _, key := range keys { if _, ok := features[key]; !ok { diff --git a/vendor/github.com/weaveworks/mesh/connection_maker.go b/vendor/github.com/weaveworks/mesh/connection_maker.go index 3eb02754..df8d04c5 100644 --- a/vendor/github.com/weaveworks/mesh/connection_maker.go +++ b/vendor/github.com/weaveworks/mesh/connection_maker.go @@ -93,7 +93,7 @@ func (cm *connectionMaker) InitiateConnections(peers []string, replace bool) []e } if host == "" || !isAlnum(port) { errors = append(errors, fmt.Errorf("invalid peer name %q, should be host[:port]", peer)) - } else if addr, err := net.ResolveTCPAddr("tcp4", fmt.Sprintf("%s:%s", host, port)); err != nil { + } else if addr, err := net.ResolveTCPAddr("tcp", fmt.Sprintf("%s:%s", host, port)); err != nil { errors = append(errors, err) } else { addrs[peer] = addr diff --git a/vendor/github.com/weaveworks/mesh/local_peer.go b/vendor/github.com/weaveworks/mesh/local_peer.go index 9e99aecf..7b36da10 100644 --- a/vendor/github.com/weaveworks/mesh/local_peer.go +++ b/vendor/github.com/weaveworks/mesh/local_peer.go @@ -8,13 +8,20 @@ import ( "time" ) +const ( + deferTopologyUpdateDuration = 1 * time.Second +) + // localPeer is the only "active" peer in the mesh. It extends Peer with // additional behaviors, mostly to retrieve and manage connection state. type localPeer struct { sync.RWMutex *Peer - router *Router - actionChan chan<- localPeerAction + router *Router + actionChan chan<- localPeerAction + topologyUpdates peerNameSet + timer *time.Timer + pendingTopologyUpdate bool } // The actor closure used by localPeer. @@ -23,11 +30,15 @@ type localPeerAction func() // newLocalPeer returns a usable LocalPeer. func newLocalPeer(name PeerName, nickName string, router *Router) *localPeer { actionChan := make(chan localPeerAction, ChannelSize) + topologyUpdates := make(peerNameSet) peer := &localPeer{ - Peer: newPeer(name, nickName, randomPeerUID(), 0, randomPeerShortID()), - router: router, - actionChan: actionChan, + Peer: newPeer(name, nickName, randomPeerUID(), 0, randomPeerShortID()), + router: router, + actionChan: actionChan, + topologyUpdates: topologyUpdates, + timer: time.NewTimer(deferTopologyUpdateDuration), } + peer.timer.Stop() go peer.actorLoop(actionChan) return peer } @@ -82,15 +93,15 @@ func (peer *localPeer) createConnection(localAddr string, peerAddr string, accep if err := peer.checkConnectionLimit(); err != nil { return err } - localTCPAddr, err := net.ResolveTCPAddr("tcp4", localAddr) + localTCPAddr, err := net.ResolveTCPAddr("tcp", localAddr) if err != nil { return err } - remoteTCPAddr, err := net.ResolveTCPAddr("tcp4", peerAddr) + remoteTCPAddr, err := net.ResolveTCPAddr("tcp", peerAddr) if err != nil { return err } - tcpConn, err := net.DialTCP("tcp4", localTCPAddr, remoteTCPAddr) + tcpConn, err := net.DialTCP("tcp", localTCPAddr, remoteTCPAddr) if err != nil { return err } @@ -136,6 +147,10 @@ func (peer *localPeer) encode(enc *gob.Encoder) { // ACTOR server func (peer *localPeer) actorLoop(actionChan <-chan localPeerAction) { + gossipInterval := defaultGossipInterval + if peer.router != nil { + gossipInterval = peer.router.gossipInterval() + } gossipTimer := time.Tick(gossipInterval) for { select { @@ -143,10 +158,22 @@ func (peer *localPeer) actorLoop(actionChan <-chan localPeerAction) { action() case <-gossipTimer: peer.router.sendAllGossip() + case <-peer.timer.C: + peer.broadcastPendingTopologyUpdates() } } } +func (peer *localPeer) broadcastPendingTopologyUpdates() { + peer.Lock() + gossipData := peer.topologyUpdates + peer.topologyUpdates = make(peerNameSet) + peer.pendingTopologyUpdate = false + peer.Unlock() + gossipData[peer.Peer.Name] = struct{}{} + peer.router.broadcastTopologyUpdate(gossipData) +} + func (peer *localPeer) handleAddConnection(conn ourConnection, isRestartedPeer bool) error { if peer.Peer != conn.getLocal() { panic("Attempt made to add connection to peer where peer is not the source of connection") @@ -190,7 +217,6 @@ func (peer *localPeer) handleAddConnection(conn ourConnection, isRestartedPeer b conn.logf("connection added (new peer)") peer.router.sendAllGossipDown(conn) } - peer.router.Routes.recalculate() peer.broadcastPeerUpdate(conn.Remote()) @@ -240,7 +266,15 @@ func (peer *localPeer) broadcastPeerUpdate(peers ...*Peer) { // context of a test, but that will involve significant // reworking of tests. if peer.router != nil { - peer.router.broadcastTopologyUpdate(append(peers, peer.Peer)) + peer.Lock() + defer peer.Unlock() + if !peer.pendingTopologyUpdate { + peer.timer.Reset(deferTopologyUpdateDuration) + peer.pendingTopologyUpdate = true + } + for _, p := range peers { + peer.topologyUpdates[p.Name] = struct{}{} + } } } diff --git a/vendor/github.com/weaveworks/mesh/overlay.go b/vendor/github.com/weaveworks/mesh/overlay.go index b7ad0e72..061147d6 100644 --- a/vendor/github.com/weaveworks/mesh/overlay.go +++ b/vendor/github.com/weaveworks/mesh/overlay.go @@ -108,7 +108,11 @@ func (NullOverlay) Diagnostics() interface{} { return nil } func (NullOverlay) Confirm() {} // EstablishedChannel implements OverlayConnection. -func (NullOverlay) EstablishedChannel() <-chan struct{} { return nil } +func (NullOverlay) EstablishedChannel() <-chan struct{} { + c := make(chan struct{}) + close(c) + return c +} // ErrorChannel implements OverlayConnection. func (NullOverlay) ErrorChannel() <-chan error { return nil } diff --git a/vendor/github.com/weaveworks/mesh/peers.go b/vendor/github.com/weaveworks/mesh/peers.go index 1190450e..6b174e9a 100644 --- a/vendor/github.com/weaveworks/mesh/peers.go +++ b/vendor/github.com/weaveworks/mesh/peers.go @@ -6,6 +6,11 @@ import ( "io" "math/rand" "sync" + "time" +) + +const ( + gcInterval = 1 * time.Second ) // Peers collects all of the known peers in the mesh, including ourself. @@ -18,6 +23,8 @@ type Peers struct { // Called when the mapping from short IDs to peers changes onInvalidateShortIDs []func() + timer *time.Timer + pendingGC bool } type shortIDPeers struct { @@ -60,8 +67,11 @@ func newPeers(ourself *localPeer) *Peers { ourself: ourself, byName: make(map[PeerName]*Peer), byShortID: make(map[PeerShortID]shortIDPeers), + timer: time.NewTimer(gcInterval), } peers.fetchWithDefault(ourself.Peer) + peers.timer.Stop() + go peers.actorLoop() return peers } @@ -339,6 +349,18 @@ func (peers *Peers) forEach(fun func(*Peer)) { } } +func (peers *Peers) actorLoop() { + for { + select { + case <-peers.timer.C: + peers.GarbageCollect() + peers.Lock() + peers.pendingGC = false + peers.Unlock() + } + } +} + // Merge an incoming update with our own topology. // // We add peers hitherto unknown to us, and update peers for which the @@ -363,7 +385,6 @@ func (peers *Peers) applyUpdate(update []byte) (peerNameSet, peerNameSet, error) // Now apply the updates newUpdate := peers.applyDecodedUpdate(decodedUpdate, decodedConns, &pending) - peers.garbageCollect(&pending) for _, peerRemoved := range pending.removed { delete(newUpdate, peerRemoved.Name) } @@ -373,6 +394,13 @@ func (peers *Peers) applyUpdate(update []byte) (peerNameSet, peerNameSet, error) updateNames[peer.Name] = struct{}{} } + if !peers.pendingGC { + // schedule a GarbageCollect() to run after gcInterval time period + // corresponding to all topology updates received during the period + peers.timer.Reset(gcInterval) + peers.pendingGC = true + } + return updateNames, newUpdate, nil } diff --git a/vendor/github.com/weaveworks/mesh/protocol.go b/vendor/github.com/weaveworks/mesh/protocol.go index f0c8c8a9..0d47dd5a 100644 --- a/vendor/github.com/weaveworks/mesh/protocol.go +++ b/vendor/github.com/weaveworks/mesh/protocol.go @@ -54,10 +54,10 @@ type protocolIntroConn interface { type protocolIntroParams struct { MinVersion byte MaxVersion byte + Outbound bool Features map[string]string Conn protocolIntroConn Password []byte - Outbound bool } // The results from a successful protocol intro. diff --git a/vendor/github.com/weaveworks/mesh/router.go b/vendor/github.com/weaveworks/mesh/router.go index 75b751c0..736be318 100644 --- a/vendor/github.com/weaveworks/mesh/router.go +++ b/vendor/github.com/weaveworks/mesh/router.go @@ -17,11 +17,12 @@ var ( // ChannelSize is the buffer size used by so-called actor goroutines // throughout mesh. ChannelSize = 16 + + defaultGossipInterval = 30 * time.Second ) const ( tcpHeartbeat = 30 * time.Second - gossipInterval = 30 * time.Second maxDuration = time.Duration(math.MaxInt64) acceptMaxTokens = 100 acceptTokenDelay = 100 * time.Millisecond // [2] @@ -32,11 +33,12 @@ const ( type Config struct { Host string Port int - ProtocolMinVersion byte Password []byte ConnLimit int + ProtocolMinVersion byte PeerDiscovery bool TrustedSubnets []*net.IPNet + GossipInterval *time.Duration } // Router manages communication between this peer and the rest of the mesh. @@ -99,11 +101,11 @@ func (router *Router) usingPassword() bool { } func (router *Router) listenTCP() { - localAddr, err := net.ResolveTCPAddr("tcp4", net.JoinHostPort(router.Host, fmt.Sprint(router.Port))) + localAddr, err := net.ResolveTCPAddr("tcp", net.JoinHostPort(router.Host, fmt.Sprint(router.Port))) if err != nil { panic(err) } - ln, err := net.ListenTCP("tcp4", localAddr) + ln, err := net.ListenTCP("tcp", localAddr) if err != nil { panic(err) } @@ -154,7 +156,7 @@ func (router *Router) gossipChannel(channelName string) *gossipChannel { if channel, found = router.gossipChannels[channelName]; found { return channel } - channel = newGossipChannel(channelName, router.Ourself, router.Routes, &surrogateGossiper{}, router.logger) + channel = newGossipChannel(channelName, router.Ourself, router.Routes, &surrogateGossiper{router: router}, router.logger) channel.logf("created surrogate channel") router.gossipChannels[channelName] = channel return channel @@ -170,6 +172,14 @@ func (router *Router) gossipChannelSet() map[*gossipChannel]struct{} { return channels } +func (router *Router) gossipInterval() time.Duration { + if router.Config.GossipInterval != nil { + return *router.Config.GossipInterval + } else { + return defaultGossipInterval + } +} + func (router *Router) handleGossip(tag protocolTag, payload []byte) error { decoder := gob.NewDecoder(bytes.NewReader(payload)) var channelName string @@ -221,12 +231,9 @@ func (router *Router) sendPendingGossip() bool { // BroadcastTopologyUpdate is invoked whenever there is a change to the mesh // topology, and broadcasts the new set of peers to the mesh. -func (router *Router) broadcastTopologyUpdate(update []*Peer) { - names := make(peerNameSet) - for _, p := range update { - names[p.Name] = struct{}{} - } - router.topologyGossip.GossipBroadcast(&topologyGossipData{peers: router.Peers, update: names}) +func (router *Router) broadcastTopologyUpdate(update peerNameSet) { + gossipData := &topologyGossipData{peers: router.Peers, update: update} + router.topologyGossip.GossipBroadcast(gossipData) } // OnGossipUnicast implements Gossiper, but always returns an error, as a @@ -274,7 +281,7 @@ func (router *Router) applyTopologyUpdate(update []byte) (peerNameSet, peerNameS } func (router *Router) trusts(remote *remoteConnection) bool { - if tcpAddr, err := net.ResolveTCPAddr("tcp4", remote.remoteTCPAddr); err == nil { + if tcpAddr, err := net.ResolveTCPAddr("tcp", remote.remoteTCPAddr); err == nil { for _, trustedSubnet := range router.TrustedSubnets { if trustedSubnet.Contains(tcpAddr.IP) { return true diff --git a/vendor/github.com/weaveworks/mesh/routes.go b/vendor/github.com/weaveworks/mesh/routes.go index 10ff3152..858da783 100644 --- a/vendor/github.com/weaveworks/mesh/routes.go +++ b/vendor/github.com/weaveworks/mesh/routes.go @@ -3,6 +3,7 @@ package mesh import ( "math" "sync" + "time" ) type unicastRoutes map[PeerName]PeerName @@ -11,23 +12,29 @@ type broadcastRoutes map[PeerName][]PeerName // routes aggregates unicast and broadcast routes for our peer. type routes struct { sync.RWMutex - ourself *localPeer - peers *Peers - onChange []func() - unicast unicastRoutes - unicastAll unicastRoutes // [1] - broadcast broadcastRoutes - broadcastAll broadcastRoutes // [1] - recalc chan<- *struct{} - wait chan<- chan struct{} - action chan<- func() + ourself *localPeer + peers *Peers + onChange []func() + unicast unicastRoutes + unicastAll unicastRoutes // [1] + broadcast broadcastRoutes + broadcastAll broadcastRoutes // [1] + recalcTimer *time.Timer + pendingRecalc bool + wait chan chan struct{} + action chan<- func() // [1] based on *all* connections, not just established & // symmetric ones } +const ( + // We defer recalculation requests by up to 100ms, in order to + // coalesce multiple recalcs together. + recalcDeferTime = 100 * time.Millisecond +) + // newRoutes returns a usable Routes based on the LocalPeer and existing Peers. func newRoutes(ourself *localPeer, peers *Peers) *routes { - recalculate := make(chan *struct{}, 1) wait := make(chan chan struct{}) action := make(chan func()) r := &routes{ @@ -37,11 +44,12 @@ func newRoutes(ourself *localPeer, peers *Peers) *routes { unicastAll: unicastRoutes{ourself.Name: UnknownPeerName}, broadcast: broadcastRoutes{ourself.Name: []PeerName{}}, broadcastAll: broadcastRoutes{ourself.Name: []PeerName{}}, - recalc: recalculate, + recalcTimer: time.NewTimer(time.Hour), wait: wait, action: action, } - go r.run(recalculate, wait, action) + r.recalcTimer.Stop() + go r.run(wait, action) return r } @@ -156,32 +164,47 @@ func (r *routes) randomNeighbours(except PeerName) []PeerName { // can effectively be made synchronous with a subsequent call to // EnsureRecalculated. func (r *routes) recalculate() { - // The use of a 1-capacity channel in combination with the - // non-blocking send is an optimisation that results in multiple - // requests being coalesced. - select { - case r.recalc <- nil: - default: + r.Lock() + if !r.pendingRecalc { + r.recalcTimer.Reset(recalcDeferTime) + r.pendingRecalc = true } + r.Unlock() +} + +func (r *routes) clearPendingRecalcFlag() { + r.Lock() + r.pendingRecalc = false + r.Unlock() } // EnsureRecalculated waits for any preceding Recalculate requests to finish. func (r *routes) ensureRecalculated() { - done := make(chan struct{}) + var done chan struct{} + // If another call is already waiting, wait on the same chan, otherwise make a new one + select { + case done = <-r.wait: + default: + done = make(chan struct{}) + } r.wait <- done <-done } -func (r *routes) run(recalculate <-chan *struct{}, wait <-chan chan struct{}, action <-chan func()) { +func (r *routes) run(wait <-chan chan struct{}, action <-chan func()) { for { select { - case <-recalculate: + case <-r.recalcTimer.C: + r.clearPendingRecalcFlag() r.calculate() case done := <-wait: - select { - case <-recalculate: + r.Lock() + pending := r.pendingRecalc + r.Unlock() + if pending { + <-r.recalcTimer.C + r.clearPendingRecalcFlag() r.calculate() - default: } close(done) case f := <-action: @@ -190,6 +213,8 @@ func (r *routes) run(recalculate <-chan *struct{}, wait <-chan chan struct{}, ac } } +// Calculate unicast and broadcast routes from r.ourself, and reset +// the broadcast route cache. func (r *routes) calculate() { r.peers.RLock() r.ourself.RLock() diff --git a/vendor/github.com/weaveworks/mesh/status.go b/vendor/github.com/weaveworks/mesh/status.go index a78cac3d..1c770396 100644 --- a/vendor/github.com/weaveworks/mesh/status.go +++ b/vendor/github.com/weaveworks/mesh/status.go @@ -30,7 +30,7 @@ type Status struct { func NewStatus(router *Router) *Status { return &Status{ Protocol: Protocol, - ProtocolMinVersion: ProtocolMinVersion, + ProtocolMinVersion: int(router.ProtocolMinVersion), ProtocolMaxVersion: ProtocolMaxVersion, Encryption: router.usingPassword(), PeerDiscovery: router.PeerDiscovery, diff --git a/vendor/github.com/weaveworks/mesh/surrogate_gossiper.go b/vendor/github.com/weaveworks/mesh/surrogate_gossiper.go index 89ea9e1b..4ed9f28b 100644 --- a/vendor/github.com/weaveworks/mesh/surrogate_gossiper.go +++ b/vendor/github.com/weaveworks/mesh/surrogate_gossiper.go @@ -11,6 +11,7 @@ import ( type surrogateGossiper struct { sync.Mutex prevUpdates []prevUpdate + router *Router } type prevUpdate struct { @@ -56,6 +57,10 @@ func (s *surrogateGossiper) OnGossip(update []byte) (GossipData, error) { // (this time limit is arbitrary; surrogateGossiper should pass on new gossip immediately // so there should be no reason for a duplicate to show up after a long time) updateTime := now() + gossipInterval := defaultGossipInterval + if s.router != nil { + gossipInterval = s.router.gossipInterval() + } deleteBefore := updateTime.Add(-gossipInterval) keepFrom := len(s.prevUpdates) for i, p := range s.prevUpdates {