11
"github.com/ipfs/boxo/mfs"
12
pin "github.com/ipfs/boxo/pinning/pinner"
13
"github.com/ipfs/boxo/pinning/pinner/dspinner"
14
- provider "github.com/ipfs/boxo/provider"
14
+ "github.com/ipfs/boxo/provider"
15
"github.com/ipfs/go-cid"
16
"github.com/ipfs/go-datastore"
17
"github.com/ipfs/go-datastore/query"
18
"github.com/ipfs/kubo/config"
19
"github.com/ipfs/kubo/repo"
20
irouting "github.com/ipfs/kubo/routing"
21
+ dht "github.com/libp2p/go-libp2p-kad-dht"
22
+ "github.com/libp2p/go-libp2p-kad-dht/amino"
23
+ "github.com/libp2p/go-libp2p-kad-dht/dual"
24
+ "github.com/libp2p/go-libp2p-kad-dht/fullrt"
25
+ dht_pb "github.com/libp2p/go-libp2p-kad-dht/pb"
26
+ dhtprovider "github.com/libp2p/go-libp2p-kad-dht/provider"
27
+ "github.com/libp2p/go-libp2p-kad-dht/provider/buffered"
28
+ ddhtprovider "github.com/libp2p/go-libp2p-kad-dht/provider/dual"
29
+ "github.com/libp2p/go-libp2p-kad-dht/provider/keystore"
30
+ routinghelpers "github.com/libp2p/go-libp2p-routing-helpers"
31
+ "github.com/libp2p/go-libp2p/core/host"
32
+ peer "github.com/libp2p/go-libp2p/core/peer"
33
+ "github.com/libp2p/go-libp2p/core/routing"
34
+ ma "github.com/multiformats/go-multiaddr"
35
+ mh "github.com/multiformats/go-multihash"
36
"go.uber.org/fx"
37
)
38
44
// Datastore key used to store previous reprovide strategy.
45
const reprovideStrategyKey = "/reprovideStrategy"
46
32
-func ProviderSys(reprovideInterval time.Duration, acceleratedDHTClient bool, provideWorkerCount int) fx.Option {
33
- return fx.Provide(func(lc fx.Lifecycle, cr irouting.ProvideManyRouter, repo repo.Repo) (provider.System, error) {
34
- // Initialize provider.System first, before pinner/blockstore/etc.
35
- // The KeyChanFunc will be set later via SetKeyProvider() once we have
36
- // created the pinner, blockstore and other dependencies.
37
- opts := []provider.Option{
38
- provider.Online(cr),
39
- provider.ReproviderInterval(reprovideInterval),
40
- provider.ProvideWorkerCount(provideWorkerCount),
47
+// DHTProvider is an interface for providing keys to a DHT swarm. It holds a
48
+// state of keys to be advertised, and is responsible for periodically
49
+// publishing provider records for these keys to the DHT swarm before the
50
+// records expire.
51
+type DHTProvider interface {
52
+ // StartProviding ensures keys are periodically advertised to the DHT swarm.
53
+ //
54
+ // If the `keys` aren't currently being reprovided, they are added to the
55
+ // queue to be provided to the DHT swarm as soon as possible, and scheduled
56
+ // to be reprovided periodically. If `force` is set to true, all keys are
57
+ // provided to the DHT swarm, regardless of whether they were already being
58
+ // reprovided in the past. `keys` keep being reprovided until `StopProviding`
59
+ // is called.
60
+ //
61
+ // This operation is asynchronous, it returns as soon as the `keys` are added
62
+ // to the provide queue, and provides happens asynchronously.
63
+ //
64
+ // Returns an error if the keys couldn't be added to the provide queue. This
65
+ // can happen if the provider is closed or if the node is currently Offline
66
+ // (either never bootstrapped, or disconnected since more than `OfflineDelay`).
67
+ // The schedule and provide queue depend on the network size, hence recent
68
+ // network connectivity is essential.
69
+ StartProviding(force bool, keys ...mh.Multihash) error
70
+ // ProvideOnce sends provider records for the specified keys to the DHT swarm
71
+ // only once. It does not automatically reprovide those keys afterward.
72
+ //
73
+ // Add the supplied multihashes to the provide queue, and return immediately.
74
+ // The provide operation happens asynchronously.
75
+ //
76
+ // Returns an error if the keys couldn't be added to the provide queue. This
77
+ // can happen if the provider is closed or if the node is currently Offline
78
+ // (either never bootstrapped, or disconnected since more than `OfflineDelay`).
79
+ // The schedule and provide queue depend on the network size, hence recent
80
+ // network connectivity is essential.
81
+ ProvideOnce(keys ...mh.Multihash) error
82
+ // Clear clears the all the keys from the provide queue and returns the number
83
+ // of keys that were cleared.
84
+ //
85
+ // The keys are not deleted from the keystore, so they will continue to be
86
+ // reprovided as scheduled.
87
+ Clear() int
88
+ // RefreshSchedule scans the Keystore for any keys that are not currently
89
+ // scheduled for reproviding. If such keys are found, it schedules their
90
+ // associated keyspace region to be reprovided.
91
+ //
92
+ // This function doesn't remove prefixes that have no keys from the schedule.
93
+ // This is done automatically during the reprovide operation if a region has no
94
+ // keys.
95
+ //
96
+ // Returns an error if the provider is closed or if the node is currently
97
+ // Offline (either never bootstrapped, or disconnected since more than
98
+ // `OfflineDelay`). The schedule depends on the network size, hence recent
99
+ // network connectivity is essential.
100
+ RefreshSchedule() error
101
+}
102
+
103
+var (
104
+ _ DHTProvider = &ddhtprovider.SweepingProvider{}
105
+ _ DHTProvider = &dhtprovider.SweepingProvider{}
106
+ _ DHTProvider = &NoopProvider{}
107
+ _ DHTProvider = &LegacyProvider{}
108
+)
109
+
110
+type NoopProvider struct{}
111
+
112
+func (r *NoopProvider) StartProviding(bool, ...mh.Multihash) error { return nil }
113
+func (r *NoopProvider) ProvideOnce(...mh.Multihash) error { return nil }
114
+func (r *NoopProvider) Clear() int { return 0 }
115
+func (r *NoopProvider) RefreshSchedule() error { return nil }
116
+
117
+// LegacyProvider is a wrapper around the boxo/provider.System. This DHT
118
+// provide system manages reprovides by bursts where it sequentially reprovides
119
+// all keys.
120
+type LegacyProvider struct {
121
+ provider.System
122
+}
123
+
124
+func (r *LegacyProvider) StartProviding(force bool, keys ...mh.Multihash) error {
125
+ return r.ProvideOnce(keys...)
126
+}
127
+
128
+func (r *LegacyProvider) ProvideOnce(keys ...mh.Multihash) error {
129
+ if many, ok := r.System.(routinghelpers.ProvideManyRouter); ok {
130
+ return many.ProvideMany(context.Background(), keys)
131
+ }
132
+
133
+ for _, k := range keys {
134
+ if err := r.Provide(context.Background(), cid.NewCidV1(cid.Raw, k), true); err != nil {
135
+ return err
136
}
42
- if !acceleratedDHTClient && reprovideInterval > 0 {
43
- // The estimation kinda suck if you are running with accelerated DHT client,
44
- // given this message is just trying to push people to use the acceleratedDHTClient
45
- // let's not report on through if it's in use
46
- opts = append(opts,
47
- provider.ThroughputReport(func(reprovide bool, complete bool, keysProvided uint, duration time.Duration) bool {
48
- avgProvideSpeed := duration / time.Duration(keysProvided)
49
- count := uint64(keysProvided)
50
-
51
- if !reprovide || !complete {
52
- // We don't know how many CIDs we have to provide, try to fetch it from the blockstore.
53
- // But don't try for too long as this might be very expensive if you have a huge datastore.
54
- ctx, cancel := context.WithTimeout(context.Background(), time.Minute*5)
55
- defer cancel()
56
-
57
- // FIXME: I want a running counter of blocks so size of blockstore can be an O(1) lookup.
58
- // Note: talk to datastore directly, as to not depend on Blockstore here.
59
- qr, err := repo.Datastore().Query(ctx, query.Query{
60
- Prefix: blockstore.BlockPrefix.String(),
61
- KeysOnly: true})
62
- if err != nil {
63
- logger.Errorf("fetching AllKeysChain in provider ThroughputReport: %v", err)
64
- return false
65
- }
66
- defer qr.Close()
67
- count = 0
68
- countLoop:
69
- for {
70
- select {
71
- case _, ok := <-qr.Next():
72
- if !ok {
73
- break countLoop
74
- }
75
- count++
76
- case <-ctx.Done():
77
- // really big blockstore mode
78
-
79
- // how many blocks would be in a 10TiB blockstore with 128KiB blocks.
80
- const probableBigBlockstore = (10 * 1024 * 1024 * 1024 * 1024) / (128 * 1024)
81
- // How long per block that lasts us.
82
- expectedProvideSpeed := reprovideInterval / probableBigBlockstore
83
- if avgProvideSpeed > expectedProvideSpeed {
84
- logger.Errorf(`
137
+ }
138
+ return nil
139
+}
140
+
141
+func (r *LegacyProvider) Clear() int {
142
+ return r.System.Clear()
143
+}
144
+
145
+func (r *LegacyProvider) RefreshSchedule() error { return nil }
146
+
147
+// LegacyProviderOpt creates a LegacyProvider to be used as provider in the
148
+// IpfsNode
149
+func LegacyProviderOpt(reprovideInterval time.Duration, strategy string, acceleratedDHTClient bool, provideWorkerCount int) fx.Option {
150
+ system := fx.Provide(
151
+ fx.Annotate(func(lc fx.Lifecycle, cr irouting.ProvideManyRouter, repo repo.Repo) (*LegacyProvider, error) {
152
+ // Initialize provider.System first, before pinner/blockstore/etc.
153
+ // The KeyChanFunc will be set later via SetKeyProvider() once we have
154
+ // created the pinner, blockstore and other dependencies.
155
+ opts := []provider.Option{
156
+ provider.Online(cr),
157
+ provider.ReproviderInterval(reprovideInterval),
158
+ provider.ProvideWorkerCount(provideWorkerCount),
159
+ }
160
+ if !acceleratedDHTClient && reprovideInterval > 0 {
161
+ // The estimation kinda suck if you are running with accelerated DHT client,
162
+ // given this message is just trying to push people to use the acceleratedDHTClient
163
+ // let's not report on through if it's in use
164
+ opts = append(opts,
165
+ provider.ThroughputReport(func(reprovide bool, complete bool, keysProvided uint, duration time.Duration) bool {
166
+ avgProvideSpeed := duration / time.Duration(keysProvided)
167
+ count := uint64(keysProvided)
168
+
169
+ if !reprovide || !complete {
170
+ // We don't know how many CIDs we have to provide, try to fetch it from the blockstore.
171
+ // But don't try for too long as this might be very expensive if you have a huge datastore.
172
+ ctx, cancel := context.WithTimeout(context.Background(), time.Minute*5)
173
+ defer cancel()
174
+
175
+ // FIXME: I want a running counter of blocks so size of blockstore can be an O(1) lookup.
176
+ // Note: talk to datastore directly, as to not depend on Blockstore here.
177
+ qr, err := repo.Datastore().Query(ctx, query.Query{
178
+ Prefix: blockstore.BlockPrefix.String(),
179
+ KeysOnly: true,
180
+ })
181
+ if err != nil {
182
+ logger.Errorf("fetching AllKeysChain in provider ThroughputReport: %v", err)
183
+ return false
184
+ }
185
+ defer qr.Close()
186
+ count = 0
187
+ countLoop:
188
+ for {
189
+ select {
190
+ case _, ok := <-qr.Next():
191
+ if !ok {
192
+ break countLoop
193
+ }
194
+ count++
195
+ case <-ctx.Done():
196
+ // really big blockstore mode
197
+
198
+ // how many blocks would be in a 10TiB blockstore with 128KiB blocks.
199
+ const probableBigBlockstore = (10 * 1024 * 1024 * 1024 * 1024) / (128 * 1024)
200
+ // How long per block that lasts us.
201
+ expectedProvideSpeed := reprovideInterval / probableBigBlockstore
202
+ if avgProvideSpeed > expectedProvideSpeed {
203
+ logger.Errorf(`
204
🔔🔔🔔 YOU MAY BE FALLING BEHIND DHT REPROVIDES! 🔔🔔🔔
205
206
⚠️ Your system might be struggling to keep up with DHT reprovides!
215
216
💡 Consider enabling the Accelerated DHT to enhance your system performance. See:
217
https://github.com/ipfs/kubo/blob/master/docs/config.md#routingaccelerateddhtclient`,
99
- keysProvided, avgProvideSpeed, avgProvideSpeed*probableBigBlockstore, reprovideInterval)
100
- return false
218
+ keysProvided, avgProvideSpeed, avgProvideSpeed*probableBigBlockstore, reprovideInterval)
219
+ return false
220
+ }
221
}
222
}
223
}
104
- }
224
106
- // How long per block that lasts us.
107
- expectedProvideSpeed := reprovideInterval
108
- if count > 0 {
109
- expectedProvideSpeed = reprovideInterval / time.Duration(count)
110
- }
225
+ // How long per block that lasts us.
226
+ expectedProvideSpeed := reprovideInterval
227
+ if count > 0 {
228
+ expectedProvideSpeed = reprovideInterval / time.Duration(count)
229
+ }
230
112
- if avgProvideSpeed > expectedProvideSpeed {
113
- logger.Errorf(`
231
+ if avgProvideSpeed > expectedProvideSpeed {
232
+ logger.Errorf(`
233
🔔🔔🔔 YOU ARE FALLING BEHIND DHT REPROVIDES! 🔔🔔🔔
234
235
⚠️ Your system is struggling to keep up with DHT reprovides!
242
243
💡 Consider enabling the Accelerated DHT to enhance your reprovide throughput. See:
244
https://github.com/ipfs/kubo/blob/master/docs/config.md#routingaccelerateddhtclient`,
126
- keysProvided, avgProvideSpeed, count, avgProvideSpeed*time.Duration(count), reprovideInterval)
127
- }
128
- return false
129
- }, sampledBatchSize))
245
+ keysProvided, avgProvideSpeed, count, avgProvideSpeed*time.Duration(count), reprovideInterval)
246
+ }
247
+ return false
248
+ }, sampledBatchSize))
249
+ }
250
+
251
+ sys, err := provider.New(repo.Datastore(), opts...)
252
+ if err != nil {
253
+ return nil, err
254
+ }
255
+ lc.Append(fx.Hook{
256
+ OnStop: func(ctx context.Context) error {
257
+ return sys.Close()
258
+ },
259
+ })
260
+
261
+ prov := &LegacyProvider{sys}
262
+ handleStrategyChange(strategy, prov, repo.Datastore())
263
+
264
+ return prov, nil
265
+ },
266
+ fx.As(new(provider.System)),
267
+ fx.As(new(DHTProvider)),
268
+ ),
269
+ )
270
+ setKeyProvider := fx.Invoke(func(lc fx.Lifecycle, system provider.System, keyProvider provider.KeyChanFunc) {
271
+ lc.Append(fx.Hook{
272
+ OnStart: func(ctx context.Context) error {
273
+ // SetKeyProvider breaks the circular dependency between provider, blockstore, and pinner.
274
+ // We cannot create the blockstore without the provider (it needs to provide blocks),
275
+ // and we cannot determine the reproviding strategy without the pinner/blockstore.
276
+ // This deferred initialization allows us to create provider.System first,
277
+ // then set the actual key provider function after all dependencies are ready.
278
+ system.SetKeyProvider(keyProvider)
279
+ return nil
280
+ },
281
+ })
282
+ })
283
+ return fx.Options(
284
+ system,
285
+ setKeyProvider,
286
+ )
287
+}
288
+
289
+type dhtImpl interface {
290
+ routing.Routing
291
+ GetClosestPeers(context.Context, string) ([]peer.ID, error)
292
+ Host() host.Host
293
+ MessageSender() dht_pb.MessageSender
294
+}
295
+type addrsFilter interface {
296
+ FilteredAddrs() []ma.Multiaddr
297
+}
298
+
299
+func SweepingProviderOpt(cfg *config.Config) fx.Option {
300
+ reprovideInterval := cfg.Reprovider.Interval.WithDefault(config.DefaultReproviderInterval)
301
+ type providerInput struct {
302
+ fx.In
303
+ DHT routing.Routing `name:"dhtc"`
304
+ Repo repo.Repo
305
+ }
306
+ sweepingReprovider := fx.Provide(func(in providerInput) (DHTProvider, *keystore.ResettableKeystore, error) {
307
+ ds := in.Repo.Datastore()
308
+ ks, err := keystore.NewResettableKeystore(ds,
309
+ keystore.WithPrefixBits(16),
310
+ keystore.WithDatastorePath("/provider/keystore"),
311
+ keystore.WithBatchSize(int(cfg.Reprovider.Sweep.KeyStoreBatchSize.WithDefault(config.DefaultReproviderSweepKeyStoreBatchSize))),
312
+ )
313
+ if err != nil {
314
+ return &NoopProvider{}, nil, err
315
+ }
316
+
317
+ bufferedProviderOpts := []buffered.Option{
318
+ buffered.WithBatchSize(1 << 10),
319
+ buffered.WithDsName("bprov"),
320
+ buffered.WithIdleWriteTime(time.Minute),
321
+ }
322
+
323
+ var impl dhtImpl
324
+ switch inDht := in.DHT.(type) {
325
+ case *dht.IpfsDHT:
326
+ if inDht != nil {
327
+ impl = inDht
328
+ }
329
+ case *dual.DHT:
330
+ if inDht != nil {
331
+ prov, err := ddhtprovider.New(inDht,
332
+ ddhtprovider.WithKeystore(ks),
333
+
334
+ ddhtprovider.WithReprovideInterval(reprovideInterval),
335
+ ddhtprovider.WithMaxReprovideDelay(time.Hour),
336
+ ddhtprovider.WithOfflineDelay(cfg.Reprovider.Sweep.OfflineDelay.WithDefault(config.DefaultReproviderSweepOfflineDelay)),
337
+ ddhtprovider.WithConnectivityCheckOnlineInterval(1*time.Minute),
338
+
339
+ ddhtprovider.WithMaxWorkers(int(cfg.Reprovider.Sweep.MaxWorkers.WithDefault(config.DefaultReproviderSweepMaxWorkers))),
340
+ ddhtprovider.WithDedicatedPeriodicWorkers(int(cfg.Reprovider.Sweep.DedicatedPeriodicWorkers.WithDefault(config.DefaultReproviderSweepDedicatedPeriodicWorkers))),
341
+ ddhtprovider.WithDedicatedBurstWorkers(int(cfg.Reprovider.Sweep.DedicatedBurstWorkers.WithDefault(config.DefaultReproviderSweepDedicatedBurstWorkers))),
342
+ ddhtprovider.WithMaxProvideConnsPerWorker(int(cfg.Reprovider.Sweep.MaxProvideConnsPerWorker.WithDefault(config.DefaultReproviderSweepMaxProvideConnsPerWorker))),
343
+ )
344
+ if err != nil {
345
+ return nil, nil, err
346
+ }
347
+ return buffered.New(prov, ds, bufferedProviderOpts...), ks, nil
348
+ }
349
+ case *fullrt.FullRT:
350
+ if inDht != nil {
351
+ impl = inDht
352
+ }
353
+ }
354
+ if impl == nil {
355
+ return &NoopProvider{}, nil, nil
356
}
357
132
- sys, err := provider.New(repo.Datastore(), opts...)
358
+ var selfAddrsFunc func() []ma.Multiaddr
359
+ if imlpFilter, ok := impl.(addrsFilter); ok {
360
+ selfAddrsFunc = imlpFilter.FilteredAddrs
361
+ } else {
362
+ selfAddrsFunc = func() []ma.Multiaddr { return impl.Host().Addrs() }
363
+ }
364
+ opts := []dhtprovider.Option{
365
+ dhtprovider.WithKeystore(ks),
366
+ dhtprovider.WithPeerID(impl.Host().ID()),
367
+ dhtprovider.WithRouter(impl),
368
+ dhtprovider.WithMessageSender(impl.MessageSender()),
369
+ dhtprovider.WithSelfAddrs(selfAddrsFunc),
370
+ dhtprovider.WithAddLocalRecord(func(h mh.Multihash) error {
371
+ return impl.Provide(context.Background(), cid.NewCidV1(cid.Raw, h), false)
372
+ }),
373
+
374
+ dhtprovider.WithReplicationFactor(amino.DefaultBucketSize),
375
+ dhtprovider.WithReprovideInterval(reprovideInterval),
376
+ dhtprovider.WithMaxReprovideDelay(time.Hour),
377
+ dhtprovider.WithOfflineDelay(cfg.Reprovider.Sweep.OfflineDelay.WithDefault(config.DefaultReproviderSweepOfflineDelay)),
378
+ dhtprovider.WithConnectivityCheckOnlineInterval(1 * time.Minute),
379
+
380
+ dhtprovider.WithMaxWorkers(int(cfg.Reprovider.Sweep.MaxWorkers.WithDefault(config.DefaultReproviderSweepMaxWorkers))),
381
+ dhtprovider.WithDedicatedPeriodicWorkers(int(cfg.Reprovider.Sweep.DedicatedPeriodicWorkers.WithDefault(config.DefaultReproviderSweepDedicatedPeriodicWorkers))),
382
+ dhtprovider.WithDedicatedBurstWorkers(int(cfg.Reprovider.Sweep.DedicatedBurstWorkers.WithDefault(config.DefaultReproviderSweepDedicatedBurstWorkers))),
383
+ dhtprovider.WithMaxProvideConnsPerWorker(int(cfg.Reprovider.Sweep.MaxProvideConnsPerWorker.WithDefault(config.DefaultReproviderSweepMaxProvideConnsPerWorker))),
384
+ }
385
+
386
+ prov, err := dhtprovider.New(opts...)
387
if err != nil {
134
- return nil, err
388
+ return &NoopProvider{}, nil, err
389
+ }
390
+ return buffered.New(prov, ds, bufferedProviderOpts...), ks, nil
391
+ })
392
+
393
+ type keystoreInput struct {
394
+ fx.In
395
+ Provider DHTProvider
396
+ Keystore *keystore.ResettableKeystore
397
+ KeyProvider provider.KeyChanFunc
398
+ }
399
+ initKeystore := fx.Invoke(func(lc fx.Lifecycle, in keystoreInput) {
400
+ var (
401
+ cancel context.CancelFunc
402
+ done = make(chan struct{})
403
+ )
404
+
405
+ syncKeystore := func(ctx context.Context) error {
406
+ kcf, err := in.KeyProvider(ctx)
407
+ if err != nil {
408
+ return err
409
+ }
410
+ if err := in.Keystore.ResetCids(ctx, kcf); err != nil {
411
+ return err
412
+ }
413
+ if err := in.Provider.RefreshSchedule(); err != nil {
414
+ logger.Infow("refreshing provider schedule", "err", err)
415
+ }
416
+ return nil
417
}
418
419
lc.Append(fx.Hook{
420
+ OnStart: func(ctx context.Context) error {
421
+ if in.Provider == nil || in.Keystore == nil {
422
+ return nil
423
+ }
424
+ // Set the KeyProvider as a garbage collection function for the
425
+ // keystore. Periodically purge the Keystore from all its keys and
426
+ // replace them with the keys that needs to be reprovided, coming from
427
+ // the KeyChanFunc. So far, this is the less worse way to remove CIDs
428
+ // that shouldn't be reprovided from the provider's state.
429
+ if err := syncKeystore(ctx); err != nil {
430
+ return err
431
+ }
432
+
433
+ gcCtx, c := context.WithCancel(context.Background())
434
+ cancel = c
435
+
436
+ go func() { // garbage collection loop for cids to reprovide
437
+ defer close(done)
438
+ ticker := time.NewTicker(reprovideInterval)
439
+ defer ticker.Stop()
440
+
441
+ for {
442
+ select {
443
+ case <-gcCtx.Done():
444
+ return
445
+ case <-ticker.C:
446
+ if err := syncKeystore(gcCtx); err != nil {
447
+ logger.Errorw("provider keystore sync", "err", err)
448
+ }
449
+ }
450
+ }
451
+ }()
452
+ return nil
453
+ },
454
OnStop: func(ctx context.Context) error {
139
- return sys.Close()
455
+ if in.Provider == nil || in.Keystore == nil {
456
+ return nil
457
+ }
458
+ if cancel != nil {
459
+ // Cancel Keystore garbage collection loop
460
+ cancel()
461
+ }
462
+ select {
463
+ case <-done:
464
+ case <-ctx.Done():
465
+ return ctx.Err()
466
+ }
467
+
468
+ // Keystore state isn't be persisted across restarts.
469
+ return in.Keystore.Empty(ctx)
470
},
471
})
142
-
143
- return sys, nil
472
})
473
+
474
+ return fx.Options(
475
+ sweepingReprovider,
476
+ initKeystore,
477
+ )
478
}
479
480
// ONLINE/OFFLINE
481
482
// OnlineProviders groups units managing provider routing records online
150
-func OnlineProviders(provide bool, providerStrategy string, reprovideInterval time.Duration, acceleratedDHTClient bool, provideWorkerCount int) fx.Option {
483
+func OnlineProviders(provide bool, cfg *config.Config) fx.Option {
484
if !provide {
485
return OfflineProviders()
486
}
487
488
+ providerStrategy := cfg.Reprovider.Strategy.WithDefault(config.DefaultReproviderStrategy)
489
+
490
strategyFlag := config.ParseReproviderStrategy(providerStrategy)
491
if strategyFlag == 0 {
492
return fx.Error(fmt.Errorf("unknown reprovider strategy %q", providerStrategy))
493
}
494
160
- return fx.Options(
495
+ opts := []fx.Option{
496
fx.Provide(setReproviderKeyProvider(providerStrategy)),
162
- ProviderSys(reprovideInterval, acceleratedDHTClient, provideWorkerCount),
163
- )
497
+ }
498
+ if cfg.Reprovider.Sweep.Enabled.WithDefault(config.DefaultReproviderSweepEnabled) {
499
+ opts = append(opts, SweepingProviderOpt(cfg))
500
+ } else {
501
+ reprovideInterval := cfg.Reprovider.Interval.WithDefault(config.DefaultReproviderInterval)
502
+ acceleratedDHTClient := cfg.Routing.AcceleratedDHTClient.WithDefault(config.DefaultAcceleratedDHTClient)
503
+ provideWorkerCount := int(cfg.Provider.WorkerCount.WithDefault(config.DefaultProviderWorkerCount))
504
+
505
+ opts = append(opts, LegacyProviderOpt(reprovideInterval, providerStrategy, acceleratedDHTClient, provideWorkerCount))
506
+ }
507
+
508
+ return fx.Options(opts...)
509
}
510
511
// OfflineProviders groups units managing provider routing records offline
512
func OfflineProviders() fx.Option {
168
- return fx.Provide(provider.NewNoopProvider)
513
+ return fx.Provide(func() DHTProvider {
514
+ return &NoopProvider{}
515
+ })
516
}
517
518
func mfsProvider(mfsRoot *mfs.Root, fetcher fetcher.Factory) provider.KeyChanFunc {
538
OfflineIPLDFetcher fetcher.Factory `name:"offlineIpldFetcher"`
539
OfflineUnixFSFetcher fetcher.Factory `name:"offlineUnixfsFetcher"`
540
MFSRoot *mfs.Root
194
- Provider provider.System
541
Repo repo.Repo
542
}
543
603
// Strategy change detection: when the reproviding strategy changes,
604
// we clear the provide queue to avoid unexpected behavior from mixing
605
// strategies. This ensures a clean transition between different providing modes.
260
-func handleStrategyChange(strategy string, provider provider.System, ds datastore.Datastore) {
606
+func handleStrategyChange(strategy string, provider DHTProvider, ds datastore.Datastore) {
607
ctx := context.Background()
608
609
previous, changed, err := detectStrategyChange(ctx, strategy, ds)
630
return func(in provStrategyIn) provStrategyOut {
631
// Create the appropriate key provider based on strategy
632
kcf := createKeyProvider(strategyFlag, in)
287
-
288
- // SetKeyProvider breaks the circular dependency between provider, blockstore, and pinner.
289
- // We cannot create the blockstore without the provider (it needs to provide blocks),
290
- // and we cannot determine the reproviding strategy without the pinner/blockstore.
291
- // This deferred initialization allows us to create provider.System first,
292
- // then set the actual key provider function after all dependencies are ready.
293
- in.Provider.SetKeyProvider(kcf)
294
-
295
- // Handle strategy changes (detection, queue clearing, persistence)
296
- handleStrategyChange(strategy, in.Provider, in.Repo.Datastore())
297
-
633
return provStrategyOut{
634
ProvidingStrategy: strategyFlag,
635
ProvidingKeyChanFunc: kcf,