feat(share/eds): Rework accessor cache

go.mod

@@ -28,7 +28,6 @@ require (
 	github.com/golang/mock v1.6.0
 	github.com/gorilla/mux v1.8.0
 	github.com/hashicorp/go-retryablehttp v0.7.4
-	github.com/hashicorp/golang-lru v1.0.2
 	github.com/imdario/mergo v0.3.16
 	github.com/ipfs/boxo v0.12.0
 	github.com/ipfs/go-block-format v0.1.2
@@ -76,6 +75,8 @@ require (
 	google.golang.org/protobuf v1.31.0
 )
 
+require github.com/hashicorp/golang-lru v1.0.2 // indirect
+
 require (
 	cloud.google.com/go v0.110.6 // indirect
 	cloud.google.com/go/compute v1.23.0 // indirect
@@ -186,7 +187,7 @@ require (
 	github.com/hashicorp/go-safetemp v1.0.0 // indirect
 	github.com/hashicorp/go-version v1.6.0 // indirect
 	github.com/hashicorp/golang-lru/arc/v2 v2.0.5 // indirect
-	github.com/hashicorp/golang-lru/v2 v2.0.5 // indirect
+	github.com/hashicorp/golang-lru/v2 v2.0.5
 	github.com/hashicorp/hcl v1.0.0 // indirect
 	github.com/hdevalence/ed25519consensus v0.0.0-20220222234857-c00d1f31bab3 // indirect
 	github.com/holiman/uint256 v1.2.2-0.20230321075855-87b91420868c // indirect

nodebuilder/store_test.go

@@ -154,6 +154,7 @@ func TestStoreRestart(t *testing.T) {
 		require.NoError(t, err)
 		_, err = eds.ReadEDS(ctx, odsReader, h)
 		require.NoError(t, err)
+		require.NoError(t, edsReader.Close())
 	}
 }
 

share/eds/blockstore.go

@@ -5,14 +5,14 @@ import (
 	"errors"
 	"fmt"
 
-	"github.com/filecoin-project/dagstore"
 	bstore "github.com/ipfs/boxo/blockstore"
 	dshelp "github.com/ipfs/boxo/datastore/dshelp"
 	blocks "github.com/ipfs/go-block-format"
 	"github.com/ipfs/go-cid"
 	"github.com/ipfs/go-datastore"
-	"github.com/ipfs/go-datastore/namespace"
 	ipld "github.com/ipfs/go-ipld-format"
+
+	"github.com/celestiaorg/celestia-node/share/eds/cache"
 )
 
 var _ bstore.Blockstore = (*blockstore)(nil)
@@ -32,18 +32,9 @@ var (
 // implementation to allow for the blockstore operations to be routed to the underlying stores.
 type blockstore struct {
 	store *Store
-	cache *blockstoreCache
 	ds    datastore.Batching
 }
 
-func newBlockstore(store *Store, cache *blockstoreCache, ds datastore.Batching) *blockstore {
-	return &blockstore{
-		store: store,
-		cache: cache,
-		ds:    namespace.Wrap(ds, blockstoreCacheKey),
-	}
-}
-
 func (bs *blockstore) Has(ctx context.Context, cid cid.Cid) (bool, error) {
 	keys, err := bs.store.dgstr.ShardsContainingMultihash(ctx, cid.Hash())
 	if errors.Is(err, ErrNotFound) || errors.Is(err, ErrNotFoundInIndex) {
@@ -63,6 +54,11 @@ func (bs *blockstore) Has(ctx context.Context, cid cid.Cid) (bool, error) {
 
 func (bs *blockstore) Get(ctx context.Context, cid cid.Cid) (blocks.Block, error) {
 	blockstr, err := bs.getReadOnlyBlockstore(ctx, cid)
+	if err == nil {
+		defer closeAndLog("blockstore", blockstr)
+		return blockstr.Get(ctx, cid)
+	}
+
 	if errors.Is(err, ErrNotFound) || errors.Is(err, ErrNotFoundInIndex) {
 		k := dshelp.MultihashToDsKey(cid.Hash())
 		blockData, err := bs.ds.Get(ctx, k)
@@ -72,15 +68,18 @@ func (bs *blockstore) Get(ctx context.Context, cid cid.Cid) (blocks.Block, error
 		// nmt's GetNode expects an ipld.ErrNotFound when a cid is not found.
 		return nil, ipld.ErrNotFound{Cid: cid}
 	}
-	if err != nil {
-		log.Debugf("failed to get blockstore for cid %s: %s", cid, err)
-		return nil, err
-	}
-	return blockstr.Get(ctx, cid)
+
+	log.Debugf("failed to get blockstore for cid %s: %s", cid, err)
+	return nil, err
 }
 
 func (bs *blockstore) GetSize(ctx context.Context, cid cid.Cid) (int, error) {
 	blockstr, err := bs.getReadOnlyBlockstore(ctx, cid)
+	if err == nil {
+		defer closeAndLog("blockstore", blockstr)
+		return blockstr.GetSize(ctx, cid)
+	}
+
 	if errors.Is(err, ErrNotFound) || errors.Is(err, ErrNotFoundInIndex) {
 		k := dshelp.MultihashToDsKey(cid.Hash())
 		size, err := bs.ds.GetSize(ctx, k)
@@ -90,10 +89,9 @@ func (bs *blockstore) GetSize(ctx context.Context, cid cid.Cid) (int, error) {
 		// nmt's GetSize expects an ipld.ErrNotFound when a cid is not found.
 		return 0, ipld.ErrNotFound{Cid: cid}
 	}
-	if err != nil {
-		return 0, err
-	}
-	return blockstr.GetSize(ctx, cid)
+
+	log.Debugf("failed to get size for cid %s: %s", cid, err)
+	return 0, err
 }
 
 func (bs *blockstore) DeleteBlock(ctx context.Context, cid cid.Cid) error {
@@ -139,7 +137,7 @@ func (bs *blockstore) HashOnRead(bool) {
 }
 
 // getReadOnlyBlockstore finds the underlying blockstore of the shard that contains the given CID.
-func (bs *blockstore) getReadOnlyBlockstore(ctx context.Context, cid cid.Cid) (dagstore.ReadBlockstore, error) {
+func (bs *blockstore) getReadOnlyBlockstore(ctx context.Context, cid cid.Cid) (*BlockstoreCloser, error) {
 	keys, err := bs.store.dgstr.ShardsContainingMultihash(ctx, cid.Hash())
 	if errors.Is(err, datastore.ErrNotFound) || errors.Is(err, ErrNotFoundInIndex) {
 		return nil, ErrNotFound
@@ -148,11 +146,28 @@ func (bs *blockstore) getReadOnlyBlockstore(ctx context.Context, cid cid.Cid) (d
 		return nil, fmt.Errorf("failed to find shards containing multihash: %w", err)
 	}
 
-	// a share can exist in multiple EDSes, so just take the first one.
+	// check if cache contains any of accessors
 	shardKey := keys[0]
-	accessor, err := bs.store.getCachedAccessor(ctx, shardKey)
+	if accessor, err := bs.store.cache.Get(shardKey); err == nil {
+		return blockstoreCloser(accessor)
+	}
+
+	// load accessor to the cache and use it as blockstoreCloser
+	accessor, err := bs.store.cache.GetOrLoad(ctx, shardKey, bs.store.getAccessor)
 	if err != nil {
 		return nil, fmt.Errorf("failed to get accessor for shard %s: %w", shardKey, err)
 	}
-	return accessor.bs, nil
+	return blockstoreCloser(accessor)
+}
+
+// blockstoreCloser constructs new BlockstoreCloser from cache.Accessor
+func blockstoreCloser(ac cache.Accessor) (*BlockstoreCloser, error) {
+	bs, err := ac.Blockstore()
+	if err != nil {
+		return nil, fmt.Errorf("eds/store: failed to get blockstore: %w", err)
+	}
+	return &BlockstoreCloser{
+		ReadBlockstore: bs,
+		Closer:         ac,
+	}, nil
 }

share/eds/blockstore_test.go

@@ -37,6 +37,9 @@ func TestBlockstore_Operations(t *testing.T) {
 	topLevelBS := edsStore.Blockstore()
 	carBS, err := edsStore.CARBlockstore(ctx, dah.Hash())
 	require.NoError(t, err)
+	defer func() {
+		require.NoError(t, carBS.Close())
+	}()
 
 	root, err := edsStore.GetDAH(ctx, dah.Hash())
 	require.NoError(t, err)