lotus/chain/store/splitstore/splitstore.go

package splitstore

import (
	"context"
	"encoding/binary"
	"errors"
	"sync"
	"sync/atomic"
	"time"

	"golang.org/x/xerrors"

	blocks "github.com/ipfs/go-block-format"
	cid "github.com/ipfs/go-cid"
	dstore "github.com/ipfs/go-datastore"
	logging "github.com/ipfs/go-log/v2"

	"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/store"
	"github.com/filecoin-project/lotus/chain/types"
)

// these are variable so that 1) lotus-soup builds 2) we can change them in tests
var (
	CompactionThreshold = 5 * build.Finality
	CompactionCold      = build.Finality
)

var baseEpochKey = dstore.NewKey("baseEpoch")

var log = logging.Logger("splitstore")

func init() {
	// TODO temporary for debugging purposes; to be removed for merge.
	logging.SetLogLevel("splitstore", "DEBUG")
}

type Config struct {
	// TrackingStore type; bolt (default) or lmdb
	TrackingStoreType string
	// LiveSet type; bloom (default), bolt, or lmdb
	LiveSetType string
	// perform full reachability analysis (expensive) for compaction
	// You should enable this option if you plan to use the splitstore without a backing coldstore
	EnableFullCompaction bool
	// EXPERIMENTAL enable pruning of unreachable objects.
	// This has not been sufficiently tested yet; only enable if you know what you are doing.
	// Only applies if you enable full compaction.
	EnableGC bool
	// full archival nodes should enable this if EnableFullCompaction is enabled
	// do NOT enable this if you synced from a snapshot.
	// Only applies if you enabled full compaction
	Archival bool
}

type SplitStore struct {
	compacting int32

	fullCompaction  bool
	enableGC        bool
	skipOldMsgs     bool
	skipMsgReceipts bool

	baseEpoch abi.ChainEpoch

	mx    sync.Mutex
	curTs *types.TipSet

	cs    *store.ChainStore
	ds    dstore.Datastore
	hot   bstore.Blockstore
	cold  bstore.Blockstore
	snoop TrackingStore

	env LiveSetEnv

	liveSetSize int64
}

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

// NewSplitStore creates a new SplitStore instance, given a path for the hotstore dbs and a cold
// blockstore. The SplitStore must be attached to the ChainStore with Start in order to trigger
// compaction.
func NewSplitStore(path string, ds dstore.Datastore, cold, hot bstore.Blockstore, cfg *Config) (*SplitStore, error) {
	// the tracking store
	snoop, err := NewTrackingStore(path, cfg.TrackingStoreType)
	if err != nil {
		return nil, err
	}

	// the liveset env
	env, err := NewLiveSetEnv(path, cfg.LiveSetType)
	if err != nil {
		snoop.Close() //nolint:errcheck
		return nil, err
	}

	// and now we can make a SplitStore
	ss := &SplitStore{
		ds:    ds,
		hot:   hot,
		cold:  cold,
		snoop: snoop,
		env:   env,

		fullCompaction:  cfg.EnableFullCompaction,
		enableGC:        cfg.EnableGC,
		skipOldMsgs:     !(cfg.EnableFullCompaction && cfg.Archival),
		skipMsgReceipts: !(cfg.EnableFullCompaction && cfg.Archival),
	}

	return ss, nil
}

// Blockstore interface
func (s *SplitStore) DeleteBlock(cid 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) Has(cid cid.Cid) (bool, error) {
	has, err := s.hot.Has(cid)

	if err != nil || has {
		return has, err
	}

	return s.cold.Has(cid)
}

func (s *SplitStore) Get(cid cid.Cid) (blocks.Block, error) {
	blk, err := s.hot.Get(cid)

	switch err {
	case nil:
		return blk, nil

	case bstore.ErrNotFound:
		return s.cold.Get(cid)

	default:
		return nil, err
	}
}

func (s *SplitStore) GetSize(cid cid.Cid) (int, error) {
	size, err := s.hot.GetSize(cid)

	switch err {
	case nil:
		return size, nil

	case bstore.ErrNotFound:
		return s.cold.GetSize(cid)

	default:
		return 0, err
	}
}

func (s *SplitStore) Put(blk blocks.Block) error {
	s.mx.Lock()
	if s.curTs == nil {
		s.mx.Unlock()
		return s.cold.Put(blk)
	}

	epoch := s.curTs.Height()
	s.mx.Unlock()

	err := s.snoop.Put(blk.Cid(), epoch)
	if err != nil {
		log.Errorf("error tracking CID in hotstore: %s; falling back to coldstore", err)
		return s.cold.Put(blk)
	}

	return s.hot.Put(blk)
}

func (s *SplitStore) PutMany(blks []blocks.Block) error {
	s.mx.Lock()
	if s.curTs == nil {
		s.mx.Unlock()
		return s.cold.PutMany(blks)
	}

	epoch := s.curTs.Height()
	s.mx.Unlock()

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

	err := s.snoop.PutBatch(batch, epoch)
	if err != nil {
		log.Errorf("error tracking CIDs in hotstore: %s; falling back to coldstore", err)
		return s.cold.PutMany(blks)
	}

	return s.hot.PutMany(blks)
}

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
	}

	ch := make(chan cid.Cid)
	go func() {
		defer cancel()
		defer close(ch)

		for _, in := range []<-chan cid.Cid{chHot, chCold} {
			for cid := range in {
				select {
				case ch <- cid:
				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(cid cid.Cid, cb func([]byte) error) error {
	err := s.hot.View(cid, cb)
	switch err {
	case bstore.ErrNotFound:
		return s.cold.View(cid, cb)

	default:
		return err
	}
}

// State tracking
func (s *SplitStore) Start(cs *store.ChainStore) error {
	s.cs = cs
	s.curTs = cs.GetHeaviestTipSet()

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

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

		err = s.setBaseEpoch(s.curTs.Height())
		if err != nil {
			return err
		}

	default:
		return err
	}

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

	return nil
}

func (s *SplitStore) Close() error {
	if atomic.LoadInt32(&s.compacting) == 1 {
		log.Warn("ongoing compaction; waiting for it to finish...")
		for atomic.LoadInt32(&s.compacting) == 1 {
			time.Sleep(time.Second)
		}
	}

	return s.env.Close()
}

func (s *SplitStore) HeadChange(revert, apply []*types.TipSet) error {
	s.mx.Lock()
	s.curTs = apply[len(apply)-1]
	epoch := s.curTs.Height()
	s.mx.Unlock()

	if !atomic.CompareAndSwapInt32(&s.compacting, 0, 1) {
		// we are currently compacting, do nothing and wait for the next head change
		return nil
	}

	if epoch-s.baseEpoch > CompactionThreshold {
		go func() {
			defer atomic.StoreInt32(&s.compacting, 0)

			log.Info("compacting splitstore")
			start := time.Now()

			s.compact()

			log.Infow("compaction done", "took", time.Since(start))
		}()
	} else {
		// no compaction necessary
		atomic.StoreInt32(&s.compacting, 0)
	}

	return nil
}

// Compaction/GC Algorithm
func (s *SplitStore) compact() {
	if s.liveSetSize == 0 {
		start := time.Now()
		log.Info("estimating live set size")
		s.estimateLiveSetSize()
		log.Infow("estimating live set size done", "took", time.Since(start), "size", s.liveSetSize)
	} else {
		log.Infow("current live set size estimate", "size", s.liveSetSize)
	}

	if s.fullCompaction {
		s.compactFull()
	} else {
		s.compactSimple()
	}
}

func (s *SplitStore) estimateLiveSetSize() {
	s.mx.Lock()
	curTs := s.curTs
	s.mx.Unlock()

	s.liveSetSize = 0
	err := s.cs.WalkSnapshot(context.Background(), curTs, 1, s.skipOldMsgs, s.skipMsgReceipts,
		func(cid cid.Cid) error {
			s.liveSetSize++
			return nil
		})

	if err != nil {
		// TODO do something better here
		panic(err)
	}
}

func (s *SplitStore) compactSimple() {
	s.mx.Lock()
	curTs := s.curTs
	s.mx.Unlock()

	coldEpoch := s.baseEpoch + CompactionCold

	log.Infow("running simple compaction", "currentEpoch", curTs.Height(), "baseEpoch", s.baseEpoch, "coldEpoch", coldEpoch)

	coldSet, err := s.env.NewLiveSet("cold", s.liveSetSize)
	if err != nil {
		// TODO do something better here
		panic(err)
	}
	defer coldSet.Close() //nolint:errcheck

	// 1. mark reachable cold objects by looking at the objects reachable only from the cold epoch
	log.Info("marking reachable cold objects")
	startMark := time.Now()

	coldTs, err := s.cs.GetTipsetByHeight(context.Background(), coldEpoch, curTs, true)
	if err != nil {
		// TODO do something better here
		panic(err)
	}

	count := int64(0)
	err = s.cs.WalkSnapshot(context.Background(), coldTs, 1, s.skipOldMsgs, s.skipMsgReceipts,
		func(cid cid.Cid) error {
			count++
			return coldSet.Mark(cid)
		})

	if err != nil {
		// TODO do something better here
		panic(err)
	}

	if count > s.liveSetSize {
		s.liveSetSize = count
	}

	log.Infow("marking done", "took", time.Since(startMark))

	// 2. move cold unreachable objects to the coldstore
	log.Info("collecting cold objects")
	startCollect := time.Now()

	cold := make(map[cid.Cid]struct{})

	// some stats for logging
	var stHot, stCold int

	// 2.1 iterate through the snoop and collect unreachable cold objects
	err = s.snoop.ForEach(func(cid cid.Cid, wrEpoch abi.ChainEpoch) error {
		// is the object stil hot?
		if wrEpoch > coldEpoch {
			// yes, stay in the hotstore
			stHot++
			return nil
		}

		// check whether it is reachable in the cold boundary
		mark, err := coldSet.Has(cid)
		if err != nil {
			return xerrors.Errorf("error checkiing cold set for %s: %w", cid, err)
		}

		if mark {
			stHot++
			return nil
		}

		// it's cold, mark it for move
		cold[cid] = struct{}{}
		stCold++
		return nil
	})

	if err != nil {
		// TODO do something better here
		panic(err)
	}

	log.Infow("collection done", "took", time.Since(startCollect))
	log.Infow("compaction stats", "hot", stHot, "cold", stCold)

	// 2.2 copy the cold objects to the coldstore
	log.Info("moving cold objects to the coldstore")
	startMove := time.Now()

	const batchSize = 1024
	batch := make([]blocks.Block, 0, batchSize)

	for cid := range cold {
		blk, err := s.hot.Get(cid)
		if err != nil {
			if err == dstore.ErrNotFound {
				// this can happen if the node is killed after we have deleted the block from the hotstore
				// but before we have deleted it from the snoop; just delete the snoop.
				err = s.snoop.Delete(cid)
				if err != nil {
					log.Errorf("error deleting cid %s from snoop: %s", cid, err)
					// TODO do something better here -- just continue?
					panic(err)
				}
			} else {
				log.Errorf("error retrieving tracked block %s from hotstore: %s", cid, err)
				// TODO do something better here -- just continue?
				panic(err)
			}

			continue
		}

		batch = append(batch, blk)
		if len(batch) == batchSize {
			err = s.cold.PutMany(batch)
			if err != nil {
				log.Errorf("error putting cold batch to coldstore: %s", err)
				// TODO do something better here -- just continue?
				panic(err)
			}
			batch = batch[:0]
		}
	}

	if len(batch) > 0 {
		err = s.cold.PutMany(batch)
		if err != nil {
			log.Errorf("error putting cold batch to coldstore: %s", err)
			// TODO do something better here -- just continue?
			panic(err)
		}
	}
	log.Infow("moving done", "took", time.Since(startMove))

	// 2.3 delete cold objects from the hotstore
	// TODO we really want batching for this!
	log.Info("purging cold objects from the hotstore")
	purgeStart := time.Now()
	for cid := range cold {
		// delete the object from the hotstore
		err = s.hot.DeleteBlock(cid)
		if err != nil {
			log.Errorf("error deleting block %s from hotstore: %s", cid, err)
			// TODO do something better here -- just continue?
			panic(err)
		}
	}
	log.Infow("purging cold from hotstore done", "took", time.Since(purgeStart))

	// 2.4 remove the snoop tracking for cold objects
	purgeStart = time.Now()
	log.Info("purging cold objects from snoop")

	err = s.snoop.DeleteBatch(cold)
	if err != nil {
		log.Errorf("error purging cold objects from snoop: %s", err)
		// TODO do something better here -- just continue?
		panic(err)
	}
	log.Infow("purging cold from snoop done", "took", time.Since(purgeStart))

	// we are done; do some housekeeping
	err = s.snoop.Sync()
	if err != nil {
		// TODO do something better here
		panic(err)
	}

	err = s.setBaseEpoch(coldEpoch)
	if err != nil {
		// TODO do something better here
		panic(err)
	}
}

func (s *SplitStore) compactFull() {
	s.mx.Lock()
	curTs := s.curTs
	s.mx.Unlock()

	epoch := curTs.Height()
	coldEpoch := s.baseEpoch + CompactionCold

	log.Infow("running full compaction", "currentEpoch", curTs.Height(), "baseEpoch", s.baseEpoch, "coldEpoch", coldEpoch)

	// create two live sets, one for marking the cold finality region
	// and one for marking the hot region
	hotSet, err := s.env.NewLiveSet("hot", s.liveSetSize)
	if err != nil {
		// TODO do something better here
		panic(err)
	}
	defer hotSet.Close() //nolint:errcheck

	coldSet, err := s.env.NewLiveSet("cold", s.liveSetSize)
	if err != nil {
		// TODO do something better here
		panic(err)
	}
	defer coldSet.Close() //nolint:errcheck

	// Phase 1: marking
	log.Info("marking live objects")
	startMark := time.Now()

	// Phase 1a: mark all reachable CIDs in the hot range
	count := int64(0)
	err = s.cs.WalkSnapshot(context.Background(), curTs, epoch-coldEpoch, s.skipOldMsgs, s.skipMsgReceipts,
		func(cid cid.Cid) error {
			count++
			return hotSet.Mark(cid)
		})

	if err != nil {
		// TODO do something better here
		panic(err)
	}

	if count > s.liveSetSize {
		s.liveSetSize = count
	}

	// Phase 1b: mark all reachable CIDs in the cold range
	coldTs, err := s.cs.GetTipsetByHeight(context.Background(), coldEpoch, curTs, true)
	if err != nil {
		// TODO do something better here
		panic(err)
	}

	count = 0
	err = s.cs.WalkSnapshot(context.Background(), coldTs, CompactionCold, s.skipOldMsgs, s.skipMsgReceipts,
		func(cid cid.Cid) error {
			count++
			return coldSet.Mark(cid)
		})

	if err != nil {
		// TODO do something better here
		panic(err)
	}

	if count > s.liveSetSize {
		s.liveSetSize = count
	}

	log.Infow("marking done", "took", time.Since(startMark))

	// Phase 2: sweep cold objects:
	// - If a cold object is reachable in the hot range, it stays in the hotstore.
	// - If a cold object is reachable in the cold range, it is moved to the coldstore.
	// - If a cold object is unreachable, it is deleted if GC is enabled, otherwise moved to the coldstore.
	startSweep := time.Now()
	log.Info("sweeping cold objects")

	// some stats for logging
	var stHot, stCold, stDead int

	cold := make(map[cid.Cid]struct{})
	dead := make(map[cid.Cid]struct{})

	// 2.1 iterate through the snoop and collect cold and dead objects
	err = s.snoop.ForEach(func(cid cid.Cid, wrEpoch abi.ChainEpoch) error {
		// is the object stil hot?
		if wrEpoch > coldEpoch {
			// yes, stay in the hotstore
			stHot++
			return nil
		}

		// the object is cold -- check whether it is reachable in the hot range
		mark, err := hotSet.Has(cid)
		if err != nil {
			return xerrors.Errorf("error checking live mark for %s: %w", cid, err)
		}

		if mark {
			// the object is reachable in the hot range, stay in the hotstore
			stHot++
			return nil
		}

		// check whether it is reachable in the cold range
		mark, err = coldSet.Has(cid)
		if err != nil {
			return xerrors.Errorf("error checkiing cold set for %s: %w", cid, err)
		}

		if s.enableGC {
			if mark {
				// the object is reachable in the cold range, move it to the cold store
				cold[cid] = struct{}{}
				stCold++
			} else {
				// the object is dead and will be deleted
				dead[cid] = struct{}{}
				stDead++
			}
		} else {
			// if GC is disabled, we move both cold and dead objects to the coldstore
			cold[cid] = struct{}{}
			if mark {
				stCold++
			} else {
				stDead++
			}
		}

		return nil
	})

	if err != nil {
		// TODO do something better here
		panic(err)
	}

	log.Infow("compaction stats", "hot", stHot, "cold", stCold, "dead", stDead)

	// 2.2 copy the cold objects to the coldstore
	log.Info("moving cold objects to the coldstore")
	startMove := time.Now()

	const batchSize = 1024
	batch := make([]blocks.Block, 0, batchSize)

	for cid := range cold {
		blk, err := s.hot.Get(cid)
		if err != nil {
			if err == dstore.ErrNotFound {
				// this can happen if the node is killed after we have deleted the block from the hotstore
				// but before we have deleted it from the snoop; just delete the snoop.
				err = s.snoop.Delete(cid)
				if err != nil {
					log.Errorf("error deleting cid %s from snoop: %s", cid, err)
					// TODO do something better here -- just continue?
					panic(err)
				}
			} else {
				log.Errorf("error retrieving tracked block %s from hotstore: %s", cid, err)
				// TODO do something better here -- just continue?
				panic(err)
			}

			continue
		}

		batch = append(batch, blk)
		if len(batch) == batchSize {
			err = s.cold.PutMany(batch)
			if err != nil {
				log.Errorf("error putting cold batch to coldstore: %s", err)
				// TODO do something better here -- just continue?
				panic(err)
			}
			batch = batch[:0]
		}
	}

	if len(batch) > 0 {
		err = s.cold.PutMany(batch)
		if err != nil {
			log.Errorf("error putting cold batch to coldstore: %s", err)
			// TODO do something better here -- just continue?
			panic(err)
		}
	}
	log.Infow("moving done", "took", time.Since(startMove))

	// 2.3 delete cold objects from the hotstore
	// TODO we really want batching for this!
	log.Info("purging cold objects from the hotstore")
	purgeStart := time.Now()
	for cid := range cold {
		// delete the object from the hotstore
		err = s.hot.DeleteBlock(cid)
		if err != nil {
			log.Errorf("error deleting block %s from hotstore: %s", cid, err)
			// TODO do something better here -- just continue?
			panic(err)
		}
	}
	log.Infow("purging cold from hotstore done", "took", time.Since(purgeStart))

	// 2.4 remove the snoop tracking for cold objects
	purgeStart = time.Now()
	log.Info("purging cold objects from snoop")

	err = s.snoop.DeleteBatch(cold)
	if err != nil {
		log.Errorf("error purging cold objects from snoop: %s", err)
		// TODO do something better here -- just continue?
		panic(err)
	}
	log.Infow("purging cold from snoop done", "took", time.Since(purgeStart))

	// 3. if we have dead objects, delete them from the hotstore and remove the tracking
	if len(dead) > 0 {
		log.Info("deleting dead objects")

		purgeStart = time.Now()
		log.Info("purging dead objects from the hotstore")
		// TODO we really want batching for this!
		for cid := range dead {
			// delete the object from the hotstore
			err = s.hot.DeleteBlock(cid)
			if err != nil {
				log.Errorf("error deleting block %s from hotstore: %s", cid, err)
				// TODO do something better here -- just continue?
				panic(err)
			}
		}
		log.Infow("purging dead from hotstore done", "took", time.Since(purgeStart))

		// remove the snoop tracking
		purgeStart := time.Now()
		log.Info("purging dead objects from snoop")

		err = s.snoop.DeleteBatch(dead)
		if err != nil {
			log.Errorf("error purging dead objects from snoop: %s", err)
			// TODO do something better here -- just continue?
			panic(err)
		}

		log.Infow("purging dead from snoop done", "took", time.Since(purgeStart))
	}

	log.Infow("sweeping done", "took", time.Since(startSweep))

	// we are done; do some housekeeping
	err = s.snoop.Sync()
	if err != nil {
		// TODO do something better here
		panic(err)
	}

	err = s.setBaseEpoch(coldEpoch)
	if err != nil {
		// TODO do something better here
		panic(err)
	}
}

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

func epochToBytes(epoch abi.ChainEpoch) []byte {
	buf := make([]byte, 16)
	n := binary.PutUvarint(buf, uint64(epoch))
	return buf[:n]
}

func bytesToEpoch(buf []byte) abi.ChainEpoch {
	epoch, n := binary.Uvarint(buf)
	if n < 0 {
		panic("bogus base epoch bytes")
	}
	return abi.ChainEpoch(epoch)
}