GO-759: WIP Flatfs file storage

core/filestorage/badgerfileindex.go

@@ -1,12 +1,13 @@
-package badgerfilestore
+package filestorage
 
 import (
 	"bytes"
 	"errors"
-	"github.com/dgraph-io/badger/v3"
-	"github.com/ipfs/go-cid"
 	"sync"
 	"time"
+
+	"github.com/dgraph-io/badger/v3"
+	"github.com/ipfs/go-cid"
 )
 
 const keyIndexPrefix = "files/indexes/"

core/filestorage/badgerfileindex_test.go

@@ -1,15 +1,16 @@
-package badgerfilestore
+package filestorage
 
 import (
 	"bytes"
 	"fmt"
+	"os"
+	"testing"
+
 	"github.com/dgraph-io/badger/v3"
 	blocks "github.com/ipfs/go-block-format"
 	"github.com/ipfs/go-cid"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	"os"
-	"testing"
 )
 
 func TestFileBadgerIndex_Add(t *testing.T) {

core/filestorage/badgerfilestore/badgerfilestore.go

@@ -1,164 +0,0 @@
-package badgerfilestore
-
-import (
-	"context"
-	"github.com/anytypeio/any-sync/commonfile/fileblockstore"
-	"github.com/anytypeio/any-sync/commonfile/fileproto"
-	"github.com/dgraph-io/badger/v3"
-	blocks "github.com/ipfs/go-block-format"
-	"github.com/ipfs/go-cid"
-	format "github.com/ipfs/go-ipld-format"
-)
-
-const keyPrefix = "files/blocks/"
-
-type FileStore interface {
-	fileblockstore.BlockStoreLocal
-	BlockAvailability(ctx context.Context, ks []cid.Cid) (availability []*fileproto.BlockAvailability, err error)
-}
-
-func NewBadgerStorage(db *badger.DB) FileStore {
-	return &badgerStorage{db: db}
-}
-
-type badgerStorage struct {
-	db *badger.DB
-}
-
-func (f *badgerStorage) Get(ctx context.Context, k cid.Cid) (b blocks.Block, err error) {
-	err = f.db.View(func(txn *badger.Txn) (e error) {
-		it, e := txn.Get(key(k))
-		if e != nil {
-			return e
-		}
-		if b, e = blockFromItem(it); e != nil {
-			return e
-		}
-		return
-	})
-	if err == badger.ErrKeyNotFound {
-		err = &format.ErrNotFound{Cid: k}
-	}
-	return
-}
-
-func (f *badgerStorage) GetMany(ctx context.Context, ks []cid.Cid) <-chan blocks.Block {
-	var res = make(chan blocks.Block)
-	go func() {
-		defer close(res)
-		_ = f.db.View(func(txn *badger.Txn) error {
-			// TODO: log errors
-			for _, k := range ks {
-				it, gerr := txn.Get(key(k))
-				if gerr != nil {
-					return gerr
-				}
-				b, berr := blockFromItem(it)
-				if berr != nil {
-					return berr
-				}
-				res <- b
-			}
-			return nil
-		})
-	}()
-	return res
-}
-
-func (f *badgerStorage) Add(ctx context.Context, bs []blocks.Block) error {
-	return f.db.Update(func(txn *badger.Txn) error {
-		for _, b := range bs {
-			if err := txn.Set(key(b.Cid()), b.RawData()); err != nil {
-				return err
-			}
-		}
-		return nil
-	})
-}
-
-func (f *badgerStorage) Delete(ctx context.Context, c cid.Cid) error {
-	return f.db.Update(func(txn *badger.Txn) error {
-		return txn.Delete(key(c))
-	})
-}
-
-func (f *badgerStorage) ExistsCids(ctx context.Context, ks []cid.Cid) (exists []cid.Cid, err error) {
-	err = f.db.View(func(txn *badger.Txn) error {
-		for _, k := range ks {
-			_, e := txn.Get(key(k))
-			if e == nil {
-				exists = append(exists, k)
-			} else if e != badger.ErrKeyNotFound {
-				return e
-			}
-		}
-		return nil
-	})
-	return
-}
-
-func (f *badgerStorage) BlockAvailability(ctx context.Context, ks []cid.Cid) (availability []*fileproto.BlockAvailability, err error) {
-	err = f.db.View(func(txn *badger.Txn) error {
-		for _, k := range ks {
-			_, e := txn.Get(key(k))
-			if e == nil {
-				availability = append(availability, &fileproto.BlockAvailability{
-					Cid:    k.Bytes(),
-					Status: fileproto.AvailabilityStatus_Exists,
-				})
-			} else if e != badger.ErrKeyNotFound {
-				return e
-			} else {
-				availability = append(availability, &fileproto.BlockAvailability{
-					Cid:    k.Bytes(),
-					Status: fileproto.AvailabilityStatus_NotExists,
-				})
-			}
-		}
-		return nil
-	})
-	return
-}
-
-func (f *badgerStorage) NotExistsBlocks(ctx context.Context, bs []blocks.Block) (notExists []blocks.Block, err error) {
-	notExists = bs[:0]
-	err = f.db.View(func(txn *badger.Txn) error {
-		for _, b := range bs {
-			_, e := txn.Get(key(b.Cid()))
-			if e == badger.ErrKeyNotFound {
-				notExists = append(notExists, b)
-			} else if e != nil {
-				return e
-			}
-		}
-		return nil
-	})
-	return
-}
-
-func key(c cid.Cid) []byte {
-	return []byte(keyPrefix + c.String())
-}
-
-func parseCID(key []byte) (cid.Cid, error) {
-	if len(key) <= len(keyPrefix) {
-		return cid.Cid{}, errInvalidKey
-	}
-	return cid.Decode(string(key[len(keyPrefix):]))
-}
-
-func blockFromItem(it *badger.Item) (b blocks.Block, err error) {
-	c, err := parseCID(it.Key())
-	if err != nil {
-		return nil, err
-	}
-	if err = it.Value(func(val []byte) error {
-		if b, err = blocks.NewBlockWithCid(val, c); err != nil {
-			return err
-		}
-		return nil
-	}); err != nil {
-		return
-	}
-	return
-}

core/filestorage/badgerfilestore/badgerfilestore_test.go

@@ -1,101 +0,0 @@
-package badgerfilestore
-
-import (
-	"context"
-	"github.com/dgraph-io/badger/v3"
-	blocks "github.com/ipfs/go-block-format"
-	"github.com/ipfs/go-cid"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-	"os"
-	"testing"
-	"time"
-)
-
-var ctx = context.Background()
-
-func TestBadgerStorage_Add(t *testing.T) {
-	tmpDir, err := os.MkdirTemp("", "badgerfilestore_*")
-	require.NoError(t, err)
-	defer os.RemoveAll(tmpDir)
-	db, err := badger.Open(badger.DefaultOptions(tmpDir))
-	require.NoError(t, err)
-	defer db.Close()
-
-	s := NewBadgerStorage(db)
-	bs := []blocks.Block{
-		blocks.NewBlock([]byte("1")),
-		blocks.NewBlock([]byte("2")),
-		blocks.NewBlock([]byte("3")),
-	}
-	assert.NoError(t, s.Add(ctx, bs))
-
-}
-
-func TestBadgerStorage_Get(t *testing.T) {
-	tmpDir, err := os.MkdirTemp("", "badgerfilestore_*")
-	require.NoError(t, err)
-	defer os.RemoveAll(tmpDir)
-	db, err := badger.Open(badger.DefaultOptions(tmpDir))
-	require.NoError(t, err)
-	defer db.Close()
-
-	s := NewBadgerStorage(db)
-	bs := []blocks.Block{
-		blocks.NewBlock([]byte("1")),
-		blocks.NewBlock([]byte("2")),
-		blocks.NewBlock([]byte("3")),
-	}
-	require.NoError(t, s.Add(ctx, bs))
-
-	cids := make([]cid.Cid, 0, len(bs))
-	for _, b := range bs {
-		cids = append(cids, b.Cid())
-	}
-	cids = append(cids, blocks.NewBlock([]byte("4")).Cid())
-
-	b, err := s.Get(ctx, bs[0].Cid())
-	require.NoError(t, err)
-	assert.Equal(t, bs[0].RawData(), b.RawData())
-
-	ecids, err := s.ExistsCids(ctx, cids)
-	require.NoError(t, err)
-	assert.Len(t, ecids, 3)
-}
-
-func TestBadgerStorage_GetMany(t *testing.T) {
-	tmpDir, err := os.MkdirTemp("", "badgerfilestore_*")
-	require.NoError(t, err)
-	defer os.RemoveAll(tmpDir)
-	db, err := badger.Open(badger.DefaultOptions(tmpDir))
-	require.NoError(t, err)
-	defer db.Close()
-
-	s := NewBadgerStorage(db)
-	bs := []blocks.Block{
-		blocks.NewBlock([]byte("1")),
-		blocks.NewBlock([]byte("2")),
-		blocks.NewBlock([]byte("3")),
-	}
-	require.NoError(t, s.Add(ctx, bs))
-
-	cids := make([]cid.Cid, 0, len(bs))
-	for _, b := range bs {
-		cids = append(cids, b.Cid())
-	}
-
-	res := s.GetMany(ctx, cids)
-	var resB []blocks.Block
-	for i := 0; i < len(bs); i++ {
-		select {
-		case b := <-res:
-			resB = append(resB, b)
-		case <-time.After(time.Second):
-			t.Error("timeout")
-			return
-		}
-	}
-	assert.Len(t, resB, 3)
-	_, ok := <-res
-	assert.False(t, ok)
-}

core/filestorage/fileservice.go

@@ -2,19 +2,21 @@ package filestorage
 
 import (
 	"context"
+	"fmt"
+
 	"github.com/anytypeio/any-sync/app"
 	"github.com/anytypeio/any-sync/app/logger"
 	"github.com/anytypeio/any-sync/commonfile/fileblockstore"
 	"github.com/anytypeio/any-sync/commonfile/fileproto"
 	"github.com/anytypeio/any-sync/commonspace/spacestorage"
 	"github.com/anytypeio/any-sync/net/rpc/server"
-	"github.com/anytypeio/go-anytype-middleware/core/filestorage/badgerfilestore"
+	blocks "github.com/ipfs/go-block-format"
+	"github.com/ipfs/go-cid"
+
 	"github.com/anytypeio/go-anytype-middleware/core/filestorage/rpcstore"
 	"github.com/anytypeio/go-anytype-middleware/pkg/lib/datastore"
 	"github.com/anytypeio/go-anytype-middleware/space"
 	"github.com/anytypeio/go-anytype-middleware/space/storage"
-	blocks "github.com/ipfs/go-block-format"
-	"github.com/ipfs/go-cid"
 )
 
 const CName = fileblockstore.CName
@@ -31,7 +33,9 @@ type FileStorage interface {
 }
 
 type fileStorage struct {
+	// TODO use cache proxy directly
 	fileblockstore.BlockStoreLocal
+
 	syncer       *syncer
 	syncerCancel context.CancelFunc
 	provider     datastore.Datastore
@@ -63,12 +67,15 @@ func (f *fileStorage) Run(ctx context.Context) (err error) {
 	if err != nil {
 		return
 	}
-	bs := badgerfilestore.NewBadgerStorage(db)
+	bs, err := newFlatStore()
+	if err != nil {
+		return fmt.Errorf("flatstore: %w", err)
+	}
 	f.handler.store = bs
 	ps := &proxyStore{
 		cache:  bs,
 		origin: f.rpcStore.NewStore(),
-		index:  badgerfilestore.NewFileBadgerIndex(db),
+		index:  NewFileBadgerIndex(db),
 	}
 	f.BlockStoreLocal = ps
 	f.syncer = &syncer{ps: ps, done: make(chan struct{})}

core/filestorage/flatstore.go

@@ -0,0 +1,116 @@
+package filestorage
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	"github.com/anytypeio/any-sync/commonfile/fileproto"
+	blocks "github.com/ipfs/go-block-format"
+	"github.com/ipfs/go-cid"
+	"github.com/ipfs/go-datastore"
+	flatfs "github.com/ipfs/go-ds-flatfs"
+	format "github.com/ipfs/go-ipld-format"
+)
+
+type flatStore struct {
+	ds *flatfs.Datastore
+}
+
+func newFlatStore() (*flatStore, error) {
+	ds, err := flatfs.CreateOrOpen("tmp/blocks/", flatfs.NextToLast(2), false)
+	if err != nil {
+		return nil, err
+	}
+	return &flatStore{ds: ds}, nil
+}
+
+func (f *flatStore) Get(ctx context.Context, k cid.Cid) (blocks.Block, error) {
+	raw, err := f.ds.Get(ctx, dskey(k))
+	if err == datastore.ErrNotFound {
+		return nil, &format.ErrNotFound{Cid: k}
+	}
+	if err != nil {
+		return nil, err
+	}
+
+	return blocks.NewBlockWithCid(raw, k)
+}
+
+func (f *flatStore) GetMany(ctx context.Context, ks []cid.Cid) <-chan blocks.Block {
+	ch := make(chan blocks.Block)
+	go func() {
+		defer close(ch)
+		for _, k := range ks {
+			b, err := f.Get(ctx, k)
+			if err != nil {
+				// TODO proper logging
+				fmt.Println("GetMany: ", k, err)
+				continue
+			}
+			ch <- b
+		}
+	}()
+	return ch
+}
+
+func dskey(c cid.Cid) datastore.Key {
+	return datastore.NewKey(strings.ToUpper(c.String()))
+}
+
+func (f *flatStore) Add(ctx context.Context, bs []blocks.Block) error {
+	for _, b := range bs {
+		if err := f.ds.Put(ctx, dskey(b.Cid()), b.RawData()); err != nil {
+			return fmt.Errorf("put %s: %w", dskey(b.Cid()), err)
+		}
+	}
+	return nil
+}
+
+func (f *flatStore) Delete(ctx context.Context, c cid.Cid) error {
+	return f.ds.Delete(ctx, dskey(c))
+}
+
+func (f *flatStore) ExistsCids(ctx context.Context, ks []cid.Cid) (exist []cid.Cid, err error) {
+	for _, k := range ks {
+		ok, err := f.ds.Has(ctx, dskey(k))
+		if err != nil {
+			return nil, err
+		}
+		if ok {
+			exist = append(exist, k)
+		}
+	}
+	return
+}
+
+func (f *flatStore) NotExistsBlocks(ctx context.Context, bs []blocks.Block) (notExist []blocks.Block, err error) {
+	for _, b := range bs {
+		ok, err := f.ds.Has(ctx, dskey(b.Cid()))
+		if err != nil {
+			return nil, err
+		}
+		if !ok {
+			notExist = append(notExist, b)
+		}
+	}
+	return
+}
+
+func (f *flatStore) BlockAvailability(ctx context.Context, ks []cid.Cid) (availability []*fileproto.BlockAvailability, err error) {
+	for _, k := range ks {
+		ok, err := f.ds.Has(ctx, dskey(k))
+		if err != nil {
+			return nil, err
+		}
+		status := fileproto.AvailabilityStatus_NotExists
+		if ok {
+			status = fileproto.AvailabilityStatus_Exists
+		}
+		availability = append(availability, &fileproto.BlockAvailability{
+			Cid:    k.Bytes(),
+			Status: status,
+		})
+	}
+	return
+}

core/filestorage/proxystore.go

@@ -2,20 +2,21 @@ package filestorage
 
 import (
 	"context"
+	"io"
+
 	"github.com/anytypeio/any-sync/commonfile/fileblockstore"
-	"github.com/anytypeio/go-anytype-middleware/core/filestorage/badgerfilestore"
-	"github.com/anytypeio/go-anytype-middleware/core/filestorage/rpcstore"
 	blocks "github.com/ipfs/go-block-format"
 	"github.com/ipfs/go-cid"
 	format "github.com/ipfs/go-ipld-format"
 	"go.uber.org/zap"
-	"io"
+
+	"github.com/anytypeio/go-anytype-middleware/core/filestorage/rpcstore"
 )
 
 type proxyStore struct {
 	cache  fileblockstore.BlockStoreLocal
 	origin rpcstore.RpcStore
-	index  *badgerfilestore.FileBadgerIndex
+	index  *FileBadgerIndex
 }
 
 func (c *proxyStore) Get(ctx context.Context, k cid.Cid) (b blocks.Block, err error) {
@@ -106,10 +107,10 @@ func (c *proxyStore) Add(ctx context.Context, bs []blocks.Block) (err error) {
 	if err = c.cache.Add(ctx, bs); err != nil {
 		return
 	}
-	indexCids := badgerfilestore.NewCids()
+	indexCids := NewCids()
 	defer indexCids.Release()
 	for _, b := range bs {
-		indexCids.Add(fileblockstore.CtxGetSpaceId(ctx), badgerfilestore.OpAdd, b.Cid())
+		indexCids.Add(fileblockstore.CtxGetSpaceId(ctx), OpAdd, b.Cid())
 	}
 	return c.index.Add(indexCids)
 }
@@ -118,9 +119,9 @@ func (c *proxyStore) Delete(ctx context.Context, k cid.Cid) error {
 	if err := c.cache.Delete(ctx, k); err != nil {
 		return err
 	}
-	indexCids := badgerfilestore.NewCids()
+	indexCids := NewCids()
 	defer indexCids.Release()
-	indexCids.Add(fileblockstore.CtxGetSpaceId(ctx), badgerfilestore.OpDelete, k)
+	indexCids.Add(fileblockstore.CtxGetSpaceId(ctx), OpDelete, k)
 	return c.index.Add(indexCids)
 }
 

core/filestorage/proxystore_test.go

@@ -3,17 +3,17 @@ package filestorage
 import (
 	"context"
 	"fmt"
-	"github.com/anytypeio/go-anytype-middleware/core/filestorage/badgerfilestore"
+	"os"
+	"sync"
+	"testing"
+	"time"
+
 	"github.com/dgraph-io/badger/v3"
 	blocks "github.com/ipfs/go-block-format"
 	"github.com/ipfs/go-cid"
 	format "github.com/ipfs/go-ipld-format"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	"os"
-	"sync"
-	"testing"
-	"time"
 )
 
 var ctx = context.Background()
@@ -274,7 +274,7 @@ func newPSFixture(t *testing.T) *psFixture {
 	fx.proxyStore = &proxyStore{
 		cache:  newTestStore(nil),
 		origin: newTestStore(nil),
-		index:  badgerfilestore.NewFileBadgerIndex(fx.db),
+		index:  NewFileBadgerIndex(fx.db),
 	}
 	return fx
 }

core/filestorage/rpchandler.go

@@ -3,18 +3,19 @@ package filestorage
 import (
 	"context"
 	"fmt"
+
 	"github.com/anytypeio/any-sync/commonfile/fileblockstore"
 	"github.com/anytypeio/any-sync/commonfile/fileproto"
 	"github.com/anytypeio/any-sync/commonfile/fileproto/fileprotoerr"
-	"github.com/anytypeio/go-anytype-middleware/core/filestorage/badgerfilestore"
-	"github.com/anytypeio/go-anytype-middleware/space/storage"
 	blocks "github.com/ipfs/go-block-format"
 	"github.com/ipfs/go-cid"
 	"go.uber.org/zap"
+
+	"github.com/anytypeio/go-anytype-middleware/space/storage"
 )
 
 type rpcHandler struct {
-	store        badgerfilestore.FileStore
+	store        *flatStore
 	spaceStorage storage.ClientStorage
 }
 

core/filestorage/syncer.go

@@ -3,14 +3,14 @@ package filestorage
 import (
 	"context"
 	"fmt"
-	"github.com/anytypeio/any-sync/commonfile/fileblockstore"
-	"github.com/anytypeio/go-anytype-middleware/core/filestorage/badgerfilestore"
-	blocks "github.com/ipfs/go-block-format"
-	"github.com/ipfs/go-cid"
-	"go.uber.org/zap"
 	"sync"
 	"sync/atomic"
 	"time"
+
+	"github.com/anytypeio/any-sync/commonfile/fileblockstore"
+	blocks "github.com/ipfs/go-block-format"
+	"github.com/ipfs/go-cid"
+	"go.uber.org/zap"
 )
 
 const syncerOpBatch = 10
@@ -55,21 +55,21 @@ func (s *syncer) sync(ctx context.Context) (doneCount int32) {
 	for _, sOps := range cids.SpaceOps {
 		if len(sOps.Load) > 0 {
 			wg.Add(1)
-			go func(opt badgerfilestore.SpaceCidOps) {
+			go func(opt SpaceCidOps) {
 				defer wg.Done()
 				doneAtomic.Add(s.load(ctx, opt))
 			}(sOps)
 		}
 		if len(sOps.Delete) > 0 {
 			wg.Add(1)
-			go func(opt badgerfilestore.SpaceCidOps) {
+			go func(opt SpaceCidOps) {
 				defer wg.Done()
 				doneAtomic.Add(s.delete(ctx, opt))
 			}(sOps)
 		}
 		if len(sOps.Add) > 0 {
 			wg.Add(1)
-			go func(opt badgerfilestore.SpaceCidOps) {
+			go func(opt SpaceCidOps) {
 				defer wg.Done()
 				doneAtomic.Add(s.add(ctx, opt))
 			}(sOps)
@@ -79,17 +79,17 @@ func (s *syncer) sync(ctx context.Context) (doneCount int32) {
 	return doneAtomic.Load()
 }
 
-func (s *syncer) load(ctx context.Context, spaceOps badgerfilestore.SpaceCidOps) (doneCount int32) {
+func (s *syncer) load(ctx context.Context, spaceOps SpaceCidOps) (doneCount int32) {
 	ctx = fileblockstore.CtxWithSpaceId(ctx, spaceOps.SpaceId)
 	res := s.ps.origin.GetMany(ctx, spaceOps.Load)
-	doneCids := badgerfilestore.NewCids()
+	doneCids := NewCids()
 	defer doneCids.Release()
 	for b := range res {
 		if err := s.ps.cache.Add(ctx, []blocks.Block{b}); err != nil {
 			log.Error("syncer: can't add to local store", zap.Error(err))
 			continue
 		}
-		doneCids.Add(spaceOps.SpaceId, badgerfilestore.OpLoad, b.Cid())
+		doneCids.Add(spaceOps.SpaceId, OpLoad, b.Cid())
 	}
 	if err := s.ps.index.Done(doneCids); err != nil {
 		log.Error("syncer: index.Done error", zap.Error(err))
@@ -100,8 +100,8 @@ func (s *syncer) load(ctx context.Context, spaceOps badgerfilestore.SpaceCidOps)
 	return
 }
 
-func (s *syncer) add(ctx context.Context, spaceOps badgerfilestore.SpaceCidOps) (doneCount int32) {
-	doneCids := badgerfilestore.NewCids()
+func (s *syncer) add(ctx context.Context, spaceOps SpaceCidOps) (doneCount int32) {
+	doneCids := NewCids()
 	defer doneCids.Release()
 	res := s.ps.cache.GetMany(ctx, spaceOps.Add)
 	var bs []blocks.Block
@@ -112,7 +112,7 @@ func (s *syncer) add(ctx context.Context, spaceOps badgerfilestore.SpaceCidOps)
 
 	successCidsCh := s.ps.origin.AddAsync(ctx, bs)
 	for doneCid := range successCidsCh {
-		doneCids.Add(spaceOps.SpaceId, badgerfilestore.OpAdd, doneCid)
+		doneCids.Add(spaceOps.SpaceId, OpAdd, doneCid)
 	}
 
 	doneCount = int32(doneCids.Len())
@@ -128,14 +128,14 @@ func (s *syncer) add(ctx context.Context, spaceOps badgerfilestore.SpaceCidOps)
 	return
 }
 
-func (s *syncer) delete(ctx context.Context, spaceOps badgerfilestore.SpaceCidOps) (doneCount int32) {
-	doneCids := badgerfilestore.NewCids()
+func (s *syncer) delete(ctx context.Context, spaceOps SpaceCidOps) (doneCount int32) {
+	doneCids := NewCids()
 	defer doneCids.Release()
 	ctx = fileblockstore.CtxWithSpaceId(ctx, spaceOps.SpaceId)
 	cids := make([]cid.Cid, len(spaceOps.Delete))
 	for i := range spaceOps.Delete {
 		cids[i] = spaceOps.Delete[i]
-		doneCids.Add(spaceOps.SpaceId, badgerfilestore.OpDelete, spaceOps.Delete[i])
+		doneCids.Add(spaceOps.SpaceId, OpDelete, spaceOps.Delete[i])
 	}
 	log.Debug(fmt.Sprintf("cids: %v", cids))
 	if err := s.ps.origin.DeleteMany(ctx, cids...); err != nil {

go.mod

@@ -111,6 +111,7 @@ require (
 	github.com/RoaringBitmap/roaring v0.9.4 // indirect
 	github.com/alecthomas/jsonschema v0.0.0-20191017121752-4bb6e3fae4f2 // indirect
 	github.com/alecthomas/units v0.0.0-20210927113745-59d0afb8317a // indirect
+	github.com/alexbrainman/goissue34681 v0.0.0-20191006012335-3fc7a47baff5 // indirect
 	github.com/andybalholm/cascadia v1.3.1 // indirect
 	github.com/anytypeio/go-chash v0.0.2 // indirect
 	github.com/benbjohnson/clock v1.3.0 // indirect
@@ -181,6 +182,7 @@ require (
 	github.com/ipfs/go-bitswap v0.11.0 // indirect
 	github.com/ipfs/go-blockservice v0.5.0 // indirect
 	github.com/ipfs/go-cidutil v0.1.0 // indirect
+	github.com/ipfs/go-ds-flatfs v0.5.1 // indirect
 	github.com/ipfs/go-fetcher v1.6.1 // indirect
 	github.com/ipfs/go-ipfs-chunker v0.0.5 // indirect
 	github.com/ipfs/go-ipfs-delay v0.0.1 // indirect

go.sum

@@ -75,6 +75,8 @@ github.com/alecthomas/units v0.0.0-20190717042225-c3de453c63f4/go.mod h1:ybxpYRF
 github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d/go.mod h1:rBZYJk541a8SKzHPHnH3zbiI+7dagKZ0cgpgrD7Fyho=
 github.com/alecthomas/units v0.0.0-20210927113745-59d0afb8317a h1:E/8AP5dFtMhl5KPJz66Kt9G0n+7Sn41Fy1wv9/jHOrc=
 github.com/alecthomas/units v0.0.0-20210927113745-59d0afb8317a/go.mod h1:OMCwj8VM1Kc9e19TLln2VL61YJF0x1XFtfdL4JdbSyE=
+github.com/alexbrainman/goissue34681 v0.0.0-20191006012335-3fc7a47baff5 h1:iW0a5ljuFxkLGPNem5Ui+KBjFJzKg4Fv2fnxe4dvzpM=
+github.com/alexbrainman/goissue34681 v0.0.0-20191006012335-3fc7a47baff5/go.mod h1:Y2QMoi1vgtOIfc+6DhrMOGkLoGzqSV2rKp4Sm+opsyA=
 github.com/andybalholm/cascadia v1.1.0/go.mod h1:GsXiBklL0woXo1j/WYWtSYYC4ouU9PqHO0sqidkEA4Y=
 github.com/andybalholm/cascadia v1.2.0/go.mod h1:YCyR8vOZT9aZ1CHEd8ap0gMVm2aFgxBp0T0eFw1RUQY=
 github.com/andybalholm/cascadia v1.3.1 h1:nhxRkql1kdYCc8Snf7D5/D3spOX+dBgjA6u8x004T2c=
@@ -628,6 +630,8 @@ github.com/ipfs/go-ds-badger v0.2.1/go.mod h1:Tx7l3aTph3FMFrRS838dcSJh+jjA7cX9Dr
 github.com/ipfs/go-ds-badger v0.2.3/go.mod h1:pEYw0rgg3FIrywKKnL+Snr+w/LjJZVMTBRn4FS6UHUk=
 github.com/ipfs/go-ds-badger v0.3.1-0.20211203181443-9205b3779632 h1:2bhtMNCuMSrD6iErf8f04V7/yFTjXIQBk5EHcHmcrCc=
 github.com/ipfs/go-ds-badger v0.3.1-0.20211203181443-9205b3779632/go.mod h1:1ke6mXNqeV8K3y5Ak2bAA0osoTfmxUdupVCGm4QUIek=
+github.com/ipfs/go-ds-flatfs v0.5.1 h1:ZCIO/kQOS/PSh3vcF1H6a8fkRGS7pOfwfPdx4n/KJH4=
+github.com/ipfs/go-ds-flatfs v0.5.1/go.mod h1:RWTV7oZD/yZYBKdbVIFXTX2fdY2Tbvl94NsWqmoyAX4=
 github.com/ipfs/go-ds-leveldb v0.0.1/go.mod h1:feO8V3kubwsEF22n0YRQCffeb79OOYIykR4L04tMOYc=
 github.com/ipfs/go-ds-leveldb v0.4.1/go.mod h1:jpbku/YqBSsBc1qgME8BkWS4AxzF2cEu1Ii2r79Hh9s=
 github.com/ipfs/go-ds-leveldb v0.4.2/go.mod h1:jpbku/YqBSsBc1qgME8BkWS4AxzF2cEu1Ii2r79Hh9s=