@cryptotaxi247 / kubo / commits / 0fd2f80be

Initial DI node implementation

License: MIT Signed-off-by: Łukasz Magiera <magik6k@gmail.com>

Łukasz Magiera committed Mar 28, 2019 at 01:49 UTC 0fd2f80be71c526999b3d05d4faca20e8541b363
3 files changed +855 -22
core/builder.go
+73 -12
@@ -5,7 +5,9 @@ import (
5 "crypto/rand"
6 "encoding/base64"
7 "errors"
8 + "github.com/ipfs/go-ipfs/p2p"
9 "github.com/ipfs/go-ipfs/provider"
10 + "go.uber.org/fx"
11 "os"
12 "syscall"
13 "time"
@@ -25,19 +27,15 @@ import (
27 cfg "github.com/ipfs/go-ipfs-config"
28 offline "github.com/ipfs/go-ipfs-exchange-offline"
29 offroute "github.com/ipfs/go-ipfs-routing/offline"
28 - ipns "github.com/ipfs/go-ipns"
30 dag "github.com/ipfs/go-merkledag"
31 metrics "github.com/ipfs/go-metrics-interface"
32 resolver "github.com/ipfs/go-path/resolver"
33 uio "github.com/ipfs/go-unixfs/io"
33 - goprocessctx "github.com/jbenet/goprocess/context"
34 libp2p "github.com/libp2p/go-libp2p"
35 ci "github.com/libp2p/go-libp2p-crypto"
36 p2phost "github.com/libp2p/go-libp2p-host"
37 peer "github.com/libp2p/go-libp2p-peer"
38 pstore "github.com/libp2p/go-libp2p-peerstore"
39 - pstoremem "github.com/libp2p/go-libp2p-peerstore/pstoremem"
40 - record "github.com/libp2p/go-libp2p-record"
39 )
40
41 type BuildCfg struct {
@@ -55,7 +53,7 @@ type BuildCfg struct {
53 // DO NOT SET THIS UNLESS YOU'RE TESTING.
54 DisableEncryptedConnections bool
55
58 - // If NilRepo is set, a repo backed by a nil datastore will be constructed
56 + // If NilRepo is set, a Repo backed by a nil datastore will be constructed
57 NilRepo bool
58
59 Routing RoutingOption
@@ -73,7 +71,7 @@ func (cfg *BuildCfg) getOpt(key string) bool {
71
72 func (cfg *BuildCfg) fillDefaults() error {
73 if cfg.Repo != nil && cfg.NilRepo {
76 - return errors.New("cannot set a repo and specify nilrepo at the same time")
74 + return errors.New("cannot set a Repo and specify nilrepo at the same time")
75 }
76
77 if cfg.Repo == nil {
@@ -142,7 +140,66 @@ func NewNode(ctx context.Context, cfg *BuildCfg) (*IpfsNode, error) {
140
141 ctx = metrics.CtxScope(ctx, "ipfs")
142
143 + repoOption := fx.Provide(func(lc fx.Lifecycle) repo.Repo {
144 + lc.Append(fx.Hook{
145 + OnStop: func(ctx context.Context) error {
146 + return cfg.Repo.Close()
147 + },
148 + })
149 +
150 + return cfg.Repo
151 + })
152 +
153 + // TODO: Remove this, use only for passing node config
154 + cfgOption := fx.Provide(func() *BuildCfg {
155 + return cfg
156 + })
157 +
158 n := &IpfsNode{
159 + ctx: ctx,
160 + }
161 +
162 + app := fx.New(
163 + repoOption,
164 + cfgOption,
165 +
166 + fx.Provide(repoConfig),
167 + fx.Provide(identity),
168 + fx.Provide(privateKey),
169 +
170 + fx.Provide(peerstore),
171 + fx.Provide(baseBlockstoreCtor),
172 + fx.Provide(gcBlockstoreCtor),
173 +
174 + fx.Provide(recordValidator),
175 +
176 + ipfsp2p,
177 +
178 + fx.Invoke(setupSharding),
179 +
180 + fx.Provide(onlineExchangeCtor), // TODO: offline
181 + fx.Provide(onlineNamesysCtor), // TODO: ^^
182 + fx.Provide(bserv.New),
183 + fx.Provide(onlineDagCtor),
184 + fx.Provide(resolver.NewBasicResolver),
185 +
186 + fx.Provide(pinning),
187 + fx.Provide(files),
188 +
189 + fx.Provide(providerQueue),
190 + fx.Provide(providerCtor),
191 + fx.Provide(reproviderCtor),
192 + fx.Invoke(reprovider),
193 +
194 + fx.Provide(p2p.NewP2P),
195 +
196 + fx.Invoke(ipnsRepublisher),
197 + fx.Invoke(provider.Provider.Run),
198 +
199 + fx.Extract(n),
200 + )
201 +
202 +/* n := &IpfsNode{
203 IsOnline: cfg.Online,
204 Repo: cfg.Repo,
205 ctx: ctx,
@@ -153,16 +210,19 @@ func NewNode(ctx context.Context, cfg *BuildCfg) (*IpfsNode, error) {
210 "pk": record.PublicKeyValidator{},
211 "ipns": ipns.Validator{KeyBook: n.Peerstore},
212 }
213 +*/
214 + // TODO: port to lifetimes
215 + // n.proc = goprocessctx.WithContextAndTeardown(ctx, n.teardown)
216
157 - // TODO: this is a weird circular-ish dependency, rework it
158 - n.proc = goprocessctx.WithContextAndTeardown(ctx, n.teardown)
159 -
160 - if err := setupNode(ctx, n, cfg); err != nil {
217 + /*if err := setupNode(ctx, n, cfg); err != nil {
218 n.Close()
219 return nil, err
163 - }
220 + }*/
221 + if app.Err() != nil {
222 + return nil, app.Err()
223 + }
224
165 - return n, nil
225 + return n, app.Start(ctx)
226 }
227
228 func isTooManyFDError(err error) bool {
@@ -247,6 +307,7 @@ func setupNode(ctx context.Context, n *IpfsNode, cfg *BuildCfg) error {
307 hostOption = func(ctx context.Context, id peer.ID, ps pstore.Peerstore, options ...libp2p.Option) (p2phost.Host, error) {
308 return innerHostOption(ctx, id, ps, append(options, libp2p.NoSecurity)...)
309 }
310 + // TODO: shouldn't this be Errorf to guarantee visibility?
311 log.Warningf(`Your IPFS node has been configured to run WITHOUT ENCRYPTED CONNECTIONS.
312 You will not be able to connect to any nodes configured to use encrypted connections`)
313 }
core/core.go
+8 -10
@@ -79,8 +79,6 @@ import (
79 mamask "github.com/whyrusleeping/multiaddr-filter"
80 )
81
82 -const IpnsValidatorTag = "ipns"
83 -
82 const kReprovideFrequency = time.Hour * 12
83 const discoveryConnTimeout = time.Second * 30
84 const DefaultIpnsCacheSize = 128
@@ -101,9 +99,9 @@ type IpfsNode struct {
99
100 // Local node
101 Pinning pin.Pinner // the pinning manager
104 - Mounts Mounts // current mount state, if any.
102 + Mounts Mounts `optional:"true"` // current mount state, if any.
103 PrivateKey ic.PrivKey // the local node's private Key
106 - PNetFingerprint []byte // fingerprint of private network
104 + PNetFingerprint PNetFingerprint // fingerprint of private network
105
106 // Services
107 Peerstore pstore.Peerstore // storage for other Peer instances
@@ -115,21 +113,21 @@ type IpfsNode struct {
113 DAG ipld.DAGService // the merkle dag service, get/add objects.
114 Resolver *resolver.Resolver // the path resolution system
115 Reporter metrics.Reporter
118 - Discovery discovery.Service
116 + Discovery discovery.Service `optional:"true"`
117 FilesRoot *mfs.Root
118 RecordValidator record.Validator
119
120 // Online
121 PeerHost p2phost.Host // the network host (server+client)
124 - Bootstrapper io.Closer // the periodic bootstrapper
122 + Bootstrapper io.Closer `optional:"true"` // the periodic bootstrapper
123 Routing routing.IpfsRouting // the routing system. recommend ipfs-dht
124 Exchange exchange.Interface // the block exchange + strategy (bitswap)
125 Namesys namesys.NameSystem // the name system, resolves paths to hashes
126 Provider provider.Provider // the value provider system
127 Reprovider *rp.Reprovider // the value reprovider system
130 - IpnsRepub *ipnsrp.Republisher
128 + IpnsRepub *ipnsrp.Republisher `optional:"true"`
129
132 - AutoNAT *autonat.AutoNATService
130 + AutoNAT *autonat.AutoNATService `optional:"true"`
131 PubSub *pubsub.PubSub
132 PSRouter *psrouter.PubsubValueStore
133 DHT *dht.IpfsDHT
@@ -139,8 +137,8 @@ type IpfsNode struct {
137 ctx context.Context
138
139 // Flags
142 - IsOnline bool // Online is set when networking is enabled.
143 - IsDaemon bool // Daemon is set when running on a long-running daemon.
140 + IsOnline bool `optional:"true"` // Online is set when networking is enabled.
141 + IsDaemon bool `optional:"true"` // Daemon is set when running on a long-running daemon.
142 }
143
144 // Mounts defines what the node's mount state is. This should
core/ncore.go new
+774
@@ -0,0 +1,774 @@
1 +package core
2 +
3 +import (
4 + "bytes"
5 + "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"
30 + "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 + "github.com/libp2p/go-libp2p-kad-dht"
35 + "github.com/libp2p/go-libp2p-metrics"
36 + pstore "github.com/libp2p/go-libp2p-peerstore"
37 + "github.com/libp2p/go-libp2p-peerstore/pstoremem"
38 + "github.com/libp2p/go-libp2p-pnet"
39 + "github.com/libp2p/go-libp2p-pubsub"
40 + psrouter "github.com/libp2p/go-libp2p-pubsub-router"
41 + quic "github.com/libp2p/go-libp2p-quic-transport"
42 + "github.com/libp2p/go-libp2p-record"
43 + "github.com/libp2p/go-libp2p-routing"
44 + rhelpers "github.com/libp2p/go-libp2p-routing-helpers"
45 + "github.com/libp2p/go-libp2p/p2p/discovery"
46 + rhost "github.com/libp2p/go-libp2p/p2p/host/routed"
47 + "go.uber.org/fx"
48 + "time"
49 +
50 + "github.com/ipfs/go-ipfs/repo"
51 +
52 + retry "github.com/ipfs/go-datastore/retrystore"
53 + iconfig "github.com/ipfs/go-ipfs-config"
54 + uio "github.com/ipfs/go-unixfs/io"
55 + ic "github.com/libp2p/go-libp2p-crypto"
56 + p2phost "github.com/libp2p/go-libp2p-host"
57 + "github.com/libp2p/go-libp2p-peer"
58 + mamask "github.com/whyrusleeping/multiaddr-filter"
59 +)
60 +
61 +func repoConfig(repo repo.Repo) (*iconfig.Config, error) {
62 + return repo.Config()
63 +}
64 +
65 +func identity(cfg *iconfig.Config) (peer.ID, error) {
66 + cid := cfg.Identity.PeerID
67 + if cid == "" {
68 + return "", errors.New("identity was not set in config (was 'ipfs init' run?)")
69 + }
70 + if len(cid) == 0 {
71 + return "", errors.New("no peer ID in config! (was 'ipfs init' run?)")
72 + }
73 +
74 + id, err := peer.IDB58Decode(cid)
75 + if err != nil {
76 + return "", fmt.Errorf("peer ID invalid: %s", err)
77 + }
78 +
79 + return id, nil
80 +}
81 +
82 +func peerstore(id peer.ID, sk ic.PrivKey) pstore.Peerstore {
83 + ps := pstoremem.NewPeerstore()
84 +
85 + if sk != nil {
86 + ps.AddPrivKey(id, sk)
87 + ps.AddPubKey(id, sk.GetPublic())
88 + }
89 +
90 + return ps
91 +}
92 +
93 +func privateKey(cfg *iconfig.Config, id peer.ID) (ic.PrivKey, error) {
94 + if cfg.Identity.PrivKey == "" {
95 + return nil, nil
96 + }
97 +
98 + sk, err := cfg.Identity.DecodePrivateKey("passphrase todo!")
99 + if err != nil {
100 + return nil, err
101 + }
102 +
103 + id2, err := peer.IDFromPrivateKey(sk)
104 + if err != nil {
105 + return nil, err
106 + }
107 +
108 + if id2 != id {
109 + return nil, fmt.Errorf("private key in config does not match id: %s != %s", id, id2)
110 + }
111 + return sk, nil
112 +}
113 +
114 +func baseBlockstoreCtor(repo repo.Repo, cfg *iconfig.Config, bcfg *BuildCfg, lc fx.Lifecycle) (bs bstore.Blockstore, err error) {
115 + rds := &retry.Datastore{
116 + Batching: repo.Datastore(),
117 + Delay: time.Millisecond * 200,
118 + Retries: 6,
119 + TempErrFunc: isTooManyFDError,
120 + }
121 + // hash security
122 + bs = bstore.NewBlockstore(rds)
123 + bs = &verifbs.VerifBS{Blockstore: bs}
124 +
125 + opts := bstore.DefaultCacheOpts()
126 + opts.HasBloomFilterSize = cfg.Datastore.BloomFilterSize
127 + if !bcfg.Permanent {
128 + opts.HasBloomFilterSize = 0
129 + }
130 +
131 + if !bcfg.NilRepo {
132 + ctx, cancel := context.WithCancel(context.TODO()) //TODO: needed for mertics
133 +
134 + lc.Append(fx.Hook{
135 + OnStop: func(context context.Context) error {
136 + cancel()
137 + return nil
138 + },
139 + })
140 + bs, err = bstore.CachedBlockstore(ctx, bs, opts)
141 + if err != nil {
142 + return nil, err
143 + }
144 + }
145 +
146 + bs = bstore.NewIdStore(bs)
147 + bs = cidv0v1.NewBlockstore(bs)
148 +
149 + if cfg.Datastore.HashOnRead { // TODO: review: this is how it was done originally, is there a reason we can't just pass this directly?
150 + bs.HashOnRead(true)
151 + }
152 +
153 + return
154 +}
155 +
156 +func gcBlockstoreCtor(repo repo.Repo, bs bstore.Blockstore, cfg *iconfig.Config) (gclocker bstore.GCLocker, gcbs bstore.GCBlockstore, fstore *filestore.Filestore) {
157 + gclocker = bstore.NewGCLocker()
158 + gcbs = bstore.NewGCBlockstore(bs, gclocker)
159 +
160 + if cfg.Experimental.FilestoreEnabled || cfg.Experimental.UrlstoreEnabled {
161 + // hash security
162 + fstore = filestore.NewFilestore(bs, repo.FileManager()) //TODO: mark optional
163 + gcbs = bstore.NewGCBlockstore(fstore, gclocker)
164 + gcbs = &verifbs.VerifBSGC{GCBlockstore: gcbs}
165 + }
166 + return
167 +}
168 +
169 +func recordValidator(ps pstore.Peerstore) record.Validator {
170 + return record.NamespacedValidator{
171 + "pk": record.PublicKeyValidator{},
172 + "ipns": ipns.Validator{KeyBook: ps},
173 + }
174 +}
175 +
176 +////////////////////
177 +// libp2p related
178 +
179 +////////////////////
180 +// libp2p
181 +
182 +var ipfsp2p = fx.Options(
183 + fx.Provide(p2pAddrFilters),
184 + fx.Provide(p2pBandwidthCounter),
185 + fx.Provide(p2pPNet),
186 + fx.Provide(p2pAddrsFactory),
187 + fx.Provide(p2pConnectionManager),
188 + fx.Provide(p2pSmuxTransport),
189 + fx.Provide(p2pNatPortMap),
190 + fx.Provide(p2pRelay),
191 + fx.Provide(p2pAutoRealy),
192 + fx.Provide(p2pDefaultTransports),
193 + fx.Provide(p2pQUIC),
194 +
195 + fx.Provide(p2pHostOption),
196 + fx.Provide(p2pHost),
197 + fx.Provide(p2pOnlineRouting),
198 +
199 + fx.Provide(pubsubCtor),
200 + fx.Provide(newDiscoveryHandler),
201 +
202 + fx.Invoke(autoNATService),
203 + fx.Invoke(p2pPNetChecker),
204 + fx.Invoke(startListening),
205 + fx.Invoke(setupDiscovery),
206 +)
207 +
208 +func p2pHostOption(bcfg *BuildCfg) (hostOption HostOption, err error) {
209 + hostOption = bcfg.Host
210 + if bcfg.DisableEncryptedConnections {
211 + innerHostOption := hostOption
212 + hostOption = func(ctx context.Context, id peer.ID, ps pstore.Peerstore, options ...libp2p.Option) (p2phost.Host, error) {
213 + return innerHostOption(ctx, id, ps, append(options, libp2p.NoSecurity)...)
214 + }
215 + // TODO: shouldn't this be Errorf to guarantee visibility?
216 + log.Warningf(`Your IPFS node has been configured to run WITHOUT ENCRYPTED CONNECTIONS.
217 + You will not be able to connect to any nodes configured to use encrypted connections`)
218 + }
219 + return hostOption, nil
220 +}
221 +
222 +func p2pAddrFilters(cfg *iconfig.Config) (opts libp2pOpts, err error) {
223 + for _, s := range cfg.Swarm.AddrFilters {
224 + f, err := mamask.NewMask(s)
225 + if err != nil {
226 + return opts, fmt.Errorf("incorrectly formatted address filter in config: %s", s)
227 + }
228 + opts.Opts = append(opts.Opts, libp2p.FilterAddresses(f))
229 + }
230 + return opts, nil
231 +}
232 +
233 +func p2pBandwidthCounter(cfg *iconfig.Config) (opts libp2pOpts, reporter metrics.Reporter) {
234 + reporter = metrics.NewBandwidthCounter()
235 +
236 + if !cfg.Swarm.DisableBandwidthMetrics {
237 + opts.Opts = append(opts.Opts, libp2p.BandwidthReporter(reporter))
238 + }
239 + return opts, reporter
240 +}
241 +
242 +type libp2pOpts struct {
243 + fx.Out
244 +
245 + Opts []libp2p.Option `group:"libp2p"`
246 +}
247 +
248 +type PNetFingerprint []byte // TODO: find some better place
249 +func p2pPNet(repo repo.Repo) (opts libp2pOpts, fp PNetFingerprint, err error) {
250 + swarmkey, err := repo.SwarmKey()
251 + if err != nil || swarmkey == nil {
252 + return opts, nil, err
253 + }
254 +
255 + protec, err := pnet.NewProtector(bytes.NewReader(swarmkey))
256 + if err != nil {
257 + return opts, nil, fmt.Errorf("failed to configure private network: %s", err)
258 + }
259 + fp = protec.Fingerprint()
260 +
261 + opts.Opts = append(opts.Opts, libp2p.PrivateNetwork(protec))
262 + return opts, fp, nil
263 +}
264 +
265 +func p2pPNetChecker(repo repo.Repo, ph p2phost.Host, lc fx.Lifecycle) error {
266 + // TODO: better check?
267 + swarmkey, err := repo.SwarmKey()
268 + if err != nil || swarmkey == nil {
269 + return err
270 + }
271 +
272 + done := make(chan struct{})
273 + lc.Append(fx.Hook{
274 + OnStart: func(_ context.Context) error {
275 + go func() {
276 + t := time.NewTicker(30 * time.Second)
277 + <-t.C // swallow one tick
278 + for {
279 + select {
280 + case <-t.C:
281 + if len(ph.Network().Peers()) == 0 {
282 + log.Warning("We are in private network and have no peers.")
283 + log.Warning("This might be configuration mistake.")
284 + }
285 + case <-done:
286 + return
287 + }
288 + }
289 + }()
290 + return nil
291 + },
292 + OnStop: func(_ context.Context) error {
293 + close(done)
294 + return nil
295 + },
296 + })
297 + return nil
298 +}
299 +
300 +func p2pAddrsFactory(cfg *iconfig.Config) (opts libp2pOpts, err error) {
301 + addrsFactory, err := makeAddrsFactory(cfg.Addresses)
302 + if err != nil {
303 + return opts, err
304 + }
305 + if !cfg.Swarm.DisableRelay {
306 + addrsFactory = composeAddrsFactory(addrsFactory, filterRelayAddrs)
307 + }
308 + opts.Opts = append(opts.Opts, libp2p.AddrsFactory(addrsFactory))
309 + return
310 +}
311 +
312 +func p2pConnectionManager(cfg *iconfig.Config) (opts libp2pOpts, err error) {
313 + connm, err := constructConnMgr(cfg.Swarm.ConnMgr)
314 + if err != nil {
315 + return opts, err
316 + }
317 +
318 + opts.Opts = append(opts.Opts, libp2p.ConnectionManager(connm))
319 + return
320 +}
321 +
322 +func p2pSmuxTransport(bcfg *BuildCfg) (opts libp2pOpts, err error) {
323 + opts.Opts = append(opts.Opts, makeSmuxTransportOption(bcfg.getOpt("mplex")))
324 + return
325 +}
326 +
327 +func p2pNatPortMap(cfg *iconfig.Config) (opts libp2pOpts, err error) {
328 + if !cfg.Swarm.DisableNatPortMap {
329 + opts.Opts = append(opts.Opts, libp2p.NATPortMap())
330 + }
331 + return
332 +}
333 +
334 +func p2pRelay(cfg *iconfig.Config) (opts libp2pOpts, err error) {
335 + if cfg.Swarm.DisableRelay {
336 + // Enabled by default.
337 + opts.Opts = append(opts.Opts, libp2p.DisableRelay())
338 + } else {
339 + relayOpts := []circuit.RelayOpt{circuit.OptDiscovery}
340 + if cfg.Swarm.EnableRelayHop {
341 + relayOpts = append(relayOpts, circuit.OptHop)
342 + }
343 + opts.Opts = append(opts.Opts, libp2p.EnableRelay(relayOpts...))
344 + }
345 + return
346 +}
347 +
348 +func p2pAutoRealy(cfg *iconfig.Config) (opts libp2pOpts, err error) {
349 + // enable autorelay
350 + if cfg.Swarm.EnableAutoRelay {
351 + opts.Opts = append(opts.Opts, libp2p.EnableAutoRelay())
352 + }
353 + return
354 +}
355 +
356 +func p2pDefaultTransports() (opts libp2pOpts, err error) {
357 + opts.Opts = append(opts.Opts, libp2p.DefaultTransports)
358 + return
359 +}
360 +
361 +func p2pQUIC(cfg *iconfig.Config) (opts libp2pOpts, err error) {
362 + if cfg.Experimental.QUIC {
363 + opts.Opts = append(opts.Opts, libp2p.Transport(quic.NewTransport))
364 + }
365 + return
366 +}
367 +
368 +type p2pHostIn struct {
369 + fx.In
370 +
371 + BCfg *BuildCfg
372 + Repo repo.Repo
373 + Validator record.Validator
374 + HostOption HostOption
375 + ID peer.ID
376 + Peerstore pstore.Peerstore
377 +
378 + Opts [][]libp2p.Option `group:"libp2p"`
379 +}
380 +
381 +type BaseRouting routing.IpfsRouting
382 +type p2pHostOut struct {
383 + fx.Out
384 +
385 + Host p2phost.Host
386 + Routing BaseRouting
387 + IpfsDHT *dht.IpfsDHT
388 +}
389 +
390 +// TODO: move some of this into params struct
391 +func p2pHost(lc fx.Lifecycle, params p2pHostIn) (out p2pHostOut, err error) {
392 + opts := []libp2p.Option{libp2p.NoListenAddrs}
393 + for _, o := range params.Opts {
394 + opts = append(opts, o...)
395 + }
396 +
397 + ctx, cancel := context.WithCancel(context.TODO())
398 + lc.Append(fx.Hook{
399 + OnStop: func(_ context.Context) error {
400 + cancel()
401 + return nil
402 + },
403 + })
404 +
405 + opts = append(opts, libp2p.Routing(func(h p2phost.Host) (routing.PeerRouting, error) {
406 + r, err := params.BCfg.Routing(ctx, h, params.Repo.Datastore(), params.Validator)
407 + out.Routing = r
408 + return r, err
409 + }))
410 +
411 + out.Host, err = params.HostOption(ctx, params.ID, params.Peerstore, opts...)
412 +
413 + // this code is necessary just for tests: mock network constructions
414 + // ignore the libp2p constructor options that actually construct the routing!
415 + if out.Routing == nil {
416 + r, err := params.BCfg.Routing(ctx, out.Host, params.Repo.Datastore(), params.Validator)
417 + if err != nil {
418 + return p2pHostOut{}, err
419 + }
420 + out.Routing = r
421 + out.Host = rhost.Wrap(out.Host, out.Routing)
422 + }
423 +
424 + // TODO: break this up into more DI units
425 + // TODO: I'm not a fan of type assertions like this but the
426 + // `RoutingOption` system doesn't currently provide access to the
427 + // IpfsNode.
428 + //
429 + // Ideally, we'd do something like:
430 + //
431 + // 1. Add some fancy method to introspect into tiered routers to extract
432 + // things like the pubsub router or the DHT (complicated, messy,
433 + // probably not worth it).
434 + // 2. Pass the IpfsNode into the RoutingOption (would also remove the
435 + // PSRouter case below.
436 + // 3. Introduce some kind of service manager? (my personal favorite but
437 + // that requires a fair amount of work).
438 + if dht, ok := out.Routing.(*dht.IpfsDHT); ok {
439 + out.IpfsDHT = dht
440 + }
441 +
442 + return out, err
443 +}
444 +
445 +type p2pRoutingIn struct {
446 + fx.In
447 +
448 + BCfg *BuildCfg
449 + Repo repo.Repo
450 + Validator record.Validator
451 + Host p2phost.Host
452 + PubSub *pubsub.PubSub
453 +
454 + BaseRouting BaseRouting
455 +}
456 +
457 +type p2pRoutingOut struct {
458 + fx.Out
459 +
460 + IpfsRouting routing.IpfsRouting
461 + PSRouter *psrouter.PubsubValueStore //TODO: optional
462 +}
463 +
464 +func p2pOnlineRouting(lc fx.Lifecycle, in p2pRoutingIn) (out p2pRoutingOut) {
465 + out.IpfsRouting = in.BaseRouting
466 +
467 + if in.BCfg.getOpt("ipnsps") {
468 + out.PSRouter = psrouter.NewPubsubValueStore(
469 + lifecycleCtx(lc),
470 + in.Host,
471 + in.BaseRouting,
472 + in.PubSub,
473 + in.Validator,
474 + )
475 +
476 + out.IpfsRouting = rhelpers.Tiered{
477 + Routers: []routing.IpfsRouting{
478 + // Always check pubsub first.
479 + &rhelpers.Compose{
480 + ValueStore: &rhelpers.LimitedValueStore{
481 + ValueStore: out.PSRouter,
482 + Namespaces: []string{"ipns"},
483 + },
484 + },
485 + in.BaseRouting,
486 + },
487 + Validator: in.Validator,
488 + }
489 + }
490 + return out
491 +}
492 +
493 +////////////
494 +// P2P services
495 +
496 +func autoNATService(lc fx.Lifecycle, cfg *iconfig.Config, host p2phost.Host) error {
497 + if !cfg.Swarm.EnableAutoNATService {
498 + return nil
499 + }
500 + var opts []libp2p.Option
501 + if cfg.Experimental.QUIC {
502 + opts = append(opts, libp2p.DefaultTransports, libp2p.Transport(quic.NewTransport))
503 + }
504 +
505 + _, err := autonat.NewAutoNATService(lifecycleCtx(lc), host, opts...)
506 + return err
507 +}
508 +
509 +func pubsubCtor(lc fx.Lifecycle, host p2phost.Host, bcfg *BuildCfg, cfg *iconfig.Config) (service *pubsub.PubSub, err error) {
510 + if !(bcfg.getOpt("pubsub") || bcfg.getOpt("ipnsps")) {
511 + return nil, nil // TODO: mark optional
512 + }
513 +
514 + var pubsubOptions []pubsub.Option
515 + if cfg.Pubsub.DisableSigning {
516 + pubsubOptions = append(pubsubOptions, pubsub.WithMessageSigning(false))
517 + }
518 +
519 + if cfg.Pubsub.StrictSignatureVerification {
520 + pubsubOptions = append(pubsubOptions, pubsub.WithStrictSignatureVerification(true))
521 + }
522 +
523 + switch cfg.Pubsub.Router {
524 + case "":
525 + fallthrough
526 + case "floodsub":
527 + service, err = pubsub.NewFloodSub(lifecycleCtx(lc), host, pubsubOptions...)
528 +
529 + case "gossipsub":
530 + service, err = pubsub.NewGossipSub(lifecycleCtx(lc), host, pubsubOptions...)
531 +
532 + default:
533 + err = fmt.Errorf("Unknown pubsub router %s", cfg.Pubsub.Router)
534 + }
535 +
536 + return service, err
537 +}
538 +
539 +////////////
540 +// Offline services
541 +
542 +// offline.Exchange
543 +// offroute.NewOfflineRouter
544 +
545 +func offlineNamesysCtor(rt routing.IpfsRouting, repo repo.Repo) (namesys.NameSystem, error) {
546 + return namesys.NewNameSystem(rt, repo.Datastore(), 0), nil
547 +}
548 +
549 +
550 +////////////
551 +// IPFS services
552 +
553 +func pinning(bstore bstore.Blockstore, ds format.DAGService, repo repo.Repo) (pin.Pinner, error) {
554 + internalDag := merkledag.NewDAGService(bserv.New(bstore, offline.Exchange(bstore)))
555 + pinning, err := pin.LoadPinner(repo.Datastore(), ds, internalDag)
556 + if err != nil {
557 + // TODO: we should move towards only running 'NewPinner' explicitly on
558 + // node init instead of implicitly here as a result of the pinner keys
559 + // not being found in the datastore.
560 + // this is kinda sketchy and could cause data loss
561 + pinning = pin.NewPinner(repo.Datastore(), ds, internalDag)
562 + }
563 +
564 + return pinning, nil
565 +}
566 +
567 +func onlineDagCtor(bs bserv.BlockService) format.DAGService {
568 + return merkledag.NewDAGService(bs)
569 +}
570 +
571 +func onlineExchangeCtor(lc fx.Lifecycle, host p2phost.Host, rt routing.IpfsRouting, bs bstore.Blockstore) exchange.Interface {
572 + bitswapNetwork := bsnet.NewFromIpfsHost(host, rt)
573 + return bitswap.New(lifecycleCtx(lc), bitswapNetwork, bs)
574 +}
575 +
576 +func onlineNamesysCtor(rt routing.IpfsRouting, repo repo.Repo, cfg *iconfig.Config) (namesys.NameSystem, error) {
577 + cs := cfg.Ipns.ResolveCacheSize
578 + if cs == 0 {
579 + cs = DefaultIpnsCacheSize
580 + }
581 + if cs < 0 {
582 + return nil, fmt.Errorf("cannot specify negative resolve cache size")
583 + }
584 + return namesys.NewNameSystem(rt, repo.Datastore(), cs), nil
585 +}
586 +
587 +func ipnsRepublisher(lc fx.Lifecycle, cfg *iconfig.Config, namesys namesys.NameSystem, repo repo.Repo, privKey ic.PrivKey) error {
588 + repub := ipnsrp.NewRepublisher(namesys, repo.Datastore(), privKey, repo.Keystore())
589 +
590 + if cfg.Ipns.RepublishPeriod != "" {
591 + d, err := time.ParseDuration(cfg.Ipns.RepublishPeriod)
592 + if err != nil {
593 + return fmt.Errorf("failure to parse config setting IPNS.RepublishPeriod: %s", err)
594 + }
595 +
596 + if !u.Debug && (d < time.Minute || d > (time.Hour*24)) {
597 + return fmt.Errorf("config setting IPNS.RepublishPeriod is not between 1min and 1day: %s", d)
598 + }
599 +
600 + repub.Interval = d
601 + }
602 +
603 + if cfg.Ipns.RecordLifetime != "" {
604 + d, err := time.ParseDuration(cfg.Ipns.RecordLifetime)
605 + if err != nil {
606 + return fmt.Errorf("failure to parse config setting IPNS.RecordLifetime: %s", err)
607 + }
608 +
609 + repub.RecordLifetime = d
610 + }
611 +
612 + lcGoProc(lc, repub.Run)
613 + return nil
614 +}
615 +
616 +type discoveryHandler struct {
617 + ctx context.Context
618 + host p2phost.Host
619 +}
620 +
621 +func (dh *discoveryHandler) HandlePeerFound(p pstore.PeerInfo) {
622 + log.Warning("trying peer info: ", p)
623 + ctx, cancel := context.WithTimeout(dh.ctx, discoveryConnTimeout)
624 + defer cancel()
625 + if err := dh.host.Connect(ctx, p); err != nil {
626 + log.Warning("Failed to connect to peer found by discovery: ", err)
627 + }
628 +}
629 +
630 +func newDiscoveryHandler(lc fx.Lifecycle, host p2phost.Host) *discoveryHandler {
631 + return &discoveryHandler{
632 + ctx: lifecycleCtx(lc),
633 + host: host,
634 + }
635 +}
636 +
637 +func setupDiscovery(lc fx.Lifecycle, cfg *iconfig.Config, host p2phost.Host, handler *discoveryHandler) error {
638 + if cfg.Discovery.MDNS.Enabled {
639 + mdns := cfg.Discovery.MDNS
640 + if mdns.Interval == 0 {
641 + mdns.Interval = 5
642 + }
643 + service, err := discovery.NewMdnsService(lifecycleCtx(lc), host, time.Duration(mdns.Interval)*time.Second, discovery.ServiceTag)
644 + if err != nil {
645 + log.Error("mdns error: ", err)
646 + return nil
647 + }
648 + service.RegisterNotifee(handler)
649 + }
650 + return nil
651 +}
652 +
653 +func providerQueue(lc fx.Lifecycle, repo repo.Repo) (*provider.Queue, error) {
654 + return provider.NewQueue(lifecycleCtx(lc), "provider-v1", repo.Datastore())
655 +}
656 +
657 +func providerCtor(lc fx.Lifecycle, queue *provider.Queue, rt routing.IpfsRouting) provider.Provider {
658 + return provider.NewProvider(lifecycleCtx(lc), queue, rt)
659 +}
660 +
661 +func reproviderCtor(lc fx.Lifecycle, cfg *iconfig.Config, bs bstore.Blockstore, ds format.DAGService, pinning pin.Pinner, rt routing.IpfsRouting) (*rp.Reprovider, error) {
662 + var keyProvider rp.KeyChanFunc
663 +
664 + switch cfg.Reprovider.Strategy {
665 + case "all":
666 + fallthrough
667 + case "":
668 + keyProvider = rp.NewBlockstoreProvider(bs)
669 + case "roots":
670 + keyProvider = rp.NewPinnedProvider(pinning, ds, true)
671 + case "pinned":
672 + keyProvider = rp.NewPinnedProvider(pinning, ds, false)
673 + default:
674 + return nil, fmt.Errorf("unknown reprovider strategy '%s'", cfg.Reprovider.Strategy)
675 + }
676 + return rp.NewReprovider(lifecycleCtx(lc), rt, keyProvider), nil
677 +}
678 +
679 +func reprovider(cfg *iconfig.Config, reprovider *rp.Reprovider) error {
680 + reproviderInterval := kReprovideFrequency
681 + if cfg.Reprovider.Interval != "" {
682 + dur, err := time.ParseDuration(cfg.Reprovider.Interval)
683 + if err != nil {
684 + return err
685 + }
686 +
687 + reproviderInterval = dur
688 + }
689 +
690 + go reprovider.Run(reproviderInterval)
691 + return nil
692 +}
693 +
694 +func files(lc fx.Lifecycle, repo repo.Repo, dag format.DAGService) (*mfs.Root, error) {
695 + dsk := ds.NewKey("/local/filesroot")
696 + pf := func(ctx context.Context, c cid.Cid) error {
697 + return repo.Datastore().Put(dsk, c.Bytes())
698 + }
699 +
700 + var nd *merkledag.ProtoNode
701 + val, err := repo.Datastore().Get(dsk)
702 + ctx := lifecycleCtx(lc)
703 +
704 + switch {
705 + case err == ds.ErrNotFound || val == nil:
706 + nd = ft.EmptyDirNode()
707 + err := dag.Add(ctx, nd)
708 + if err != nil {
709 + return nil, fmt.Errorf("failure writing to dagstore: %s", err)
710 + }
711 + case err == nil:
712 + c, err := cid.Cast(val)
713 + if err != nil {
714 + return nil, err
715 + }
716 +
717 + rnd, err := dag.Get(ctx, c)
718 + if err != nil {
719 + return nil, fmt.Errorf("error loading filesroot from DAG: %s", err)
720 + }
721 +
722 + pbnd, ok := rnd.(*merkledag.ProtoNode)
723 + if !ok {
724 + return nil, merkledag.ErrNotProtobuf
725 + }
726 +
727 + nd = pbnd
728 + default:
729 + return nil, err
730 + }
731 +
732 + return mfs.NewRoot(ctx, dag, nd, pf)
733 +}
734 +
735 +// TODO !!!!!!!!
736 +func bootstrap(n IpfsNode) error {
737 + return n.Bootstrap(DefaultBootstrapConfig)
738 +}
739 +
740 +////////////
741 +// Hacks
742 +
743 +// lifecycleCtx creates a context which will be cancelled when lifecycle stops
744 +//
745 +// This is a hack which we need because most of our services use contexts in a
746 +// wrong way
747 +func lifecycleCtx(lc fx.Lifecycle) context.Context {
748 + ctx, cancel := context.WithCancel(context.TODO()) // TODO: really wire this context up, things (like metrics) may depend on it
749 + lc.Append(fx.Hook{
750 + OnStop: func(_ context.Context) error {
751 + cancel()
752 + return nil
753 + },
754 + })
755 + return ctx
756 +}
757 +
758 +func lcGoProc(lc fx.Lifecycle, processFunc goprocess.ProcessFunc) {
759 + proc := goprocess.Background()
760 + lc.Append(fx.Hook{
761 + OnStart: func(ctx context.Context) error {
762 + proc.Go(processFunc)
763 + return nil
764 + },
765 + OnStop: func(ctx context.Context) error {
766 + return proc.Close() // todo: respect ctx
767 + },
768 + })
769 +}
770 +
771 +func setupSharding(cfg *iconfig.Config) {
772 + // TEMP: setting global sharding switch here
773 + uio.UseHAMTSharding = cfg.Experimental.ShardingEnabled
774 +}