feat!: only consider backbone nodes for core protocols

.github/workflows/beekeeper.yml

@@ -16,6 +16,7 @@ jobs:
       SETUP_CONTRACT_IMAGE_TAG: "0.2.0"
       BEEKEEPER_BRANCH: "master"
       BEEKEEPER_METRICS_ENABLED: false
+      REACHABILITY_OVERRIDE_PUBLIC: true
     runs-on: ubuntu-latest
     steps:
       - name: Setup Go
@@ -130,6 +131,7 @@ jobs:
       SETUP_CONTRACT_IMAGE_TAG: "0.2.0"
       BEEKEEPER_BRANCH: "master"
       BEEKEEPER_METRICS_ENABLED: false
+      REACHABILITY_OVERRIDE_PUBLIC: true
     runs-on: ubuntu-latest
     steps:
       - name: Setup Go
@@ -214,6 +216,7 @@ jobs:
       SETUP_CONTRACT_IMAGE_TAG: "0.2.0"
       BEEKEEPER_BRANCH: "master"
       BEEKEEPER_METRICS_ENABLED: false
+      REACHABILITY_OVERRIDE_PUBLIC: true
     runs-on: ubuntu-latest
     steps:
       - name: Setup Go

Makefile

@@ -8,6 +8,7 @@ BEEKEEPER_USE_SUDO ?= false
 BEEKEEPER_CLUSTER ?= local
 BEELOCAL_BRANCH ?= main
 BEEKEEPER_BRANCH ?= master
+REACHABILITY_OVERRIDE_PUBLIC ?= false
 
 GO_MIN_VERSION ?= "1.17"
 GO_BUILD_VERSION ?= "1.17.2"
@@ -25,7 +26,8 @@ LDFLAGS ?= -s -w \
 -X github.com/ethersphere/bee.commitHash="$(COMMIT_HASH)" \
 -X github.com/ethersphere/bee.commitTime="$(COMMIT_TIME)" \
 -X github.com/ethersphere/bee/pkg/api.Version="$(BEE_API_VERSION)" \
--X github.com/ethersphere/bee/pkg/debugapi.Version="$(BEE_DEBUG_API_VERSION)"
+-X github.com/ethersphere/bee/pkg/debugapi.Version="$(BEE_DEBUG_API_VERSION)" \
+-X github.com/ethersphere/bee/pkg/p2p/libp2p.reachabilityOverridePublic="$(REACHABILITY_OVERRIDE_PUBLIC)"
 
 .PHONY: all
 all: build lint vet test-race binary

pkg/chainsyncer/chainsyncer.go

@@ -157,7 +157,7 @@ func (c *ChainSyncer) manage() {
 				atomic.AddInt32(&positives, 1)
 			}(p)
 			return false, false, nil
-		})
+		}, topology.Filter{Reachable: true})
 
 		// wait for all operations to finish
 		wg.Wait()

pkg/p2p/libp2p/connections_test.go

@@ -23,8 +23,15 @@ import (
 	"github.com/ethersphere/bee/pkg/swarm"
 	"github.com/ethersphere/bee/pkg/swarm/test"
 	"github.com/ethersphere/bee/pkg/topology/lightnode"
+	"github.com/libp2p/go-eventbus"
+	libp2pm "github.com/libp2p/go-libp2p"
+	"github.com/libp2p/go-libp2p-core/event"
+	"github.com/libp2p/go-libp2p-core/host"
 	"github.com/libp2p/go-libp2p-core/mux"
+	"github.com/libp2p/go-libp2p-core/network"
 	libp2ppeer "github.com/libp2p/go-libp2p-core/peer"
+	swarmt "github.com/libp2p/go-libp2p-swarm/testing"
+	bhost "github.com/libp2p/go-libp2p/p2p/host/basic"
 	ma "github.com/multiformats/go-multiaddr"
 )
 
@@ -824,6 +831,7 @@ func TestWithBlocklistStreams(t *testing.T) {
 }
 
 func TestUserAgentLogging(t *testing.T) {
+
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
@@ -859,6 +867,55 @@ func TestUserAgentLogging(t *testing.T) {
 	testUserAgentLogLine(t, s2Logs, "(outbound)")
 }
 
+func TestReachabilityUpdate(t *testing.T) {
+	s1, _ := newService(t, 1, libp2pServiceOpts{
+		libp2pOpts: libp2p.WithHostFactory(
+			func(ctx context.Context, _ ...libp2pm.Option) (host.Host, error) {
+				return bhost.NewHost(context.TODO(), swarmt.GenSwarm(t, context.TODO()), &bhost.HostOpts{})
+			},
+		),
+	})
+	defer s1.Close()
+
+	emitReachabilityChanged, _ := s1.Host().EventBus().Emitter(new(event.EvtLocalReachabilityChanged), eventbus.Stateful)
+
+	firstUpdate := make(chan struct{})
+	s1.SetPickyNotifier(mockReachabilityNotifier(func(status p2p.ReachabilityStatus) {
+		if status == p2p.ReachabilityStatusPublic {
+			close(firstUpdate)
+		}
+	}))
+
+	err := emitReachabilityChanged.Emit(event.EvtLocalReachabilityChanged{Reachability: network.ReachabilityPublic})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	select {
+	case <-firstUpdate:
+	case <-time.After(time.Second):
+		t.Fatalf("test timed out")
+	}
+
+	secondUpdate := make(chan struct{})
+	s1.SetPickyNotifier(mockReachabilityNotifier(func(status p2p.ReachabilityStatus) {
+		if status == p2p.ReachabilityStatusPrivate {
+			close(secondUpdate)
+		}
+	}))
+
+	err = emitReachabilityChanged.Emit(event.EvtLocalReachabilityChanged{Reachability: network.ReachabilityPrivate})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	select {
+	case <-secondUpdate:
+	case <-time.After(time.Second):
+		t.Fatalf("test timed out")
+	}
+}
+
 func testUserAgentLogLine(t *testing.T, logs *buffer, substring string) {
 	t.Helper()
 
@@ -990,11 +1047,13 @@ func checkAddressbook(t *testing.T, ab addressbook.Getter, overlay swarm.Address
 }
 
 type notifiee struct {
-	connected    cFunc
-	disconnected dFunc
-	pick         bool
-	announce     announceFunc
-	announceTo   announceToFunc
+	connected          cFunc
+	disconnected       dFunc
+	pick               bool
+	announce           announceFunc
+	announceTo         announceToFunc
+	updateReachability reachabilityFunc
+	reachable          reachableFunc
 }
 
 func (n *notifiee) Connected(c context.Context, p p2p.Peer, f bool) error {
@@ -1005,7 +1064,7 @@ func (n *notifiee) Disconnected(p p2p.Peer) {
 	n.disconnected(p)
 }
 
-func (n *notifiee) Pick(p p2p.Peer) bool {
+func (n *notifiee) Pick(p2p.Peer) bool {
 	return n.pick
 }
 
@@ -1017,22 +1076,62 @@ func (n *notifiee) AnnounceTo(ctx context.Context, a, b swarm.Address, full bool
 	return n.announceTo(ctx, a, b, full)
 }
 
+func (n *notifiee) UpdateReachability(status p2p.ReachabilityStatus) {
+	n.updateReachability(status)
+}
+
+func (n *notifiee) Reachable(addr swarm.Address, status p2p.ReachabilityStatus) {
+	n.reachable(addr, status)
+}
+
 func mockNotifier(c cFunc, d dFunc, pick bool) p2p.PickyNotifier {
-	return &notifiee{connected: c, disconnected: d, pick: pick, announce: noopAnnounce, announceTo: noopAnnounceTo}
+	return &notifiee{
+		connected:          c,
+		disconnected:       d,
+		pick:               pick,
+		announce:           noopAnnounce,
+		announceTo:         noopAnnounceTo,
+		updateReachability: noopReachability,
+		reachable:          noopReachable,
+	}
 }
 
 func mockAnnouncingNotifier(a announceFunc, at announceToFunc) p2p.PickyNotifier {
-	return &notifiee{connected: noopCf, disconnected: noopDf, pick: true, announce: a, announceTo: at}
+	return &notifiee{
+		connected:          noopCf,
+		disconnected:       noopDf,
+		pick:               true,
+		announce:           a,
+		announceTo:         at,
+		updateReachability: noopReachability,
+		reachable:          noopReachable,
+	}
+}
+
+func mockReachabilityNotifier(r reachabilityFunc) p2p.PickyNotifier {
+	return &notifiee{
+		connected:          noopCf,
+		disconnected:       noopDf,
+		pick:               true,
+		announce:           noopAnnounce,
+		announceTo:         noopAnnounceTo,
+		updateReachability: r,
+		reachable:          noopReachable,
+	}
 }
 
 type (
-	cFunc          func(context.Context, p2p.Peer, bool) error
-	dFunc          func(p2p.Peer)
-	announceFunc   func(context.Context, swarm.Address, bool) error
-	announceToFunc func(context.Context, swarm.Address, swarm.Address, bool) error
+	cFunc            func(context.Context, p2p.Peer, bool) error
+	dFunc            func(p2p.Peer)
+	announceFunc     func(context.Context, swarm.Address, bool) error
+	announceToFunc   func(context.Context, swarm.Address, swarm.Address, bool) error
+	reachabilityFunc func(p2p.ReachabilityStatus)
+	reachableFunc    func(swarm.Address, p2p.ReachabilityStatus)
 )
 
 var noopCf = func(context.Context, p2p.Peer, bool) error { return nil }
 var noopDf = func(p2p.Peer) {}
 var noopAnnounce = func(context.Context, swarm.Address, bool) error { return nil }
 var noopAnnounceTo = func(context.Context, swarm.Address, swarm.Address, bool) error { return nil }
+var noopReachability = func(p2p.ReachabilityStatus) {}
+var noopReachable = func(swarm.Address, p2p.ReachabilityStatus) {}

pkg/p2p/libp2p/export_test.go

@@ -22,6 +22,10 @@ func (s *Service) NewStreamForPeerID(peerID libp2ppeer.ID, protocolName, protoco
 	return s.newStreamForPeerID(context.Background(), peerID, protocolName, protocolVersion, streamName)
 }
 
+func (s *Service) Host() host.Host {
+	return s.host
+}
+
 type StaticAddressResolver = staticAddressResolver
 
 var NewStaticAddressResolver = newStaticAddressResolver

pkg/p2p/libp2p/internal/reacher/metrics.go

@@ -0,0 +1,38 @@
+// Copyright 2021 The Swarm Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package reacher
+
+import (
+	m "github.com/ethersphere/bee/pkg/metrics"
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+type metrics struct {
+	Pings    prometheus.CounterVec
+	PingTime prometheus.HistogramVec
+}
+
+func newMetrics() metrics {
+	subsystem := "reacher"
+
+	return metrics{
+		Pings: *prometheus.NewCounterVec(prometheus.CounterOpts{
+			Namespace: m.Namespace,
+			Subsystem: subsystem,
+			Name:      "pings",
+			Help:      "Ping counter.",
+		}, []string{"status"}),
+		PingTime: *prometheus.NewHistogramVec(prometheus.HistogramOpts{
+			Namespace: m.Namespace,
+			Subsystem: subsystem,
+			Name:      "ping_timer",
+			Help:      "Ping timer.",
+		}, []string{"status"}),
+	}
+}
+
+func (s *reacher) Metrics() []prometheus.Collector {
+	return m.PrometheusCollectorsFromFields(s.metrics)
+}