lotus/node/repo/memrepo.go

package repo

import (
	"context"
	"encoding/json"
	"os"
	"path/filepath"
	"sync"

	"github.com/google/uuid"
	"github.com/ipfs/go-datastore"
	"github.com/ipfs/go-datastore/namespace"
	dssync "github.com/ipfs/go-datastore/sync"
	"github.com/multiformats/go-multiaddr"
	"golang.org/x/xerrors"

	"github.com/filecoin-project/lotus/blockstore"
	"github.com/filecoin-project/lotus/chain/types"
	"github.com/filecoin-project/lotus/node/config"
	"github.com/filecoin-project/lotus/storage/sealer/fsutil"
	"github.com/filecoin-project/lotus/storage/sealer/storiface"
)

type MemRepo struct {
	api struct {
		sync.Mutex
		ma    multiaddr.Multiaddr
		token []byte
	}

	repoLock chan struct{}
	token    *byte

	datastore  datastore.Datastore
	keystore   map[string]types.KeyInfo
	blockstore blockstore.Blockstore

	sc      *storiface.StorageConfig
	tempDir string

	// holds the current config value
	config struct {
		sync.Mutex
		val interface{}
	}
}

type lockedMemRepo struct {
	mem *MemRepo
	t   RepoType
	sync.RWMutex

	token *byte
}

func (lmem *lockedMemRepo) RepoType() RepoType {
	return lmem.t
}

func (lmem *lockedMemRepo) GetStorage() (storiface.StorageConfig, error) {
	if err := lmem.checkToken(); err != nil {
		return storiface.StorageConfig{}, err
	}

	if lmem.mem.sc == nil {
		lmem.mem.sc = &storiface.StorageConfig{StoragePaths: []storiface.LocalPath{
			{Path: lmem.Path()},
		}}
	}

	return *lmem.mem.sc, nil
}

func (lmem *lockedMemRepo) SetStorage(c func(*storiface.StorageConfig)) error {
	if err := lmem.checkToken(); err != nil {
		return err
	}

	_, _ = lmem.GetStorage()

	c(lmem.mem.sc)
	return nil
}

func (lmem *lockedMemRepo) Stat(path string) (fsutil.FsStat, error) {
	return fsutil.Statfs(path)
}

func (lmem *lockedMemRepo) DiskUsage(path string) (int64, error) {
	si, err := fsutil.FileSize(path)
	if err != nil {
		return 0, err
	}
	return si.OnDisk, nil
}

func (lmem *lockedMemRepo) Path() string {
	lmem.Lock()
	defer lmem.Unlock()

	if lmem.mem.tempDir != "" {
		return lmem.mem.tempDir
	}

	t, err := os.MkdirTemp(os.TempDir(), "lotus-memrepo-temp-")
	if err != nil {
		panic(err) // only used in tests, probably fine
	}

	if _, ok := lmem.t.(SupportsStagingDeals); ok {
		// this is required due to the method makeDealStaging from cmd/lotus-storage-miner/init.go
		// deal-staging is the directory deal files are staged in before being sealed into sectors
		// for offline deal flow.
		if err := os.MkdirAll(filepath.Join(t, "deal-staging"), 0755); err != nil {
			panic(err)
		}
	}
	if lmem.t == StorageMiner || lmem.t == Worker {
		lmem.initSectorStore(t)
	}

	lmem.mem.tempDir = t
	return t
}

func (lmem *lockedMemRepo) initSectorStore(t string) {
	if err := config.WriteStorageFile(filepath.Join(t, fsStorageConfig), storiface.StorageConfig{
		StoragePaths: []storiface.LocalPath{
			{Path: t},
		}}); err != nil {
		panic(err)
	}

	b, err := json.MarshalIndent(&storiface.LocalStorageMeta{
		ID:       storiface.ID(uuid.New().String()),
		Weight:   10,
		CanSeal:  true,
		CanStore: true,
	}, "", "  ")
	if err != nil {
		panic(err)
	}

	if err := os.WriteFile(filepath.Join(t, "sectorstore.json"), b, 0644); err != nil {
		panic(err)
	}
}

var _ Repo = &MemRepo{}

// MemRepoOptions contains options for memory repo
type MemRepoOptions struct {
	Ds       datastore.Datastore
	KeyStore map[string]types.KeyInfo
}

// NewMemory creates new memory based repo with provided options.
// opts can be nil, it  will be replaced with defaults.
// Any field in opts can be nil, they will be replaced by defaults.
func NewMemory(opts *MemRepoOptions) *MemRepo {
	if opts == nil {
		opts = &MemRepoOptions{}
	}
	if opts.Ds == nil {
		opts.Ds = dssync.MutexWrap(datastore.NewMapDatastore())
	}
	if opts.KeyStore == nil {
		opts.KeyStore = make(map[string]types.KeyInfo)
	}

	return &MemRepo{
		repoLock:   make(chan struct{}, 1),
		blockstore: blockstore.WrapIDStore(blockstore.NewMemorySync()),
		datastore:  opts.Ds,
		keystore:   opts.KeyStore,
	}
}

func (mem *MemRepo) APIEndpoint() (multiaddr.Multiaddr, error) {
	mem.api.Lock()
	defer mem.api.Unlock()
	if mem.api.ma == nil {
		return nil, ErrNoAPIEndpoint
	}
	return mem.api.ma, nil
}

func (mem *MemRepo) APIToken() ([]byte, error) {
	mem.api.Lock()
	defer mem.api.Unlock()
	if mem.api.ma == nil {
		return nil, ErrNoAPIToken
	}
	return mem.api.token, nil
}

func (mem *MemRepo) Lock(t RepoType) (LockedRepo, error) {
	select {
	case mem.repoLock <- struct{}{}:
	default:
		return nil, ErrRepoAlreadyLocked
	}
	mem.token = new(byte)

	return &lockedMemRepo{
		mem:   mem,
		t:     t,
		token: mem.token,
	}, nil
}

func (mem *MemRepo) Cleanup() {
	mem.api.Lock()
	defer mem.api.Unlock()

	if mem.tempDir != "" {
		if err := os.RemoveAll(mem.tempDir); err != nil {
			log.Errorw("cleanup test memrepo", "error", err)
		}
		mem.tempDir = ""
	}
}

func (lmem *lockedMemRepo) Readonly() bool {
	return false
}

func (lmem *lockedMemRepo) checkToken() error {
	lmem.RLock()
	defer lmem.RUnlock()
	if lmem.mem.token != lmem.token {
		return ErrClosedRepo
	}
	return nil
}

func (lmem *lockedMemRepo) Close() error {
	if err := lmem.checkToken(); err != nil {
		return err
	}
	lmem.Lock()
	defer lmem.Unlock()

	if lmem.mem.token != lmem.token {
		return ErrClosedRepo
	}

	lmem.mem.token = nil
	lmem.mem.api.Lock()
	lmem.mem.api.ma = nil
	lmem.mem.api.Unlock()
	<-lmem.mem.repoLock // unlock
	return nil
}

func (lmem *lockedMemRepo) Datastore(_ context.Context, ns string) (datastore.Batching, error) {
	if err := lmem.checkToken(); err != nil {
		return nil, err
	}

	return namespace.Wrap(lmem.mem.datastore, datastore.NewKey(ns)), nil
}

func (lmem *lockedMemRepo) Blockstore(ctx context.Context, domain BlockstoreDomain) (blockstore.Blockstore, error) {
	if domain != UniversalBlockstore {
		return nil, ErrInvalidBlockstoreDomain
	}
	return lmem.mem.blockstore, nil
}

func (lmem *lockedMemRepo) SplitstorePath() (string, error) {
	splitstorePath := filepath.Join(lmem.Path(), "splitstore")
	if err := os.MkdirAll(splitstorePath, 0755); err != nil {
		return "", err
	}
	return splitstorePath, nil
}

func (lmem *lockedMemRepo) SqlitePath() (string, error) {
	sqlitePath := filepath.Join(lmem.Path(), "sqlite")
	if err := os.MkdirAll(sqlitePath, 0755); err != nil {
		return "", err
	}
	return sqlitePath, nil
}

func (lmem *lockedMemRepo) ListDatastores(ns string) ([]int64, error) {
	return nil, nil
}

func (lmem *lockedMemRepo) DeleteDatastore(ns string) error {
	/** poof **/
	return nil
}

func (lmem *lockedMemRepo) Config() (interface{}, error) {
	if err := lmem.checkToken(); err != nil {
		return nil, err
	}

	lmem.mem.config.Lock()
	defer lmem.mem.config.Unlock()

	if lmem.mem.config.val == nil {
		lmem.mem.config.val = lmem.t.Config()
	}

	return lmem.mem.config.val, nil
}

func (lmem *lockedMemRepo) SetConfig(c func(interface{})) error {
	if err := lmem.checkToken(); err != nil {
		return err
	}

	lmem.mem.config.Lock()
	defer lmem.mem.config.Unlock()

	if lmem.mem.config.val == nil {
		lmem.mem.config.val = lmem.t.Config()
	}

	c(lmem.mem.config.val)

	return nil
}

func (lmem *lockedMemRepo) SetAPIEndpoint(ma multiaddr.Multiaddr) error {
	if err := lmem.checkToken(); err != nil {
		return err
	}
	lmem.mem.api.Lock()
	lmem.mem.api.ma = ma
	lmem.mem.api.Unlock()
	return nil
}

func (lmem *lockedMemRepo) SetAPIToken(token []byte) error {
	if err := lmem.checkToken(); err != nil {
		return err
	}
	lmem.mem.api.Lock()
	lmem.mem.api.token = token
	lmem.mem.api.Unlock()
	return nil
}

func (lmem *lockedMemRepo) KeyStore() (types.KeyStore, error) {
	if err := lmem.checkToken(); err != nil {
		return nil, err
	}
	return lmem, nil
}

// Implement KeyStore on the same instance

// List lists all the keys stored in the KeyStore
func (lmem *lockedMemRepo) List() ([]string, error) {
	if err := lmem.checkToken(); err != nil {
		return nil, err
	}
	lmem.RLock()
	defer lmem.RUnlock()

	res := make([]string, 0, len(lmem.mem.keystore))
	for k := range lmem.mem.keystore {
		res = append(res, k)
	}
	return res, nil
}

// Get gets a key out of keystore and returns types.KeyInfo coresponding to named key
func (lmem *lockedMemRepo) Get(name string) (types.KeyInfo, error) {
	if err := lmem.checkToken(); err != nil {
		return types.KeyInfo{}, err
	}
	lmem.RLock()
	defer lmem.RUnlock()

	key, ok := lmem.mem.keystore[name]
	if !ok {
		return types.KeyInfo{}, xerrors.Errorf("getting key '%s': %w", name, types.ErrKeyInfoNotFound)
	}
	return key, nil
}

// Put saves key info under given name
func (lmem *lockedMemRepo) Put(name string, key types.KeyInfo) error {
	if err := lmem.checkToken(); err != nil {
		return err
	}
	lmem.Lock()
	defer lmem.Unlock()

	_, isThere := lmem.mem.keystore[name]
	if isThere {
		return xerrors.Errorf("putting key '%s': %w", name, types.ErrKeyExists)
	}

	lmem.mem.keystore[name] = key
	return nil
}

func (lmem *lockedMemRepo) Delete(name string) error {
	if err := lmem.checkToken(); err != nil {
		return err
	}
	lmem.Lock()
	defer lmem.Unlock()

	_, isThere := lmem.mem.keystore[name]
	if !isThere {
		return xerrors.Errorf("deleting key '%s': %w", name, types.ErrKeyInfoNotFound)
	}
	delete(lmem.mem.keystore, name)
	return nil
}