Add reprovider

License: MIT
Signed-off-by: Michael Avila <davidmichaelavila@gmail.com>

core/builder.go

@@ -277,14 +277,19 @@ func setupNode(ctx context.Context, n *IpfsNode, cfg *BuildCfg) error {
 	n.Resolver = resolver.NewBasicResolver(n.DAG)
 
 	// Provider
-	queue, err := provider.NewQueue(ctx, "provider-v1", n.Repo.Datastore())
+	queueP, err := provider.NewQueue(ctx, "provider-v1", n.Repo.Datastore())
 	if err != nil {
 		return err
 	}
-
 	tracker := provider.NewTracker(n.Repo.Datastore())
+	n.Provider = provider.NewProvider(ctx, queueP, tracker, n.Routing)
 
-	n.Provider = provider.NewProvider(ctx, queue, tracker, n.Routing)
+	// Reprovider
+	queueR, err := provider.NewQueue(ctx, "reprovider-v1", n.Repo.Datastore())
+	if err != nil {
+		return err
+	}
+	n.Reprovider = provider.NewReprovider(ctx, queueR, tracker, time.Minute, time.Hour*12, n.Blockstore, n.Routing)
 
 	if cfg.Online {
 		if err := n.startLateOnlineServices(ctx); err != nil {

core/core.go

@@ -21,7 +21,6 @@ import (
 	"time"
 
 	version "github.com/ipfs/go-ipfs"
-	rp "github.com/ipfs/go-ipfs/exchange/reprovide"
 	filestore "github.com/ipfs/go-ipfs/filestore"
 	mount "github.com/ipfs/go-ipfs/fuse/mount"
 	namesys "github.com/ipfs/go-ipfs/namesys"
@@ -120,13 +119,13 @@ type IpfsNode struct {
 	RecordValidator record.Validator
 
 	// Online
-	PeerHost     p2phost.Host        // the network host (server+client)
-	Bootstrapper io.Closer           // the periodic bootstrapper
-	Routing      routing.IpfsRouting // the routing system. recommend ipfs-dht
-	Exchange     exchange.Interface  // the block exchange + strategy (bitswap)
-	Namesys      namesys.NameSystem  // the name system, resolves paths to hashes
+	PeerHost     p2phost.Host         // the network host (server+client)
+	Bootstrapper io.Closer            // the periodic bootstrapper
+	Routing      routing.IpfsRouting  // the routing system. recommend ipfs-dht
+	Exchange     exchange.Interface   // the block exchange + strategy (bitswap)
+	Namesys      namesys.NameSystem   // the name system, resolves paths to hashes
 	Provider     provider.Provider   // the value provider system
-	Reprovider   *rp.Reprovider      // the value reprovider system
+	Reprovider   *provider.Reprovider // the value reprovider system
 	IpnsRepub    *ipnsrp.Republisher
 
 	AutoNAT  *autonat.AutoNATService
@@ -321,44 +320,13 @@ func constructConnMgr(cfg config.ConnMgr) (ifconnmgr.ConnManager, error) {
 }
 
 func (n *IpfsNode) startLateOnlineServices(ctx context.Context) error {
-	cfg, err := n.Repo.Config()
-	if err != nil {
-		return err
-	}
-
 	// Provider
 
 	n.Provider.Run()
 
 	// Reprovider
 
-	var keyProvider rp.KeyChanFunc
-
-	switch cfg.Reprovider.Strategy {
-	case "all":
-		fallthrough
-	case "":
-		keyProvider = rp.NewBlockstoreProvider(n.Blockstore)
-	case "roots":
-		keyProvider = rp.NewPinnedProvider(n.Pinning, n.DAG, true)
-	case "pinned":
-		keyProvider = rp.NewPinnedProvider(n.Pinning, n.DAG, false)
-	default:
-		return fmt.Errorf("unknown reprovider strategy '%s'", cfg.Reprovider.Strategy)
-	}
-	n.Reprovider = rp.NewReprovider(ctx, n.Routing, keyProvider)
-
-	reproviderInterval := kReprovideFrequency
-	if cfg.Reprovider.Interval != "" {
-		dur, err := time.ParseDuration(cfg.Reprovider.Interval)
-		if err != nil {
-			return err
-		}
-
-		reproviderInterval = dur
-	}
-
-	go n.Reprovider.Run(reproviderInterval)
+	n.Reprovider.Run()
 
 	return nil
 }

provider/reprovider.go

@@ -16,8 +16,8 @@ type Reprovider struct {
 	ctx            context.Context
 	queue          *Queue
 	tracker        *Tracker
-	tick           time.Duration
 	initialTick    time.Duration
+	tick           time.Duration
 	blockstore     blockstore.Blockstore
 	contentRouting routing.ContentRouting
 	trigger        chan struct{}
@@ -28,13 +28,13 @@ type Reprovider struct {
 // Reprovider periodically re-announces the cids that have been provided. These
 // reprovides can be run on an interval and/or manually. Reprovider also untracks
 // cids that are no longer in the blockstore.
-func NewReprovider(ctx context.Context, queue *Queue, tracker *Tracker, tick time.Duration, initialTick time.Duration, blockstore blockstore.Blockstore, contentRouting routing.ContentRouting) *Reprovider {
+func NewReprovider(ctx context.Context, queue *Queue, tracker *Tracker, initialTick time.Duration, tick time.Duration, blockstore blockstore.Blockstore, contentRouting routing.ContentRouting) *Reprovider {
 	return &Reprovider{
 		ctx:            ctx,
 		queue:          queue,
 		tracker:        tracker,
-		tick:           tick,
 		initialTick:    initialTick,
+		tick:           tick,
 		blockstore:     blockstore,
 		contentRouting: contentRouting,
 		trigger:        make(chan struct{}),