Cleanup core package
License: MIT Signed-off-by: Łukasz Magiera <magik6k@gmail.com>
Łukasz Magiera committed
Apr 3, 2019 at 03:44 UTC
d35dac70f0b0b47c7813c6380deb8974179bdb32
24 files changed
+1312
-1208
cmd/ipfs/daemon.go
+4
-3
@@ -20,6 +20,7 @@ import (
20
coreapi "github.com/ipfs/go-ipfs/core/coreapi"
21
corehttp "github.com/ipfs/go-ipfs/core/corehttp"
22
corerepo "github.com/ipfs/go-ipfs/core/corerepo"
23
+ "github.com/ipfs/go-ipfs/core/node"
24
nodeMount "github.com/ipfs/go-ipfs/fuse/node"
25
fsrepo "github.com/ipfs/go-ipfs/repo/fsrepo"
26
migrate "github.com/ipfs/go-ipfs/repo/fsrepo/migrations"
@@ -323,11 +324,11 @@ func daemonFunc(req *cmds.Request, re cmds.ResponseEmitter, env cmds.Environment
324
case routingOptionSupernodeKwd:
325
return errors.New("supernode routing was never fully implemented and has been removed")
326
case routingOptionDHTClientKwd:
326
- ncfg.Routing = core.DHTClientOption
327
+ ncfg.Routing = node.DHTClientOption
328
case routingOptionDHTKwd:
328
- ncfg.Routing = core.DHTOption
329
+ ncfg.Routing = node.DHTOption
330
case routingOptionNoneKwd:
330
- ncfg.Routing = core.NilRouterOption
331
+ ncfg.Routing = node.NilRouterOption
332
default:
333
return fmt.Errorf("unrecognized routing option: %s", routingOption)
334
}
core/bootstrap/bootstrap.go
renamed
+50
-45
@@ -1,4 +1,4 @@
1
-package core
1
+package bootstrap
2
3
import (
4
"context"
@@ -9,19 +9,23 @@ import (
9
"sync"
10
"time"
11
12
- math2 "github.com/ipfs/go-ipfs/thirdparty/math2"
13
- lgbl "github.com/libp2p/go-libp2p-loggables"
14
-
12
config "github.com/ipfs/go-ipfs-config"
16
- goprocess "github.com/jbenet/goprocess"
17
- procctx "github.com/jbenet/goprocess/context"
18
- periodicproc "github.com/jbenet/goprocess/periodic"
19
- host "github.com/libp2p/go-libp2p-host"
20
- inet "github.com/libp2p/go-libp2p-net"
21
- peer "github.com/libp2p/go-libp2p-peer"
22
- pstore "github.com/libp2p/go-libp2p-peerstore"
13
+ logging "github.com/ipfs/go-log"
14
+ "github.com/jbenet/goprocess"
15
+ "github.com/jbenet/goprocess/context"
16
+ "github.com/jbenet/goprocess/periodic"
17
+ "github.com/libp2p/go-libp2p-host"
18
+ "github.com/libp2p/go-libp2p-loggables"
19
+ "github.com/libp2p/go-libp2p-net"
20
+ "github.com/libp2p/go-libp2p-peer"
21
+ "github.com/libp2p/go-libp2p-peerstore"
22
+ "github.com/libp2p/go-libp2p-routing"
23
+
24
+ "github.com/ipfs/go-ipfs/thirdparty/math2"
25
)
26
27
+var log = logging.Logger("bootstrap")
28
+
29
// ErrNotEnoughBootstrapPeers signals that we do not have enough bootstrap
30
// peers to bootstrap correctly.
31
var ErrNotEnoughBootstrapPeers = errors.New("not enough bootstrap peers to bootstrap")
@@ -29,7 +33,6 @@ var ErrNotEnoughBootstrapPeers = errors.New("not enough bootstrap peers to boots
33
// BootstrapConfig specifies parameters used in an IpfsNode's network
34
// bootstrapping process.
35
type BootstrapConfig struct {
32
-
36
// MinPeerThreshold governs whether to bootstrap more connections. If the
37
// node has less open connections than this number, it will open connections
38
// to the bootstrap nodes. From there, the routing system should be able
@@ -50,7 +53,7 @@ type BootstrapConfig struct {
53
// BootstrapPeers is a function that returns a set of bootstrap peers
54
// for the bootstrap process to use. This makes it possible for clients
55
// to control the peers the process uses at any moment.
53
- BootstrapPeers func() []pstore.PeerInfo
56
+ BootstrapPeers func() []peerstore.PeerInfo
57
}
58
59
// DefaultBootstrapConfig specifies default sane parameters for bootstrapping.
@@ -60,9 +63,9 @@ var DefaultBootstrapConfig = BootstrapConfig{
63
ConnectionTimeout: (30 * time.Second) / 3, // Perod / 3
64
}
65
63
-func BootstrapConfigWithPeers(pis []pstore.PeerInfo) BootstrapConfig {
66
+func BootstrapConfigWithPeers(pis []peerstore.PeerInfo) BootstrapConfig {
67
cfg := DefaultBootstrapConfig
65
- cfg.BootstrapPeers = func() []pstore.PeerInfo {
68
+ cfg.BootstrapPeers = func() []peerstore.PeerInfo {
69
return pis
70
}
71
return cfg
@@ -72,7 +75,7 @@ func BootstrapConfigWithPeers(pis []pstore.PeerInfo) BootstrapConfig {
75
// check the number of open connections and -- if there are too few -- initiate
76
// connections to well-known bootstrap peers. It also kicks off subsystem
77
// bootstrapping (i.e. routing).
75
-func Bootstrap(n *IpfsNode, cfg BootstrapConfig) (io.Closer, error) {
78
+func Bootstrap(id peer.ID, host host.Host, rt routing.IpfsRouting, cfg BootstrapConfig) (io.Closer, error) {
79
80
// make a signal to wait for one bootstrap round to complete.
81
doneWithRound := make(chan struct{})
@@ -85,12 +88,12 @@ func Bootstrap(n *IpfsNode, cfg BootstrapConfig) (io.Closer, error) {
88
89
// the periodic bootstrap function -- the connection supervisor
90
periodic := func(worker goprocess.Process) {
88
- ctx := procctx.OnClosingContext(worker)
89
- defer log.EventBegin(ctx, "periodicBootstrap", n.Identity).Done()
91
+ ctx := goprocessctx.OnClosingContext(worker)
92
+ defer log.EventBegin(ctx, "periodicBootstrap", id).Done()
93
91
- if err := bootstrapRound(ctx, n.PeerHost, cfg); err != nil {
92
- log.Event(ctx, "bootstrapError", n.Identity, lgbl.Error(err))
93
- log.Debugf("%s bootstrap error: %s", n.Identity, err)
94
+ if err := bootstrapRound(ctx, host, cfg); err != nil {
95
+ log.Event(ctx, "bootstrapError", id, loggables.Error(err))
96
+ log.Debugf("%s bootstrap error: %s", id, err)
97
}
98
99
<-doneWithRound
@@ -101,9 +104,9 @@ func Bootstrap(n *IpfsNode, cfg BootstrapConfig) (io.Closer, error) {
104
proc.Go(periodic) // run one right now.
105
106
// kick off Routing.Bootstrap
104
- if n.Routing != nil {
105
- ctx := procctx.OnClosingContext(proc)
106
- if err := n.Routing.Bootstrap(ctx); err != nil {
107
+ if rt != nil {
108
+ ctx := goprocessctx.OnClosingContext(proc)
109
+ if err := rt.Bootstrap(ctx); err != nil {
110
proc.Close()
111
return nil, err
112
}
@@ -134,9 +137,9 @@ func bootstrapRound(ctx context.Context, host host.Host, cfg BootstrapConfig) er
137
numToDial := cfg.MinPeerThreshold - len(connected)
138
139
// filter out bootstrap nodes we are already connected to
137
- var notConnected []pstore.PeerInfo
140
+ var notConnected []peerstore.PeerInfo
141
for _, p := range peers {
139
- if host.Network().Connectedness(p.ID) != inet.Connected {
142
+ if host.Network().Connectedness(p.ID) != net.Connected {
143
notConnected = append(notConnected, p)
144
}
145
}
@@ -155,7 +158,7 @@ func bootstrapRound(ctx context.Context, host host.Host, cfg BootstrapConfig) er
158
return bootstrapConnect(ctx, host, randSubset)
159
}
160
158
-func bootstrapConnect(ctx context.Context, ph host.Host, peers []pstore.PeerInfo) error {
161
+func bootstrapConnect(ctx context.Context, ph host.Host, peers []peerstore.PeerInfo) error {
162
if len(peers) < 1 {
163
return ErrNotEnoughBootstrapPeers
164
}
@@ -170,12 +173,12 @@ func bootstrapConnect(ctx context.Context, ph host.Host, peers []pstore.PeerInfo
173
// Also, performed asynchronously for dial speed.
174
175
wg.Add(1)
173
- go func(p pstore.PeerInfo) {
176
+ go func(p peerstore.PeerInfo) {
177
defer wg.Done()
178
defer log.EventBegin(ctx, "bootstrapDial", ph.ID(), p.ID).Done()
179
log.Debugf("%s bootstrapping to %s", ph.ID(), p.ID)
180
178
- ph.Peerstore().AddAddrs(p.ID, p.Addrs, pstore.PermanentAddrTTL)
181
+ ph.Peerstore().AddAddrs(p.ID, p.Addrs, peerstore.PermanentAddrTTL)
182
if err := ph.Connect(ctx, p); err != nil {
183
log.Event(ctx, "bootstrapDialFailed", p.ID)
184
log.Debugf("failed to bootstrap with %v: %s", p.ID, err)
@@ -204,12 +207,26 @@ func bootstrapConnect(ctx context.Context, ph host.Host, peers []pstore.PeerInfo
207
return nil
208
}
209
207
-func toPeerInfos(bpeers []config.BootstrapPeer) []pstore.PeerInfo {
208
- pinfos := make(map[peer.ID]*pstore.PeerInfo)
210
+func randomSubsetOfPeers(in []peerstore.PeerInfo, max int) []peerstore.PeerInfo {
211
+ n := math2.IntMin(max, len(in))
212
+ var out []peerstore.PeerInfo
213
+ for _, val := range rand.Perm(len(in)) {
214
+ out = append(out, in[val])
215
+ if len(out) >= n {
216
+ break
217
+ }
218
+ }
219
+ return out
220
+}
221
+
222
+type Peers []config.BootstrapPeer
223
+
224
+func (bpeers Peers) ToPeerInfos() []peerstore.PeerInfo {
225
+ pinfos := make(map[peer.ID]*peerstore.PeerInfo)
226
for _, bootstrap := range bpeers {
227
pinfo, ok := pinfos[bootstrap.ID()]
228
if !ok {
212
- pinfo = new(pstore.PeerInfo)
229
+ pinfo = new(peerstore.PeerInfo)
230
pinfos[bootstrap.ID()] = pinfo
231
pinfo.ID = bootstrap.ID()
232
}
@@ -217,22 +234,10 @@ func toPeerInfos(bpeers []config.BootstrapPeer) []pstore.PeerInfo {
234
pinfo.Addrs = append(pinfo.Addrs, bootstrap.Transport())
235
}
236
220
- var peers []pstore.PeerInfo
237
+ var peers []peerstore.PeerInfo
238
for _, pinfo := range pinfos {
239
peers = append(peers, *pinfo)
240
}
241
242
return peers
243
}
227
-
228
-func randomSubsetOfPeers(in []pstore.PeerInfo, max int) []pstore.PeerInfo {
229
- n := math2.IntMin(max, len(in))
230
- var out []pstore.PeerInfo
231
- for _, val := range rand.Perm(len(in)) {
232
- out = append(out, in[val])
233
- if len(out) >= n {
234
- break
235
- }
236
- }
237
- return out
238
-}
core/bootstrap/bootstrap_test.go
renamed
+2
-2
@@ -1,4 +1,4 @@
1
-package core
1
+package bootstrap
2
3
import (
4
"fmt"
@@ -49,7 +49,7 @@ func TestMultipleAddrsPerPeer(t *testing.T) {
49
bsps = append(bsps, bsp1, bsp2)
50
}
51
52
- pinfos := toPeerInfos(bsps)
52
+ pinfos := Peers.ToPeerInfos(bsps)
53
if len(pinfos) != len(bsps)/2 {
54
t.Fatal("expected fewer peers")
55
}
core/builder.go
+20
-110
@@ -5,57 +5,24 @@ import (
5
"crypto/rand"
6
"encoding/base64"
7
"errors"
8
- "os"
9
- "syscall"
8
9
"go.uber.org/fx"
10
13
- "github.com/ipfs/go-ipfs/p2p"
14
- "github.com/ipfs/go-ipfs/provider"
11
+ "github.com/ipfs/go-ipfs/core/bootstrap"
12
+ "github.com/ipfs/go-ipfs/core/node"
13
14
repo "github.com/ipfs/go-ipfs/repo"
15
16
ds "github.com/ipfs/go-datastore"
17
dsync "github.com/ipfs/go-datastore/sync"
18
cfg "github.com/ipfs/go-ipfs-config"
21
- offline "github.com/ipfs/go-ipfs-exchange-offline"
22
- offroute "github.com/ipfs/go-ipfs-routing/offline"
19
metrics "github.com/ipfs/go-metrics-interface"
20
resolver "github.com/ipfs/go-path/resolver"
21
ci "github.com/libp2p/go-libp2p-crypto"
22
peer "github.com/libp2p/go-libp2p-peer"
23
)
24
29
-type BuildCfg struct {
30
- // If online is set, the node will have networking enabled
31
- Online bool
32
-
33
- // ExtraOpts is a map of extra options used to configure the ipfs nodes creation
34
- ExtraOpts map[string]bool
35
-
36
- // If permanent then node should run more expensive processes
37
- // that will improve performance in long run
38
- Permanent bool
39
-
40
- // DisableEncryptedConnections disables connection encryption *entirely*.
41
- // DO NOT SET THIS UNLESS YOU'RE TESTING.
42
- DisableEncryptedConnections bool
43
-
44
- // If NilRepo is set, a Repo backed by a nil datastore will be constructed
45
- NilRepo bool
46
-
47
- Routing RoutingOption
48
- Host HostOption
49
- Repo repo.Repo
50
-}
51
-
52
-func (cfg *BuildCfg) getOpt(key string) bool {
53
- if cfg.ExtraOpts == nil {
54
- return false
55
- }
56
-
57
- return cfg.ExtraOpts[key]
58
-}
25
+type BuildCfg node.BuildCfg
26
27
func (cfg *BuildCfg) fillDefaults() error {
28
if cfg.Repo != nil && cfg.NilRepo {
@@ -77,11 +44,11 @@ func (cfg *BuildCfg) fillDefaults() error {
44
}
45
46
if cfg.Routing == nil {
80
- cfg.Routing = DHTOption
47
+ cfg.Routing = node.DHTOption
48
}
49
50
if cfg.Host == nil {
84
- cfg.Host = DefaultHostOption
51
+ cfg.Host = node.DefaultHostOption
52
}
53
54
return nil
@@ -115,8 +82,6 @@ func defaultRepo(dstore repo.Datastore) (repo.Repo, error) {
82
}, nil
83
}
84
118
-type MetricsCtx context.Context
119
-
85
// NewNode constructs and returns an IpfsNode using the given cfg.
86
func NewNode(ctx context.Context, cfg *BuildCfg) (*IpfsNode, error) {
87
if cfg == nil {
@@ -141,12 +106,12 @@ func NewNode(ctx context.Context, cfg *BuildCfg) (*IpfsNode, error) {
106
})
107
108
// TODO: Remove this, use only for passing node config
144
- cfgOption := fx.Provide(func() *BuildCfg {
145
- return cfg
109
+ cfgOption := fx.Provide(func() *node.BuildCfg {
110
+ return (*node.BuildCfg)(cfg)
111
})
112
148
- metricsCtx := fx.Provide(func() MetricsCtx {
149
- return MetricsCtx(ctx)
113
+ metricsCtx := fx.Provide(func() node.MetricsCtx {
114
+ return node.MetricsCtx(ctx)
115
})
116
117
params := fx.Options(
@@ -155,58 +120,12 @@ func NewNode(ctx context.Context, cfg *BuildCfg) (*IpfsNode, error) {
120
metricsCtx,
121
)
122
158
- storage := fx.Options(
159
- fx.Provide(repoConfig),
160
- fx.Provide(datastoreCtor),
161
- fx.Provide(baseBlockstoreCtor),
162
- fx.Provide(gcBlockstoreCtor),
163
- )
164
-
165
- ident := fx.Options(
166
- fx.Provide(identity),
167
- fx.Provide(privateKey),
168
- fx.Provide(peerstore),
169
- )
170
-
171
- ipns := fx.Options(
172
- fx.Provide(recordValidator),
173
- )
174
-
175
- providers := fx.Options(
176
- fx.Provide(providerQueue),
177
- fx.Provide(providerCtor),
178
- fx.Provide(reproviderCtor),
179
-
180
- fx.Invoke(reprovider),
181
- fx.Invoke(provider.Provider.Run),
182
- )
183
-
184
- online := fx.Options(
185
- fx.Provide(onlineExchangeCtor),
186
- fx.Provide(onlineNamesysCtor),
187
-
188
- fx.Invoke(ipnsRepublisher),
189
-
190
- fx.Provide(p2p.NewP2P),
191
-
192
- ipfsp2p,
193
- providers,
194
- )
195
- if !cfg.Online {
196
- online = fx.Options(
197
- fx.Provide(offline.Exchange),
198
- fx.Provide(offlineNamesysCtor),
199
- fx.Provide(offroute.NewOfflineRouter),
200
- fx.Provide(provider.NewOfflineProvider),
201
- )
202
- }
203
-
123
core := fx.Options(
205
- fx.Provide(blockServiceCtor),
206
- fx.Provide(dagCtor),
124
+ fx.Provide(node.BlockServiceCtor),
125
+ fx.Provide(node.DagCtor),
126
fx.Provide(resolver.NewBasicResolver),
208
- fx.Provide(pinning),
209
- fx.Provide(files),
127
+ fx.Provide(node.Pinning),
128
+ fx.Provide(node.Files),
129
)
130
131
n := &IpfsNode{
@@ -214,16 +133,16 @@ func NewNode(ctx context.Context, cfg *BuildCfg) (*IpfsNode, error) {
133
}
134
135
app := fx.New(
136
+ fx.NopLogger,
137
fx.Provide(baseProcess),
138
139
params,
220
- storage,
221
- ident,
222
- ipns,
223
- online,
140
+ node.Storage,
141
+ node.Identity,
142
+ node.IPNS,
143
+ node.Networked(cfg.Online),
144
145
fx.Invoke(setupSharding),
226
- fx.NopLogger,
146
147
core,
148
@@ -248,19 +167,10 @@ func NewNode(ctx context.Context, cfg *BuildCfg) (*IpfsNode, error) {
167
return nil, err
168
}
169
251
- // TODO: DI-ify bootstrap
170
+ // TODO: How soon will bootstrap move to libp2p?
171
if !cfg.Online {
172
return n, nil
173
}
174
256
- return n, n.Bootstrap(DefaultBootstrapConfig)
257
-}
258
-
259
-func isTooManyFDError(err error) bool {
260
- perr, ok := err.(*os.PathError)
261
- if ok && perr.Err == syscall.EMFILE {
262
- return true
263
- }
264
-
265
- return false
175
+ return n, n.Bootstrap(bootstrap.DefaultBootstrapConfig)
176
}
core/core.go
+10
-211
@@ -11,16 +11,13 @@ package core
11
12
import (
13
"context"
14
- "fmt"
14
"io"
16
- "io/ioutil"
17
- "os"
18
- "strings"
19
- "time"
15
16
"go.uber.org/fx"
17
18
version "github.com/ipfs/go-ipfs"
19
+ "github.com/ipfs/go-ipfs/core/bootstrap"
20
+ "github.com/ipfs/go-ipfs/core/node"
21
rp "github.com/ipfs/go-ipfs/exchange/reprovide"
22
"github.com/ipfs/go-ipfs/filestore"
23
"github.com/ipfs/go-ipfs/fuse/mount"
@@ -32,27 +29,18 @@ import (
29
"github.com/ipfs/go-ipfs/repo"
30
31
bserv "github.com/ipfs/go-blockservice"
35
- "github.com/ipfs/go-cid"
36
- ds "github.com/ipfs/go-datastore"
32
bstore "github.com/ipfs/go-ipfs-blockstore"
38
- config "github.com/ipfs/go-ipfs-config"
33
exchange "github.com/ipfs/go-ipfs-exchange-interface"
40
- nilrouting "github.com/ipfs/go-ipfs-routing/none"
34
ipld "github.com/ipfs/go-ipld-format"
35
logging "github.com/ipfs/go-log"
43
- "github.com/ipfs/go-merkledag"
36
"github.com/ipfs/go-mfs"
37
"github.com/ipfs/go-path/resolver"
46
- ft "github.com/ipfs/go-unixfs"
38
"github.com/jbenet/goprocess"
48
- "github.com/libp2p/go-libp2p"
39
autonat "github.com/libp2p/go-libp2p-autonat-svc"
50
- circuit "github.com/libp2p/go-libp2p-circuit"
40
ic "github.com/libp2p/go-libp2p-crypto"
41
p2phost "github.com/libp2p/go-libp2p-host"
42
ifconnmgr "github.com/libp2p/go-libp2p-interface-connmgr"
43
dht "github.com/libp2p/go-libp2p-kad-dht"
55
- dhtopts "github.com/libp2p/go-libp2p-kad-dht/opts"
44
metrics "github.com/libp2p/go-libp2p-metrics"
45
peer "github.com/libp2p/go-libp2p-peer"
46
pstore "github.com/libp2p/go-libp2p-peerstore"
@@ -63,18 +51,8 @@ import (
51
"github.com/libp2p/go-libp2p/p2p/discovery"
52
p2pbhost "github.com/libp2p/go-libp2p/p2p/host/basic"
53
"github.com/libp2p/go-libp2p/p2p/protocol/identify"
66
- mafilter "github.com/libp2p/go-maddr-filter"
67
- smux "github.com/libp2p/go-stream-muxer"
68
- ma "github.com/multiformats/go-multiaddr"
69
- mplex "github.com/whyrusleeping/go-smux-multiplex"
70
- yamux "github.com/whyrusleeping/go-smux-yamux"
71
- mamask "github.com/whyrusleeping/multiaddr-filter"
54
)
55
74
-const kReprovideFrequency = time.Hour * 12
75
-const discoveryConnTimeout = time.Second * 30
76
-const DefaultIpnsCacheSize = 128
77
-
56
var log = logging.Logger("core")
57
58
func init() {
@@ -90,16 +68,16 @@ type IpfsNode struct {
68
Repo repo.Repo
69
70
// Local node
93
- Pinning pin.Pinner // the pinning manager
94
- Mounts Mounts `optional:"true"` // current mount state, if any.
95
- PrivateKey ic.PrivKey // the local node's private Key
96
- PNetFingerprint PNetFingerprint `optional:"true"` // fingerprint of private network
71
+ Pinning pin.Pinner // the pinning manager
72
+ Mounts Mounts `optional:"true"` // current mount state, if any.
73
+ PrivateKey ic.PrivKey // the local node's private Key
74
+ PNetFingerprint node.PNetFingerprint `optional:"true"` // fingerprint of private network
75
76
// Services
77
Peerstore pstore.Peerstore `optional:"true"` // storage for other Peer instances
78
Blockstore bstore.GCBlockstore // the block store (lower level)
79
Filestore *filestore.Filestore // the filestore blockstore
102
- BaseBlocks BaseBlocks // the raw blockstore, no filestore wrapping
80
+ BaseBlocks node.BaseBlocks // the raw blockstore, no filestore wrapping
81
GCLocker bstore.GCLocker // the locker used to protect the blockstore during gc
82
Blocks bserv.BlockService // the block service, get/add blocks.
83
DAG ipld.DAGService // the merkle dag service, get/add objects.
@@ -143,94 +121,6 @@ type Mounts struct {
121
Ipns mount.Mount
122
}
123
146
-func makeAddrsFactory(cfg config.Addresses) (p2pbhost.AddrsFactory, error) {
147
- var annAddrs []ma.Multiaddr
148
- for _, addr := range cfg.Announce {
149
- maddr, err := ma.NewMultiaddr(addr)
150
- if err != nil {
151
- return nil, err
152
- }
153
- annAddrs = append(annAddrs, maddr)
154
- }
155
-
156
- filters := mafilter.NewFilters()
157
- noAnnAddrs := map[string]bool{}
158
- for _, addr := range cfg.NoAnnounce {
159
- f, err := mamask.NewMask(addr)
160
- if err == nil {
161
- filters.AddDialFilter(f)
162
- continue
163
- }
164
- maddr, err := ma.NewMultiaddr(addr)
165
- if err != nil {
166
- return nil, err
167
- }
168
- noAnnAddrs[maddr.String()] = true
169
- }
170
-
171
- return func(allAddrs []ma.Multiaddr) []ma.Multiaddr {
172
- var addrs []ma.Multiaddr
173
- if len(annAddrs) > 0 {
174
- addrs = annAddrs
175
- } else {
176
- addrs = allAddrs
177
- }
178
-
179
- var out []ma.Multiaddr
180
- for _, maddr := range addrs {
181
- // check for exact matches
182
- ok := noAnnAddrs[maddr.String()]
183
- // check for /ipcidr matches
184
- if !ok && !filters.AddrBlocked(maddr) {
185
- out = append(out, maddr)
186
- }
187
- }
188
- return out
189
- }, nil
190
-}
191
-
192
-func makeSmuxTransportOption(mplexExp bool) libp2p.Option {
193
- const yamuxID = "/yamux/1.0.0"
194
- const mplexID = "/mplex/6.7.0"
195
-
196
- ymxtpt := &yamux.Transport{
197
- AcceptBacklog: 512,
198
- ConnectionWriteTimeout: time.Second * 10,
199
- KeepAliveInterval: time.Second * 30,
200
- EnableKeepAlive: true,
201
- MaxStreamWindowSize: uint32(16 * 1024 * 1024), // 16MiB
202
- LogOutput: ioutil.Discard,
203
- }
204
-
205
- if os.Getenv("YAMUX_DEBUG") != "" {
206
- ymxtpt.LogOutput = os.Stderr
207
- }
208
-
209
- muxers := map[string]smux.Transport{yamuxID: ymxtpt}
210
- if mplexExp {
211
- muxers[mplexID] = mplex.DefaultTransport
212
- }
213
-
214
- // Allow muxer preference order overriding
215
- order := []string{yamuxID, mplexID}
216
- if prefs := os.Getenv("LIBP2P_MUX_PREFS"); prefs != "" {
217
- order = strings.Fields(prefs)
218
- }
219
-
220
- opts := make([]libp2p.Option, 0, len(order))
221
- for _, id := range order {
222
- tpt, ok := muxers[id]
223
- if !ok {
224
- log.Warning("unknown or duplicate muxer in LIBP2P_MUX_PREFS: %s", id)
225
- continue
226
- }
227
- delete(muxers, id)
228
- opts = append(opts, libp2p.Muxer(id, tpt))
229
- }
230
-
231
- return libp2p.ChainOptions(opts...)
232
-}
233
-
124
// Close calls Close() on the App object
125
func (n *IpfsNode) Close() error {
126
return n.app.Stop(n.ctx)
@@ -245,7 +135,7 @@ func (n *IpfsNode) Context() context.Context {
135
}
136
137
// Bootstrap will set and call the IpfsNodes bootstrap function.
248
-func (n *IpfsNode) Bootstrap(cfg BootstrapConfig) error {
138
+func (n *IpfsNode) Bootstrap(cfg bootstrap.BootstrapConfig) error {
139
// TODO what should return value be when in offlineMode?
140
if n.Routing == nil {
141
return nil
@@ -269,7 +159,7 @@ func (n *IpfsNode) Bootstrap(cfg BootstrapConfig) error {
159
}
160
161
var err error
272
- n.Bootstrapper, err = Bootstrap(n, cfg)
162
+ n.Bootstrapper, err = bootstrap.Bootstrap(n.Identity, n.PeerHost, n.Routing, cfg)
163
return err
164
}
165
@@ -283,20 +173,7 @@ func (n *IpfsNode) loadBootstrapPeers() ([]pstore.PeerInfo, error) {
173
if err != nil {
174
return nil, err
175
}
286
- return toPeerInfos(parsed), nil
287
-}
288
-
289
-func listenAddresses(cfg *config.Config) ([]ma.Multiaddr, error) {
290
- var listen []ma.Multiaddr
291
- for _, addr := range cfg.Addresses.Swarm {
292
- maddr, err := ma.NewMultiaddr(addr)
293
- if err != nil {
294
- return nil, fmt.Errorf("failure to parse config.Addresses.Swarm: %s", cfg.Addresses.Swarm)
295
- }
296
- listen = append(listen, maddr)
297
- }
298
-
299
- return listen, nil
176
+ return bootstrap.Peers.ToPeerInfos(parsed), nil
177
}
178
179
type ConstructPeerHostOpts struct {
@@ -306,81 +183,3 @@ type ConstructPeerHostOpts struct {
183
EnableRelayHop bool
184
ConnectionManager ifconnmgr.ConnManager
185
}
309
-
310
-type HostOption func(ctx context.Context, id peer.ID, ps pstore.Peerstore, options ...libp2p.Option) (p2phost.Host, error)
311
-
312
-var DefaultHostOption HostOption = constructPeerHost
313
-
314
-// isolates the complex initialization steps
315
-func constructPeerHost(ctx context.Context, id peer.ID, ps pstore.Peerstore, options ...libp2p.Option) (p2phost.Host, error) {
316
- pkey := ps.PrivKey(id)
317
- if pkey == nil {
318
- return nil, fmt.Errorf("missing private key for node ID: %s", id.Pretty())
319
- }
320
- options = append([]libp2p.Option{libp2p.Identity(pkey), libp2p.Peerstore(ps)}, options...)
321
- return libp2p.New(ctx, options...)
322
-}
323
-
324
-func filterRelayAddrs(addrs []ma.Multiaddr) []ma.Multiaddr {
325
- var raddrs []ma.Multiaddr
326
- for _, addr := range addrs {
327
- _, err := addr.ValueForProtocol(circuit.P_CIRCUIT)
328
- if err == nil {
329
- continue
330
- }
331
- raddrs = append(raddrs, addr)
332
- }
333
- return raddrs
334
-}
335
-
336
-func composeAddrsFactory(f, g p2pbhost.AddrsFactory) p2pbhost.AddrsFactory {
337
- return func(addrs []ma.Multiaddr) []ma.Multiaddr {
338
- return f(g(addrs))
339
- }
340
-}
341
-
342
-// startListening on the network addresses
343
-func startListening(host p2phost.Host, cfg *config.Config) error {
344
- listenAddrs, err := listenAddresses(cfg)
345
- if err != nil {
346
- return err
347
- }
348
-
349
- // Actually start listening:
350
- if err := host.Network().Listen(listenAddrs...); err != nil {
351
- return err
352
- }
353
-
354
- // list out our addresses
355
- addrs, err := host.Network().InterfaceListenAddresses()
356
- if err != nil {
357
- return err
358
- }
359
- log.Infof("Swarm listening at: %s", addrs)
360
- return nil
361
-}
362
-
363
-func constructDHTRouting(ctx context.Context, host p2phost.Host, dstore ds.Batching, validator record.Validator) (routing.IpfsRouting, error) {
364
- return dht.New(
365
- ctx, host,
366
- dhtopts.Datastore(dstore),
367
- dhtopts.Validator(validator),
368
- )
369
-}
370
-
371
-func constructClientDHTRouting(ctx context.Context, host p2phost.Host, dstore ds.Batching, validator record.Validator) (routing.IpfsRouting, error) {
372
- return dht.New(
373
- ctx, host,
374
- dhtopts.Client(true),
375
- dhtopts.Datastore(dstore),
376
- dhtopts.Validator(validator),
377
- )
378
-}
379
-
380
-type RoutingOption func(context.Context, p2phost.Host, ds.Batching, record.Validator) (routing.IpfsRouting, error)
381
-
382
-type DiscoveryOption func(context.Context, p2phost.Host) (discovery.Service, error)
383
-
384
-var DHTOption RoutingOption = constructDHTRouting
385
-var DHTClientOption RoutingOption = constructClientDHTRouting
386
-var NilRouterOption RoutingOption = nilrouting.ConstructNilRouting
core/coreapi/coreapi.go
+2
-1
@@ -19,6 +19,7 @@ import (
19
"fmt"
20
21
"github.com/ipfs/go-ipfs/core"
22
+ "github.com/ipfs/go-ipfs/core/node"
23
"github.com/ipfs/go-ipfs/namesys"
24
"github.com/ipfs/go-ipfs/pin"
25
"github.com/ipfs/go-ipfs/provider"
@@ -207,7 +208,7 @@ func (api *CoreAPI) WithOptions(opts ...options.ApiOption) (coreiface.CoreAPI, e
208
209
cs := cfg.Ipns.ResolveCacheSize
210
if cs == 0 {
210
- cs = core.DefaultIpnsCacheSize
211
+ cs = node.DefaultIpnsCacheSize
212
}
213
if cs < 0 {
214
return nil, fmt.Errorf("cannot specify negative resolve cache size")
core/coreapi/test/api_test.go
+2
-1
@@ -8,6 +8,7 @@ import (
8
"path/filepath"
9
"testing"
10
11
+ "github.com/ipfs/go-ipfs/core/bootstrap"
12
"github.com/ipfs/go-ipfs/filestore"
13
14
"github.com/ipfs/go-ipfs/core"
@@ -101,7 +102,7 @@ func (NodeProvider) MakeAPISwarm(ctx context.Context, fullIdentity bool, n int)
102
return nil, err
103
}
104
104
- bsinf := core.BootstrapConfigWithPeers(
105
+ bsinf := bootstrap.BootstrapConfigWithPeers(
106
[]pstore.PeerInfo{
107
nodes[0].Peerstore.PeerInfo(nodes[0].Identity),
108
},
core/mock/mock.go
+2
-1
@@ -5,6 +5,7 @@ import (
5
6
commands "github.com/ipfs/go-ipfs/commands"
7
core "github.com/ipfs/go-ipfs/core"
8
+ "github.com/ipfs/go-ipfs/core/node"
9
"github.com/ipfs/go-ipfs/repo"
10
11
datastore "github.com/ipfs/go-datastore"
@@ -29,7 +30,7 @@ func NewMockNode() (*core.IpfsNode, error) {
30
})
31
}
32
32
-func MockHostOption(mn mocknet.Mocknet) core.HostOption {
33
+func MockHostOption(mn mocknet.Mocknet) node.HostOption {
34
return func(ctx context.Context, id peer.ID, ps pstore.Peerstore, _ ...libp2p.Option) (host.Host, error) {
35
return mn.AddPeerWithPeerstore(id, ps)
36
}
core/ncore.go
+4
-827
@@ -1,839 +1,21 @@
1
package core
2
3
import (
4
- "bytes"
4
"context"
6
- "errors"
7
- "fmt"
8
- "github.com/ipfs/go-bitswap"
9
- bsnet "github.com/ipfs/go-bitswap/network"
10
- bserv "github.com/ipfs/go-blockservice"
11
- "github.com/ipfs/go-cid"
12
- ds "github.com/ipfs/go-datastore"
13
- bstore "github.com/ipfs/go-ipfs-blockstore"
14
- exchange "github.com/ipfs/go-ipfs-exchange-interface"
15
- "github.com/ipfs/go-ipfs-exchange-offline"
16
- u "github.com/ipfs/go-ipfs-util"
17
- rp "github.com/ipfs/go-ipfs/exchange/reprovide"
18
- "github.com/ipfs/go-ipfs/filestore"
19
- "github.com/ipfs/go-ipfs/namesys"
20
- ipnsrp "github.com/ipfs/go-ipfs/namesys/republisher"
21
- "github.com/ipfs/go-ipfs/pin"
22
- "github.com/ipfs/go-ipfs/provider"
23
- "github.com/ipfs/go-ipfs/thirdparty/cidv0v1"
24
- "github.com/ipfs/go-ipfs/thirdparty/verifbs"
25
- "github.com/ipfs/go-ipld-format"
26
- "github.com/ipfs/go-ipns"
27
- merkledag "github.com/ipfs/go-merkledag"
28
- "github.com/ipfs/go-mfs"
29
- ft "github.com/ipfs/go-unixfs"
5
+
6
"github.com/jbenet/goprocess"
31
- "github.com/libp2p/go-libp2p"
32
- "github.com/libp2p/go-libp2p-autonat-svc"
33
- circuit "github.com/libp2p/go-libp2p-circuit"
34
- connmgr "github.com/libp2p/go-libp2p-connmgr"
35
- "github.com/libp2p/go-libp2p-kad-dht"
36
- "github.com/libp2p/go-libp2p-metrics"
37
- pstore "github.com/libp2p/go-libp2p-peerstore"
38
- "github.com/libp2p/go-libp2p-peerstore/pstoremem"
39
- "github.com/libp2p/go-libp2p-pnet"
40
- "github.com/libp2p/go-libp2p-pubsub"
41
- psrouter "github.com/libp2p/go-libp2p-pubsub-router"
42
- quic "github.com/libp2p/go-libp2p-quic-transport"
43
- "github.com/libp2p/go-libp2p-record"
44
- "github.com/libp2p/go-libp2p-routing"
45
- rhelpers "github.com/libp2p/go-libp2p-routing-helpers"
46
- "github.com/libp2p/go-libp2p/p2p/discovery"
47
- rhost "github.com/libp2p/go-libp2p/p2p/host/routed"
7
"go.uber.org/fx"
49
- "time"
50
-
51
- "github.com/ipfs/go-ipfs/repo"
8
53
- retry "github.com/ipfs/go-datastore/retrystore"
9
iconfig "github.com/ipfs/go-ipfs-config"
10
uio "github.com/ipfs/go-unixfs/io"
56
- ic "github.com/libp2p/go-libp2p-crypto"
57
- p2phost "github.com/libp2p/go-libp2p-host"
58
- "github.com/libp2p/go-libp2p-peer"
59
- mamask "github.com/whyrusleeping/multiaddr-filter"
11
)
12
62
-func repoConfig(repo repo.Repo) (*iconfig.Config, error) {
63
- return repo.Config()
64
-}
65
-
66
-func identity(cfg *iconfig.Config) (peer.ID, error) {
67
- cid := cfg.Identity.PeerID
68
- if cid == "" {
69
- return "", errors.New("identity was not set in config (was 'ipfs init' run?)")
70
- }
71
- if len(cid) == 0 {
72
- return "", errors.New("no peer ID in config! (was 'ipfs init' run?)")
73
- }
74
-
75
- id, err := peer.IDB58Decode(cid)
76
- if err != nil {
77
- return "", fmt.Errorf("peer ID invalid: %s", err)
78
- }
79
-
80
- return id, nil
81
-}
82
-
83
-func peerstore(id peer.ID, sk ic.PrivKey) pstore.Peerstore {
84
- ps := pstoremem.NewPeerstore()
85
-
86
- if sk != nil {
87
- ps.AddPrivKey(id, sk)
88
- ps.AddPubKey(id, sk.GetPublic())
89
- }
90
-
91
- return ps
92
-}
93
-
94
-func privateKey(cfg *iconfig.Config, id peer.ID) (ic.PrivKey, error) {
95
- if cfg.Identity.PrivKey == "" {
96
- return nil, nil
97
- }
98
-
99
- sk, err := cfg.Identity.DecodePrivateKey("passphrase todo!")
100
- if err != nil {
101
- return nil, err
102
- }
103
-
104
- id2, err := peer.IDFromPrivateKey(sk)
105
- if err != nil {
106
- return nil, err
107
- }
108
-
109
- if id2 != id {
110
- return nil, fmt.Errorf("private key in config does not match id: %s != %s", id, id2)
111
- }
112
- return sk, nil
113
-}
114
-
115
-func datastoreCtor(repo repo.Repo) ds.Datastore {
116
- return repo.Datastore()
117
-}
118
-
119
-type BaseBlocks bstore.Blockstore
120
-
121
-func baseBlockstoreCtor(mctx MetricsCtx, repo repo.Repo, cfg *iconfig.Config, bcfg *BuildCfg, lc fx.Lifecycle) (bs BaseBlocks, err error) {
122
- rds := &retry.Datastore{
123
- Batching: repo.Datastore(),
124
- Delay: time.Millisecond * 200,
125
- Retries: 6,
126
- TempErrFunc: isTooManyFDError,
127
- }
128
- // hash security
129
- bs = bstore.NewBlockstore(rds)
130
- bs = &verifbs.VerifBS{Blockstore: bs}
131
-
132
- opts := bstore.DefaultCacheOpts()
133
- opts.HasBloomFilterSize = cfg.Datastore.BloomFilterSize
134
- if !bcfg.Permanent {
135
- opts.HasBloomFilterSize = 0
136
- }
137
-
138
- if !bcfg.NilRepo {
139
- ctx, cancel := context.WithCancel(mctx)
140
-
141
- lc.Append(fx.Hook{
142
- OnStop: func(context context.Context) error {
143
- cancel()
144
- return nil
145
- },
146
- })
147
- bs, err = bstore.CachedBlockstore(ctx, bs, opts)
148
- if err != nil {
149
- return nil, err
150
- }
151
- }
152
-
153
- bs = bstore.NewIdStore(bs)
154
- bs = cidv0v1.NewBlockstore(bs)
155
-
156
- if cfg.Datastore.HashOnRead { // TODO: review: this is how it was done originally, is there a reason we can't just pass this directly?
157
- bs.HashOnRead(true)
158
- }
159
-
160
- return
161
-}
162
-
163
-func gcBlockstoreCtor(lc fx.Lifecycle, repo repo.Repo, bb BaseBlocks, cfg *iconfig.Config) (gclocker bstore.GCLocker, gcbs bstore.GCBlockstore, bs bstore.Blockstore, fstore *filestore.Filestore) {
164
- gclocker = bstore.NewGCLocker()
165
- gcbs = bstore.NewGCBlockstore(bb, gclocker)
166
-
167
- if cfg.Experimental.FilestoreEnabled || cfg.Experimental.UrlstoreEnabled {
168
- // hash security
169
- fstore = filestore.NewFilestore(bb, repo.FileManager()) //TODO: mark optional
170
- gcbs = bstore.NewGCBlockstore(fstore, gclocker)
171
- gcbs = &verifbs.VerifBSGC{GCBlockstore: gcbs}
172
- }
173
- bs = gcbs
174
- return
175
-}
176
-
177
-func blockServiceCtor(lc fx.Lifecycle, bs bstore.Blockstore, rem exchange.Interface) bserv.BlockService {
178
- bsvc := bserv.New(bs, rem)
179
-
180
- lc.Append(fx.Hook{
181
- OnStop: func(ctx context.Context) error {
182
- return bsvc.Close()
183
- },
184
- })
185
-
186
- return bsvc
187
-}
188
-
189
-func recordValidator(ps pstore.Peerstore) record.Validator {
190
- return record.NamespacedValidator{
191
- "pk": record.PublicKeyValidator{},
192
- "ipns": ipns.Validator{KeyBook: ps},
193
- }
194
-}
195
-
196
-////////////////////
197
-// libp2p related
198
-
13
////////////////////
14
// libp2p
15
202
-var ipfsp2p = fx.Options(
203
- fx.Provide(p2pAddrFilters),
204
- fx.Provide(p2pBandwidthCounter),
205
- fx.Provide(p2pPNet),
206
- fx.Provide(p2pAddrsFactory),
207
- fx.Provide(p2pConnectionManager),
208
- fx.Provide(p2pSmuxTransport),
209
- fx.Provide(p2pNatPortMap),
210
- fx.Provide(p2pRelay),
211
- fx.Provide(p2pAutoRealy),
212
- fx.Provide(p2pDefaultTransports),
213
- fx.Provide(p2pQUIC),
214
-
215
- fx.Provide(p2pHostOption),
216
- fx.Provide(p2pHost),
217
- fx.Provide(p2pOnlineRouting),
218
-
219
- fx.Provide(pubsubCtor),
220
- fx.Provide(newDiscoveryHandler),
221
-
222
- fx.Invoke(autoNATService),
223
- fx.Invoke(p2pPNetChecker),
224
- fx.Invoke(startListening),
225
- fx.Invoke(setupDiscovery),
226
-)
227
-
228
-func p2pHostOption(bcfg *BuildCfg) (hostOption HostOption, err error) {
229
- hostOption = bcfg.Host
230
- if bcfg.DisableEncryptedConnections {
231
- innerHostOption := hostOption
232
- hostOption = func(ctx context.Context, id peer.ID, ps pstore.Peerstore, options ...libp2p.Option) (p2phost.Host, error) {
233
- return innerHostOption(ctx, id, ps, append(options, libp2p.NoSecurity)...)
234
- }
235
- // TODO: shouldn't this be Errorf to guarantee visibility?
236
- log.Warningf(`Your IPFS node has been configured to run WITHOUT ENCRYPTED CONNECTIONS.
237
- You will not be able to connect to any nodes configured to use encrypted connections`)
238
- }
239
- return hostOption, nil
240
-}
241
-
242
-func p2pAddrFilters(cfg *iconfig.Config) (opts libp2pOpts, err error) {
243
- for _, s := range cfg.Swarm.AddrFilters {
244
- f, err := mamask.NewMask(s)
245
- if err != nil {
246
- return opts, fmt.Errorf("incorrectly formatted address filter in config: %s", s)
247
- }
248
- opts.Opts = append(opts.Opts, libp2p.FilterAddresses(f))
249
- }
250
- return opts, nil
251
-}
252
-
253
-func p2pBandwidthCounter(cfg *iconfig.Config) (opts libp2pOpts, reporter metrics.Reporter) {
254
- reporter = metrics.NewBandwidthCounter()
255
-
256
- if !cfg.Swarm.DisableBandwidthMetrics {
257
- opts.Opts = append(opts.Opts, libp2p.BandwidthReporter(reporter))
258
- }
259
- return opts, reporter
260
-}
261
-
262
-type libp2pOpts struct {
263
- fx.Out
264
-
265
- Opts []libp2p.Option `group:"libp2p"`
266
-}
267
-
268
-type PNetFingerprint []byte // TODO: find some better place
269
-func p2pPNet(repo repo.Repo) (opts libp2pOpts, fp PNetFingerprint, err error) {
270
- swarmkey, err := repo.SwarmKey()
271
- if err != nil || swarmkey == nil {
272
- return opts, nil, err
273
- }
274
-
275
- protec, err := pnet.NewProtector(bytes.NewReader(swarmkey))
276
- if err != nil {
277
- return opts, nil, fmt.Errorf("failed to configure private network: %s", err)
278
- }
279
- fp = protec.Fingerprint()
280
-
281
- opts.Opts = append(opts.Opts, libp2p.PrivateNetwork(protec))
282
- return opts, fp, nil
283
-}
284
-
285
-func p2pPNetChecker(repo repo.Repo, ph p2phost.Host, lc fx.Lifecycle) error {
286
- // TODO: better check?
287
- swarmkey, err := repo.SwarmKey()
288
- if err != nil || swarmkey == nil {
289
- return err
290
- }
291
-
292
- done := make(chan struct{})
293
- lc.Append(fx.Hook{
294
- OnStart: func(_ context.Context) error {
295
- go func() {
296
- t := time.NewTicker(30 * time.Second)
297
- <-t.C // swallow one tick
298
- for {
299
- select {
300
- case <-t.C:
301
- if len(ph.Network().Peers()) == 0 {
302
- log.Warning("We are in private network and have no peers.")
303
- log.Warning("This might be configuration mistake.")
304
- }
305
- case <-done:
306
- return
307
- }
308
- }
309
- }()
310
- return nil
311
- },
312
- OnStop: func(_ context.Context) error {
313
- close(done)
314
- return nil
315
- },
316
- })
317
- return nil
318
-}
319
-
320
-func p2pAddrsFactory(cfg *iconfig.Config) (opts libp2pOpts, err error) {
321
- addrsFactory, err := makeAddrsFactory(cfg.Addresses)
322
- if err != nil {
323
- return opts, err
324
- }
325
- if !cfg.Swarm.DisableRelay {
326
- addrsFactory = composeAddrsFactory(addrsFactory, filterRelayAddrs)
327
- }
328
- opts.Opts = append(opts.Opts, libp2p.AddrsFactory(addrsFactory))
329
- return
330
-}
331
-
332
-func p2pConnectionManager(cfg *iconfig.Config) (opts libp2pOpts, err error) {
333
- grace := iconfig.DefaultConnMgrGracePeriod
334
- low := iconfig.DefaultConnMgrHighWater
335
- high := iconfig.DefaultConnMgrHighWater
336
-
337
- switch cfg.Swarm.ConnMgr.Type {
338
- case "":
339
- // 'default' value is the basic connection manager
340
- return
341
- case "none":
342
- return opts, nil
343
- case "basic":
344
- grace, err = time.ParseDuration(cfg.Swarm.ConnMgr.GracePeriod)
345
- if err != nil {
346
- return opts, fmt.Errorf("parsing Swarm.ConnMgr.GracePeriod: %s", err)
347
- }
348
-
349
- low = cfg.Swarm.ConnMgr.LowWater
350
- high = cfg.Swarm.ConnMgr.HighWater
351
- default:
352
- return opts, fmt.Errorf("unrecognized ConnMgr.Type: %q", cfg.Swarm.ConnMgr.Type)
353
- }
354
-
355
- cm := connmgr.NewConnManager(low, high, grace)
356
- opts.Opts = append(opts.Opts, libp2p.ConnectionManager(cm))
357
- return
358
-}
359
-
360
-func p2pSmuxTransport(bcfg *BuildCfg) (opts libp2pOpts, err error) {
361
- opts.Opts = append(opts.Opts, makeSmuxTransportOption(bcfg.getOpt("mplex")))
362
- return
363
-}
364
-
365
-func p2pNatPortMap(cfg *iconfig.Config) (opts libp2pOpts, err error) {
366
- if !cfg.Swarm.DisableNatPortMap {
367
- opts.Opts = append(opts.Opts, libp2p.NATPortMap())
368
- }
369
- return
370
-}
371
-
372
-func p2pRelay(cfg *iconfig.Config) (opts libp2pOpts, err error) {
373
- if cfg.Swarm.DisableRelay {
374
- // Enabled by default.
375
- opts.Opts = append(opts.Opts, libp2p.DisableRelay())
376
- } else {
377
- relayOpts := []circuit.RelayOpt{circuit.OptDiscovery}
378
- if cfg.Swarm.EnableRelayHop {
379
- relayOpts = append(relayOpts, circuit.OptHop)
380
- }
381
- opts.Opts = append(opts.Opts, libp2p.EnableRelay(relayOpts...))
382
- }
383
- return
384
-}
385
-
386
-func p2pAutoRealy(cfg *iconfig.Config) (opts libp2pOpts, err error) {
387
- // enable autorelay
388
- if cfg.Swarm.EnableAutoRelay {
389
- opts.Opts = append(opts.Opts, libp2p.EnableAutoRelay())
390
- }
391
- return
392
-}
393
-
394
-func p2pDefaultTransports() (opts libp2pOpts, err error) {
395
- opts.Opts = append(opts.Opts, libp2p.DefaultTransports)
396
- return
397
-}
398
-
399
-func p2pQUIC(cfg *iconfig.Config) (opts libp2pOpts, err error) {
400
- if cfg.Experimental.QUIC {
401
- opts.Opts = append(opts.Opts, libp2p.Transport(quic.NewTransport))
402
- }
403
- return
404
-}
405
-
406
-type p2pHostIn struct {
407
- fx.In
408
-
409
- BCfg *BuildCfg
410
- Repo repo.Repo
411
- Validator record.Validator
412
- HostOption HostOption
413
- ID peer.ID
414
- Peerstore pstore.Peerstore
415
-
416
- Opts [][]libp2p.Option `group:"libp2p"`
417
-}
418
-
419
-type BaseRouting routing.IpfsRouting
420
-type p2pHostOut struct {
421
- fx.Out
422
-
423
- Host p2phost.Host
424
- Routing BaseRouting
425
- IpfsDHT *dht.IpfsDHT
426
-}
427
-
428
-// TODO: move some of this into params struct
429
-func p2pHost(mctx MetricsCtx, lc fx.Lifecycle, params p2pHostIn) (out p2pHostOut, err error) {
430
- opts := []libp2p.Option{libp2p.NoListenAddrs}
431
- for _, o := range params.Opts {
432
- opts = append(opts, o...)
433
- }
434
-
435
- ctx, cancel := context.WithCancel(mctx)
436
- lc.Append(fx.Hook{
437
- OnStop: func(_ context.Context) error {
438
- cancel()
439
- return nil
440
- },
441
- })
442
-
443
- opts = append(opts, libp2p.Routing(func(h p2phost.Host) (routing.PeerRouting, error) {
444
- r, err := params.BCfg.Routing(ctx, h, params.Repo.Datastore(), params.Validator)
445
- out.Routing = r
446
- return r, err
447
- }))
448
-
449
- out.Host, err = params.HostOption(ctx, params.ID, params.Peerstore, opts...)
450
- if err != nil {
451
- return p2pHostOut{}, err
452
- }
453
-
454
- // this code is necessary just for tests: mock network constructions
455
- // ignore the libp2p constructor options that actually construct the routing!
456
- if out.Routing == nil {
457
- r, err := params.BCfg.Routing(ctx, out.Host, params.Repo.Datastore(), params.Validator)
458
- if err != nil {
459
- return p2pHostOut{}, err
460
- }
461
- out.Routing = r
462
- out.Host = rhost.Wrap(out.Host, out.Routing)
463
- }
464
-
465
- lc.Append(fx.Hook{
466
- OnStop: func(ctx context.Context) error {
467
- return out.Host.Close()
468
- },
469
- })
470
-
471
- // TODO: break this up into more DI units
472
- // TODO: I'm not a fan of type assertions like this but the
473
- // `RoutingOption` system doesn't currently provide access to the
474
- // IpfsNode.
475
- //
476
- // Ideally, we'd do something like:
477
- //
478
- // 1. Add some fancy method to introspect into tiered routers to extract
479
- // things like the pubsub router or the DHT (complicated, messy,
480
- // probably not worth it).
481
- // 2. Pass the IpfsNode into the RoutingOption (would also remove the
482
- // PSRouter case below.
483
- // 3. Introduce some kind of service manager? (my personal favorite but
484
- // that requires a fair amount of work).
485
- if dht, ok := out.Routing.(*dht.IpfsDHT); ok {
486
- out.IpfsDHT = dht
487
-
488
- lc.Append(fx.Hook{
489
- OnStop: func(ctx context.Context) error {
490
- return out.IpfsDHT.Close()
491
- },
492
- })
493
- }
494
-
495
- return out, err
496
-}
497
-
498
-type p2pRoutingIn struct {
499
- fx.In
500
-
501
- BCfg *BuildCfg
502
- Repo repo.Repo
503
- Validator record.Validator
504
- Host p2phost.Host
505
- PubSub *pubsub.PubSub
506
-
507
- BaseRouting BaseRouting
508
-}
509
-
510
-type p2pRoutingOut struct {
511
- fx.Out
512
-
513
- IpfsRouting routing.IpfsRouting
514
- PSRouter *psrouter.PubsubValueStore //TODO: optional
515
-}
516
-
517
-func p2pOnlineRouting(mctx MetricsCtx, lc fx.Lifecycle, in p2pRoutingIn) (out p2pRoutingOut) {
518
- out.IpfsRouting = in.BaseRouting
519
-
520
- if in.BCfg.getOpt("ipnsps") {
521
- out.PSRouter = psrouter.NewPubsubValueStore(
522
- lifecycleCtx(mctx, lc),
523
- in.Host,
524
- in.BaseRouting,
525
- in.PubSub,
526
- in.Validator,
527
- )
528
-
529
- out.IpfsRouting = rhelpers.Tiered{
530
- Routers: []routing.IpfsRouting{
531
- // Always check pubsub first.
532
- &rhelpers.Compose{
533
- ValueStore: &rhelpers.LimitedValueStore{
534
- ValueStore: out.PSRouter,
535
- Namespaces: []string{"ipns"},
536
- },
537
- },
538
- in.BaseRouting,
539
- },
540
- Validator: in.Validator,
541
- }
542
- }
543
- return out
544
-}
545
-
546
-////////////
547
-// P2P services
548
-
549
-func autoNATService(mctx MetricsCtx, lc fx.Lifecycle, cfg *iconfig.Config, host p2phost.Host) error {
550
- if !cfg.Swarm.EnableAutoNATService {
551
- return nil
552
- }
553
- var opts []libp2p.Option
554
- if cfg.Experimental.QUIC {
555
- opts = append(opts, libp2p.DefaultTransports, libp2p.Transport(quic.NewTransport))
556
- }
557
-
558
- _, err := autonat.NewAutoNATService(lifecycleCtx(mctx, lc), host, opts...)
559
- return err
560
-}
561
-
562
-func pubsubCtor(mctx MetricsCtx, lc fx.Lifecycle, host p2phost.Host, bcfg *BuildCfg, cfg *iconfig.Config) (service *pubsub.PubSub, err error) {
563
- if !(bcfg.getOpt("pubsub") || bcfg.getOpt("ipnsps")) {
564
- return nil, nil // TODO: mark optional
565
- }
566
-
567
- var pubsubOptions []pubsub.Option
568
- if cfg.Pubsub.DisableSigning {
569
- pubsubOptions = append(pubsubOptions, pubsub.WithMessageSigning(false))
570
- }
571
-
572
- if cfg.Pubsub.StrictSignatureVerification {
573
- pubsubOptions = append(pubsubOptions, pubsub.WithStrictSignatureVerification(true))
574
- }
575
-
576
- switch cfg.Pubsub.Router {
577
- case "":
578
- fallthrough
579
- case "floodsub":
580
- service, err = pubsub.NewFloodSub(lifecycleCtx(mctx, lc), host, pubsubOptions...)
581
-
582
- case "gossipsub":
583
- service, err = pubsub.NewGossipSub(lifecycleCtx(mctx, lc), host, pubsubOptions...)
584
-
585
- default:
586
- err = fmt.Errorf("Unknown pubsub router %s", cfg.Pubsub.Router)
587
- }
588
-
589
- return service, err
590
-}
591
-
592
-////////////
593
-// Offline services
594
-
595
-// offline.Exchange
596
-// offroute.NewOfflineRouter
597
-
598
-func offlineNamesysCtor(rt routing.IpfsRouting, repo repo.Repo) (namesys.NameSystem, error) {
599
- return namesys.NewNameSystem(rt, repo.Datastore(), 0), nil
600
-}
601
-
602
-////////////
603
-// IPFS services
604
-
605
-func pinning(bstore bstore.Blockstore, ds format.DAGService, repo repo.Repo) (pin.Pinner, error) {
606
- internalDag := merkledag.NewDAGService(bserv.New(bstore, offline.Exchange(bstore)))
607
- pinning, err := pin.LoadPinner(repo.Datastore(), ds, internalDag)
608
- if err != nil {
609
- // TODO: we should move towards only running 'NewPinner' explicitly on
610
- // node init instead of implicitly here as a result of the pinner keys
611
- // not being found in the datastore.
612
- // this is kinda sketchy and could cause data loss
613
- pinning = pin.NewPinner(repo.Datastore(), ds, internalDag)
614
- }
615
-
616
- return pinning, nil
617
-}
618
-
619
-func dagCtor(bs bserv.BlockService) format.DAGService {
620
- return merkledag.NewDAGService(bs)
621
-}
622
-
623
-func onlineExchangeCtor(mctx MetricsCtx, lc fx.Lifecycle, host p2phost.Host, rt routing.IpfsRouting, bs bstore.GCBlockstore) exchange.Interface {
624
- bitswapNetwork := bsnet.NewFromIpfsHost(host, rt)
625
- exch := bitswap.New(lifecycleCtx(mctx, lc), bitswapNetwork, bs)
626
- lc.Append(fx.Hook{
627
- OnStop: func(ctx context.Context) error {
628
- return exch.Close()
629
- },
630
- })
631
- return exch
632
-}
633
-
634
-func onlineNamesysCtor(rt routing.IpfsRouting, repo repo.Repo, cfg *iconfig.Config) (namesys.NameSystem, error) {
635
- cs := cfg.Ipns.ResolveCacheSize
636
- if cs == 0 {
637
- cs = DefaultIpnsCacheSize
638
- }
639
- if cs < 0 {
640
- return nil, fmt.Errorf("cannot specify negative resolve cache size")
641
- }
642
- return namesys.NewNameSystem(rt, repo.Datastore(), cs), nil
643
-}
644
-
645
-func ipnsRepublisher(lc lcProcess, cfg *iconfig.Config, namesys namesys.NameSystem, repo repo.Repo, privKey ic.PrivKey) error {
646
- repub := ipnsrp.NewRepublisher(namesys, repo.Datastore(), privKey, repo.Keystore())
647
-
648
- if cfg.Ipns.RepublishPeriod != "" {
649
- d, err := time.ParseDuration(cfg.Ipns.RepublishPeriod)
650
- if err != nil {
651
- return fmt.Errorf("failure to parse config setting IPNS.RepublishPeriod: %s", err)
652
- }
653
-
654
- if !u.Debug && (d < time.Minute || d > (time.Hour*24)) {
655
- return fmt.Errorf("config setting IPNS.RepublishPeriod is not between 1min and 1day: %s", d)
656
- }
657
-
658
- repub.Interval = d
659
- }
660
-
661
- if cfg.Ipns.RecordLifetime != "" {
662
- d, err := time.ParseDuration(cfg.Ipns.RecordLifetime)
663
- if err != nil {
664
- return fmt.Errorf("failure to parse config setting IPNS.RecordLifetime: %s", err)
665
- }
666
-
667
- repub.RecordLifetime = d
668
- }
669
-
670
- lc.Run(repub.Run)
671
- return nil
672
-}
673
-
674
-type discoveryHandler struct {
675
- ctx context.Context
676
- host p2phost.Host
677
-}
678
-
679
-func (dh *discoveryHandler) HandlePeerFound(p pstore.PeerInfo) {
680
- log.Warning("trying peer info: ", p)
681
- ctx, cancel := context.WithTimeout(dh.ctx, discoveryConnTimeout)
682
- defer cancel()
683
- if err := dh.host.Connect(ctx, p); err != nil {
684
- log.Warning("Failed to connect to peer found by discovery: ", err)
685
- }
686
-}
687
-
688
-func newDiscoveryHandler(mctx MetricsCtx, lc fx.Lifecycle, host p2phost.Host) *discoveryHandler {
689
- return &discoveryHandler{
690
- ctx: lifecycleCtx(mctx, lc),
691
- host: host,
692
- }
693
-}
694
-
695
-func setupDiscovery(mctx MetricsCtx, lc fx.Lifecycle, cfg *iconfig.Config, host p2phost.Host, handler *discoveryHandler) error {
696
- if cfg.Discovery.MDNS.Enabled {
697
- mdns := cfg.Discovery.MDNS
698
- if mdns.Interval == 0 {
699
- mdns.Interval = 5
700
- }
701
- service, err := discovery.NewMdnsService(lifecycleCtx(mctx, lc), host, time.Duration(mdns.Interval)*time.Second, discovery.ServiceTag)
702
- if err != nil {
703
- log.Error("mdns error: ", err)
704
- return nil
705
- }
706
- service.RegisterNotifee(handler)
707
- }
708
- return nil
709
-}
710
-
711
-func providerQueue(mctx MetricsCtx, lc fx.Lifecycle, repo repo.Repo) (*provider.Queue, error) {
712
- return provider.NewQueue(lifecycleCtx(mctx, lc), "provider-v1", repo.Datastore())
713
-}
714
-
715
-func providerCtor(mctx MetricsCtx, lc fx.Lifecycle, queue *provider.Queue, rt routing.IpfsRouting) provider.Provider {
716
- return provider.NewProvider(lifecycleCtx(mctx, lc), queue, rt)
717
-}
718
-
719
-func reproviderCtor(mctx MetricsCtx, lc fx.Lifecycle, cfg *iconfig.Config, bs BaseBlocks, ds format.DAGService, pinning pin.Pinner, rt routing.IpfsRouting) (*rp.Reprovider, error) {
720
- var keyProvider rp.KeyChanFunc
721
-
722
- switch cfg.Reprovider.Strategy {
723
- case "all":
724
- fallthrough
725
- case "":
726
- keyProvider = rp.NewBlockstoreProvider(bs)
727
- case "roots":
728
- keyProvider = rp.NewPinnedProvider(pinning, ds, true)
729
- case "pinned":
730
- keyProvider = rp.NewPinnedProvider(pinning, ds, false)
731
- default:
732
- return nil, fmt.Errorf("unknown reprovider strategy '%s'", cfg.Reprovider.Strategy)
733
- }
734
- return rp.NewReprovider(lifecycleCtx(mctx, lc), rt, keyProvider), nil
735
-}
736
-
737
-func reprovider(cfg *iconfig.Config, reprovider *rp.Reprovider) error {
738
- reproviderInterval := kReprovideFrequency
739
- if cfg.Reprovider.Interval != "" {
740
- dur, err := time.ParseDuration(cfg.Reprovider.Interval)
741
- if err != nil {
742
- return err
743
- }
744
-
745
- reproviderInterval = dur
746
- }
747
-
748
- go reprovider.Run(reproviderInterval)
749
- return nil
750
-}
751
-
752
-func files(mctx MetricsCtx, lc fx.Lifecycle, repo repo.Repo, dag format.DAGService) (*mfs.Root, error) {
753
- dsk := ds.NewKey("/local/filesroot")
754
- pf := func(ctx context.Context, c cid.Cid) error {
755
- return repo.Datastore().Put(dsk, c.Bytes())
756
- }
757
-
758
- var nd *merkledag.ProtoNode
759
- val, err := repo.Datastore().Get(dsk)
760
- ctx := lifecycleCtx(mctx, lc)
761
-
762
- switch {
763
- case err == ds.ErrNotFound || val == nil:
764
- nd = ft.EmptyDirNode()
765
- err := dag.Add(ctx, nd)
766
- if err != nil {
767
- return nil, fmt.Errorf("failure writing to dagstore: %s", err)
768
- }
769
- case err == nil:
770
- c, err := cid.Cast(val)
771
- if err != nil {
772
- return nil, err
773
- }
774
-
775
- rnd, err := dag.Get(ctx, c)
776
- if err != nil {
777
- return nil, fmt.Errorf("error loading filesroot from DAG: %s", err)
778
- }
779
-
780
- pbnd, ok := rnd.(*merkledag.ProtoNode)
781
- if !ok {
782
- return nil, merkledag.ErrNotProtobuf
783
- }
784
-
785
- nd = pbnd
786
- default:
787
- return nil, err
788
- }
789
-
790
- root, err := mfs.NewRoot(ctx, dag, nd, pf)
791
-
792
- lc.Append(fx.Hook{
793
- OnStop: func(ctx context.Context) error {
794
- return root.Close()
795
- },
796
- })
797
-
798
- return root, err
799
-}
800
-
801
-////////////
802
-// Hacks
803
-
804
-// lifecycleCtx creates a context which will be cancelled when lifecycle stops
805
-//
806
-// This is a hack which we need because most of our services use contexts in a
807
-// wrong way
808
-func lifecycleCtx(mctx MetricsCtx, lc fx.Lifecycle) context.Context {
809
- ctx, cancel := context.WithCancel(mctx)
810
- lc.Append(fx.Hook{
811
- OnStop: func(_ context.Context) error {
812
- cancel()
813
- return nil
814
- },
815
- })
816
- return ctx
817
-}
818
-
819
-type lcProcess struct {
820
- fx.In
821
-
822
- LC fx.Lifecycle
823
- Proc goprocess.Process
824
-}
825
-
826
-func (lp *lcProcess) Run(f goprocess.ProcessFunc) {
827
- proc := make(chan goprocess.Process, 1)
828
- lp.LC.Append(fx.Hook{
829
- OnStart: func(ctx context.Context) error {
830
- proc <- lp.Proc.Go(f)
831
- return nil
832
- },
833
- OnStop: func(ctx context.Context) error {
834
- return (<-proc).Close() // todo: respect ctx, somehow
835
- },
836
- })
16
+func setupSharding(cfg *iconfig.Config) {
17
+ // TEMP: setting global sharding switch here
18
+ uio.UseHAMTSharding = cfg.Experimental.ShardingEnabled
19
}
20
21
func baseProcess(lc fx.Lifecycle) goprocess.Process {
@@ -845,8 +27,3 @@ func baseProcess(lc fx.Lifecycle) goprocess.Process {
27
})
28
return p
29
}
848
-
849
-func setupSharding(cfg *iconfig.Config) {
850
- // TEMP: setting global sharding switch here
851
- uio.UseHAMTSharding = cfg.Experimental.ShardingEnabled
852
-}
core/node/builder.go
new
+36
@@ -0,0 +1,36 @@
1
+package node
2
+
3
+import (
4
+ "github.com/ipfs/go-ipfs/repo"
5
+)
6
+
7
+type BuildCfg struct {
8
+ // If online is set, the node will have networking enabled
9
+ Online bool
10
+
11
+ // ExtraOpts is a map of extra options used to configure the ipfs nodes creation
12
+ ExtraOpts map[string]bool
13
+
14
+ // If permanent then node should run more expensive processes
15
+ // that will improve performance in long run
16
+ Permanent bool
17
+
18
+ // DisableEncryptedConnections disables connection encryption *entirely*.
19
+ // DO NOT SET THIS UNLESS YOU'RE TESTING.
20
+ DisableEncryptedConnections bool
21
+
22
+ // If NilRepo is set, a Repo backed by a nil datastore will be constructed
23
+ NilRepo bool
24
+
25
+ Routing RoutingOption
26
+ Host HostOption
27
+ Repo repo.Repo
28
+}
29
+
30
+func (cfg *BuildCfg) getOpt(key string) bool {
31
+ if cfg.ExtraOpts == nil {
32
+ return false
33
+ }
34
+
35
+ return cfg.ExtraOpts[key]
36
+}
core/node/core.go
new
+117
@@ -0,0 +1,117 @@
1
+package node
2
+
3
+import (
4
+ "context"
5
+ "fmt"
6
+
7
+ "github.com/ipfs/go-bitswap"
8
+ "github.com/ipfs/go-bitswap/network"
9
+ "github.com/ipfs/go-blockservice"
10
+ "github.com/ipfs/go-cid"
11
+ "github.com/ipfs/go-datastore"
12
+ blockstore "github.com/ipfs/go-ipfs-blockstore"
13
+ exchange "github.com/ipfs/go-ipfs-exchange-interface"
14
+ offline "github.com/ipfs/go-ipfs-exchange-offline"
15
+ format "github.com/ipfs/go-ipld-format"
16
+ "github.com/ipfs/go-merkledag"
17
+ "github.com/ipfs/go-mfs"
18
+ "github.com/ipfs/go-unixfs"
19
+ host "github.com/libp2p/go-libp2p-host"
20
+ routing "github.com/libp2p/go-libp2p-routing"
21
+ "go.uber.org/fx"
22
+
23
+ "github.com/ipfs/go-ipfs/pin"
24
+ "github.com/ipfs/go-ipfs/repo"
25
+)
26
+
27
+func BlockServiceCtor(lc fx.Lifecycle, bs blockstore.Blockstore, rem exchange.Interface) blockservice.BlockService {
28
+ bsvc := blockservice.New(bs, rem)
29
+
30
+ lc.Append(fx.Hook{
31
+ OnStop: func(ctx context.Context) error {
32
+ return bsvc.Close()
33
+ },
34
+ })
35
+
36
+ return bsvc
37
+}
38
+
39
+func Pinning(bstore blockstore.Blockstore, ds format.DAGService, repo repo.Repo) (pin.Pinner, error) {
40
+ internalDag := merkledag.NewDAGService(blockservice.New(bstore, offline.Exchange(bstore)))
41
+ pinning, err := pin.LoadPinner(repo.Datastore(), ds, internalDag)
42
+ if err != nil {
43
+ // TODO: we should move towards only running 'NewPinner' explicitly on
44
+ // node init instead of implicitly here as a result of the pinner keys
45
+ // not being found in the datastore.
46
+ // this is kinda sketchy and could cause data loss
47
+ pinning = pin.NewPinner(repo.Datastore(), ds, internalDag)
48
+ }
49
+
50
+ return pinning, nil
51
+}
52
+
53
+func DagCtor(bs blockservice.BlockService) format.DAGService {
54
+ return merkledag.NewDAGService(bs)
55
+}
56
+
57
+func OnlineExchangeCtor(mctx MetricsCtx, lc fx.Lifecycle, host host.Host, rt routing.IpfsRouting, bs blockstore.GCBlockstore) exchange.Interface {
58
+ bitswapNetwork := network.NewFromIpfsHost(host, rt)
59
+ exch := bitswap.New(lifecycleCtx(mctx, lc), bitswapNetwork, bs)
60
+ lc.Append(fx.Hook{
61
+ OnStop: func(ctx context.Context) error {
62
+ return exch.Close()
63
+ },
64
+ })
65
+ return exch
66
+}
67
+
68
+func Files(mctx MetricsCtx, lc fx.Lifecycle, repo repo.Repo, dag format.DAGService) (*mfs.Root, error) {
69
+ dsk := datastore.NewKey("/local/filesroot")
70
+ pf := func(ctx context.Context, c cid.Cid) error {
71
+ return repo.Datastore().Put(dsk, c.Bytes())
72
+ }
73
+
74
+ var nd *merkledag.ProtoNode
75
+ val, err := repo.Datastore().Get(dsk)
76
+ ctx := lifecycleCtx(mctx, lc)
77
+
78
+ switch {
79
+ case err == datastore.ErrNotFound || val == nil:
80
+ nd = unixfs.EmptyDirNode()
81
+ err := dag.Add(ctx, nd)
82
+ if err != nil {
83
+ return nil, fmt.Errorf("failure writing to dagstore: %s", err)
84
+ }
85
+ case err == nil:
86
+ c, err := cid.Cast(val)
87
+ if err != nil {
88
+ return nil, err
89
+ }
90
+
91
+ rnd, err := dag.Get(ctx, c)
92
+ if err != nil {
93
+ return nil, fmt.Errorf("error loading filesroot from DAG: %s", err)
94
+ }
95
+
96
+ pbnd, ok := rnd.(*merkledag.ProtoNode)
97
+ if !ok {
98
+ return nil, merkledag.ErrNotProtobuf
99
+ }
100
+
101
+ nd = pbnd
102
+ default:
103
+ return nil, err
104
+ }
105
+
106
+ root, err := mfs.NewRoot(ctx, dag, nd, pf)
107
+
108
+ lc.Append(fx.Hook{
109
+ OnStop: func(ctx context.Context) error {
110
+ return root.Close()
111
+ },
112
+ })
113
+
114
+ return root, err
115
+}
116
+
117
+type MetricsCtx context.Context
core/node/discovery.go
new
+51
@@ -0,0 +1,51 @@
1
+package node
2
+
3
+import (
4
+ "context"
5
+ "time"
6
+
7
+ "github.com/ipfs/go-ipfs-config"
8
+ "github.com/libp2p/go-libp2p-host"
9
+ "github.com/libp2p/go-libp2p-peerstore"
10
+ "github.com/libp2p/go-libp2p/p2p/discovery"
11
+ "go.uber.org/fx"
12
+)
13
+
14
+const discoveryConnTimeout = time.Second * 30
15
+
16
+type discoveryHandler struct {
17
+ ctx context.Context
18
+ host host.Host
19
+}
20
+
21
+func (dh *discoveryHandler) HandlePeerFound(p peerstore.PeerInfo) {
22
+ log.Warning("trying peer info: ", p)
23
+ ctx, cancel := context.WithTimeout(dh.ctx, discoveryConnTimeout)
24
+ defer cancel()
25
+ if err := dh.host.Connect(ctx, p); err != nil {
26
+ log.Warning("Failed to connect to peer found by discovery: ", err)
27
+ }
28
+}
29
+
30
+func NewDiscoveryHandler(mctx MetricsCtx, lc fx.Lifecycle, host host.Host) *discoveryHandler {
31
+ return &discoveryHandler{
32
+ ctx: lifecycleCtx(mctx, lc),
33
+ host: host,
34
+ }
35
+}
36
+
37
+func SetupDiscovery(mctx MetricsCtx, lc fx.Lifecycle, cfg *config.Config, host host.Host, handler *discoveryHandler) error {
38
+ if cfg.Discovery.MDNS.Enabled {
39
+ mdns := cfg.Discovery.MDNS
40
+ if mdns.Interval == 0 {
41
+ mdns.Interval = 5
42
+ }
43
+ service, err := discovery.NewMdnsService(lifecycleCtx(mctx, lc), host, time.Duration(mdns.Interval)*time.Second, discovery.ServiceTag)
44
+ if err != nil {
45
+ log.Error("mdns error: ", err)
46
+ return nil
47
+ }
48
+ service.RegisterNotifee(handler)
49
+ }
50
+ return nil
51
+}
core/node/groups.go
new
+88
@@ -0,0 +1,88 @@
1
+package node
2
+
3
+import (
4
+ offline "github.com/ipfs/go-ipfs-exchange-offline"
5
+ "go.uber.org/fx"
6
+
7
+ offroute "github.com/ipfs/go-ipfs-routing/offline"
8
+ "github.com/ipfs/go-ipfs/p2p"
9
+ "github.com/ipfs/go-ipfs/provider"
10
+)
11
+
12
+var LibP2P = fx.Options(
13
+ fx.Provide(P2PAddrFilters),
14
+ fx.Provide(P2PBandwidthCounter),
15
+ fx.Provide(P2PPNet),
16
+ fx.Provide(P2PAddrsFactory),
17
+ fx.Provide(P2PConnectionManager),
18
+ fx.Provide(P2PSmuxTransport),
19
+ fx.Provide(P2PNatPortMap),
20
+ fx.Provide(P2PRelay),
21
+ fx.Provide(P2PAutoRealy),
22
+ fx.Provide(P2PDefaultTransports),
23
+ fx.Provide(P2PQUIC),
24
+
25
+ fx.Provide(P2PHostOption),
26
+ fx.Provide(P2PHost),
27
+ fx.Provide(P2POnlineRouting),
28
+
29
+ fx.Provide(Pubsub),
30
+ fx.Provide(NewDiscoveryHandler),
31
+
32
+ fx.Invoke(AutoNATService),
33
+ fx.Invoke(P2PPNetChecker),
34
+ fx.Invoke(StartListening),
35
+ fx.Invoke(SetupDiscovery),
36
+)
37
+
38
+var Storage = fx.Options(
39
+ fx.Provide(RepoConfig),
40
+ fx.Provide(DatastoreCtor),
41
+ fx.Provide(BaseBlockstoreCtor),
42
+ fx.Provide(GcBlockstoreCtor),
43
+)
44
+
45
+var Identity = fx.Options(
46
+ fx.Provide(PeerID),
47
+ fx.Provide(PrivateKey),
48
+ fx.Provide(Peerstore),
49
+)
50
+
51
+var IPNS = fx.Options(
52
+ fx.Provide(RecordValidator),
53
+)
54
+
55
+var Providers = fx.Options(
56
+ fx.Provide(ProviderQueue),
57
+ fx.Provide(ProviderCtor),
58
+ fx.Provide(ReproviderCtor),
59
+
60
+ fx.Invoke(Reprovider),
61
+ fx.Invoke(provider.Provider.Run),
62
+)
63
+
64
+var Online = fx.Options(
65
+ fx.Provide(OnlineExchangeCtor),
66
+ fx.Provide(OnlineNamesysCtor),
67
+
68
+ fx.Invoke(IpnsRepublisher),
69
+
70
+ fx.Provide(p2p.NewP2P),
71
+
72
+ LibP2P,
73
+ Providers,
74
+)
75
+
76
+var Offline = fx.Options(
77
+ fx.Provide(offline.Exchange),
78
+ fx.Provide(OfflineNamesysCtor),
79
+ fx.Provide(offroute.NewOfflineRouter),
80
+ fx.Provide(provider.NewOfflineProvider),
81
+)
82
+
83
+func Networked(online bool) fx.Option {
84
+ if online {
85
+ return Online
86
+ }
87
+ return Offline
88
+}
core/node/helpers.go
new
+43
@@ -0,0 +1,43 @@
1
+package node
2
+
3
+import (
4
+ "context"
5
+
6
+ "github.com/jbenet/goprocess"
7
+ "go.uber.org/fx"
8
+)
9
+
10
+// lifecycleCtx creates a context which will be cancelled when lifecycle stops
11
+//
12
+// This is a hack which we need because most of our services use contexts in a
13
+// wrong way
14
+func lifecycleCtx(mctx MetricsCtx, lc fx.Lifecycle) context.Context {
15
+ ctx, cancel := context.WithCancel(mctx)
16
+ lc.Append(fx.Hook{
17
+ OnStop: func(_ context.Context) error {
18
+ cancel()
19
+ return nil
20
+ },
21
+ })
22
+ return ctx
23
+}
24
+
25
+type lcProcess struct {
26
+ fx.In
27
+
28
+ LC fx.Lifecycle
29
+ Proc goprocess.Process
30
+}
31
+
32
+func (lp *lcProcess) Run(f goprocess.ProcessFunc) {
33
+ proc := make(chan goprocess.Process, 1)
34
+ lp.LC.Append(fx.Hook{
35
+ OnStart: func(ctx context.Context) error {
36
+ proc <- lp.Proc.Go(f)
37
+ return nil
38
+ },
39
+ OnStop: func(ctx context.Context) error {
40
+ return (<-proc).Close() // todo: respect ctx, somehow
41
+ },
42
+ })
43
+}
core/node/identity.go
new
+48
@@ -0,0 +1,48 @@
1
+package node
2
+
3
+import (
4
+ "errors"
5
+ "fmt"
6
+
7
+ "github.com/ipfs/go-ipfs-config"
8
+ "github.com/libp2p/go-libp2p-crypto"
9
+ "github.com/libp2p/go-libp2p-peer"
10
+)
11
+
12
+func PeerID(cfg *config.Config) (peer.ID, error) {
13
+ cid := cfg.Identity.PeerID
14
+ if cid == "" {
15
+ return "", errors.New("identity was not set in config (was 'ipfs init' run?)")
16
+ }
17
+ if len(cid) == 0 {
18
+ return "", errors.New("no peer ID in config! (was 'ipfs init' run?)")
19
+ }
20
+
21
+ id, err := peer.IDB58Decode(cid)
22
+ if err != nil {
23
+ return "", fmt.Errorf("peer ID invalid: %s", err)
24
+ }
25
+
26
+ return id, nil
27
+}
28
+
29
+func PrivateKey(cfg *config.Config, id peer.ID) (crypto.PrivKey, error) {
30
+ if cfg.Identity.PrivKey == "" {
31
+ return nil, nil
32
+ }
33
+
34
+ sk, err := cfg.Identity.DecodePrivateKey("passphrase todo!")
35
+ if err != nil {
36
+ return nil, err
37
+ }
38
+
39
+ id2, err := peer.IDFromPrivateKey(sk)
40
+ if err != nil {
41
+ return nil, err
42
+ }
43
+
44
+ if id2 != id {
45
+ return nil, fmt.Errorf("private key in config does not match id: %s != %s", id, id2)
46
+ }
47
+ return sk, nil
48
+}
core/node/ipns.go
new
+71
@@ -0,0 +1,71 @@
1
+package node
2
+
3
+import (
4
+ "fmt"
5
+ "time"
6
+
7
+ "github.com/ipfs/go-ipfs-config"
8
+ "github.com/ipfs/go-ipfs-util"
9
+ "github.com/ipfs/go-ipns"
10
+ "github.com/libp2p/go-libp2p-crypto"
11
+ "github.com/libp2p/go-libp2p-peerstore"
12
+ "github.com/libp2p/go-libp2p-record"
13
+ "github.com/libp2p/go-libp2p-routing"
14
+
15
+ "github.com/ipfs/go-ipfs/namesys"
16
+ "github.com/ipfs/go-ipfs/namesys/republisher"
17
+ "github.com/ipfs/go-ipfs/repo"
18
+)
19
+
20
+const DefaultIpnsCacheSize = 128
21
+
22
+func RecordValidator(ps peerstore.Peerstore) record.Validator {
23
+ return record.NamespacedValidator{
24
+ "pk": record.PublicKeyValidator{},
25
+ "ipns": ipns.Validator{KeyBook: ps},
26
+ }
27
+}
28
+
29
+func OfflineNamesysCtor(rt routing.IpfsRouting, repo repo.Repo) (namesys.NameSystem, error) {
30
+ return namesys.NewNameSystem(rt, repo.Datastore(), 0), nil
31
+}
32
+
33
+func OnlineNamesysCtor(rt routing.IpfsRouting, repo repo.Repo, cfg *config.Config) (namesys.NameSystem, error) {
34
+ cs := cfg.Ipns.ResolveCacheSize
35
+ if cs == 0 {
36
+ cs = DefaultIpnsCacheSize
37
+ }
38
+ if cs < 0 {
39
+ return nil, fmt.Errorf("cannot specify negative resolve cache size")
40
+ }
41
+ return namesys.NewNameSystem(rt, repo.Datastore(), cs), nil
42
+}
43
+
44
+func IpnsRepublisher(lc lcProcess, cfg *config.Config, namesys namesys.NameSystem, repo repo.Repo, privKey crypto.PrivKey) error {
45
+ repub := republisher.NewRepublisher(namesys, repo.Datastore(), privKey, repo.Keystore())
46
+
47
+ if cfg.Ipns.RepublishPeriod != "" {
48
+ d, err := time.ParseDuration(cfg.Ipns.RepublishPeriod)
49
+ if err != nil {
50
+ return fmt.Errorf("failure to parse config setting IPNS.RepublishPeriod: %s", err)
51
+ }
52
+
53
+ if !util.Debug && (d < time.Minute || d > (time.Hour*24)) {
54
+ return fmt.Errorf("config setting IPNS.RepublishPeriod is not between 1min and 1day: %s", d)
55
+ }
56
+
57
+ repub.Interval = d
58
+ }
59
+
60
+ if cfg.Ipns.RecordLifetime != "" {
61
+ d, err := time.ParseDuration(cfg.Ipns.RecordLifetime)
62
+ if err != nil {
63
+ return fmt.Errorf("failure to parse config setting IPNS.RecordLifetime: %s", err)
64
+ }
65
+
66
+ repub.RecordLifetime = d
67
+ }
68
+
69
+ lc.Run(repub.Run)
70
+ return nil
71
+}
core/node/libp2p.go
new
+597
@@ -0,0 +1,597 @@
1
+package node
2
+
3
+import (
4
+ "bytes"
5
+ "context"
6
+ "fmt"
7
+ "io/ioutil"
8
+ "os"
9
+ "strings"
10
+ "time"
11
+
12
+ "github.com/ipfs/go-datastore"
13
+ "github.com/ipfs/go-ipfs-config"
14
+ nilrouting "github.com/ipfs/go-ipfs-routing/none"
15
+ logging "github.com/ipfs/go-log"
16
+ "github.com/libp2p/go-libp2p"
17
+ "github.com/libp2p/go-libp2p-autonat-svc"
18
+ "github.com/libp2p/go-libp2p-circuit"
19
+ circuit "github.com/libp2p/go-libp2p-circuit"
20
+ "github.com/libp2p/go-libp2p-connmgr"
21
+ "github.com/libp2p/go-libp2p-crypto"
22
+ "github.com/libp2p/go-libp2p-host"
23
+ "github.com/libp2p/go-libp2p-kad-dht"
24
+ dhtopts "github.com/libp2p/go-libp2p-kad-dht/opts"
25
+ "github.com/libp2p/go-libp2p-metrics"
26
+ "github.com/libp2p/go-libp2p-peer"
27
+ "github.com/libp2p/go-libp2p-peerstore"
28
+ "github.com/libp2p/go-libp2p-peerstore/pstoremem"
29
+ "github.com/libp2p/go-libp2p-pnet"
30
+ "github.com/libp2p/go-libp2p-pubsub"
31
+ "github.com/libp2p/go-libp2p-pubsub-router"
32
+ "github.com/libp2p/go-libp2p-quic-transport"
33
+ "github.com/libp2p/go-libp2p-record"
34
+ "github.com/libp2p/go-libp2p-routing"
35
+ "github.com/libp2p/go-libp2p-routing-helpers"
36
+ p2pbhost "github.com/libp2p/go-libp2p/p2p/host/basic"
37
+ "github.com/libp2p/go-libp2p/p2p/host/routed"
38
+ mafilter "github.com/libp2p/go-maddr-filter"
39
+ smux "github.com/libp2p/go-stream-muxer"
40
+ ma "github.com/multiformats/go-multiaddr"
41
+ mplex "github.com/whyrusleeping/go-smux-multiplex"
42
+ yamux "github.com/whyrusleeping/go-smux-yamux"
43
+ "github.com/whyrusleeping/multiaddr-filter"
44
+ mamask "github.com/whyrusleeping/multiaddr-filter"
45
+ "go.uber.org/fx"
46
+
47
+ "github.com/ipfs/go-ipfs/repo"
48
+)
49
+
50
+var log = logging.Logger("node")
51
+
52
+type HostOption func(ctx context.Context, id peer.ID, ps peerstore.Peerstore, options ...libp2p.Option) (host.Host, error)
53
+type RoutingOption func(context.Context, host.Host, datastore.Batching, record.Validator) (routing.IpfsRouting, error)
54
+
55
+var DefaultHostOption HostOption = constructPeerHost
56
+
57
+// isolates the complex initialization steps
58
+func constructPeerHost(ctx context.Context, id peer.ID, ps peerstore.Peerstore, options ...libp2p.Option) (host.Host, error) {
59
+ pkey := ps.PrivKey(id)
60
+ if pkey == nil {
61
+ return nil, fmt.Errorf("missing private key for node ID: %s", id.Pretty())
62
+ }
63
+ options = append([]libp2p.Option{libp2p.Identity(pkey), libp2p.Peerstore(ps)}, options...)
64
+ return libp2p.New(ctx, options...)
65
+}
66
+
67
+func constructDHTRouting(ctx context.Context, host host.Host, dstore datastore.Batching, validator record.Validator) (routing.IpfsRouting, error) {
68
+ return dht.New(
69
+ ctx, host,
70
+ dhtopts.Datastore(dstore),
71
+ dhtopts.Validator(validator),
72
+ )
73
+}
74
+
75
+func constructClientDHTRouting(ctx context.Context, host host.Host, dstore datastore.Batching, validator record.Validator) (routing.IpfsRouting, error) {
76
+ return dht.New(
77
+ ctx, host,
78
+ dhtopts.Client(true),
79
+ dhtopts.Datastore(dstore),
80
+ dhtopts.Validator(validator),
81
+ )
82
+}
83
+
84
+var DHTOption RoutingOption = constructDHTRouting
85
+var DHTClientOption RoutingOption = constructClientDHTRouting
86
+var NilRouterOption RoutingOption = nilrouting.ConstructNilRouting
87
+
88
+func Peerstore(id peer.ID, sk crypto.PrivKey) peerstore.Peerstore {
89
+ ps := pstoremem.NewPeerstore()
90
+
91
+ if sk != nil {
92
+ ps.AddPrivKey(id, sk)
93
+ ps.AddPubKey(id, sk.GetPublic())
94
+ }
95
+
96
+ return ps
97
+}
98
+
99
+func P2PAddrFilters(cfg *config.Config) (opts Libp2pOpts, err error) {
100
+ for _, s := range cfg.Swarm.AddrFilters {
101
+ f, err := mask.NewMask(s)
102
+ if err != nil {
103
+ return opts, fmt.Errorf("incorrectly formatted address filter in config: %s", s)
104
+ }
105
+ opts.Opts = append(opts.Opts, libp2p.FilterAddresses(f))
106
+ }
107
+ return opts, nil
108
+}
109
+
110
+func P2PBandwidthCounter(cfg *config.Config) (opts Libp2pOpts, reporter metrics.Reporter) {
111
+ reporter = metrics.NewBandwidthCounter()
112
+
113
+ if !cfg.Swarm.DisableBandwidthMetrics {
114
+ opts.Opts = append(opts.Opts, libp2p.BandwidthReporter(reporter))
115
+ }
116
+ return opts, reporter
117
+}
118
+
119
+type Libp2pOpts struct {
120
+ fx.Out
121
+
122
+ Opts []libp2p.Option `group:"libp2p"`
123
+}
124
+
125
+type PNetFingerprint []byte // TODO: find some better place
126
+func P2PPNet(repo repo.Repo) (opts Libp2pOpts, fp PNetFingerprint, err error) {
127
+ swarmkey, err := repo.SwarmKey()
128
+ if err != nil || swarmkey == nil {
129
+ return opts, nil, err
130
+ }
131
+
132
+ protec, err := pnet.NewProtector(bytes.NewReader(swarmkey))
133
+ if err != nil {
134
+ return opts, nil, fmt.Errorf("failed to configure private network: %s", err)
135
+ }
136
+ fp = protec.Fingerprint()
137
+
138
+ opts.Opts = append(opts.Opts, libp2p.PrivateNetwork(protec))
139
+ return opts, fp, nil
140
+}
141
+
142
+func P2PPNetChecker(repo repo.Repo, ph host.Host, lc fx.Lifecycle) error {
143
+ // TODO: better check?
144
+ swarmkey, err := repo.SwarmKey()
145
+ if err != nil || swarmkey == nil {
146
+ return err
147
+ }
148
+
149
+ done := make(chan struct{})
150
+ lc.Append(fx.Hook{
151
+ OnStart: func(_ context.Context) error {
152
+ go func() {
153
+ t := time.NewTicker(30 * time.Second)
154
+ <-t.C // swallow one tick
155
+ for {
156
+ select {
157
+ case <-t.C:
158
+ if len(ph.Network().Peers()) == 0 {
159
+ log.Warning("We are in private network and have no peers.")
160
+ log.Warning("This might be configuration mistake.")
161
+ }
162
+ case <-done:
163
+ return
164
+ }
165
+ }
166
+ }()
167
+ return nil
168
+ },
169
+ OnStop: func(_ context.Context) error {
170
+ close(done)
171
+ return nil
172
+ },
173
+ })
174
+ return nil
175
+}
176
+
177
+func makeAddrsFactory(cfg config.Addresses) (p2pbhost.AddrsFactory, error) {
178
+ var annAddrs []ma.Multiaddr
179
+ for _, addr := range cfg.Announce {
180
+ maddr, err := ma.NewMultiaddr(addr)
181
+ if err != nil {
182
+ return nil, err
183
+ }
184
+ annAddrs = append(annAddrs, maddr)
185
+ }
186
+
187
+ filters := mafilter.NewFilters()
188
+ noAnnAddrs := map[string]bool{}
189
+ for _, addr := range cfg.NoAnnounce {
190
+ f, err := mamask.NewMask(addr)
191
+ if err == nil {
192
+ filters.AddDialFilter(f)
193
+ continue
194
+ }
195
+ maddr, err := ma.NewMultiaddr(addr)
196
+ if err != nil {
197
+ return nil, err
198
+ }
199
+ noAnnAddrs[maddr.String()] = true
200
+ }
201
+
202
+ return func(allAddrs []ma.Multiaddr) []ma.Multiaddr {
203
+ var addrs []ma.Multiaddr
204
+ if len(annAddrs) > 0 {
205
+ addrs = annAddrs
206
+ } else {
207
+ addrs = allAddrs
208
+ }
209
+
210
+ var out []ma.Multiaddr
211
+ for _, maddr := range addrs {
212
+ // check for exact matches
213
+ ok := noAnnAddrs[maddr.String()]
214
+ // check for /ipcidr matches
215
+ if !ok && !filters.AddrBlocked(maddr) {
216
+ out = append(out, maddr)
217
+ }
218
+ }
219
+ return out
220
+ }, nil
221
+}
222
+
223
+func P2PAddrsFactory(cfg *config.Config) (opts Libp2pOpts, err error) {
224
+ addrsFactory, err := makeAddrsFactory(cfg.Addresses)
225
+ if err != nil {
226
+ return opts, err
227
+ }
228
+ if !cfg.Swarm.DisableRelay {
229
+ addrsFactory = composeAddrsFactory(addrsFactory, filterRelayAddrs)
230
+ }
231
+ opts.Opts = append(opts.Opts, libp2p.AddrsFactory(addrsFactory))
232
+ return
233
+}
234
+
235
+func filterRelayAddrs(addrs []ma.Multiaddr) []ma.Multiaddr {
236
+ var raddrs []ma.Multiaddr
237
+ for _, addr := range addrs {
238
+ _, err := addr.ValueForProtocol(circuit.P_CIRCUIT)
239
+ if err == nil {
240
+ continue
241
+ }
242
+ raddrs = append(raddrs, addr)
243
+ }
244
+ return raddrs
245
+}
246
+
247
+func composeAddrsFactory(f, g p2pbhost.AddrsFactory) p2pbhost.AddrsFactory {
248
+ return func(addrs []ma.Multiaddr) []ma.Multiaddr {
249
+ return f(g(addrs))
250
+ }
251
+}
252
+
253
+func P2PConnectionManager(cfg *config.Config) (opts Libp2pOpts, err error) {
254
+ grace := config.DefaultConnMgrGracePeriod
255
+ low := config.DefaultConnMgrHighWater
256
+ high := config.DefaultConnMgrHighWater
257
+
258
+ switch cfg.Swarm.ConnMgr.Type {
259
+ case "":
260
+ // 'default' value is the basic connection manager
261
+ return
262
+ case "none":
263
+ return opts, nil
264
+ case "basic":
265
+ grace, err = time.ParseDuration(cfg.Swarm.ConnMgr.GracePeriod)
266
+ if err != nil {
267
+ return opts, fmt.Errorf("parsing Swarm.ConnMgr.GracePeriod: %s", err)
268
+ }
269
+
270
+ low = cfg.Swarm.ConnMgr.LowWater
271
+ high = cfg.Swarm.ConnMgr.HighWater
272
+ default:
273
+ return opts, fmt.Errorf("unrecognized ConnMgr.Type: %q", cfg.Swarm.ConnMgr.Type)
274
+ }
275
+
276
+ cm := connmgr.NewConnManager(low, high, grace)
277
+ opts.Opts = append(opts.Opts, libp2p.ConnectionManager(cm))
278
+ return
279
+}
280
+
281
+func makeSmuxTransportOption(mplexExp bool) libp2p.Option {
282
+ const yamuxID = "/yamux/1.0.0"
283
+ const mplexID = "/mplex/6.7.0"
284
+
285
+ ymxtpt := &yamux.Transport{
286
+ AcceptBacklog: 512,
287
+ ConnectionWriteTimeout: time.Second * 10,
288
+ KeepAliveInterval: time.Second * 30,
289
+ EnableKeepAlive: true,
290
+ MaxStreamWindowSize: uint32(16 * 1024 * 1024), // 16MiB
291
+ LogOutput: ioutil.Discard,
292
+ }
293
+
294
+ if os.Getenv("YAMUX_DEBUG") != "" {
295
+ ymxtpt.LogOutput = os.Stderr
296
+ }
297
+
298
+ muxers := map[string]smux.Transport{yamuxID: ymxtpt}
299
+ if mplexExp {
300
+ muxers[mplexID] = mplex.DefaultTransport
301
+ }
302
+
303
+ // Allow muxer preference order overriding
304
+ order := []string{yamuxID, mplexID}
305
+ if prefs := os.Getenv("LIBP2P_MUX_PREFS"); prefs != "" {
306
+ order = strings.Fields(prefs)
307
+ }
308
+
309
+ opts := make([]libp2p.Option, 0, len(order))
310
+ for _, id := range order {
311
+ tpt, ok := muxers[id]
312
+ if !ok {
313
+ log.Warning("unknown or duplicate muxer in LIBP2P_MUX_PREFS: %s", id)
314
+ continue
315
+ }
316
+ delete(muxers, id)
317
+ opts = append(opts, libp2p.Muxer(id, tpt))
318
+ }
319
+
320
+ return libp2p.ChainOptions(opts...)
321
+}
322
+
323
+func P2PSmuxTransport(bcfg *BuildCfg) (opts Libp2pOpts, err error) {
324
+ opts.Opts = append(opts.Opts, makeSmuxTransportOption(bcfg.getOpt("mplex")))
325
+ return
326
+}
327
+
328
+func P2PNatPortMap(cfg *config.Config) (opts Libp2pOpts, err error) {
329
+ if !cfg.Swarm.DisableNatPortMap {
330
+ opts.Opts = append(opts.Opts, libp2p.NATPortMap())
331
+ }
332
+ return
333
+}
334
+
335
+func P2PRelay(cfg *config.Config) (opts Libp2pOpts, err error) {
336
+ if cfg.Swarm.DisableRelay {
337
+ // Enabled by default.
338
+ opts.Opts = append(opts.Opts, libp2p.DisableRelay())
339
+ } else {
340
+ relayOpts := []relay.RelayOpt{relay.OptDiscovery}
341
+ if cfg.Swarm.EnableRelayHop {
342
+ relayOpts = append(relayOpts, relay.OptHop)
343
+ }
344
+ opts.Opts = append(opts.Opts, libp2p.EnableRelay(relayOpts...))
345
+ }
346
+ return
347
+}
348
+
349
+func P2PAutoRealy(cfg *config.Config) (opts Libp2pOpts, err error) {
350
+ // enable autorelay
351
+ if cfg.Swarm.EnableAutoRelay {
352
+ opts.Opts = append(opts.Opts, libp2p.EnableAutoRelay())
353
+ }
354
+ return
355
+}
356
+
357
+func P2PDefaultTransports() (opts Libp2pOpts, err error) {
358
+ opts.Opts = append(opts.Opts, libp2p.DefaultTransports)
359
+ return
360
+}
361
+
362
+func P2PQUIC(cfg *config.Config) (opts Libp2pOpts, err error) {
363
+ if cfg.Experimental.QUIC {
364
+ opts.Opts = append(opts.Opts, libp2p.Transport(libp2pquic.NewTransport))
365
+ }
366
+ return
367
+}
368
+
369
+type P2PHostIn struct {
370
+ fx.In
371
+
372
+ BCfg *BuildCfg
373
+ Repo repo.Repo
374
+ Validator record.Validator
375
+ HostOption HostOption
376
+ ID peer.ID
377
+ Peerstore peerstore.Peerstore
378
+
379
+ Opts [][]libp2p.Option `group:"libp2p"`
380
+}
381
+
382
+type BaseRouting routing.IpfsRouting
383
+type P2PHostOut struct {
384
+ fx.Out
385
+
386
+ Host host.Host
387
+ Routing BaseRouting
388
+ IpfsDHT *dht.IpfsDHT
389
+}
390
+
391
+// TODO: move some of this into params struct
392
+func P2PHost(mctx MetricsCtx, lc fx.Lifecycle, params P2PHostIn) (out P2PHostOut, err error) {
393
+ opts := []libp2p.Option{libp2p.NoListenAddrs}
394
+ for _, o := range params.Opts {
395
+ opts = append(opts, o...)
396
+ }
397
+
398
+ ctx, cancel := context.WithCancel(mctx)
399
+ lc.Append(fx.Hook{
400
+ OnStop: func(_ context.Context) error {
401
+ cancel()
402
+ return nil
403
+ },
404
+ })
405
+
406
+ opts = append(opts, libp2p.Routing(func(h host.Host) (routing.PeerRouting, error) {
407
+ r, err := params.BCfg.Routing(ctx, h, params.Repo.Datastore(), params.Validator)
408
+ out.Routing = r
409
+ return r, err
410
+ }))
411
+
412
+ out.Host, err = params.HostOption(ctx, params.ID, params.Peerstore, opts...)
413
+ if err != nil {
414
+ return P2PHostOut{}, err
415
+ }
416
+
417
+ // this code is necessary just for tests: mock network constructions
418
+ // ignore the libp2p constructor options that actually construct the routing!
419
+ if out.Routing == nil {
420
+ r, err := params.BCfg.Routing(ctx, out.Host, params.Repo.Datastore(), params.Validator)
421
+ if err != nil {
422
+ return P2PHostOut{}, err
423
+ }
424
+ out.Routing = r
425
+ out.Host = routedhost.Wrap(out.Host, out.Routing)
426
+ }
427
+
428
+ lc.Append(fx.Hook{
429
+ OnStop: func(ctx context.Context) error {
430
+ return out.Host.Close()
431
+ },
432
+ })
433
+
434
+ // TODO: break this up into more DI units
435
+ // TODO: I'm not a fan of type assertions like this but the
436
+ // `RoutingOption` system doesn't currently provide access to the
437
+ // IpfsNode.
438
+ //
439
+ // Ideally, we'd do something like:
440
+ //
441
+ // 1. Add some fancy method to introspect into tiered routers to extract
442
+ // things like the pubsub router or the DHT (complicated, messy,
443
+ // probably not worth it).
444
+ // 2. Pass the IpfsNode into the RoutingOption (would also remove the
445
+ // PSRouter case below.
446
+ // 3. Introduce some kind of service manager? (my personal favorite but
447
+ // that requires a fair amount of work).
448
+ if dht, ok := out.Routing.(*dht.IpfsDHT); ok {
449
+ out.IpfsDHT = dht
450
+
451
+ lc.Append(fx.Hook{
452
+ OnStop: func(ctx context.Context) error {
453
+ return out.IpfsDHT.Close()
454
+ },
455
+ })
456
+ }
457
+
458
+ return out, err
459
+}
460
+
461
+type p2pRoutingIn struct {
462
+ fx.In
463
+
464
+ BCfg *BuildCfg
465
+ Repo repo.Repo
466
+ Validator record.Validator
467
+ Host host.Host
468
+ PubSub *pubsub.PubSub
469
+
470
+ BaseRouting BaseRouting
471
+}
472
+
473
+type p2pRoutingOut struct {
474
+ fx.Out
475
+
476
+ IpfsRouting routing.IpfsRouting
477
+ PSRouter *namesys.PubsubValueStore // TODO: optional
478
+}
479
+
480
+func P2POnlineRouting(mctx MetricsCtx, lc fx.Lifecycle, in p2pRoutingIn) (out p2pRoutingOut) {
481
+ out.IpfsRouting = in.BaseRouting
482
+
483
+ if in.BCfg.getOpt("ipnsps") {
484
+ out.PSRouter = namesys.NewPubsubValueStore(
485
+ lifecycleCtx(mctx, lc),
486
+ in.Host,
487
+ in.BaseRouting,
488
+ in.PubSub,
489
+ in.Validator,
490
+ )
491
+
492
+ out.IpfsRouting = routinghelpers.Tiered{
493
+ Routers: []routing.IpfsRouting{
494
+ // Always check pubsub first.
495
+ &routinghelpers.Compose{
496
+ ValueStore: &routinghelpers.LimitedValueStore{
497
+ ValueStore: out.PSRouter,
498
+ Namespaces: []string{"ipns"},
499
+ },
500
+ },
501
+ in.BaseRouting,
502
+ },
503
+ Validator: in.Validator,
504
+ }
505
+ }
506
+ return out
507
+}
508
+
509
+func AutoNATService(mctx MetricsCtx, lc fx.Lifecycle, cfg *config.Config, host host.Host) error {
510
+ if !cfg.Swarm.EnableAutoNATService {
511
+ return nil
512
+ }
513
+ var opts []libp2p.Option
514
+ if cfg.Experimental.QUIC {
515
+ opts = append(opts, libp2p.DefaultTransports, libp2p.Transport(libp2pquic.NewTransport))
516
+ }
517
+
518
+ _, err := autonat.NewAutoNATService(lifecycleCtx(mctx, lc), host, opts...)
519
+ return err
520
+}
521
+
522
+func Pubsub(mctx MetricsCtx, lc fx.Lifecycle, host host.Host, bcfg *BuildCfg, cfg *config.Config) (service *pubsub.PubSub, err error) {
523
+ if !(bcfg.getOpt("pubsub") || bcfg.getOpt("ipnsps")) {
524
+ return nil, nil // TODO: mark optional
525
+ }
526
+
527
+ var pubsubOptions []pubsub.Option
528
+ if cfg.Pubsub.DisableSigning {
529
+ pubsubOptions = append(pubsubOptions, pubsub.WithMessageSigning(false))
530
+ }
531
+
532
+ if cfg.Pubsub.StrictSignatureVerification {
533
+ pubsubOptions = append(pubsubOptions, pubsub.WithStrictSignatureVerification(true))
534
+ }
535
+
536
+ switch cfg.Pubsub.Router {
537
+ case "":
538
+ fallthrough
539
+ case "floodsub":
540
+ service, err = pubsub.NewFloodSub(lifecycleCtx(mctx, lc), host, pubsubOptions...)
541
+
542
+ case "gossipsub":
543
+ service, err = pubsub.NewGossipSub(lifecycleCtx(mctx, lc), host, pubsubOptions...)
544
+
545
+ default:
546
+ err = fmt.Errorf("Unknown pubsub router %s", cfg.Pubsub.Router)
547
+ }
548
+
549
+ return service, err
550
+}
551
+
552
+func listenAddresses(cfg *config.Config) ([]ma.Multiaddr, error) {
553
+ var listen []ma.Multiaddr
554
+ for _, addr := range cfg.Addresses.Swarm {
555
+ maddr, err := ma.NewMultiaddr(addr)
556
+ if err != nil {
557
+ return nil, fmt.Errorf("failure to parse config.Addresses.Swarm: %s", cfg.Addresses.Swarm)
558
+ }
559
+ listen = append(listen, maddr)
560
+ }
561
+
562
+ return listen, nil
563
+}
564
+
565
+func StartListening(host host.Host, cfg *config.Config) error {
566
+ listenAddrs, err := listenAddresses(cfg)
567
+ if err != nil {
568
+ return err
569
+ }
570
+
571
+ // Actually start listening:
572
+ if err := host.Network().Listen(listenAddrs...); err != nil {
573
+ return err
574
+ }
575
+
576
+ // list out our addresses
577
+ addrs, err := host.Network().InterfaceListenAddresses()
578
+ if err != nil {
579
+ return err
580
+ }
581
+ log.Infof("Swarm listening at: %s", addrs)
582
+ return nil
583
+}
584
+
585
+func P2PHostOption(bcfg *BuildCfg) (hostOption HostOption, err error) {
586
+ hostOption = bcfg.Host
587
+ if bcfg.DisableEncryptedConnections {
588
+ innerHostOption := hostOption
589
+ hostOption = func(ctx context.Context, id peer.ID, ps peerstore.Peerstore, options ...libp2p.Option) (host.Host, error) {
590
+ return innerHostOption(ctx, id, ps, append(options, libp2p.NoSecurity)...)
591
+ }
592
+ // TODO: shouldn't this be Errorf to guarantee visibility?
593
+ log.Warningf(`Your IPFS node has been configured to run WITHOUT ENCRYPTED CONNECTIONS.
594
+ You will not be able to connect to any nodes configured to use encrypted connections`)
595
+ }
596
+ return hostOption, nil
597
+}
core/node/provider.go
new
+59
@@ -0,0 +1,59 @@
1
+package node
2
+
3
+import (
4
+ "fmt"
5
+ "time"
6
+
7
+ "github.com/ipfs/go-ipfs-config"
8
+ "github.com/ipfs/go-ipld-format"
9
+ "github.com/libp2p/go-libp2p-routing"
10
+ "go.uber.org/fx"
11
+
12
+ "github.com/ipfs/go-ipfs/exchange/reprovide"
13
+ "github.com/ipfs/go-ipfs/pin"
14
+ "github.com/ipfs/go-ipfs/provider"
15
+ "github.com/ipfs/go-ipfs/repo"
16
+)
17
+
18
+const kReprovideFrequency = time.Hour * 12
19
+
20
+func ProviderQueue(mctx MetricsCtx, lc fx.Lifecycle, repo repo.Repo) (*provider.Queue, error) {
21
+ return provider.NewQueue(lifecycleCtx(mctx, lc), "provider-v1", repo.Datastore())
22
+}
23
+
24
+func ProviderCtor(mctx MetricsCtx, lc fx.Lifecycle, queue *provider.Queue, rt routing.IpfsRouting) provider.Provider {
25
+ return provider.NewProvider(lifecycleCtx(mctx, lc), queue, rt)
26
+}
27
+
28
+func ReproviderCtor(mctx MetricsCtx, lc fx.Lifecycle, cfg *config.Config, bs BaseBlocks, ds format.DAGService, pinning pin.Pinner, rt routing.IpfsRouting) (*reprovide.Reprovider, error) {
29
+ var keyProvider reprovide.KeyChanFunc
30
+
31
+ switch cfg.Reprovider.Strategy {
32
+ case "all":
33
+ fallthrough
34
+ case "":
35
+ keyProvider = reprovide.NewBlockstoreProvider(bs)
36
+ case "roots":
37
+ keyProvider = reprovide.NewPinnedProvider(pinning, ds, true)
38
+ case "pinned":
39
+ keyProvider = reprovide.NewPinnedProvider(pinning, ds, false)
40
+ default:
41
+ return nil, fmt.Errorf("unknown reprovider strategy '%s'", cfg.Reprovider.Strategy)
42
+ }
43
+ return reprovide.NewReprovider(lifecycleCtx(mctx, lc), rt, keyProvider), nil
44
+}
45
+
46
+func Reprovider(cfg *config.Config, reprovider *reprovide.Reprovider) error {
47
+ reproviderInterval := kReprovideFrequency
48
+ if cfg.Reprovider.Interval != "" {
49
+ dur, err := time.ParseDuration(cfg.Reprovider.Interval)
50
+ if err != nil {
51
+ return err
52
+ }
53
+
54
+ reproviderInterval = dur
55
+ }
56
+
57
+ go reprovider.Run(reproviderInterval)
58
+ return nil
59
+}
core/node/storage.go
new
+94
@@ -0,0 +1,94 @@
1
+package node
2
+
3
+import (
4
+ "context"
5
+ "os"
6
+ "syscall"
7
+ "time"
8
+
9
+ "github.com/ipfs/go-datastore"
10
+ "github.com/ipfs/go-datastore/retrystore"
11
+ blockstore "github.com/ipfs/go-ipfs-blockstore"
12
+ config "github.com/ipfs/go-ipfs-config"
13
+ "go.uber.org/fx"
14
+
15
+ "github.com/ipfs/go-ipfs/filestore"
16
+ "github.com/ipfs/go-ipfs/repo"
17
+ "github.com/ipfs/go-ipfs/thirdparty/cidv0v1"
18
+ "github.com/ipfs/go-ipfs/thirdparty/verifbs"
19
+)
20
+
21
+func isTooManyFDError(err error) bool {
22
+ perr, ok := err.(*os.PathError)
23
+ if ok && perr.Err == syscall.EMFILE {
24
+ return true
25
+ }
26
+
27
+ return false
28
+}
29
+
30
+func RepoConfig(repo repo.Repo) (*config.Config, error) {
31
+ return repo.Config()
32
+}
33
+
34
+func DatastoreCtor(repo repo.Repo) datastore.Datastore {
35
+ return repo.Datastore()
36
+}
37
+
38
+type BaseBlocks blockstore.Blockstore
39
+
40
+func BaseBlockstoreCtor(mctx MetricsCtx, repo repo.Repo, cfg *config.Config, bcfg *BuildCfg, lc fx.Lifecycle) (bs BaseBlocks, err error) {
41
+ rds := &retrystore.Datastore{
42
+ Batching: repo.Datastore(),
43
+ Delay: time.Millisecond * 200,
44
+ Retries: 6,
45
+ TempErrFunc: isTooManyFDError,
46
+ }
47
+ // hash security
48
+ bs = blockstore.NewBlockstore(rds)
49
+ bs = &verifbs.VerifBS{Blockstore: bs}
50
+
51
+ opts := blockstore.DefaultCacheOpts()
52
+ opts.HasBloomFilterSize = cfg.Datastore.BloomFilterSize
53
+ if !bcfg.Permanent {
54
+ opts.HasBloomFilterSize = 0
55
+ }
56
+
57
+ if !bcfg.NilRepo {
58
+ ctx, cancel := context.WithCancel(mctx)
59
+
60
+ lc.Append(fx.Hook{
61
+ OnStop: func(context context.Context) error {
62
+ cancel()
63
+ return nil
64
+ },
65
+ })
66
+ bs, err = blockstore.CachedBlockstore(ctx, bs, opts)
67
+ if err != nil {
68
+ return nil, err
69
+ }
70
+ }
71
+
72
+ bs = blockstore.NewIdStore(bs)
73
+ bs = cidv0v1.NewBlockstore(bs)
74
+
75
+ if cfg.Datastore.HashOnRead { // TODO: review: this is how it was done originally, is there a reason we can't just pass this directly?
76
+ bs.HashOnRead(true)
77
+ }
78
+
79
+ return
80
+}
81
+
82
+func GcBlockstoreCtor(repo repo.Repo, bb BaseBlocks, cfg *config.Config) (gclocker blockstore.GCLocker, gcbs blockstore.GCBlockstore, bs blockstore.Blockstore, fstore *filestore.Filestore) {
83
+ gclocker = blockstore.NewGCLocker()
84
+ gcbs = blockstore.NewGCBlockstore(bb, gclocker)
85
+
86
+ if cfg.Experimental.FilestoreEnabled || cfg.Experimental.UrlstoreEnabled {
87
+ // hash security
88
+ fstore = filestore.NewFilestore(bb, repo.FileManager()) // TODO: mark optional
89
+ gcbs = blockstore.NewGCBlockstore(fstore, gclocker)
90
+ gcbs = &verifbs.VerifBSGC{GCBlockstore: gcbs}
91
+ }
92
+ bs = gcbs
93
+ return
94
+}
namesys/republisher/repub_test.go
+2
-1
@@ -7,6 +7,7 @@ import (
7
"time"
8
9
"github.com/ipfs/go-ipfs/core"
10
+ "github.com/ipfs/go-ipfs/core/bootstrap"
11
mock "github.com/ipfs/go-ipfs/core/mock"
12
namesys "github.com/ipfs/go-ipfs/namesys"
13
. "github.com/ipfs/go-ipfs/namesys/republisher"
@@ -45,7 +46,7 @@ func TestRepublish(t *testing.T) {
46
t.Fatal(err)
47
}
48
48
- bsinf := core.BootstrapConfigWithPeers(
49
+ bsinf := bootstrap.BootstrapConfigWithPeers(
50
[]pstore.PeerInfo{
51
nodes[0].Peerstore.PeerInfo(nodes[0].Identity),
52
},
test/integration/addcat_test.go
+3
-2
@@ -12,6 +12,7 @@ import (
12
"time"
13
14
"github.com/ipfs/go-ipfs/core"
15
+ "github.com/ipfs/go-ipfs/core/bootstrap"
16
"github.com/ipfs/go-ipfs/core/coreapi"
17
mock "github.com/ipfs/go-ipfs/core/mock"
18
"github.com/ipfs/go-ipfs/thirdparty/unit"
@@ -140,10 +141,10 @@ func DirectAddCat(data []byte, conf testutil.LatencyConfig) error {
141
bs1 := []pstore.PeerInfo{adder.Peerstore.PeerInfo(adder.Identity)}
142
bs2 := []pstore.PeerInfo{catter.Peerstore.PeerInfo(catter.Identity)}
143
143
- if err := catter.Bootstrap(core.BootstrapConfigWithPeers(bs1)); err != nil {
144
+ if err := catter.Bootstrap(bootstrap.BootstrapConfigWithPeers(bs1)); err != nil {
145
return err
146
}
146
- if err := adder.Bootstrap(core.BootstrapConfigWithPeers(bs2)); err != nil {
147
+ if err := adder.Bootstrap(bootstrap.BootstrapConfigWithPeers(bs2)); err != nil {
148
return err
149
}
150
test/integration/bench_cat_test.go
+3
-2
@@ -9,6 +9,7 @@ import (
9
"testing"
10
11
"github.com/ipfs/go-ipfs/core"
12
+ "github.com/ipfs/go-ipfs/core/bootstrap"
13
"github.com/ipfs/go-ipfs/core/coreapi"
14
mock "github.com/ipfs/go-ipfs/core/mock"
15
"github.com/ipfs/go-ipfs/thirdparty/unit"
@@ -83,10 +84,10 @@ func benchCat(b *testing.B, data []byte, conf testutil.LatencyConfig) error {
84
bs1 := []pstore.PeerInfo{adder.Peerstore.PeerInfo(adder.Identity)}
85
bs2 := []pstore.PeerInfo{catter.Peerstore.PeerInfo(catter.Identity)}
86
86
- if err := catter.Bootstrap(core.BootstrapConfigWithPeers(bs1)); err != nil {
87
+ if err := catter.Bootstrap(bootstrap.BootstrapConfigWithPeers(bs1)); err != nil {
88
return err
89
}
89
- if err := adder.Bootstrap(core.BootstrapConfigWithPeers(bs2)); err != nil {
90
+ if err := adder.Bootstrap(bootstrap.BootstrapConfigWithPeers(bs2)); err != nil {
91
return err
92
}
93
test/integration/bitswap_wo_routing_test.go
+2
-1
@@ -8,6 +8,7 @@ import (
8
"github.com/ipfs/go-block-format"
9
"github.com/ipfs/go-ipfs/core"
10
"github.com/ipfs/go-ipfs/core/mock"
11
+ "github.com/ipfs/go-ipfs/core/node"
12
13
cid "github.com/ipfs/go-cid"
14
mocknet "github.com/libp2p/go-libp2p/p2p/net/mock"
@@ -26,7 +27,7 @@ func TestBitswapWithoutRouting(t *testing.T) {
27
n, err := core.NewNode(ctx, &core.BuildCfg{
28
Online: true,
29
Host: coremock.MockHostOption(mn),
29
- Routing: core.NilRouterOption, // no routing
30
+ Routing: node.NilRouterOption, // no routing
31
})
32
if err != nil {
33
t.Fatal(err)
test/integration/three_legged_cat_test.go
+2
-1
@@ -10,6 +10,7 @@ import (
10
"time"
11
12
core "github.com/ipfs/go-ipfs/core"
13
+ bootstrap2 "github.com/ipfs/go-ipfs/core/bootstrap"
14
"github.com/ipfs/go-ipfs/core/coreapi"
15
mock "github.com/ipfs/go-ipfs/core/mock"
16
"github.com/ipfs/go-ipfs/thirdparty/unit"
@@ -118,7 +119,7 @@ func RunThreeLeggedCat(data []byte, conf testutil.LatencyConfig) error {
119
}
120
121
bis := bootstrap.Peerstore.PeerInfo(bootstrap.PeerHost.ID())
121
- bcfg := core.BootstrapConfigWithPeers([]pstore.PeerInfo{bis})
122
+ bcfg := bootstrap2.BootstrapConfigWithPeers([]pstore.PeerInfo{bis})
123
if err := adder.Bootstrap(bcfg); err != nil {
124
return err
125
}