feat: plumb through datastore contexts

batched/system.go

@@ -259,10 +259,10 @@ func (s *BatchProvidingSystem) Run() {
 				s.lastReprovideBatchSize = len(keys)
 				s.lastReprovideDuration = dur
 
-				if err := s.ds.Put(lastReprovideKey, storeTime(time.Now())); err != nil {
+				if err := s.ds.Put(s.ctx, lastReprovideKey, storeTime(time.Now())); err != nil {
 					log.Errorf("could not store last reprovide time: %v", err)
 				}
-				if err := s.ds.Sync(lastReprovideKey); err != nil {
+				if err := s.ds.Sync(s.ctx, lastReprovideKey); err != nil {
 					log.Errorf("could not perform sync of last reprovide time: %v", err)
 				}
 			}
@@ -374,7 +374,7 @@ reprovideCidLoop:
 }
 
 func (s *BatchProvidingSystem) getLastReprovideTime() (time.Time, error) {
-	val, err := s.ds.Get(lastReprovideKey)
+	val, err := s.ds.Get(s.ctx, lastReprovideKey)
 	if errors.Is(err, datastore.ErrNotFound) {
 		return time.Time{}, nil
 	}

go.mod

@@ -6,20 +6,24 @@ retract [v1.0.0, v1.0.1]
 
 require (
 	github.com/cenkalti/backoff v2.2.1+incompatible
-	github.com/ipfs/go-block-format v0.0.2
-	github.com/ipfs/go-blockservice v0.1.4
-	github.com/ipfs/go-cid v0.0.7
+	github.com/ipfs/go-block-format v0.0.3
+	github.com/ipfs/go-blockservice v0.2.0
+	github.com/ipfs/go-cid v0.1.0
 	github.com/ipfs/go-cidutil v0.0.2
-	github.com/ipfs/go-datastore v0.4.5
+	github.com/ipfs/go-datastore v0.5.0
 	github.com/ipfs/go-fetcher v1.5.0
-	github.com/ipfs/go-ipfs-blockstore v0.1.4
+	github.com/ipfs/go-ipfs-blockstore v0.2.0
 	github.com/ipfs/go-ipfs-blocksutil v0.0.1
-	github.com/ipfs/go-ipfs-exchange-offline v0.0.1
-	github.com/ipfs/go-ipfs-routing v0.1.0
-	github.com/ipfs/go-log v1.0.4
+	github.com/ipfs/go-ipfs-exchange-offline v0.1.0
+	github.com/ipfs/go-ipfs-routing v0.2.0
+	github.com/ipfs/go-log v1.0.5
 	github.com/ipfs/go-verifcid v0.0.1
 	github.com/ipld/go-ipld-prime v0.11.0
+	github.com/klauspost/cpuid/v2 v2.0.9 // indirect
 	github.com/libp2p/go-libp2p-core v0.8.5
 	github.com/libp2p/go-libp2p-testing v0.4.0
-	github.com/multiformats/go-multihash v0.0.15
+	github.com/multiformats/go-base32 v0.0.4 // indirect
+	github.com/multiformats/go-multihash v0.0.16
+	golang.org/x/crypto v0.0.0-20210921155107-089bfa567519 // indirect
+	golang.org/x/sys v0.0.0-20211025112917-711f33c9992c // indirect
 )

queue/queue.go

@@ -97,7 +97,7 @@ func (q *Queue) work() {
 					c, err = cid.Parse(head.Value)
 					if err != nil {
 						log.Warningf("error parsing queue entry cid with key (%s), removing it from queue: %s", head.Key, err)
-						err = q.ds.Delete(k)
+						err = q.ds.Delete(q.ctx, k)
 						if err != nil {
 							log.Errorf("error deleting queue entry with key (%s), due to error (%s), stopping provider", head.Key, err)
 							return
@@ -120,12 +120,12 @@ func (q *Queue) work() {
 				keyPath := fmt.Sprintf("%d/%s", time.Now().UnixNano(), c.String())
 				nextKey := datastore.NewKey(keyPath)
 
-				if err := q.ds.Put(nextKey, toQueue.Bytes()); err != nil {
+				if err := q.ds.Put(q.ctx, nextKey, toQueue.Bytes()); err != nil {
 					log.Errorf("Failed to enqueue cid: %s", err)
 					continue
 				}
 			case dequeue <- c:
-				err := q.ds.Delete(k)
+				err := q.ds.Delete(q.ctx, k)
 
 				if err != nil {
 					log.Errorf("Failed to delete queued cid %s with key %s: %s", c, k, err)
@@ -141,7 +141,7 @@ func (q *Queue) work() {
 
 func (q *Queue) getQueueHead() (*query.Entry, error) {
 	qry := query.Query{Orders: []query.Order{query.OrderByKey{}}, Limit: 1}
-	results, err := q.ds.Query(qry)
+	results, err := q.ds.Query(q.ctx, qry)
 	if err != nil {
 		return nil, err
 	}

queue/queue_test.go

@@ -8,7 +8,7 @@ import (
 	"github.com/ipfs/go-cid"
 	"github.com/ipfs/go-datastore"
 	"github.com/ipfs/go-datastore/sync"
-	"github.com/ipfs/go-ipfs-blocksutil"
+	blocksutil "github.com/ipfs/go-ipfs-blocksutil"
 )
 
 var blockGenerator = blocksutil.NewBlockGenerator()
@@ -72,7 +72,7 @@ func TestMangledData(t *testing.T) {
 
 	// put bad data in the queue
 	queueKey := datastore.NewKey("/test/0")
-	err = queue.ds.Put(queueKey, []byte("borked"))
+	err = queue.ds.Put(ctx, queueKey, []byte("borked"))
 	if err != nil {
 		t.Fatal(err)
 	}

simple/reprovide.go

@@ -45,7 +45,7 @@ type Reprovider struct {
 }
 
 // NewReprovider creates new Reprovider instance.
-func NewReprovider(ctx context.Context, reprovideIniterval time.Duration, rsys routing.ContentRouting, keyProvider KeyChanFunc) *Reprovider {
+func NewReprovider(ctx context.Context, reprovideInterval time.Duration, rsys routing.ContentRouting, keyProvider KeyChanFunc) *Reprovider {
 	ctx, cancel := context.WithCancel(ctx)
 	return &Reprovider{
 		ctx:      ctx,
@@ -55,7 +55,7 @@ func NewReprovider(ctx context.Context, reprovideIniterval time.Duration, rsys r
 
 		rsys:        rsys,
 		keyProvider: keyProvider,
-		tick:        reprovideIniterval,
+		tick:        reprovideInterval,
 	}
 }
 

simple/reprovide_test.go

@@ -48,7 +48,7 @@ func setupDag(t *testing.T) (nodes []cid.Cid, bstore blockstore.Blockstore) {
 			t.Fatal(err)
 		}
 		blk := toBlock(t, nb.Build())
-		err = bstore.Put(blk)
+		err = bstore.Put(context.Background(), blk)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -60,7 +60,7 @@ func setupDag(t *testing.T) (nodes []cid.Cid, bstore blockstore.Blockstore) {
 			t.Fatal(err)
 		}
 		blk = toBlock(t, nd)
-		err = bstore.Put(blk)
+		err = bstore.Put(context.Background(), blk)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -117,6 +117,7 @@ func testReprovide(t *testing.T, trigger func(r *Reprovider, ctx context.Context
 
 	keyProvider := NewBlockstoreProvider(bstore)
 	reprov := NewReprovider(ctx, time.Hour, clA, keyProvider)
+	reprov.Trigger(context.Background())
 	err := trigger(reprov, ctx)
 	if err != nil {
 		t.Fatal(err)