splitstore.go

package splitstore

import (
	"context"
	"errors"
	"os"
	"sync"
	"sync/atomic"
	"time"

	blocks "github.com/ipfs/go-block-format"
	"github.com/ipfs/go-cid"
	dstore "github.com/ipfs/go-datastore"
	ipld "github.com/ipfs/go-ipld-format"
	logging "github.com/ipfs/go-log/v2"
	"go.opencensus.io/stats"
	"go.uber.org/multierr"
	"golang.org/x/xerrors"

	"github.com/filecoin-project/go-state-types/abi"

	bstore "github.com/filecoin-project/lotus/blockstore"
	"github.com/filecoin-project/lotus/build"
	"github.com/filecoin-project/lotus/chain/stmgr"
	"github.com/filecoin-project/lotus/chain/types"
	"github.com/filecoin-project/lotus/metrics"
)

var (
	// baseEpochKey stores the base epoch (last compaction epoch) in the
	// metadata store.
	baseEpochKey = dstore.NewKey("/splitstore/baseEpoch")

	// warmupEpochKey stores whether a hot store warmup has been performed.
	// On first start, the splitstore will walk the state tree and will copy
	// all active blocks into the hotstore.
	warmupEpochKey = dstore.NewKey("/splitstore/warmupEpoch")

	// markSetSizeKey stores the current estimate for the mark set size.
	// this is first computed at warmup and updated in every compaction
	markSetSizeKey = dstore.NewKey("/splitstore/markSetSize")

	// compactionIndexKey stores the compaction index (serial number)
	compactionIndexKey = dstore.NewKey("/splitstore/compactionIndex")

	// stores the prune index (serial number)
	pruneIndexKey = dstore.NewKey("/splitstore/pruneIndex")

	// stores the base epoch of last prune in the metadata store
	pruneEpochKey = dstore.NewKey("/splitstore/pruneEpoch")

	log = logging.Logger("splitstore")

	errClosing = errors.New("splitstore is closing")

	// set this to true if you are debugging the splitstore to enable debug logging
	enableDebugLog = false
	// set this to true if you want to track origin stack traces in the write log
	enableDebugLogWriteTraces = false

	// upgradeBoundary is the boundary before and after an upgrade where we suppress compaction
	upgradeBoundary = build.Finality
)

type CompactType int

const (
	none CompactType = iota
	warmup
	hot
	cold
	check
)

func init() {
	if os.Getenv("LOTUS_SPLITSTORE_DEBUG_LOG") == "1" {
		enableDebugLog = true
	}

	if os.Getenv("LOTUS_SPLITSTORE_DEBUG_LOG_WRITE_TRACES") == "1" {
		enableDebugLogWriteTraces = true
	}
}

type Config struct {
	// MarkSetType is the type of mark set to use.
	//
	// The default value is "map", which uses an in-memory map-backed markset.
	// If you are constrained in memory (i.e. compaction runs out of memory), you
	// can use "badger", which will use a disk-backed markset using badger.
	// Note that compaction will take quite a bit longer when using the "badger" option,
	// but that shouldn't really matter (as long as it is under 7.5hrs).
	MarkSetType string

	// DiscardColdBlocks indicates whether to skip moving cold blocks to the coldstore.
	// If the splitstore is running with a noop coldstore then this option is set to true
	// which skips moving (as it is a noop, but still takes time to read all the cold objects)
	// and directly purges cold blocks.
	DiscardColdBlocks bool

	// HotstoreMessageRetention indicates the hotstore retention policy for messages.
	// It has the following semantics:
	// - a value of 0 will only retain messages within the compaction boundary (4 finalities)
	// - a positive integer indicates the number of finalities, outside the compaction boundary,
	//   for which messages will be retained in the hotstore.
	HotStoreMessageRetention uint64

	// HotstoreFullGCFrequency indicates how frequently (in terms of compactions) to garbage collect
	// the hotstore using full (moving) GC if supported by the hotstore.
	// A value of 0 disables full GC entirely.
	// A positive value is the number of compactions before a full GC is performed;
	// a value of 1 will perform full GC in every compaction.
	HotStoreFullGCFrequency uint64

	// EnableColdStoreAutoPrune turns on compaction of the cold store i.e. pruning
	// where hotstore compaction occurs every finality epochs pruning happens every 3 finalities
	// Default is false
	EnableColdStoreAutoPrune bool

	// ColdStoreFullGCFrequency specifies how often to performa a full (moving) GC on the coldstore.
	// Only applies if auto prune is enabled.  A value of 0 disables while a value of 1 will do
	// full GC in every prune.
	// Default is 7 (about once every a week)
	ColdStoreFullGCFrequency uint64

	// ColdStoreRetention specifies the retention policy for data reachable from the chain, in
	// finalities beyond the compaction boundary, default is 0, -1 retains everything
	ColdStoreRetention int64
}

// ChainAccessor allows the Splitstore to access the chain. It will most likely
// be a ChainStore at runtime.
type ChainAccessor interface {
	GetTipsetByHeight(context.Context, abi.ChainEpoch, *types.TipSet, bool) (*types.TipSet, error)
	GetHeaviestTipSet() *types.TipSet
	SubscribeHeadChanges(change func(revert []*types.TipSet, apply []*types.TipSet) error)
}

// upgradeRange is a precomputed epoch range during which we shouldn't compact so as to not
// interfere with an upgrade
type upgradeRange struct {
	start, end abi.ChainEpoch
}

// hotstore is the interface that must be satisfied by the hot blockstore; it is an extension
// of the Blockstore interface with the traits we need for compaction.
type hotstore interface {
	bstore.Blockstore
	bstore.BlockstoreIterator
}

type SplitStore struct {
	compacting  int32       // flag for when compaction is in progress
	compactType CompactType // compaction type, protected by compacting atomic, only meaningful when compacting == 1
	closing     int32       // the splitstore is closing

	cfg  *Config
	path string

	mx          sync.Mutex
	warmupEpoch abi.ChainEpoch // protected by mx
	baseEpoch   abi.ChainEpoch // protected by compaction lock
	pruneEpoch  abi.ChainEpoch // protected by compaction lock

	headChangeMx sync.Mutex

	chain ChainAccessor
	ds    dstore.Datastore
	cold  bstore.Blockstore
	hot   hotstore

	upgrades []upgradeRange

	markSetEnv  MarkSetEnv
	markSetSize int64

	compactionIndex int64
	pruneIndex      int64

	ctx    context.Context
	cancel func()

	debug *debugLog

	// transactional protection for concurrent read/writes during compaction
	txnLk           sync.RWMutex
	txnViewsMx      sync.Mutex
	txnViewsCond    sync.Cond
	txnViews        int
	txnViewsWaiting bool
	txnActive       bool
	txnRefsMx       sync.Mutex
	txnRefs         map[cid.Cid]struct{}
	txnMissing      map[cid.Cid]struct{}
	txnMarkSet      MarkSet
	txnSyncMx       sync.Mutex
	txnSyncCond     sync.Cond
	txnSync         bool

	// background cold object reification
	reifyWorkers    sync.WaitGroup
	reifyMx         sync.Mutex
	reifyCond       sync.Cond
	reifyPend       map[cid.Cid]struct{}
	reifyInProgress map[cid.Cid]struct{}

	// registered protectors
	protectors []func(func(cid.Cid) error) error
}

var _ bstore.Blockstore = (*SplitStore)(nil)

// Open opens an existing splistore, or creates a new splitstore. The splitstore
// is backed by the provided hot and cold stores. The returned SplitStore MUST be
// attached to the ChainStore with Start in order to trigger compaction.
func Open(path string, ds dstore.Datastore, hot, cold bstore.Blockstore, cfg *Config) (*SplitStore, error) {
	// hot blockstore must support the hotstore interface
	hots, ok := hot.(hotstore)
	if !ok {
		// be specific about what is missing
		if _, ok := hot.(bstore.BlockstoreIterator); !ok {
			return nil, xerrors.Errorf("hot blockstore does not support efficient iteration: %T", hot)
		}

		return nil, xerrors.Errorf("hot blockstore does not support the necessary traits: %T", hot)
	}

	// the markset env
	markSetEnv, err := OpenMarkSetEnv(path, cfg.MarkSetType)
	if err != nil {
		return nil, err
	}

	// and now we can make a SplitStore
	ss := &SplitStore{
		cfg:        cfg,
		path:       path,
		ds:         ds,
		cold:       cold,
		hot:        hots,
		markSetEnv: markSetEnv,
	}

	ss.txnViewsCond.L = &ss.txnViewsMx
	ss.txnSyncCond.L = &ss.txnSyncMx
	ss.ctx, ss.cancel = context.WithCancel(context.Background())

	ss.reifyCond.L = &ss.reifyMx
	ss.reifyPend = make(map[cid.Cid]struct{})
	ss.reifyInProgress = make(map[cid.Cid]struct{})

	if enableDebugLog {
		ss.debug, err = openDebugLog(path)
		if err != nil {
			return nil, err
		}
	}

	if ss.checkpointExists() {
		log.Info("found compaction checkpoint; resuming compaction")
		if err := ss.completeCompaction(); err != nil {
			markSetEnv.Close() //nolint:errcheck
			return nil, xerrors.Errorf("error resuming compaction: %w", err)
		}
	}
	if ss.pruneCheckpointExists() {
		log.Info("found prune checkpoint; resuming prune")
		if err := ss.completePrune(); err != nil {
			markSetEnv.Close() //nolint:errcheck
			return nil, xerrors.Errorf("error resuming prune: %w", err)
		}
	}

	return ss, nil
}

// Blockstore interface
func (s *SplitStore) DeleteBlock(_ context.Context, _ cid.Cid) error {
	// afaict we don't seem to be using this method, so it's not implemented
	return errors.New("DeleteBlock not implemented on SplitStore; don't do this Luke!") //nolint
}

func (s *SplitStore) DeleteMany(_ context.Context, _ []cid.Cid) error {
	// afaict we don't seem to be using this method, so it's not implemented
	return errors.New("DeleteMany not implemented on SplitStore; don't do this Luke!") //nolint
}

func (s *SplitStore) Has(ctx context.Context, cid cid.Cid) (bool, error) {
	if isIdentiyCid(cid) {
		return true, nil
	}

	s.txnLk.RLock()
	defer s.txnLk.RUnlock()

	// critical section
	if s.txnMarkSet != nil {
		has, err := s.txnMarkSet.Has(cid)
		if err != nil {
			return false, err
		}

		if has {
			return s.has(cid)
		}
		switch s.compactType {
		case hot:
			return s.cold.Has(ctx, cid)
		case cold:
			return s.hot.Has(ctx, cid)
		default:
			return false, xerrors.Errorf("invalid compaction type %d, only hot and cold allowed for critical section", s.compactType)
		}
	}

	has, err := s.hot.Has(ctx, cid)

	if err != nil {
		return has, err
	}

	if has {
		s.trackTxnRef(cid)
		return true, nil
	}

	has, err = s.cold.Has(ctx, cid)
	if has {
		s.trackTxnRef(cid)
		if bstore.IsHotView(ctx) {
			s.reifyColdObject(cid)
		}
	}

	return has, err

}

func (s *SplitStore) Get(ctx context.Context, cid cid.Cid) (blocks.Block, error) {
	if isIdentiyCid(cid) {
		data, err := decodeIdentityCid(cid)
		if err != nil {
			return nil, err
		}

		return blocks.NewBlockWithCid(data, cid)
	}

	s.txnLk.RLock()
	defer s.txnLk.RUnlock()

	// critical section
	if s.txnMarkSet != nil {
		has, err := s.txnMarkSet.Has(cid)
		if err != nil {
			return nil, err
		}

		if has {
			return s.get(cid)
		}
		switch s.compactType {
		case hot:
			return s.cold.Get(ctx, cid)
		case cold:
			return s.hot.Get(ctx, cid)
		default:
			return nil, xerrors.Errorf("invalid compaction type %d, only hot and cold allowed for critical section", s.compactType)
		}
	}

	blk, err := s.hot.Get(ctx, cid)

	switch {
	case err == nil:
		s.trackTxnRef(cid)
		return blk, nil

	case ipld.IsNotFound(err):
		if s.isWarm() {
			s.debug.LogReadMiss(cid)
		}

		blk, err = s.cold.Get(ctx, cid)
		if err == nil {
			s.trackTxnRef(cid)
			if bstore.IsHotView(ctx) {
				s.reifyColdObject(cid)
			}

			stats.Record(s.ctx, metrics.SplitstoreMiss.M(1))
		}
		return blk, err

	default:
		return nil, err
	}
}

func (s *SplitStore) GetSize(ctx context.Context, cid cid.Cid) (int, error) {
	if isIdentiyCid(cid) {
		data, err := decodeIdentityCid(cid)
		if err != nil {
			return 0, err
		}

		return len(data), nil
	}

	s.txnLk.RLock()
	defer s.txnLk.RUnlock()

	// critical section
	if s.txnMarkSet != nil {
		has, err := s.txnMarkSet.Has(cid)
		if err != nil {
			return 0, err
		}

		if has {
			return s.getSize(cid)
		}
		switch s.compactType {
		case hot:
			return s.cold.GetSize(ctx, cid)
		case cold:
			return s.hot.GetSize(ctx, cid)
		default:
			return 0, xerrors.Errorf("invalid compaction type %d, only hot and cold allowed for critical section", s.compactType)
		}
	}

	size, err := s.hot.GetSize(ctx, cid)

	switch {
	case err == nil:
		s.trackTxnRef(cid)
		return size, nil

	case ipld.IsNotFound(err):
		if s.isWarm() {
			s.debug.LogReadMiss(cid)
		}

		size, err = s.cold.GetSize(ctx, cid)
		if err == nil {
			s.trackTxnRef(cid)
			if bstore.IsHotView(ctx) {
				s.reifyColdObject(cid)
			}

			stats.Record(s.ctx, metrics.SplitstoreMiss.M(1))
		}
		return size, err

	default:
		return 0, err
	}
}

func (s *SplitStore) Put(ctx context.Context, blk blocks.Block) error {
	if isIdentiyCid(blk.Cid()) {
		return nil
	}

	s.txnLk.RLock()
	defer s.txnLk.RUnlock()

	err := s.hot.Put(ctx, blk)
	if err != nil {
		return err
	}

	s.debug.LogWrite(blk)

	// critical section
	if s.txnMarkSet != nil && s.compactType == hot { // puts only touch hot store
		s.markLiveRefs([]cid.Cid{blk.Cid()})
		return nil
	}
	s.trackTxnRef(blk.Cid())

	return nil
}

func (s *SplitStore) PutMany(ctx context.Context, blks []blocks.Block) error {
	// filter identites
	idcids := 0
	for _, blk := range blks {
		if isIdentiyCid(blk.Cid()) {
			idcids++
		}
	}

	if idcids > 0 {
		if idcids == len(blks) {
			// it's all identities
			return nil
		}

		filtered := make([]blocks.Block, 0, len(blks)-idcids)
		for _, blk := range blks {
			if isIdentiyCid(blk.Cid()) {
				continue
			}
			filtered = append(filtered, blk)
		}

		blks = filtered
	}

	batch := make([]cid.Cid, 0, len(blks))
	for _, blk := range blks {
		batch = append(batch, blk.Cid())
	}

	s.txnLk.RLock()
	defer s.txnLk.RUnlock()

	err := s.hot.PutMany(ctx, blks)
	if err != nil {
		return err
	}

	s.debug.LogWriteMany(blks)

	// critical section
	if s.txnMarkSet != nil && s.compactType == hot { // puts only touch hot store
		s.markLiveRefs(batch)
		return nil
	}
	s.trackTxnRefMany(batch)

	return nil
}

func (s *SplitStore) AllKeysChan(ctx context.Context) (<-chan cid.Cid, error) {
	ctx, cancel := context.WithCancel(ctx)

	chHot, err := s.hot.AllKeysChan(ctx)
	if err != nil {
		cancel()
		return nil, err
	}

	chCold, err := s.cold.AllKeysChan(ctx)
	if err != nil {
		cancel()
		return nil, err
	}

	seen := cid.NewSet()
	ch := make(chan cid.Cid, 8) // buffer is arbitrary, just enough to avoid context switches
	go func() {
		defer cancel()
		defer close(ch)

		for _, in := range []<-chan cid.Cid{chHot, chCold} {
			for c := range in {
				// ensure we only emit each key once
				if !seen.Visit(c) {
					continue
				}

				select {
				case ch <- c:
				case <-ctx.Done():
					return
				}
			}
		}
	}()

	return ch, nil
}

func (s *SplitStore) HashOnRead(enabled bool) {
	s.hot.HashOnRead(enabled)
	s.cold.HashOnRead(enabled)
}

func (s *SplitStore) View(ctx context.Context, cid cid.Cid, cb func([]byte) error) error {
	if isIdentiyCid(cid) {
		data, err := decodeIdentityCid(cid)
		if err != nil {
			return err
		}

		return cb(data)
	}

	// critical section
	s.txnLk.RLock() // the lock is released in protectView if we are not in critical section
	if s.txnMarkSet != nil {
		has, err := s.txnMarkSet.Has(cid)
		s.txnLk.RUnlock()

		if err != nil {
			return err
		}

		if has {
			return s.view(cid, cb)
		}
		switch s.compactType {
		case hot:
			return s.cold.View(ctx, cid, cb)
		case cold:
			return s.hot.View(ctx, cid, cb)
		default:
			return xerrors.Errorf("invalid compaction type %d, only hot and cold allowed for critical section", s.compactType)
		}
	}

	// views are (optimistically) protected two-fold:
	// - if there is an active transaction, then the reference is protected.
	// - if there is no active transaction, active views are tracked in a
	//   wait group and compaction is inhibited from starting until they
	//   have all completed.  this is necessary to ensure that a (very) long-running
	//   view can't have its data pointer deleted, which would be catastrophic.
	//   Note that we can't just RLock for the duration of the view, as this could
	//    lead to deadlock with recursive views.
	s.protectView(cid)
	defer s.viewDone()

	err := s.hot.View(ctx, cid, cb)
	if ipld.IsNotFound(err) {
		if s.isWarm() {
			s.debug.LogReadMiss(cid)
		}

		err = s.cold.View(ctx, cid, cb)
		if err == nil {
			if bstore.IsHotView(ctx) {
				s.reifyColdObject(cid)
			}

			stats.Record(s.ctx, metrics.SplitstoreMiss.M(1))
		}
		return err
	}
	return err
}

func (s *SplitStore) isWarm() bool {
	s.mx.Lock()
	defer s.mx.Unlock()
	return s.warmupEpoch > 0
}

// State tracking
func (s *SplitStore) Start(chain ChainAccessor, us stmgr.UpgradeSchedule) error {
	s.chain = chain
	curTs := chain.GetHeaviestTipSet()

	// precompute the upgrade boundaries
	s.upgrades = make([]upgradeRange, 0, len(us))
	for _, upgrade := range us {
		boundary := upgrade.Height
		for _, pre := range upgrade.PreMigrations {
			preMigrationBoundary := upgrade.Height - pre.StartWithin
			if preMigrationBoundary < boundary {
				boundary = preMigrationBoundary
			}
		}

		upgradeStart := boundary - upgradeBoundary
		upgradeEnd := upgrade.Height + upgradeBoundary

		s.upgrades = append(s.upgrades, upgradeRange{start: upgradeStart, end: upgradeEnd})
	}

	// should we warmup
	warmup := false

	// load base epoch from metadata ds
	// if none, then use current epoch because it's a fresh start
	bs, err := s.ds.Get(s.ctx, baseEpochKey)
	switch err {
	case nil:
		s.baseEpoch = bytesToEpoch(bs)

	case dstore.ErrNotFound:
		if curTs == nil {
			// this can happen in some tests
			break
		}

		err = s.setBaseEpoch(curTs.Height())
		if err != nil {
			return xerrors.Errorf("error saving base epoch: %w", err)
		}

	default:
		return xerrors.Errorf("error loading base epoch: %w", err)
	}

	// load prune epoch from metadata ds
	bs, err = s.ds.Get(s.ctx, pruneEpochKey)
	switch err {
	case nil:
		s.pruneEpoch = bytesToEpoch(bs)
	case dstore.ErrNotFound:
		if curTs == nil {
			//this can happen in some tests
			break
		}
		if err := s.setPruneEpoch(curTs.Height()); err != nil {
			return xerrors.Errorf("error saving prune epoch: %w", err)
		}
	default:
		return xerrors.Errorf("error loading prune epoch: %w", err)
	}

	// load warmup epoch from metadata ds
	bs, err = s.ds.Get(s.ctx, warmupEpochKey)
	switch err {
	case nil:
		s.warmupEpoch = bytesToEpoch(bs)

	case dstore.ErrNotFound:
		warmup = true

	default:
		return xerrors.Errorf("error loading warmup epoch: %w", err)
	}

	// load markSetSize from metadata ds to provide a size hint for marksets
	bs, err = s.ds.Get(s.ctx, markSetSizeKey)
	switch err {
	case nil:
		s.markSetSize = bytesToInt64(bs)

	case dstore.ErrNotFound:
	default:
		return xerrors.Errorf("error loading mark set size: %w", err)
	}

	// load compactionIndex from metadata ds to provide a hint as to when to perform moving gc
	bs, err = s.ds.Get(s.ctx, compactionIndexKey)
	switch err {
	case nil:
		s.compactionIndex = bytesToInt64(bs)

	case dstore.ErrNotFound:
		// this is potentially an upgrade from splitstore v0; schedule a warmup as v0 has
		// some issues with hot references leaking into the coldstore.
		warmup = true
	default:
		return xerrors.Errorf("error loading compaction index: %w", err)
	}

	log.Infow("starting splitstore", "baseEpoch", s.baseEpoch, "warmupEpoch", s.warmupEpoch)

	if warmup {
		err = s.warmup(curTs)
		if err != nil {
			return xerrors.Errorf("error starting warmup: %w", err)
		}
	}

	// spawn the reifier
	go s.reifyOrchestrator()

	// watch the chain
	chain.SubscribeHeadChanges(s.HeadChange)

	return nil
}

func (s *SplitStore) AddProtector(protector func(func(cid.Cid) error) error) {
	s.mx.Lock()
	defer s.mx.Unlock()

	s.protectors = append(s.protectors, protector)
}

func (s *SplitStore) Close() error {
	if !atomic.CompareAndSwapInt32(&s.closing, 0, 1) {
		// already closing
		return nil
	}

	if atomic.LoadInt32(&s.compacting) == 1 {
		s.txnSyncMx.Lock()
		s.txnSync = true
		s.txnSyncCond.Broadcast()
		s.txnSyncMx.Unlock()

		log.Warn("close with ongoing compaction in progress; waiting for it to finish...")
		for atomic.LoadInt32(&s.compacting) == 1 {
			time.Sleep(time.Second)
		}
	}

	s.reifyCond.Broadcast()
	s.reifyWorkers.Wait()
	s.cancel()
	return multierr.Combine(s.markSetEnv.Close(), s.debug.Close())
}

func (s *SplitStore) checkClosing() error {
	if atomic.LoadInt32(&s.closing) == 1 {
		return xerrors.Errorf("splitstore is closing")
	}

	return nil
}

func (s *SplitStore) setBaseEpoch(epoch abi.ChainEpoch) error {
	s.baseEpoch = epoch
	return s.ds.Put(s.ctx, baseEpochKey, epochToBytes(epoch))
}

func (s *SplitStore) setPruneEpoch(epoch abi.ChainEpoch) error {
	s.pruneEpoch = epoch
	return s.ds.Put(s.ctx, pruneEpochKey, epochToBytes(epoch))
}