Filestore: add Provider option to provide filestore blocks. (#1042)

* Filestore: add Provider option to provide filestore blocks.

A Filestore is a Blockstore wrapper. When I moved providing resposibilities to Blockstore/Pinner/MFS... I forgot the Filestore. In Particular, the Filestore writes all blocks to the Blockstore except those of FilestoreNode type, which are serialized and written to the datastore directly via the Filemanager. The data serialized is not the block itself, but a pointer to a filesystem object and an offset.

So FilestoreNodes escaped providing even when the blockstore is configured to provide... Not anymore.

Author: hsanjuan

CHANGELOG.md

@@ -40,6 +40,7 @@ The following emojis are used to highlight certain changes:
 - `bitswap`: Updated to use `verifcid.DefaultMaxDigestSize` for `MaximumHashLength` constant
   - The default `MaximumAllowedCid` limit for incoming CIDs can be adjusted using `bitswap.MaxCidSize` or `server.MaxCidSize` options
 - 🛠 `bitswap/client`: The `RebroadcastDelay` option now takes a `time.Duration` value. This is a potentially BREAKING CHANGE. The time-varying functionality of `delay.Delay` was never used, so it was replaced with a fixed duration value. This also removes the `github.com/ipfs/go-ipfs-delay` dependency.
+- `filestore`: Support providing filestore-blocks. A new `provider.MultihashProvider` parameter has been added to `filestore.New()`. When used, the blocks handled by the Filestore's `FileManager` will be provided on write (Put and PutMany).
 
 ### Removed
 

blockstore/blockstore.go

@@ -250,11 +250,12 @@ func (bs *blockstore) PutMany(ctx context.Context, blocks []blocks.Block) error
 		return err
 	}
 
-	var hashes []multihash.Multihash
-	for _, block := range blocks {
-		hashes = append(hashes, block.Cid().Hash())
-	}
 	if bs.provider != nil {
+		var hashes []multihash.Multihash
+		for _, block := range blocks {
+			hashes = append(hashes, block.Cid().Hash())
+		}
+		logger.Debugf("blockstore: provide %d hashes", len(hashes))
 		if err := bs.provider.StartProviding(false, hashes...); err != nil {
 			logger.Warnf("blockstore: error while providing blocks: %s", err)
 		}

filestore/filestore.go

@@ -13,11 +13,13 @@ import (
 
 	blockstore "github.com/ipfs/boxo/blockstore"
 	posinfo "github.com/ipfs/boxo/filestore/posinfo"
+	"github.com/ipfs/boxo/provider"
 	blocks "github.com/ipfs/go-block-format"
 	cid "github.com/ipfs/go-cid"
 	dsq "github.com/ipfs/go-datastore/query"
 	ipld "github.com/ipfs/go-ipld-format"
 	logging "github.com/ipfs/go-log/v2"
+	"github.com/multiformats/go-multihash"
 )
 
 var logger = logging.Logger("filestore")
@@ -31,8 +33,9 @@ var (
 // to store regular blocks and a special Blockstore called
 // FileManager to store blocks which data exists in an external file.
 type Filestore struct {
-	fm *FileManager
-	bs blockstore.Blockstore
+	fm       *FileManager
+	bs       blockstore.Blockstore
+	provider provider.MultihashProvider
 }
 
 // FileManager returns the FileManager in Filestore.
@@ -45,9 +48,12 @@ func (f *Filestore) MainBlockstore() blockstore.Blockstore {
 	return f.bs
 }
 
-// NewFilestore creates one using the given Blockstore and FileManager.
-func NewFilestore(bs blockstore.Blockstore, fm *FileManager) *Filestore {
-	return &Filestore{fm, bs}
+// NewFilestore creates one using the given Blockstore and FileManager. An
+// optional MultihashProvider can be used to provide blocks written to the
+// FileManager (blocks written to the normal blockstore will be provided by it
+// if it has been initialized with the blockstore.Provider option).
+func NewFilestore(bs blockstore.Blockstore, fm *FileManager, prov provider.MultihashProvider) *Filestore {
+	return &Filestore{fm, bs, prov}
 }
 
 // AllKeysChan returns a channel from which to read the keys stored in
@@ -193,7 +199,17 @@ func (f *Filestore) Put(ctx context.Context, b blocks.Block) error {
 
 	switch b := b.(type) {
 	case *posinfo.FilestoreNode:
-		return f.fm.Put(ctx, b)
+		err = f.fm.Put(ctx, b)
+		if err != nil {
+			return err
+		}
+		if f.provider != nil {
+			logger.Debugf("filestore: provide %s", b.Cid())
+			if err := f.provider.StartProviding(false, b.Cid().Hash()); err != nil {
+				logger.Warnf("filestore: error while providing %s: %s", b.Cid(), err)
+			}
+		}
+		return nil
 	default:
 		return f.bs.Put(ctx, b)
 	}
@@ -235,6 +251,17 @@ func (f *Filestore) PutMany(ctx context.Context, bs []blocks.Block) error {
 		if err != nil {
 			return err
 		}
+
+		if f.provider != nil {
+			var hashes []multihash.Multihash
+			for _, n := range fstores {
+				hashes = append(hashes, n.Node.Cid().Hash())
+			}
+			logger.Debugf("filestore: provide %d hashes", len(hashes))
+			if err := f.provider.StartProviding(false, hashes...); err != nil {
+				logger.Warnf("filestore: error while providing hashes: %s", err)
+			}
+		}
 	}
 	return nil
 }

filestore/filestore_test.go

@@ -25,7 +25,7 @@ func newTestFilestore(t *testing.T, option ...Option) (string, *Filestore) {
 	fm.AllowFiles = true
 
 	bs := blockstore.NewBlockstore(mds)
-	fstore := NewFilestore(bs, fm)
+	fstore := NewFilestore(bs, fm, nil)
 	return testdir, fstore
 }