lotus/blockstore/fallback.go

106 lines
2.6 KiB
Go
Raw Normal View History

package blockstore
import (
"context"
"sync"
"time"
blocks "github.com/ipfs/go-block-format"
"github.com/ipfs/go-cid"
2022-06-14 15:00:51 +00:00
"golang.org/x/xerrors"
)
// UnwrapFallbackStore takes a blockstore, and returns the underlying blockstore
// if it was a FallbackStore. Otherwise, it just returns the supplied store
// unmodified.
func UnwrapFallbackStore(bs Blockstore) (Blockstore, bool) {
if fbs, ok := bs.(*FallbackStore); ok {
return fbs.Blockstore, true
}
return bs, false
}
// FallbackStore is a read-through store that queries another (potentially
// remote) source if the block is not found locally. If the block is found
// during the fallback, it stores it in the local store.
type FallbackStore struct {
Blockstore
2021-01-29 23:24:44 +00:00
lk sync.RWMutex
// missFn is the function that will be invoked on a local miss to pull the
// block from elsewhere.
missFn func(context.Context, cid.Cid) (blocks.Block, error)
}
var _ Blockstore = (*FallbackStore)(nil)
2021-01-29 23:24:44 +00:00
func (fbs *FallbackStore) SetFallback(missFn func(context.Context, cid.Cid) (blocks.Block, error)) {
fbs.lk.Lock()
defer fbs.lk.Unlock()
2021-01-29 23:24:44 +00:00
fbs.missFn = missFn
}
func (fbs *FallbackStore) getFallback(c cid.Cid) (blocks.Block, error) {
log.Warnf("fallbackstore: block not found locally, fetching from the network; cid: %s", c)
fbs.lk.RLock()
defer fbs.lk.RUnlock()
2021-01-29 23:24:44 +00:00
if fbs.missFn == nil {
// FallbackStore wasn't configured yet (chainstore/bitswap aren't up yet)
// Wait for a bit and retry
fbs.lk.RUnlock()
time.Sleep(5 * time.Second)
fbs.lk.RLock()
2021-01-29 23:24:44 +00:00
if fbs.missFn == nil {
log.Errorw("fallbackstore: missFn not configured yet")
return nil, ErrNotFound
}
}
ctx, cancel := context.WithTimeout(context.TODO(), 120*time.Second)
defer cancel()
2021-01-29 23:24:44 +00:00
b, err := fbs.missFn(ctx, c)
if err != nil {
return nil, err
}
// chain bitswap puts blocks in temp blockstore which is cleaned up
// every few min (to drop any messages we fetched but don't want)
// in this case we want to keep this block around
if err := fbs.Put(ctx, b); err != nil {
return nil, xerrors.Errorf("persisting fallback-fetched block: %w", err)
}
return b, nil
}
func (fbs *FallbackStore) Get(ctx context.Context, c cid.Cid) (blocks.Block, error) {
b, err := fbs.Blockstore.Get(ctx, c)
switch err {
case nil:
return b, nil
case ErrNotFound:
return fbs.getFallback(c)
default:
return b, err
}
}
func (fbs *FallbackStore) GetSize(ctx context.Context, c cid.Cid) (int, error) {
sz, err := fbs.Blockstore.GetSize(ctx, c)
switch err {
case nil:
return sz, nil
case ErrNotFound:
b, err := fbs.getFallback(c)
if err != nil {
return 0, err
}
return len(b.RawData()), nil
default:
return sz, err
}
}