feat: plumb through context changes

dsindex/indexer.go

@@ -75,7 +75,7 @@ func (x *indexer) Add(ctx context.Context, key, value string) error {
 		return ErrEmptyValue
 	}
 	dsKey := ds.NewKey(encode(key)).ChildString(encode(value))
-	return x.dstore.Put(dsKey, []byte{})
+	return x.dstore.Put(ctx, dsKey, []byte{})
 }
 
 func (x *indexer) Delete(ctx context.Context, key, value string) error {
@@ -85,7 +85,7 @@ func (x *indexer) Delete(ctx context.Context, key, value string) error {
 	if value == "" {
 		return ErrEmptyValue
 	}
-	return x.dstore.Delete(ds.NewKey(encode(key)).ChildString(encode(value)))
+	return x.dstore.Delete(ctx, ds.NewKey(encode(key)).ChildString(encode(value)))
 }
 
 func (x *indexer) DeleteKey(ctx context.Context, key string) (int, error) {
@@ -108,7 +108,7 @@ func (x *indexer) ForEach(ctx context.Context, key string, fn func(key, value st
 		Prefix:   key,
 		KeysOnly: true,
 	}
-	results, err := x.dstore.Query(q)
+	results, err := x.dstore.Query(ctx, q)
 	if err != nil {
 		return err
 	}
@@ -145,7 +145,7 @@ func (x *indexer) HasValue(ctx context.Context, key, value string) (bool, error)
 	if value == "" {
 		return false, ErrEmptyValue
 	}
-	return x.dstore.Has(ds.NewKey(encode(key)).ChildString(encode(value)))
+	return x.dstore.Has(ctx, ds.NewKey(encode(key)).ChildString(encode(value)))
 }
 
 func (x *indexer) HasAny(ctx context.Context, key string) (bool, error) {
@@ -238,7 +238,7 @@ func (x *indexer) deletePrefix(ctx context.Context, prefix string) (int, error)
 	}
 
 	for i := range ents {
-		err = x.dstore.Delete(ds.NewKey(ents[i].Key))
+		err = x.dstore.Delete(ctx, ds.NewKey(ents[i].Key))
 		if err != nil {
 			return 0, err
 		}
@@ -252,7 +252,7 @@ func (x *indexer) queryPrefix(ctx context.Context, prefix string) ([]query.Entry
 		Prefix:   prefix,
 		KeysOnly: true,
 	}
-	results, err := x.dstore.Query(q)
+	results, err := x.dstore.Query(ctx, q)
 	if err != nil {
 		return nil, err
 	}

dspinner/pin.go

@@ -142,7 +142,7 @@ func New(ctx context.Context, dstore ds.Datastore, dserv ipld.DAGService) (*pinn
 		dstore:    dstore,
 	}
 
-	data, err := dstore.Get(dirtyKey)
+	data, err := dstore.Get(ctx, dirtyKey)
 	if err != nil {
 		if err == ds.ErrNotFound {
 			return p, nil
@@ -268,7 +268,7 @@ func (p *pinner) addPin(ctx context.Context, c cid.Cid, mode ipfspinner.Mode, na
 	p.setDirty(ctx)
 
 	// Store the pin
-	err = p.dstore.Put(pp.dsKey(), pinData)
+	err = p.dstore.Put(ctx, pp.dsKey(), pinData)
 	if err != nil {
 		return "", err
 	}
@@ -332,7 +332,7 @@ func (p *pinner) removePin(ctx context.Context, pp *pin) error {
 
 	// The pin is removed last so that an incomplete remove is detected by a
 	// pin that has a missing index.
-	err = p.dstore.Delete(pp.dsKey())
+	err = p.dstore.Delete(ctx, pp.dsKey())
 	if err != nil {
 		return err
 	}
@@ -669,7 +669,7 @@ func (p *pinner) removePinsForCid(ctx context.Context, c cid.Cid, mode ipfspinne
 
 // loadPin loads a single pin from the datastore.
 func (p *pinner) loadPin(ctx context.Context, pid string) (*pin, error) {
-	pinData, err := p.dstore.Get(ds.NewKey(path.Join(pinKeyPath, pid)))
+	pinData, err := p.dstore.Get(ctx, ds.NewKey(path.Join(pinKeyPath, pid)))
 	if err != nil {
 		return nil, err
 	}
@@ -804,7 +804,7 @@ func (p *pinner) flushPins(ctx context.Context, force bool) error {
 	if !p.autoSync && !force {
 		return nil
 	}
-	if err := p.dstore.Sync(ds.NewKey(basePath)); err != nil {
+	if err := p.dstore.Sync(ctx, ds.NewKey(basePath)); err != nil {
 		return fmt.Errorf("cannot sync pin state: %v", err)
 	}
 	p.setClean(ctx)
@@ -909,12 +909,12 @@ func (p *pinner) setDirty(ctx context.Context) {
 	}
 
 	data := []byte{1}
-	err := p.dstore.Put(dirtyKey, data)
+	err := p.dstore.Put(ctx, dirtyKey, data)
 	if err != nil {
 		log.Errorf("failed to set pin dirty flag: %s", err)
 		return
 	}
-	err = p.dstore.Sync(dirtyKey)
+	err = p.dstore.Sync(ctx, dirtyKey)
 	if err != nil {
 		log.Errorf("failed to sync pin dirty flag: %s", err)
 	}
@@ -928,12 +928,12 @@ func (p *pinner) setClean(ctx context.Context) {
 	}
 
 	data := []byte{0}
-	err := p.dstore.Put(dirtyKey, data)
+	err := p.dstore.Put(ctx, dirtyKey, data)
 	if err != nil {
 		log.Errorf("failed to set clear dirty flag: %s", err)
 		return
 	}
-	if err = p.dstore.Sync(dirtyKey); err != nil {
+	if err = p.dstore.Sync(ctx, dirtyKey); err != nil {
 		log.Errorf("failed to sync cleared pin dirty flag: %s", err)
 		return
 	}
@@ -951,7 +951,7 @@ func (p *pinner) rebuildIndexes(ctx context.Context) error {
 	q := query.Query{
 		Prefix: pinKeyPath,
 	}
-	results, err := p.dstore.Query(q)
+	results, err := p.dstore.Query(ctx, q)
 	if err != nil {
 		return err
 	}

dspinner/pin_test.go

@@ -309,7 +309,7 @@ func TestPinnerBasic(t *testing.T) {
 	if pp.Cid != ak {
 		t.Error("loaded pin has wrong cid")
 	}
-	err = p.dstore.Delete(pp.dsKey())
+	err = p.dstore.Delete(ctx, pp.dsKey())
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -703,7 +703,7 @@ func TestLoadDirty(t *testing.T) {
 	p.setDirty(ctx)
 
 	// Verify dirty
-	data, err := dstore.Get(dirtyKey)
+	data, err := dstore.Get(ctx, dirtyKey)
 	if err != nil {
 		t.Fatalf("could not read dirty flag: %v", err)
 	}
@@ -727,7 +727,7 @@ func TestLoadDirty(t *testing.T) {
 	}
 
 	// Verify not dirty
-	data, err = dstore.Get(dirtyKey)
+	data, err = dstore.Get(ctx, dirtyKey)
 	if err != nil {
 		t.Fatalf("could not read dirty flag: %v", err)
 	}
@@ -923,7 +923,7 @@ type batchWrap struct {
 	ds.Datastore
 }
 
-func (d *batchWrap) Batch() (ds.Batch, error) {
+func (d *batchWrap) Batch(_ context.Context) (ds.Batch, error) {
 	return ds.NewBasicBatch(d), nil
 }
 
@@ -957,7 +957,7 @@ func BenchmarkLoad(b *testing.B) {
 
 	b.Run("RebuildTrue", func(b *testing.B) {
 		for i := 0; i < b.N; i++ {
-			err = dstore.Put(dirtyKey, []byte{1})
+			err = dstore.Put(ctx, dirtyKey, []byte{1})
 			if err != nil {
 				panic(err.Error())
 			}
@@ -971,7 +971,7 @@ func BenchmarkLoad(b *testing.B) {
 
 	b.Run("RebuildFalse", func(b *testing.B) {
 		for i := 0; i < b.N; i++ {
-			err = dstore.Put(dirtyKey, []byte{0})
+			err = dstore.Put(ctx, dirtyKey, []byte{0})
 			if err != nil {
 				panic(err.Error())
 			}
@@ -1161,7 +1161,7 @@ func BenchmarkRebuild(b *testing.B) {
 		b.Run(fmt.Sprintf("Rebuild %d", pins), func(b *testing.B) {
 			b.ReportAllocs()
 			for i := 0; i < b.N; i++ {
-				err = dstore.Put(dirtyKey, []byte{1})
+				err = dstore.Put(ctx, dirtyKey, []byte{1})
 				if err != nil {
 					panic(err.Error())
 				}
@@ -1252,7 +1252,7 @@ func TestCidIndex(t *testing.T) {
 	q := query.Query{
 		Prefix: pinKeyPath,
 	}
-	results, err := pinner.dstore.Query(q)
+	results, err := pinner.dstore.Query(ctx, q)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -1335,7 +1335,7 @@ func TestRebuild(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	err = pinner.dstore.Delete(pp.dsKey())
+	err = pinner.dstore.Delete(ctx, pp.dsKey())
 	if err != nil {
 		t.Fatal(err)
 	}

go.mod

@@ -3,17 +3,17 @@ module github.com/ipfs/go-ipfs-pinner
 go 1.15
 
 require (
-	github.com/ipfs/go-blockservice v0.1.4
+	github.com/ipfs/go-blockservice v0.2.0
 	github.com/ipfs/go-cid v0.0.7
-	github.com/ipfs/go-datastore v0.4.5
-	github.com/ipfs/go-ds-badger v0.2.6
-	github.com/ipfs/go-ds-leveldb v0.4.2
-	github.com/ipfs/go-ipfs-blockstore v0.1.4
-	github.com/ipfs/go-ipfs-exchange-offline v0.0.1
+	github.com/ipfs/go-datastore v0.5.0
+	github.com/ipfs/go-ds-badger v0.3.0
+	github.com/ipfs/go-ds-leveldb v0.5.0
+	github.com/ipfs/go-ipfs-blockstore v0.2.0
+	github.com/ipfs/go-ipfs-exchange-offline v0.1.0
 	github.com/ipfs/go-ipfs-util v0.0.2
 	github.com/ipfs/go-ipld-format v0.2.0
-	github.com/ipfs/go-log v1.0.4
-	github.com/ipfs/go-merkledag v0.3.0
+	github.com/ipfs/go-log v1.0.5
+	github.com/ipfs/go-merkledag v0.5.0
 	github.com/multiformats/go-multibase v0.0.3
-	github.com/polydawn/refmt v0.0.0-20190807091052-3d65705ee9f1
+	github.com/polydawn/refmt v0.0.0-20201211092308-30ac6d18308e
 )