chore: migrate peering to ipfs/boxo (#10157)
Co-authored-by: Henrique Dias <hacdias@gmail.com>
Andrew Gillis committed
Oct 31, 2023 at 06:45 UTC
ab7630fcd497c3761ecff76d35e1e9987dbbbec2
4 files changed
+4
-500
core/core.go
+1
-1
@@ -49,13 +49,13 @@ import (
49
50
"github.com/ipfs/boxo/namesys"
51
ipnsrp "github.com/ipfs/boxo/namesys/republisher"
52
+ "github.com/ipfs/boxo/peering"
53
"github.com/ipfs/kubo/config"
54
"github.com/ipfs/kubo/core/bootstrap"
55
"github.com/ipfs/kubo/core/node"
56
"github.com/ipfs/kubo/core/node/libp2p"
57
"github.com/ipfs/kubo/fuse/mount"
58
"github.com/ipfs/kubo/p2p"
58
- "github.com/ipfs/kubo/peering"
59
"github.com/ipfs/kubo/repo"
60
irouting "github.com/ipfs/kubo/routing"
61
)
core/node/peering.go
+3
-2
@@ -3,7 +3,7 @@ package node
3
import (
4
"context"
5
6
- "github.com/ipfs/kubo/peering"
6
+ "github.com/ipfs/boxo/peering"
7
"github.com/libp2p/go-libp2p/core/host"
8
"github.com/libp2p/go-libp2p/core/peer"
9
"go.uber.org/fx"
@@ -18,7 +18,8 @@ func Peering(lc fx.Lifecycle, host host.Host) *peering.PeeringService {
18
return ps.Start()
19
},
20
OnStop: func(context.Context) error {
21
- return ps.Stop()
21
+ ps.Stop()
22
+ return nil
23
},
24
})
25
return ps
peering/peering.go
deleted
-325
@@ -1,325 +0,0 @@
1
-package peering
2
-
3
-import (
4
- "context"
5
- "errors"
6
- "math/rand"
7
- "strconv"
8
- "sync"
9
- "time"
10
-
11
- "github.com/ipfs/go-log"
12
- "github.com/libp2p/go-libp2p/core/host"
13
- "github.com/libp2p/go-libp2p/core/network"
14
- "github.com/libp2p/go-libp2p/core/peer"
15
- "github.com/multiformats/go-multiaddr"
16
-)
17
-
18
-// Seed the random number generator.
19
-//
20
-// We don't need good randomness, but we do need randomness.
21
-const (
22
- // maxBackoff is the maximum time between reconnect attempts.
23
- maxBackoff = 10 * time.Minute
24
- // The backoff will be cut off when we get within 10% of the actual max.
25
- // If we go over the max, we'll adjust the delay down to a random value
26
- // between 90-100% of the max backoff.
27
- maxBackoffJitter = 10 // %
28
- connmgrTag = "ipfs-peering"
29
- // This needs to be sufficient to prevent two sides from simultaneously
30
- // dialing.
31
- initialDelay = 5 * time.Second
32
-)
33
-
34
-var logger = log.Logger("peering")
35
-
36
-type State uint
37
-
38
-func (s State) String() string {
39
- switch s {
40
- case StateInit:
41
- return "init"
42
- case StateRunning:
43
- return "running"
44
- case StateStopped:
45
- return "stopped"
46
- default:
47
- return "unknown peering state: " + strconv.FormatUint(uint64(s), 10)
48
- }
49
-}
50
-
51
-const (
52
- StateInit State = iota
53
- StateRunning
54
- StateStopped
55
-)
56
-
57
-// peerHandler keeps track of all state related to a specific "peering" peer.
58
-type peerHandler struct {
59
- peer peer.ID
60
- host host.Host
61
- ctx context.Context
62
- cancel context.CancelFunc
63
-
64
- mu sync.Mutex
65
- addrs []multiaddr.Multiaddr
66
- reconnectTimer *time.Timer
67
-
68
- nextDelay time.Duration
69
-}
70
-
71
-// setAddrs sets the addresses for this peer.
72
-func (ph *peerHandler) setAddrs(addrs []multiaddr.Multiaddr) {
73
- // Not strictly necessary, but it helps to not trust the calling code.
74
- addrCopy := make([]multiaddr.Multiaddr, len(addrs))
75
- copy(addrCopy, addrs)
76
-
77
- ph.mu.Lock()
78
- defer ph.mu.Unlock()
79
- ph.addrs = addrCopy
80
-}
81
-
82
-// getAddrs returns a shared slice of addresses for this peer. Do not modify.
83
-func (ph *peerHandler) getAddrs() []multiaddr.Multiaddr {
84
- ph.mu.Lock()
85
- defer ph.mu.Unlock()
86
- return ph.addrs
87
-}
88
-
89
-// stop permanently stops the peer handler.
90
-func (ph *peerHandler) stop() {
91
- ph.cancel()
92
-
93
- ph.mu.Lock()
94
- defer ph.mu.Unlock()
95
- if ph.reconnectTimer != nil {
96
- ph.reconnectTimer.Stop()
97
- ph.reconnectTimer = nil
98
- }
99
-}
100
-
101
-func (ph *peerHandler) nextBackoff() time.Duration {
102
- if ph.nextDelay < maxBackoff {
103
- ph.nextDelay += ph.nextDelay/2 + time.Duration(rand.Int63n(int64(ph.nextDelay)))
104
- }
105
-
106
- // If we've gone over the max backoff, reduce it under the max.
107
- if ph.nextDelay > maxBackoff {
108
- ph.nextDelay = maxBackoff
109
- // randomize the backoff a bit (10%).
110
- ph.nextDelay -= time.Duration(rand.Int63n(int64(maxBackoff) * maxBackoffJitter / 100))
111
- }
112
-
113
- return ph.nextDelay
114
-}
115
-
116
-func (ph *peerHandler) reconnect() {
117
- // Try connecting
118
- addrs := ph.getAddrs()
119
- logger.Debugw("reconnecting", "peer", ph.peer, "addrs", addrs)
120
-
121
- err := ph.host.Connect(ph.ctx, peer.AddrInfo{ID: ph.peer, Addrs: addrs})
122
- if err != nil {
123
- logger.Debugw("failed to reconnect", "peer", ph.peer, "error", err)
124
- // Ok, we failed. Extend the timeout.
125
- ph.mu.Lock()
126
- if ph.reconnectTimer != nil {
127
- // Only counts if the reconnectTimer still exists. If not, a
128
- // connection _was_ somehow established.
129
- ph.reconnectTimer.Reset(ph.nextBackoff())
130
- }
131
- // Otherwise, someone else has stopped us so we can assume that
132
- // we're either connected or someone else will start us.
133
- ph.mu.Unlock()
134
- }
135
-
136
- // Always call this. We could have connected since we processed the
137
- // error.
138
- ph.stopIfConnected()
139
-}
140
-
141
-func (ph *peerHandler) stopIfConnected() {
142
- ph.mu.Lock()
143
- defer ph.mu.Unlock()
144
-
145
- if ph.reconnectTimer != nil && ph.host.Network().Connectedness(ph.peer) == network.Connected {
146
- logger.Debugw("successfully reconnected", "peer", ph.peer)
147
- ph.reconnectTimer.Stop()
148
- ph.reconnectTimer = nil
149
- ph.nextDelay = initialDelay
150
- }
151
-}
152
-
153
-// startIfDisconnected is the inverse of stopIfConnected.
154
-func (ph *peerHandler) startIfDisconnected() {
155
- ph.mu.Lock()
156
- defer ph.mu.Unlock()
157
-
158
- if ph.reconnectTimer == nil && ph.host.Network().Connectedness(ph.peer) != network.Connected {
159
- logger.Debugw("disconnected from peer", "peer", ph.peer)
160
- // Always start with a short timeout so we can stagger things a bit.
161
- ph.reconnectTimer = time.AfterFunc(ph.nextBackoff(), ph.reconnect)
162
- }
163
-}
164
-
165
-// PeeringService maintains connections to specified peers, reconnecting on
166
-// disconnect with a back-off.
167
-type PeeringService struct {
168
- host host.Host
169
-
170
- mu sync.RWMutex
171
- peers map[peer.ID]*peerHandler
172
- state State
173
-}
174
-
175
-// NewPeeringService constructs a new peering service. Peers can be added and
176
-// removed immediately, but connections won't be formed until `Start` is called.
177
-func NewPeeringService(host host.Host) *PeeringService {
178
- return &PeeringService{host: host, peers: make(map[peer.ID]*peerHandler)}
179
-}
180
-
181
-// Start starts the peering service, connecting and maintaining connections to
182
-// all registered peers. It returns an error if the service has already been
183
-// stopped.
184
-func (ps *PeeringService) Start() error {
185
- ps.mu.Lock()
186
- defer ps.mu.Unlock()
187
-
188
- switch ps.state {
189
- case StateInit:
190
- logger.Infow("starting")
191
- case StateRunning:
192
- return nil
193
- case StateStopped:
194
- return errors.New("already stopped")
195
- }
196
- ps.host.Network().Notify((*netNotifee)(ps))
197
- ps.state = StateRunning
198
- for _, handler := range ps.peers {
199
- go handler.startIfDisconnected()
200
- }
201
- return nil
202
-}
203
-
204
-// GetState get the State of the PeeringService.
205
-func (ps *PeeringService) GetState() State {
206
- ps.mu.RLock()
207
- defer ps.mu.RUnlock()
208
- return ps.state
209
-}
210
-
211
-// Stop stops the peering service.
212
-func (ps *PeeringService) Stop() error {
213
- ps.host.Network().StopNotify((*netNotifee)(ps))
214
- ps.mu.Lock()
215
- defer ps.mu.Unlock()
216
-
217
- switch ps.state {
218
- case StateInit, StateRunning:
219
- logger.Infow("stopping")
220
- for _, handler := range ps.peers {
221
- handler.stop()
222
- }
223
- ps.state = StateStopped
224
- }
225
- return nil
226
-}
227
-
228
-// AddPeer adds a peer to the peering service. This function may be safely
229
-// called at any time: before the service is started, while running, or after it
230
-// stops.
231
-//
232
-// Add peer may also be called multiple times for the same peer. The new
233
-// addresses will replace the old.
234
-func (ps *PeeringService) AddPeer(info peer.AddrInfo) {
235
- ps.mu.Lock()
236
- defer ps.mu.Unlock()
237
-
238
- if handler, ok := ps.peers[info.ID]; ok {
239
- logger.Infow("updating addresses", "peer", info.ID, "addrs", info.Addrs)
240
- handler.setAddrs(info.Addrs)
241
- } else {
242
- logger.Infow("peer added", "peer", info.ID, "addrs", info.Addrs)
243
- ps.host.ConnManager().Protect(info.ID, connmgrTag)
244
-
245
- handler = &peerHandler{
246
- host: ps.host,
247
- peer: info.ID,
248
- addrs: info.Addrs,
249
- nextDelay: initialDelay,
250
- }
251
- handler.ctx, handler.cancel = context.WithCancel(context.Background())
252
- ps.peers[info.ID] = handler
253
- switch ps.state {
254
- case StateRunning:
255
- go handler.startIfDisconnected()
256
- case StateStopped:
257
- // We still construct everything in this state because
258
- // it's easier to reason about. But we should still free
259
- // resources.
260
- handler.cancel()
261
- }
262
- }
263
-}
264
-
265
-// ListPeers lists peers in the peering service.
266
-func (ps *PeeringService) ListPeers() []peer.AddrInfo {
267
- ps.mu.RLock()
268
- defer ps.mu.RUnlock()
269
-
270
- out := make([]peer.AddrInfo, 0, len(ps.peers))
271
- for id, addrs := range ps.peers {
272
- ai := peer.AddrInfo{ID: id}
273
- ai.Addrs = append(ai.Addrs, addrs.addrs...)
274
- out = append(out, ai)
275
- }
276
- return out
277
-}
278
-
279
-// RemovePeer removes a peer from the peering service. This function may be
280
-// safely called at any time: before the service is started, while running, or
281
-// after it stops.
282
-func (ps *PeeringService) RemovePeer(id peer.ID) {
283
- ps.mu.Lock()
284
- defer ps.mu.Unlock()
285
-
286
- if handler, ok := ps.peers[id]; ok {
287
- logger.Infow("peer removed", "peer", id)
288
- ps.host.ConnManager().Unprotect(id, connmgrTag)
289
-
290
- handler.stop()
291
- delete(ps.peers, id)
292
- }
293
-}
294
-
295
-type netNotifee PeeringService
296
-
297
-func (nn *netNotifee) Connected(_ network.Network, c network.Conn) {
298
- ps := (*PeeringService)(nn)
299
-
300
- p := c.RemotePeer()
301
- ps.mu.RLock()
302
- defer ps.mu.RUnlock()
303
-
304
- if handler, ok := ps.peers[p]; ok {
305
- // use a goroutine to avoid blocking events.
306
- go handler.stopIfConnected()
307
- }
308
-}
309
-
310
-func (nn *netNotifee) Disconnected(_ network.Network, c network.Conn) {
311
- ps := (*PeeringService)(nn)
312
-
313
- p := c.RemotePeer()
314
- ps.mu.RLock()
315
- defer ps.mu.RUnlock()
316
-
317
- if handler, ok := ps.peers[p]; ok {
318
- // use a goroutine to avoid blocking events.
319
- go handler.startIfDisconnected()
320
- }
321
-}
322
-func (nn *netNotifee) OpenedStream(network.Network, network.Stream) {}
323
-func (nn *netNotifee) ClosedStream(network.Network, network.Stream) {}
324
-func (nn *netNotifee) Listen(network.Network, multiaddr.Multiaddr) {}
325
-func (nn *netNotifee) ListenClose(network.Network, multiaddr.Multiaddr) {}
peering/peering_test.go
deleted
-172
@@ -1,172 +0,0 @@
1
-package peering
2
-
3
-import (
4
- "context"
5
- "testing"
6
- "time"
7
-
8
- "github.com/libp2p/go-libp2p"
9
- "github.com/libp2p/go-libp2p/core/host"
10
- "github.com/libp2p/go-libp2p/core/network"
11
- "github.com/libp2p/go-libp2p/core/peer"
12
- "github.com/libp2p/go-libp2p/p2p/net/connmgr"
13
-
14
- "github.com/stretchr/testify/require"
15
-)
16
-
17
-func newNode(t *testing.T) host.Host {
18
- cm, err := connmgr.NewConnManager(1, 100, connmgr.WithGracePeriod(0))
19
- require.NoError(t, err)
20
- h, err := libp2p.New(
21
- libp2p.ListenAddrStrings("/ip4/127.0.0.1/tcp/0"),
22
- // We'd like to set the connection manager low water to 0, but
23
- // that would disable the connection manager.
24
- libp2p.ConnectionManager(cm),
25
- )
26
- require.NoError(t, err)
27
- return h
28
-}
29
-
30
-func TestPeeringService(t *testing.T) {
31
- ctx, cancel := context.WithCancel(context.Background())
32
- defer cancel()
33
-
34
- h1 := newNode(t)
35
- ps1 := NewPeeringService(h1)
36
-
37
- h2 := newNode(t)
38
- h3 := newNode(t)
39
- h4 := newNode(t)
40
-
41
- // peer 1 -> 2
42
- ps1.AddPeer(peer.AddrInfo{ID: h2.ID(), Addrs: h2.Addrs()})
43
- require.Contains(t, ps1.ListPeers(), peer.AddrInfo{ID: h2.ID(), Addrs: h2.Addrs()})
44
-
45
- // We haven't started so we shouldn't have any peers.
46
- require.Never(t, func() bool {
47
- return len(h1.Network().Peers()) > 0
48
- }, 100*time.Millisecond, 1*time.Second, "expected host 1 to have no peers")
49
-
50
- // Use p4 to take up the one slot we have in the connection manager.
51
- for _, h := range []host.Host{h1, h2} {
52
- require.NoError(t, h.Connect(ctx, peer.AddrInfo{ID: h4.ID(), Addrs: h4.Addrs()}))
53
- h.ConnManager().TagPeer(h4.ID(), "sticky-peer", 1000)
54
- }
55
-
56
- // Now start.
57
- require.NoError(t, ps1.Start())
58
- // starting twice is fine.
59
- require.NoError(t, ps1.Start())
60
-
61
- // We should eventually connect.
62
- t.Logf("waiting for h1 to connect to h2")
63
- require.Eventually(t, func() bool {
64
- return h1.Network().Connectedness(h2.ID()) == network.Connected
65
- }, 30*time.Second, 10*time.Millisecond)
66
-
67
- // Now explicitly connect to h3.
68
- t.Logf("waiting for h1's connection to h3 to work")
69
- require.NoError(t, h1.Connect(ctx, peer.AddrInfo{ID: h3.ID(), Addrs: h3.Addrs()}))
70
- require.Eventually(t, func() bool {
71
- return h1.Network().Connectedness(h3.ID()) == network.Connected
72
- }, 30*time.Second, 100*time.Millisecond)
73
-
74
- require.Len(t, h1.Network().Peers(), 3)
75
-
76
- // force a disconnect
77
- h1.ConnManager().TrimOpenConns(ctx)
78
-
79
- // Should disconnect from h3.
80
- t.Logf("waiting for h1's connection to h3 to disconnect")
81
- require.Eventually(t, func() bool {
82
- return h1.Network().Connectedness(h3.ID()) != network.Connected
83
- }, 5*time.Second, 10*time.Millisecond)
84
-
85
- // Should remain connected to p2
86
- require.Never(t, func() bool {
87
- return h1.Network().Connectedness(h2.ID()) != network.Connected
88
- }, 5*time.Second, 1*time.Second)
89
-
90
- // Now force h2 to disconnect (we have an asymmetric peering).
91
- conns := h2.Network().ConnsToPeer(h1.ID())
92
- require.NotEmpty(t, conns)
93
- h2.ConnManager().TrimOpenConns(ctx)
94
-
95
- // All conns to peer should eventually close.
96
- t.Logf("waiting for all connections to close")
97
- for _, c := range conns {
98
- require.Eventually(t, func() bool {
99
- s, err := c.NewStream(context.Background())
100
- if s != nil {
101
- _ = s.Reset()
102
- }
103
- return err != nil
104
- }, 5*time.Second, 10*time.Millisecond)
105
- }
106
-
107
- // Should eventually re-connect.
108
- require.Eventually(t, func() bool {
109
- return h1.Network().Connectedness(h2.ID()) == network.Connected
110
- }, 30*time.Second, 1*time.Second)
111
-
112
- // Unprotect 2 from 1.
113
- ps1.RemovePeer(h2.ID())
114
- require.NotContains(t, ps1.ListPeers(), peer.AddrInfo{ID: h2.ID(), Addrs: h2.Addrs()})
115
-
116
- // Trim connections.
117
- h1.ConnManager().TrimOpenConns(ctx)
118
-
119
- // Should disconnect
120
- t.Logf("waiting for h1 to disconnect from h2")
121
- require.Eventually(t, func() bool {
122
- return h1.Network().Connectedness(h2.ID()) != network.Connected
123
- }, 5*time.Second, 10*time.Millisecond)
124
-
125
- // Should never reconnect.
126
- t.Logf("ensuring h1 is not connected to h2 again")
127
- require.Never(t, func() bool {
128
- return h1.Network().Connectedness(h2.ID()) == network.Connected
129
- }, 20*time.Second, 1*time.Second)
130
-
131
- // Until added back
132
- ps1.AddPeer(peer.AddrInfo{ID: h2.ID(), Addrs: h2.Addrs()})
133
- require.Contains(t, ps1.ListPeers(), peer.AddrInfo{ID: h2.ID(), Addrs: h2.Addrs()})
134
- ps1.AddPeer(peer.AddrInfo{ID: h3.ID(), Addrs: h3.Addrs()})
135
- require.Contains(t, ps1.ListPeers(), peer.AddrInfo{ID: h3.ID(), Addrs: h3.Addrs()})
136
- t.Logf("wait for h1 to connect to h2 and h3 again")
137
- require.Eventually(t, func() bool {
138
- return h1.Network().Connectedness(h2.ID()) == network.Connected
139
- }, 30*time.Second, 1*time.Second)
140
- require.Eventually(t, func() bool {
141
- return h1.Network().Connectedness(h3.ID()) == network.Connected
142
- }, 30*time.Second, 1*time.Second)
143
-
144
- // Should be able to repeatedly stop.
145
- require.NoError(t, ps1.Stop())
146
- require.NoError(t, ps1.Stop())
147
-
148
- // Adding and removing should work after stopping.
149
- ps1.AddPeer(peer.AddrInfo{ID: h4.ID(), Addrs: h4.Addrs()})
150
- require.Contains(t, ps1.ListPeers(), peer.AddrInfo{ID: h4.ID(), Addrs: h4.Addrs()})
151
- ps1.RemovePeer(h2.ID())
152
- require.NotContains(t, ps1.ListPeers(), peer.AddrInfo{ID: h2.ID(), Addrs: h2.Addrs()})
153
-}
154
-
155
-func TestNextBackoff(t *testing.T) {
156
- minMaxBackoff := (100 - maxBackoffJitter) / 100 * maxBackoff
157
- for x := 0; x < 1000; x++ {
158
- ph := peerHandler{nextDelay: time.Second}
159
- for min, max := time.Second*3/2, time.Second*5/2; min < minMaxBackoff; min, max = min*3/2, max*5/2 {
160
- b := ph.nextBackoff()
161
- if b > max || b < min {
162
- t.Errorf("expected backoff %s to be between %s and %s", b, min, max)
163
- }
164
- }
165
- for i := 0; i < 100; i++ {
166
- b := ph.nextBackoff()
167
- if b < minMaxBackoff || b > maxBackoff {
168
- t.Fatal("failed to stay within max bounds")
169
- }
170
- }
171
- }
172
-}