feat(bootstrap): save connected peers as backup bootstrap peers (#8856)
* feat(bootstrap): save connected peers as backup temporary bootstrap ones * fix: do not add duplicated oldSavedPeers, not using tags, reuse randomizeList * test: add regression test * chore: add changelog --------- Co-authored-by: Henrique Dias <hacdias@gmail.com> Co-authored-by: Marcin Rataj <lidel@lidel.org>
Lucas Molas committed
May 25, 2023 at 09:39 UTC
63561f3baf63524ce7d147f67c0c4b4e0ddc5bc9
6 files changed
+323
-56
config/internal.go
+4
-3
@@ -2,9 +2,10 @@ package config
2
3
type Internal struct {
4
// All marked as omitempty since we are expecting to make changes to all subcomponents of Internal
5
- Bitswap *InternalBitswap `json:",omitempty"`
6
- UnixFSShardingSizeThreshold *OptionalString `json:",omitempty"`
7
- Libp2pForceReachability *OptionalString `json:",omitempty"`
5
+ Bitswap *InternalBitswap `json:",omitempty"`
6
+ UnixFSShardingSizeThreshold *OptionalString `json:",omitempty"`
7
+ Libp2pForceReachability *OptionalString `json:",omitempty"`
8
+ BackupBootstrapInterval *OptionalDuration `json:",omitempty"`
9
}
10
11
type InternalBitswap struct {
core/bootstrap/bootstrap.go
+180
-49
@@ -3,16 +3,16 @@ package bootstrap
3
import (
4
"context"
5
"errors"
6
- "fmt"
6
"io"
7
"math/rand"
8
"sync"
9
+ "sync/atomic"
10
"time"
11
12
logging "github.com/ipfs/go-log"
13
"github.com/jbenet/goprocess"
14
- "github.com/jbenet/goprocess/context"
15
- "github.com/jbenet/goprocess/periodic"
14
+ goprocessctx "github.com/jbenet/goprocess/context"
15
+ periodicproc "github.com/jbenet/goprocess/periodic"
16
"github.com/libp2p/go-libp2p/core/host"
17
"github.com/libp2p/go-libp2p/core/network"
18
"github.com/libp2p/go-libp2p/core/peer"
@@ -50,13 +50,26 @@ type BootstrapConfig struct {
50
// for the bootstrap process to use. This makes it possible for clients
51
// to control the peers the process uses at any moment.
52
BootstrapPeers func() []peer.AddrInfo
53
+
54
+ // BackupBootstrapInterval governs the periodic interval at which the node will
55
+ // attempt to save connected nodes to use as temporary bootstrap peers.
56
+ BackupBootstrapInterval time.Duration
57
+
58
+ // MaxBackupBootstrapSize controls the maximum number of peers we're saving
59
+ // as backup bootstrap peers.
60
+ MaxBackupBootstrapSize int
61
+
62
+ SaveBackupBootstrapPeers func(context.Context, []peer.AddrInfo)
63
+ LoadBackupBootstrapPeers func(context.Context) []peer.AddrInfo
64
}
65
66
// DefaultBootstrapConfig specifies default sane parameters for bootstrapping.
67
var DefaultBootstrapConfig = BootstrapConfig{
57
- MinPeerThreshold: 4,
58
- Period: 30 * time.Second,
59
- ConnectionTimeout: (30 * time.Second) / 3, // Perod / 3
68
+ MinPeerThreshold: 4,
69
+ Period: 30 * time.Second,
70
+ ConnectionTimeout: (30 * time.Second) / 3, // Perod / 3
71
+ BackupBootstrapInterval: 1 * time.Hour,
72
+ MaxBackupBootstrapSize: 20,
73
}
74
75
func BootstrapConfigWithPeers(pis []peer.AddrInfo) BootstrapConfig {
@@ -90,6 +103,9 @@ func Bootstrap(id peer.ID, host host.Host, rt routing.Routing, cfg BootstrapConf
103
log.Debugf("%s bootstrap error: %s", id, err)
104
}
105
106
+ // Exit the first call (triggered independently by `proc.Go`, not `Tick`)
107
+ // only after being done with the *single* Routing.Bootstrap call. Following
108
+ // periodic calls (`Tick`) will not block on this.
109
<-doneWithRound
110
}
111
@@ -108,9 +124,100 @@ func Bootstrap(id peer.ID, host host.Host, rt routing.Routing, cfg BootstrapConf
124
125
doneWithRound <- struct{}{}
126
close(doneWithRound) // it no longer blocks periodic
127
+
128
+ startSavePeersAsTemporaryBootstrapProc(cfg, host, proc)
129
+
130
return proc, nil
131
}
132
133
+// Aside of the main bootstrap process we also run a secondary one that saves
134
+// connected peers as a backup measure if we can't connect to the official
135
+// bootstrap ones. These peers will serve as *temporary* bootstrap nodes.
136
+func startSavePeersAsTemporaryBootstrapProc(cfg BootstrapConfig, host host.Host, bootstrapProc goprocess.Process) {
137
+ savePeersFn := func(worker goprocess.Process) {
138
+ ctx := goprocessctx.OnClosingContext(worker)
139
+
140
+ if err := saveConnectedPeersAsTemporaryBootstrap(ctx, host, cfg); err != nil {
141
+ log.Debugf("saveConnectedPeersAsTemporaryBootstrap error: %s", err)
142
+ }
143
+ }
144
+ savePeersProc := periodicproc.Tick(cfg.BackupBootstrapInterval, savePeersFn)
145
+
146
+ // When the main bootstrap process ends also terminate the 'save connected
147
+ // peers' ones. Coupling the two seems the easiest way to handle this backup
148
+ // process without additional complexity.
149
+ go func() {
150
+ <-bootstrapProc.Closing()
151
+ savePeersProc.Close()
152
+ }()
153
+
154
+ // Run the first round now (after the first bootstrap process has finished)
155
+ // as the SavePeersPeriod can be much longer than bootstrap.
156
+ savePeersProc.Go(savePeersFn)
157
+}
158
+
159
+func saveConnectedPeersAsTemporaryBootstrap(ctx context.Context, host host.Host, cfg BootstrapConfig) error {
160
+ // Randomize the list of connected peers, we don't prioritize anyone.
161
+ connectedPeers := randomizeList(host.Network().Peers())
162
+
163
+ bootstrapPeers := cfg.BootstrapPeers()
164
+ backupPeers := make([]peer.AddrInfo, 0, cfg.MaxBackupBootstrapSize)
165
+
166
+ // Choose peers to save and filter out the ones that are already bootstrap nodes.
167
+ for _, p := range connectedPeers {
168
+ found := false
169
+ for _, bootstrapPeer := range bootstrapPeers {
170
+ if p == bootstrapPeer.ID {
171
+ found = true
172
+ break
173
+ }
174
+ }
175
+ if !found {
176
+ backupPeers = append(backupPeers, peer.AddrInfo{
177
+ ID: p,
178
+ Addrs: host.Network().Peerstore().Addrs(p),
179
+ })
180
+ }
181
+
182
+ if len(backupPeers) >= cfg.MaxBackupBootstrapSize {
183
+ break
184
+ }
185
+ }
186
+
187
+ // If we didn't reach the target number use previously stored connected peers.
188
+ if len(backupPeers) < cfg.MaxBackupBootstrapSize {
189
+ oldSavedPeers := cfg.LoadBackupBootstrapPeers(ctx)
190
+ log.Debugf("missing %d peers to reach backup bootstrap target of %d, trying from previous list of %d saved peers",
191
+ cfg.MaxBackupBootstrapSize-len(backupPeers), cfg.MaxBackupBootstrapSize, len(oldSavedPeers))
192
+
193
+ // Add some of the old saved peers. Ensure we don't duplicate them.
194
+ for _, p := range oldSavedPeers {
195
+ found := false
196
+ for _, sp := range backupPeers {
197
+ if p.ID == sp.ID {
198
+ found = true
199
+ break
200
+ }
201
+ }
202
+
203
+ if !found {
204
+ backupPeers = append(backupPeers, p)
205
+ }
206
+
207
+ if len(backupPeers) >= cfg.MaxBackupBootstrapSize {
208
+ break
209
+ }
210
+ }
211
+ }
212
+
213
+ cfg.SaveBackupBootstrapPeers(ctx, backupPeers)
214
+ log.Debugf("saved %d peers (of %d target) as bootstrap backup in the config", len(backupPeers), cfg.MaxBackupBootstrapSize)
215
+ return nil
216
+}
217
+
218
+// Connect to as many peers needed to reach the BootstrapConfig.MinPeerThreshold.
219
+// Peers can be original bootstrap or temporary ones (drawn from a list of
220
+// persisted previously connected peers).
221
func bootstrapRound(ctx context.Context, host host.Host, cfg BootstrapConfig) error {
222
223
ctx, cancel := context.WithTimeout(ctx, cfg.ConnectionTimeout)
@@ -127,35 +234,58 @@ func bootstrapRound(ctx context.Context, host host.Host, cfg BootstrapConfig) er
234
id, len(connected), cfg.MinPeerThreshold)
235
return nil
236
}
130
- numToDial := cfg.MinPeerThreshold - len(connected)
237
+ numToDial := cfg.MinPeerThreshold - len(connected) // numToDial > 0
238
132
- // filter out bootstrap nodes we are already connected to
133
- var notConnected []peer.AddrInfo
134
- for _, p := range peers {
135
- if host.Network().Connectedness(p.ID) != network.Connected {
136
- notConnected = append(notConnected, p)
239
+ if len(peers) > 0 {
240
+ numToDial -= int(peersConnect(ctx, host, peers, numToDial, true))
241
+ if numToDial <= 0 {
242
+ return nil
243
}
244
}
245
140
- // if connected to all bootstrap peer candidates, exit
141
- if len(notConnected) < 1 {
142
- log.Debugf("%s no more bootstrap peers to create %d connections", id, numToDial)
143
- return ErrNotEnoughBootstrapPeers
246
+ log.Debugf("not enough bootstrap peers to fill the remaining target of %d connections, trying backup list", numToDial)
247
+
248
+ tempBootstrapPeers := cfg.LoadBackupBootstrapPeers(ctx)
249
+ if len(tempBootstrapPeers) > 0 {
250
+ numToDial -= int(peersConnect(ctx, host, tempBootstrapPeers, numToDial, false))
251
+ if numToDial <= 0 {
252
+ return nil
253
+ }
254
}
255
146
- // connect to a random susbset of bootstrap candidates
147
- randSubset := randomSubsetOfPeers(notConnected, numToDial)
256
+ log.Debugf("tried both original bootstrap peers and temporary ones but still missing target of %d connections", numToDial)
257
149
- log.Debugf("%s bootstrapping to %d nodes: %s", id, numToDial, randSubset)
150
- return bootstrapConnect(ctx, host, randSubset)
258
+ return ErrNotEnoughBootstrapPeers
259
}
260
153
-func bootstrapConnect(ctx context.Context, ph host.Host, peers []peer.AddrInfo) error {
154
- if len(peers) < 1 {
155
- return ErrNotEnoughBootstrapPeers
156
- }
261
+// Attempt to make `needed` connections from the `availablePeers` list. Mark
262
+// peers as either `permanent` or temporary when adding them to the Peerstore.
263
+// Return the number of connections completed. We eagerly over-connect in parallel,
264
+// so we might connect to more than needed.
265
+// (We spawn as many routines and attempt connections as the number of availablePeers,
266
+// but this list comes from restricted sets of original or temporary bootstrap
267
+// nodes which will keep it under a sane value.)
268
+func peersConnect(ctx context.Context, ph host.Host, availablePeers []peer.AddrInfo, needed int, permanent bool) uint64 {
269
+ peers := randomizeList(availablePeers)
270
+
271
+ // Monitor the number of connections and stop if we reach the target.
272
+ var connected uint64
273
+ ctx, cancel := context.WithCancel(ctx)
274
+ defer cancel()
275
+ go func() {
276
+ for {
277
+ select {
278
+ case <-ctx.Done():
279
+ return
280
+ case <-time.After(1 * time.Second):
281
+ if int(atomic.LoadUint64(&connected)) >= needed {
282
+ cancel()
283
+ return
284
+ }
285
+ }
286
+ }
287
+ }()
288
158
- errs := make(chan error, len(peers))
289
var wg sync.WaitGroup
290
for _, p := range peers {
291
@@ -164,45 +294,46 @@ func bootstrapConnect(ctx context.Context, ph host.Host, peers []peer.AddrInfo)
294
// fail/abort due to an expiring context.
295
// Also, performed asynchronously for dial speed.
296
297
+ if int(atomic.LoadUint64(&connected)) >= needed {
298
+ cancel()
299
+ break
300
+ }
301
+
302
wg.Add(1)
303
go func(p peer.AddrInfo) {
304
defer wg.Done()
305
+
306
+ // Skip addresses belonging to a peer we're already connected to.
307
+ // (Not a guarantee but a best-effort policy.)
308
+ if ph.Network().Connectedness(p.ID) == network.Connected {
309
+ return
310
+ }
311
log.Debugf("%s bootstrapping to %s", ph.ID(), p.ID)
312
172
- ph.Peerstore().AddAddrs(p.ID, p.Addrs, peerstore.PermanentAddrTTL)
313
if err := ph.Connect(ctx, p); err != nil {
174
- log.Debugf("failed to bootstrap with %v: %s", p.ID, err)
175
- errs <- err
314
+ if ctx.Err() != context.Canceled {
315
+ log.Debugf("failed to bootstrap with %v: %s", p.ID, err)
316
+ }
317
return
318
}
319
+ if permanent {
320
+ // We're connecting to an original bootstrap peer, mark it as
321
+ // a permanent address (Connect will register it as TempAddrTTL).
322
+ ph.Peerstore().AddAddrs(p.ID, p.Addrs, peerstore.PermanentAddrTTL)
323
+ }
324
+
325
log.Infof("bootstrapped with %v", p.ID)
326
+ atomic.AddUint64(&connected, 1)
327
}(p)
328
}
329
wg.Wait()
330
183
- // our failure condition is when no connection attempt succeeded.
184
- // So drain the errs channel, counting the results.
185
- close(errs)
186
- count := 0
187
- var err error
188
- for err = range errs {
189
- if err != nil {
190
- count++
191
- }
192
- }
193
- if count == len(peers) {
194
- return fmt.Errorf("failed to bootstrap. %s", err)
195
- }
196
- return nil
331
+ return connected
332
}
333
199
-func randomSubsetOfPeers(in []peer.AddrInfo, max int) []peer.AddrInfo {
200
- if max > len(in) {
201
- max = len(in)
202
- }
203
-
204
- out := make([]peer.AddrInfo, max)
205
- for i, val := range rand.Perm(len(in))[:max] {
334
+func randomizeList[T any](in []T) []T {
335
+ out := make([]T, len(in))
336
+ for i, val := range rand.Perm(len(in)) {
337
out[i] = in[val]
338
}
339
return out
core/bootstrap/bootstrap_test.go
+3
-3
@@ -7,9 +7,9 @@ import (
7
"github.com/libp2p/go-libp2p/core/test"
8
)
9
10
-func TestSubsetWhenMaxIsGreaterThanLengthOfSlice(t *testing.T) {
10
+func TestRandomizeAddressList(t *testing.T) {
11
var ps []peer.AddrInfo
12
- sizeofSlice := 100
12
+ sizeofSlice := 10
13
for i := 0; i < sizeofSlice; i++ {
14
pid, err := test.RandPeerID()
15
if err != nil {
@@ -18,7 +18,7 @@ func TestSubsetWhenMaxIsGreaterThanLengthOfSlice(t *testing.T) {
18
19
ps = append(ps, peer.AddrInfo{ID: pid})
20
}
21
- out := randomSubsetOfPeers(ps, 2*sizeofSlice)
21
+ out := randomizeList(ps)
22
if len(out) != len(ps) {
23
t.Fail()
24
}
core/core.go
+60
-1
@@ -11,10 +11,13 @@ package core
11
12
import (
13
"context"
14
+ "encoding/json"
15
"io"
16
+ "time"
17
18
"github.com/ipfs/boxo/filestore"
19
pin "github.com/ipfs/boxo/pinning/pinner"
20
+ "github.com/ipfs/go-datastore"
21
22
bserv "github.com/ipfs/boxo/blockservice"
23
bstore "github.com/ipfs/boxo/blockstore"
@@ -46,6 +49,7 @@ import (
49
50
"github.com/ipfs/boxo/namesys"
51
ipnsrp "github.com/ipfs/boxo/namesys/republisher"
52
+ "github.com/ipfs/kubo/config"
53
"github.com/ipfs/kubo/core/bootstrap"
54
"github.com/ipfs/kubo/core/node"
55
"github.com/ipfs/kubo/core/node/libp2p"
@@ -165,12 +169,40 @@ func (n *IpfsNode) Bootstrap(cfg bootstrap.BootstrapConfig) error {
169
return ps
170
}
171
}
172
+ if cfg.SaveBackupBootstrapPeers == nil {
173
+ cfg.SaveBackupBootstrapPeers = func(ctx context.Context, peerList []peer.AddrInfo) {
174
+ err := n.saveTempBootstrapPeers(ctx, peerList)
175
+ if err != nil {
176
+ log.Warnf("saveTempBootstrapPeers failed: %s", err)
177
+ return
178
+ }
179
+ }
180
+ }
181
+ if cfg.LoadBackupBootstrapPeers == nil {
182
+ cfg.LoadBackupBootstrapPeers = func(ctx context.Context) []peer.AddrInfo {
183
+ peerList, err := n.loadTempBootstrapPeers(ctx)
184
+ if err != nil {
185
+ log.Warnf("loadTempBootstrapPeers failed: %s", err)
186
+ return nil
187
+ }
188
+ return peerList
189
+ }
190
+ }
191
+
192
+ repoConf, err := n.Repo.Config()
193
+ if err != nil {
194
+ return err
195
+ }
196
+ if repoConf.Internal.BackupBootstrapInterval != nil {
197
+ cfg.BackupBootstrapInterval = repoConf.Internal.BackupBootstrapInterval.WithDefault(time.Hour)
198
+ }
199
169
- var err error
200
n.Bootstrapper, err = bootstrap.Bootstrap(n.Identity, n.PeerHost, n.Routing, cfg)
201
return err
202
}
203
204
+var TempBootstrapPeersKey = datastore.NewKey("/local/temp_bootstrap_peers")
205
+
206
func (n *IpfsNode) loadBootstrapPeers() ([]peer.AddrInfo, error) {
207
cfg, err := n.Repo.Config()
208
if err != nil {
@@ -180,6 +212,33 @@ func (n *IpfsNode) loadBootstrapPeers() ([]peer.AddrInfo, error) {
212
return cfg.BootstrapPeers()
213
}
214
215
+func (n *IpfsNode) saveTempBootstrapPeers(ctx context.Context, peerList []peer.AddrInfo) error {
216
+ ds := n.Repo.Datastore()
217
+ bytes, err := json.Marshal(config.BootstrapPeerStrings(peerList))
218
+ if err != nil {
219
+ return err
220
+ }
221
+
222
+ if err := ds.Put(ctx, TempBootstrapPeersKey, bytes); err != nil {
223
+ return err
224
+ }
225
+ return ds.Sync(ctx, TempBootstrapPeersKey)
226
+}
227
+
228
+func (n *IpfsNode) loadTempBootstrapPeers(ctx context.Context) ([]peer.AddrInfo, error) {
229
+ ds := n.Repo.Datastore()
230
+ bytes, err := ds.Get(ctx, TempBootstrapPeersKey)
231
+ if err != nil {
232
+ return nil, err
233
+ }
234
+
235
+ var addrs []string
236
+ if err := json.Unmarshal(bytes, &addrs); err != nil {
237
+ return nil, err
238
+ }
239
+ return config.ParseBootstrapPeers(addrs)
240
+}
241
+
242
type ConstructPeerHostOpts struct {
243
AddrsFactory p2pbhost.AddrsFactory
244
DisableNatPortMap bool
docs/changelogs/v0.21.md
+16
@@ -6,6 +6,7 @@
6
7
- [Overview](#overview)
8
- [🔦 Highlights](#-highlights)
9
+ - [Saving previously seen nodes for later bootstrapping](#saving-previously-seen-nodes-for-later-bootstrapping)
10
- [📝 Changelog](#-changelog)
11
- [👨👩👧👦 Contributors](#-contributors)
12
@@ -13,6 +14,21 @@
14
15
### 🔦 Highlights
16
17
+#### Saving previously seen nodes for later bootstrapping
18
+
19
+Kubo now stores a subset of connected peers as backup bootstrap nodes ([kubo#8856](https://github.com/ipfs/kubo/pull/8856)).
20
+These nodes are used in addition to the explicitly defined bootstrappers in the
21
+[`Bootstrap`](https://github.com/ipfs/kubo/blob/master/docs/config.md#bootstrap) configuration.
22
+
23
+This enhancement improves the resiliency of the system, as it eliminates the
24
+necessity of relying solely on the default bootstrappers operated by Protocol
25
+Labs for joining the public IPFS swarm. Previously, this level of robustness
26
+was only available in LAN contexts with [mDNS peer discovery](https://github.com/ipfs/kubo/blob/master/docs/config.md#discoverymdns)
27
+enabled.
28
+
29
+With this update, the same level of robustness is applied to peers that lack
30
+mDNS peers and solely rely on the public DHT.
31
+
32
### 📝 Changelog
33
34
### 👨👩👧👦 Contributors
test/cli/backup_bootstrap_test.go
new
+60
@@ -0,0 +1,60 @@
1
+package cli
2
+
3
+import (
4
+ "fmt"
5
+ "testing"
6
+ "time"
7
+
8
+ "github.com/ipfs/kubo/config"
9
+ "github.com/ipfs/kubo/test/cli/harness"
10
+ "github.com/stretchr/testify/assert"
11
+)
12
+
13
+func TestBackupBootstrapPeers(t *testing.T) {
14
+ nodes := harness.NewT(t).NewNodes(3).Init()
15
+ nodes.ForEachPar(func(n *harness.Node) {
16
+ n.UpdateConfig(func(cfg *config.Config) {
17
+ cfg.Bootstrap = []string{}
18
+ cfg.Addresses.Swarm = []string{fmt.Sprintf("/ip4/127.0.0.1/tcp/%d", harness.NewRandPort())}
19
+ cfg.Discovery.MDNS.Enabled = false
20
+ cfg.Internal.BackupBootstrapInterval = config.NewOptionalDuration(250 * time.Millisecond)
21
+ })
22
+ })
23
+
24
+ // Start all nodes and ensure they all have no peers.
25
+ nodes.StartDaemons()
26
+ nodes.ForEachPar(func(n *harness.Node) {
27
+ assert.Len(t, n.Peers(), 0)
28
+ })
29
+
30
+ // Connect nodes 0 and 1, ensure they know each other.
31
+ nodes[0].Connect(nodes[1])
32
+ assert.Len(t, nodes[0].Peers(), 1)
33
+ assert.Len(t, nodes[1].Peers(), 1)
34
+ assert.Len(t, nodes[2].Peers(), 0)
35
+
36
+ // Wait a bit to ensure that 0 and 1 saved their temporary bootstrap backups.
37
+ time.Sleep(time.Millisecond * 500)
38
+ nodes.StopDaemons()
39
+
40
+ // Start 1 and 2. 2 does not know anyone yet.
41
+ nodes[1].StartDaemon()
42
+ nodes[2].StartDaemon()
43
+ assert.Len(t, nodes[1].Peers(), 0)
44
+ assert.Len(t, nodes[2].Peers(), 0)
45
+
46
+ // Connect 1 and 2, ensure they know each other.
47
+ nodes[1].Connect(nodes[2])
48
+ assert.Len(t, nodes[1].Peers(), 1)
49
+ assert.Len(t, nodes[2].Peers(), 1)
50
+
51
+ // Start 0, wait a bit. Should connect to 1, and then discover 2 via the
52
+ // backup bootstrap peers.
53
+ nodes[0].StartDaemon()
54
+ time.Sleep(time.Millisecond * 500)
55
+
56
+ // Check if they're all connected.
57
+ assert.Len(t, nodes[0].Peers(), 2)
58
+ assert.Len(t, nodes[1].Peers(), 2)
59
+ assert.Len(t, nodes[2].Peers(), 2)
60
+}