wire a context in most of the ipfs data pipeline

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
 	}
@@ -153,7 +153,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) {
@@ -246,7 +246,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
 		}
@@ -260,7 +260,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

@@ -140,7 +140,7 @@ func New(ctx context.Context, dstore ds.Datastore, dserv ipld.DAGService) (ipfsp
 		dstore:    dstore,
 	}
 
-	data, err := dstore.Get(dirtyKey)
+	data, err := dstore.Get(ctx, dirtyKey)
 	if err != nil {
 		if err == ds.ErrNotFound {
 			return p, nil
@@ -272,7 +272,7 @@ func (p *pinner) addPin(ctx context.Context, c cid.Cid, mode ipfspinner.Mode, na
 	}
 
 	// Store the pin.  Pin must be stored after index for recovery to work.
-	err = p.dstore.Put(pp.dsKey(), pinData)
+	err = p.dstore.Put(ctx, pp.dsKey(), pinData)
 	if err != nil {
 		if mode == ipfspinner.Recursive {
 			p.cidRIndex.Delete(ctx, c.KeyString(), pp.Id)
@@ -293,7 +293,7 @@ func (p *pinner) removePin(ctx context.Context, pp *pin) error {
 
 	// Remove pin from datastore.  Pin must be removed before index for
 	// recovery to work.
-	err := p.dstore.Delete(pp.dsKey())
+	err := p.dstore.Delete(ctx, pp.dsKey())
 	if err != nil {
 		return err
 	}
@@ -616,7 +616,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
 	}
@@ -628,7 +628,7 @@ func (p *pinner) loadAllPins(ctx context.Context) ([]*pin, error) {
 	q := query.Query{
 		Prefix: pinKeyPath,
 	}
-	results, err := p.dstore.Query(q)
+	results, err := p.dstore.Query(ctx, q)
 	if err != nil {
 		return nil, err
 	}
@@ -838,7 +838,7 @@ func (p *pinner) Flush(ctx context.Context) error {
 	}
 
 	// Sync pins and indexes
-	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)
 	}
 
@@ -938,6 +938,6 @@ func (p *pinner) setDirty(ctx context.Context, dirty bool) {
 	if dirty {
 		data[0] = 1
 	}
-	p.dstore.Put(dirtyKey, data)
-	p.dstore.Sync(dirtyKey)
+	p.dstore.Put(ctx, dirtyKey, data)
+	p.dstore.Sync(ctx, dirtyKey)
 }

dspinner/pin_test.go

@@ -299,7 +299,7 @@ func TestPinnerBasic(t *testing.T) {
 	if pp.Cid != ak {
 		t.Error("loaded pin has wrong cid")
 	}
-	err = dsp.dstore.Delete(pp.dsKey())
+	err = dsp.dstore.Delete(ctx, pp.dsKey())
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -665,7 +665,7 @@ func TestLoadDirty(t *testing.T) {
 	cidRIndex.Add(ctx, cidBKey, "not-a-pin-id")
 
 	// 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)
 	}
@@ -688,7 +688,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)
 	}
@@ -920,7 +920,7 @@ func BenchmarkLoadRebuild(b *testing.B) {
 
 	b.Run("RebuildTrue", func(b *testing.B) {
 		for i := 0; i < b.N; i++ {
-			dstore.Put(dirtyKey, []byte{1})
+			dstore.Put(ctx, dirtyKey, []byte{1})
 
 			_, err = New(ctx, dstore, dserv)
 			if err != nil {
@@ -931,7 +931,7 @@ func BenchmarkLoadRebuild(b *testing.B) {
 
 	b.Run("RebuildFalse", func(b *testing.B) {
 		for i := 0; i < b.N; i++ {
-			dstore.Put(dirtyKey, []byte{0})
+			dstore.Put(ctx, dirtyKey, []byte{0})
 
 			_, err = New(ctx, dstore, dserv)
 			if err != nil {