move splitstore into blockstore package.

blockstore/splitstore/liveset.go

@@ -0,0 +1,33 @@
+package splitstore
+
+import (
+	"path/filepath"
+
+	"golang.org/x/xerrors"
+
+	cid "github.com/ipfs/go-cid"
+)
+
+type LiveSet interface {
+	Mark(cid.Cid) error
+	Has(cid.Cid) (bool, error)
+	Close() error
+}
+
+var markBytes = []byte{}
+
+type LiveSetEnv interface {
+	NewLiveSet(name string, sizeHint int64) (LiveSet, error)
+	Close() error
+}
+
+func NewLiveSetEnv(path string, liveSetType string) (LiveSetEnv, error) {
+	switch liveSetType {
+	case "", "bloom":
+		return NewBloomLiveSetEnv()
+	case "bolt":
+		return NewBoltLiveSetEnv(filepath.Join(path, "sweep.bolt"))
+	default:
+		return nil, xerrors.Errorf("unknown live set type %s", liveSetType)
+	}
+}

blockstore/splitstore/liveset_bloom.go

@@ -0,0 +1,77 @@
+package splitstore
+
+import (
+	"math/rand"
+
+	"golang.org/x/xerrors"
+
+	bbloom "github.com/ipfs/bbloom"
+	cid "github.com/ipfs/go-cid"
+	blake2b "github.com/minio/blake2b-simd"
+)
+
+const (
+	BloomFilterMinSize     = 10_000_000
+	BloomFilterProbability = 0.01
+)
+
+type BloomLiveSetEnv struct{}
+
+var _ LiveSetEnv = (*BloomLiveSetEnv)(nil)
+
+type BloomLiveSet struct {
+	salt []byte
+	bf   *bbloom.Bloom
+}
+
+var _ LiveSet = (*BloomLiveSet)(nil)
+
+func NewBloomLiveSetEnv() (*BloomLiveSetEnv, error) {
+	return &BloomLiveSetEnv{}, nil
+}
+
+func (e *BloomLiveSetEnv) NewLiveSet(name string, sizeHint int64) (LiveSet, error) {
+	size := int64(BloomFilterMinSize)
+	for size < sizeHint {
+		size += BloomFilterMinSize
+	}
+
+	salt := make([]byte, 4)
+	_, err := rand.Read(salt) //nolint
+	if err != nil {
+		return nil, xerrors.Errorf("error reading salt: %w", err)
+	}
+
+	bf, err := bbloom.New(float64(size), float64(BloomFilterProbability))
+	if err != nil {
+		return nil, xerrors.Errorf("error creating bloom filter: %w", err)
+	}
+
+	return &BloomLiveSet{salt: salt, bf: bf}, nil
+}
+
+func (e *BloomLiveSetEnv) Close() error {
+	return nil
+}
+
+func (s *BloomLiveSet) saltedKey(cid cid.Cid) []byte {
+	hash := cid.Hash()
+	key := make([]byte, len(s.salt)+len(hash))
+	n := copy(key, s.salt)
+	copy(key[n:], hash)
+	rehash := blake2b.Sum256(key)
+	return rehash[:]
+}
+
+func (s *BloomLiveSet) Mark(cid cid.Cid) error {
+	s.bf.Add(s.saltedKey(cid))
+	return nil
+}
+
+func (s *BloomLiveSet) Has(cid cid.Cid) (bool, error) {
+	return s.bf.Has(s.saltedKey(cid)), nil
+}
+
+func (s *BloomLiveSet) Close() error {
+	return nil
+}

blockstore/splitstore/liveset_bolt.go

@@ -0,0 +1,81 @@
+package splitstore
+
+import (
+	"time"
+
+	"golang.org/x/xerrors"
+
+	cid "github.com/ipfs/go-cid"
+	bolt "go.etcd.io/bbolt"
+)
+
+type BoltLiveSetEnv struct {
+	db *bolt.DB
+}
+
+var _ LiveSetEnv = (*BoltLiveSetEnv)(nil)
+
+type BoltLiveSet struct {
+	db       *bolt.DB
+	bucketId []byte
+}
+
+var _ LiveSet = (*BoltLiveSet)(nil)
+
+func NewBoltLiveSetEnv(path string) (*BoltLiveSetEnv, error) {
+	db, err := bolt.Open(path, 0644,
+		&bolt.Options{
+			Timeout: 1 * time.Second,
+			NoSync:  true,
+		})
+	if err != nil {
+		return nil, err
+	}
+
+	return &BoltLiveSetEnv{db: db}, nil
+}
+
+func (e *BoltLiveSetEnv) NewLiveSet(name string, hint int64) (LiveSet, error) {
+	bucketId := []byte(name)
+	err := e.db.Update(func(tx *bolt.Tx) error {
+		_, err := tx.CreateBucketIfNotExists(bucketId)
+		if err != nil {
+			return xerrors.Errorf("error creating bolt db bucket %s: %w", name, err)
+		}
+		return nil
+	})
+
+	if err != nil {
+		return nil, err
+	}
+
+	return &BoltLiveSet{db: e.db, bucketId: bucketId}, nil
+}
+
+func (e *BoltLiveSetEnv) Close() error {
+	return e.db.Close()
+}
+
+func (s *BoltLiveSet) Mark(cid cid.Cid) error {
+	return s.db.Update(func(tx *bolt.Tx) error {
+		b := tx.Bucket(s.bucketId)
+		return b.Put(cid.Hash(), markBytes)
+	})
+}
+
+func (s *BoltLiveSet) Has(cid cid.Cid) (result bool, err error) {
+	err = s.db.View(func(tx *bolt.Tx) error {
+		b := tx.Bucket(s.bucketId)
+		v := b.Get(cid.Hash())
+		result = v != nil
+		return nil
+	})
+
+	return result, err
+}
+
+func (s *BoltLiveSet) Close() error {
+	return s.db.Update(func(tx *bolt.Tx) error {
+		return tx.DeleteBucket(s.bucketId)
+	})
+}

blockstore/splitstore/liveset_test.go

@@ -0,0 +1,140 @@
+package splitstore
+
+import (
+	"os"
+	"testing"
+
+	cid "github.com/ipfs/go-cid"
+	"github.com/multiformats/go-multihash"
+)
+
+func TestBoltLiveSet(t *testing.T) {
+	testLiveSet(t, "bolt")
+}
+
+func TestBloomLiveSet(t *testing.T) {
+	testLiveSet(t, "bloom")
+}
+
+func testLiveSet(t *testing.T, lsType string) {
+	t.Helper()
+
+	path := "/tmp/liveset-test"
+
+	err := os.MkdirAll(path, 0777)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	env, err := NewLiveSetEnv(path, lsType)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer env.Close() //nolint:errcheck
+
+	hotSet, err := env.NewLiveSet("hot", 0)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	coldSet, err := env.NewLiveSet("cold", 0)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	makeCid := func(key string) cid.Cid {
+		h, err := multihash.Sum([]byte(key), multihash.SHA2_256, -1)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		return cid.NewCidV1(cid.Raw, h)
+	}
+
+	mustHave := func(s LiveSet, cid cid.Cid) {
+		has, err := s.Has(cid)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		if !has {
+			t.Fatal("mark not found")
+		}
+	}
+
+	mustNotHave := func(s LiveSet, cid cid.Cid) {
+		has, err := s.Has(cid)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		if has {
+			t.Fatal("unexpected mark")
+		}
+	}
+
+	k1 := makeCid("a")
+	k2 := makeCid("b")
+	k3 := makeCid("c")
+	k4 := makeCid("d")
+
+	hotSet.Mark(k1)  //nolint
+	hotSet.Mark(k2)  //nolint
+	coldSet.Mark(k3) //nolint
+
+	mustHave(hotSet, k1)
+	mustHave(hotSet, k2)
+	mustNotHave(hotSet, k3)
+	mustNotHave(hotSet, k4)
+
+	mustNotHave(coldSet, k1)
+	mustNotHave(coldSet, k2)
+	mustHave(coldSet, k3)
+	mustNotHave(coldSet, k4)
+
+	// close them and reopen to redo the dance
+
+	err = hotSet.Close()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	err = coldSet.Close()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	hotSet, err = env.NewLiveSet("hot", 0)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	coldSet, err = env.NewLiveSet("cold", 0)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	hotSet.Mark(k3)  //nolint
+	hotSet.Mark(k4)  //nolint
+	coldSet.Mark(k1) //nolint
+
+	mustNotHave(hotSet, k1)
+	mustNotHave(hotSet, k2)
+	mustHave(hotSet, k3)
+	mustHave(hotSet, k4)
+
+	mustHave(coldSet, k1)
+	mustNotHave(coldSet, k2)
+	mustNotHave(coldSet, k3)
+	mustNotHave(coldSet, k4)
+
+	err = hotSet.Close()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	err = coldSet.Close()
+	if err != nil {
+		t.Fatal(err)
+	}
+}

blockstore/splitstore/splitstore.go

@@ -0,0 +1,831 @@
+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)
+}

blockstore/splitstore/tracking.go

@@ -0,0 +1,30 @@
+package splitstore
+
+import (
+	"path/filepath"
+
+	"golang.org/x/xerrors"
+
+	"github.com/filecoin-project/go-state-types/abi"
+	cid "github.com/ipfs/go-cid"
+)
+
+type TrackingStore interface {
+	Put(cid.Cid, abi.ChainEpoch) error
+	PutBatch([]cid.Cid, abi.ChainEpoch) error
+	Get(cid.Cid) (abi.ChainEpoch, error)
+	Delete(cid.Cid) error
+	DeleteBatch(map[cid.Cid]struct{}) error
+	ForEach(func(cid.Cid, abi.ChainEpoch) error) error
+	Sync() error
+	Close() error
+}
+
+func NewTrackingStore(path string, trackingStoreType string) (TrackingStore, error) {
+	switch trackingStoreType {
+	case "", "bolt":
+		return NewBoltTrackingStore(filepath.Join(path, "snoop.bolt"))
+	default:
+		return nil, xerrors.Errorf("unknown tracking store type %s", trackingStoreType)
+	}
+}

blockstore/splitstore/tracking_bolt.go

@@ -0,0 +1,120 @@
+package splitstore
+
+import (
+	"time"
+
+	"golang.org/x/xerrors"
+
+	cid "github.com/ipfs/go-cid"
+	bolt "go.etcd.io/bbolt"
+
+	"github.com/filecoin-project/go-state-types/abi"
+)
+
+type BoltTrackingStore struct {
+	db       *bolt.DB
+	bucketId []byte
+}
+
+var _ TrackingStore = (*BoltTrackingStore)(nil)
+
+func NewBoltTrackingStore(path string) (*BoltTrackingStore, error) {
+	db, err := bolt.Open(path, 0644,
+		&bolt.Options{
+			Timeout: 1 * time.Second,
+			NoSync:  true,
+		})
+	if err != nil {
+		return nil, err
+	}
+
+	bucketId := []byte("snoop")
+	err = db.Update(func(tx *bolt.Tx) error {
+		_, err := tx.CreateBucketIfNotExists(bucketId)
+		if err != nil {
+			return xerrors.Errorf("error creating bolt db bucket %s: %w", string(bucketId), err)
+		}
+		return nil
+	})
+
+	if err != nil {
+		db.Close() //nolint:errcheck
+		return nil, err
+	}
+
+	return &BoltTrackingStore{db: db, bucketId: bucketId}, nil
+}
+
+func (s *BoltTrackingStore) Put(cid cid.Cid, epoch abi.ChainEpoch) error {
+	val := epochToBytes(epoch)
+	return s.db.Batch(func(tx *bolt.Tx) error {
+		b := tx.Bucket(s.bucketId)
+		return b.Put(cid.Hash(), val)
+	})
+}
+
+func (s *BoltTrackingStore) PutBatch(cids []cid.Cid, epoch abi.ChainEpoch) error {
+	val := epochToBytes(epoch)
+	return s.db.Batch(func(tx *bolt.Tx) error {
+		b := tx.Bucket(s.bucketId)
+		for _, cid := range cids {
+			err := b.Put(cid.Hash(), val)
+			if err != nil {
+				return err
+			}
+		}
+		return nil
+	})
+}
+
+func (s *BoltTrackingStore) Get(cid cid.Cid) (epoch abi.ChainEpoch, err error) {
+	err = s.db.View(func(tx *bolt.Tx) error {
+		b := tx.Bucket(s.bucketId)
+		val := b.Get(cid.Hash())
+		if val == nil {
+			return xerrors.Errorf("missing tracking epoch for %s", cid)
+		}
+		epoch = bytesToEpoch(val)
+		return nil
+	})
+	return epoch, err
+}
+
+func (s *BoltTrackingStore) Delete(cid cid.Cid) error {
+	return s.db.Batch(func(tx *bolt.Tx) error {
+		b := tx.Bucket(s.bucketId)
+		return b.Delete(cid.Hash())
+	})
+}
+
+func (s *BoltTrackingStore) DeleteBatch(cids map[cid.Cid]struct{}) error {
+	return s.db.Batch(func(tx *bolt.Tx) error {
+		b := tx.Bucket(s.bucketId)
+		for cid := range cids {
+			err := b.Delete(cid.Hash())
+			if err != nil {
+				return xerrors.Errorf("error deleting %s", cid)
+			}
+		}
+		return nil
+	})
+}
+
+func (s *BoltTrackingStore) ForEach(f func(cid.Cid, abi.ChainEpoch) error) error {
+	return s.db.View(func(tx *bolt.Tx) error {
+		b := tx.Bucket(s.bucketId)
+		return b.ForEach(func(k, v []byte) error {
+			cid := cid.NewCidV1(cid.Raw, k)
+			epoch := bytesToEpoch(v)
+			return f(cid, epoch)
+		})
+	})
+}
+
+func (s *BoltTrackingStore) Sync() error {
+	return s.db.Sync()
+}
+
+func (s *BoltTrackingStore) Close() error {
+	return s.db.Close()
+}

blockstore/splitstore/tracking_test.go

@@ -0,0 +1,132 @@
+package splitstore
+
+import (
+	"os"
+	"testing"
+
+	cid "github.com/ipfs/go-cid"
+	"github.com/multiformats/go-multihash"
+
+	"github.com/filecoin-project/go-state-types/abi"
+)
+
+func TestBoltTrackingStore(t *testing.T) {
+	testTrackingStore(t, "bolt")
+}
+
+func testTrackingStore(t *testing.T, tsType string) {
+	t.Helper()
+
+	makeCid := func(key string) cid.Cid {
+		h, err := multihash.Sum([]byte(key), multihash.SHA2_256, -1)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		return cid.NewCidV1(cid.Raw, h)
+	}
+
+	mustHave := func(s TrackingStore, cid cid.Cid, epoch abi.ChainEpoch) {
+		val, err := s.Get(cid)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		if val != epoch {
+			t.Fatal("epoch mismatch")
+		}
+	}
+
+	mustNotHave := func(s TrackingStore, cid cid.Cid) {
+		_, err := s.Get(cid)
+		if err == nil {
+			t.Fatal("expected error")
+		}
+	}
+
+	path := "/tmp/liveset-test"
+
+	err := os.MkdirAll(path, 0777)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	s, err := NewTrackingStore(path, tsType)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	k1 := makeCid("a")
+	k2 := makeCid("b")
+	k3 := makeCid("c")
+	k4 := makeCid("d")
+
+	s.Put(k1, 1) //nolint
+	s.Put(k2, 2) //nolint
+	s.Put(k3, 3) //nolint
+	s.Put(k4, 4) //nolint
+
+	mustHave(s, k1, 1)
+	mustHave(s, k2, 2)
+	mustHave(s, k3, 3)
+	mustHave(s, k4, 4)
+
+	s.Delete(k1) // nolint
+	s.Delete(k2) // nolint
+
+	mustNotHave(s, k1)
+	mustNotHave(s, k2)
+	mustHave(s, k3, 3)
+	mustHave(s, k4, 4)
+
+	s.PutBatch([]cid.Cid{k1}, 1) //nolint
+	s.PutBatch([]cid.Cid{k2}, 2) //nolint
+
+	mustHave(s, k1, 1)
+	mustHave(s, k2, 2)
+	mustHave(s, k3, 3)
+	mustHave(s, k4, 4)
+
+	allKeys := map[string]struct{}{
+		k1.String(): {},
+		k2.String(): {},
+		k3.String(): {},
+		k4.String(): {},
+	}
+
+	err = s.ForEach(func(k cid.Cid, _ abi.ChainEpoch) error {
+		_, ok := allKeys[k.String()]
+		if !ok {
+			t.Fatal("unexpected key")
+		}
+
+		delete(allKeys, k.String())
+		return nil
+	})
+
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if len(allKeys) != 0 {
+		t.Fatal("not all keys were returned")
+	}
+
+	// no close and reopen and ensure the keys still exist
+	err = s.Close()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	s, err = NewTrackingStore(path, tsType)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	mustHave(s, k1, 1)
+	mustHave(s, k2, 2)
+	mustHave(s, k3, 3)
+	mustHave(s, k4, 4)
+
+	s.Close() //nolint:errcheck
+}