[filedao] Impl merge filedao

blockchain/filedao/blockstore.go

@@ -0,0 +1,101 @@
+package filedao
+
+import (
+	"fmt"
+
+	"github.com/iotexproject/iotex-proto/golang/iotextypes"
+
+	"github.com/iotexproject/iotex-core/blockchain/block"
+	"github.com/iotexproject/iotex-core/pkg/util/byteutil"
+)
+
+const (
+	blockstoreDefaultVersion = byte(0)
+	blockstoreHeaderSize     = 8
+)
+
+type (
+	// blockstore is a byte slice that contains a block, its receipts and transaction logs
+	// the first 8 bytes is the size of the block
+	// the next n bytes is the serialized block and receipts, n is the size of the block
+	// the rest bytes is the serialized transaction logs
+	blockstore []byte
+)
+
+var (
+	errInvalidBlockstore = fmt.Errorf("invalid blockstore")
+)
+
+func convertToBlockStore(blk *block.Block) (blockstore, error) {
+	data := make(blockstore, 0)
+	s := &block.Store{
+		Block:    blk,
+		Receipts: blk.Receipts,
+	}
+	tmp, err := s.Serialize()
+	if err != nil {
+		return nil, err
+	}
+	data = append(data, byteutil.Uint64ToBytesBigEndian(uint64(len(tmp)))...)
+	data = append(data, tmp...)
+	txLog := blk.TransactionLog()
+	if txLog != nil {
+		tmp = txLog.Serialize()
+		data = append(data, tmp...)
+	}
+	return data, nil
+}
+
+func (s blockstore) Block(deser *block.Deserializer) (*block.Block, error) {
+	size := s.blockSize()
+	bs, err := deser.DeserializeBlockStore(s[blockstoreHeaderSize : size+blockstoreHeaderSize])
+	if err != nil {
+		return nil, err
+	}
+	bs.Block.Receipts = bs.Receipts
+	return bs.Block, nil
+}
+
+func (s blockstore) TransactionLogs() (*iotextypes.TransactionLogs, error) {
+	size := s.blockSize()
+	if uint64(len(s)) == size+blockstoreHeaderSize {
+		return nil, nil
+	}
+	return block.DeserializeSystemLogPb(s[size+blockstoreHeaderSize:])
+}
+
+func (s blockstore) Serialize() []byte {
+	return append([]byte{blockstoreDefaultVersion}, s...)
+}
+
+func (s *blockstore) Deserialize(data []byte) error {
+	if len(data) == 0 {
+		return errInvalidBlockstore
+	}
+	switch data[0] {
+	case blockstoreDefaultVersion:
+		bs := blockstore(data[1:])
+		if err := bs.Validate(); err != nil {
+			return err
+		}
+		*s = bs
+		return nil
+	default:
+		return errInvalidBlockstore
+	}
+}
+
+func (s blockstore) blockSize() uint64 {
+	return byteutil.BytesToUint64BigEndian(s[:blockstoreHeaderSize])
+}
+
+func (s blockstore) Validate() error {
+	if len(s) < blockstoreHeaderSize {
+		return errInvalidBlockstore
+	}
+	blkSize := s.blockSize()
+	if uint64(len(s)) < blkSize+blockstoreHeaderSize {
+		return errInvalidBlockstore
+	}
+	return nil
+}

blockchain/filedao/blockstore_test.go

@@ -0,0 +1,40 @@
+package filedao
+
+import (
+	"testing"
+
+	"github.com/iotexproject/go-pkgs/hash"
+	"github.com/stretchr/testify/require"
+
+	"github.com/iotexproject/iotex-core/blockchain/block"
+)
+
+func TestBlockStore(t *testing.T) {
+	r := require.New(t)
+	builder := block.NewTestingBuilder()
+	blk := createTestingBlock(builder, 1, hash.ZeroHash256)
+	bs, err := convertToBlockStore(blk)
+	r.NoError(err)
+	data := bs.Serialize()
+	dbs := new(blockstore)
+	r.NoError(dbs.Deserialize(data))
+	r.Equal(bs[:], (*dbs)[:], "serialized block store should be equal to deserialized block store")
+	// check deserialized block
+	deser := block.NewDeserializer(0)
+	dBlk, err := dbs.Block(deser)
+	r.NoError(err)
+	dTxLogs, err := dbs.TransactionLogs()
+	r.NoError(err)
+	r.NoError(fillTransactionLog(dBlk.Receipts, dTxLogs.Logs))
+	r.Equal(blk.Header, dBlk.Header)
+	r.Equal(blk.Body, dBlk.Body)
+	r.Equal(blk.Footer, dBlk.Footer)
+	r.Equal(len(blk.Receipts), len(dBlk.Receipts))
+	for i := range blk.Receipts {
+		r.Equal(blk.Receipts[i].Hash(), dBlk.Receipts[i].Hash())
+		r.Equal(len(blk.Receipts[i].TransactionLogs()), len(dBlk.Receipts[i].TransactionLogs()))
+		for j := range blk.Receipts[i].TransactionLogs() {
+			r.Equal(blk.Receipts[i].TransactionLogs()[j], dBlk.Receipts[i].TransactionLogs()[j])
+		}
+	}
+}

blockchain/filedao/merge.go

@@ -0,0 +1,85 @@
+package filedao
+
+import (
+	"context"
+
+	"go.uber.org/zap"
+
+	"github.com/iotexproject/iotex-core/pkg/log"
+)
+
+type (
+	// mergeDao wraps a file dao, and merge the blocks from the original dao on start
+	mergeDao struct {
+		FileDAO
+		originalDao FileDAO
+		mergeMax    uint64
+	}
+)
+
+// NewMergeDao creates a new merge dao
+func NewMergeDao(original FileDAO, new FileDAO, mergeMax uint64) FileDAO {
+	return &mergeDao{
+		FileDAO:     new,
+		originalDao: original,
+		mergeMax:    mergeMax,
+	}
+}
+
+func (md *mergeDao) Start(ctx context.Context) error {
+	if err := md.FileDAO.Start(ctx); err != nil {
+		return err
+	}
+	if err := md.originalDao.Start(context.Background()); err != nil {
+		log.L().Error("failed to start original dao", zap.Error(err))
+		return nil
+	}
+	defer md.originalDao.Stop(ctx)
+
+	height, err := md.FileDAO.Height()
+	if err != nil {
+		return err
+	}
+	orgTip, err := md.originalDao.Height()
+	if err != nil {
+		return err
+	}
+	// doing nothing if the height is larger than the original dao
+	if height >= orgTip {
+		return nil
+	}
+	// init the blocks from the original dao
+	start := uint64(1)
+	if md.mergeMax > 0 && orgTip > (md.mergeMax+1) {
+		start = orgTip - md.mergeMax
+		if d, ok := md.FileDAO.(interface{ SetStart(uint64) error }); ok {
+			if err = d.SetStart(start); err != nil {
+				return err
+			}
+		}
+	}
+	for i := start; i <= orgTip; i++ {
+		blk, err := md.originalDao.GetBlockByHeight(i)
+		if err != nil {
+			return err
+		}
+		receipts, err := md.originalDao.GetReceipts(i)
+		if err != nil {
+			return err
+		}
+		blk.Receipts = receipts
+		if md.originalDao.ContainsTransactionLog() {
+			logs, err := md.originalDao.TransactionLogs(i)
+			if err != nil {
+				return err
+			}
+			if err = fillTransactionLog(receipts, logs.Logs); err != nil {
+				return err
+			}
+		}
+		if err := md.PutBlock(ctx, blk); err != nil {
+			return err
+		}
+	}
+	return nil
+}

blockchain/filedao/merge_test.go

@@ -0,0 +1,40 @@
+package filedao
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/iotexproject/go-pkgs/hash"
+
+	"github.com/iotexproject/iotex-core/blockchain/block"
+	"github.com/iotexproject/iotex-core/db"
+)
+
+func TestMergeDao(t *testing.T) {
+	r := require.New(t)
+	cfg := db.DefaultConfig
+	cfg.DbPath = t.TempDir() + "/filedao"
+	deser := block.NewDeserializer(4689)
+	fdao, err := NewFileDAO(cfg, deser)
+	r.NoError(err)
+	ctx := context.Background()
+	r.NoError(fdao.Start(ctx))
+	r.NoError(testCommitBlocks(t, fdao, 1, 100, hash.ZeroHash256))
+	r.NoError(fdao.Stop(ctx))
+
+	fdao, err = NewFileDAO(cfg, deser)
+	r.NoError(err)
+	sdao, err := NewSizedFileDao(10, t.TempDir(), deser)
+	r.NoError(err)
+	mdao := NewMergeDao(fdao, sdao, 10)
+	r.NoError(mdao.Start(ctx))
+	_, err = mdao.GetBlockByHeight(1)
+	r.ErrorIs(err, db.ErrNotExist)
+	_, err = mdao.GetBlockByHeight(90)
+	r.ErrorIs(err, db.ErrNotExist)
+	_, err = mdao.GetBlockByHeight(91)
+	r.NoError(err)
+	r.NoError(mdao.Stop(ctx))
+}