feat(state): prune block on commit (#1404)

Co-authored-by: b00f <[email protected]>

genesis/genesis.go

@@ -2,7 +2,6 @@ package genesis
 
 import (
 	"encoding/json"
-	"fmt"
 	"os"
 	"time"
 
@@ -66,7 +65,7 @@ type genesisData struct {
 func (gen *Genesis) Hash() hash.Hash {
 	bs, err := cbor.Marshal(gen.data)
 	if err != nil {
-		panic(fmt.Errorf("could not create hash of Genesis: %w", err))
+		return hash.UndefHash
 	}
 
 	return hash.CalcHash(bs)

state/state.go

@@ -138,25 +138,6 @@ func (st *state) concreteSandbox() sandbox.Sandbox {
 }
 
 func (st *state) tryLoadLastInfo() error {
-	// Make sure the genesis doc is the same as before.
-	//
-	// This check is not strictly necessary, since the genesis state is already committed.
-	// However, it is good to perform this check to ensure that the genesis document has not been modified.
-	genStateRoot := st.calculateGenesisStateRootFromGenesisDoc()
-	committedBlockOne, err := st.store.Block(1)
-	if err != nil {
-		return err
-	}
-
-	blockOne, err := committedBlockOne.ToBlock()
-	if err != nil {
-		return err
-	}
-
-	if genStateRoot != blockOne.Header().StateRoot() {
-		return fmt.Errorf("invalid genesis doc")
-	}
-
 	logger.Debug("try to restore the last state")
 	committeeInstance, err := st.lastInfo.RestoreLastInfo(st.store, st.params.CommitteeSize)
 	if err != nil {
@@ -244,27 +225,6 @@ func (st *state) stateRoot() hash.Hash {
 	return *stateRoot
 }
 
-func (st *state) calculateGenesisStateRootFromGenesisDoc() hash.Hash {
-	accs := st.genDoc.Accounts()
-	vals := st.genDoc.Validators()
-
-	accHashes := make([]hash.Hash, len(accs))
-	valHashes := make([]hash.Hash, len(vals))
-	for _, acc := range accs {
-		accHashes[acc.Number()] = acc.Hash()
-	}
-	for _, val := range vals {
-		valHashes[val.Number()] = val.Hash()
-	}
-
-	accTree := simplemerkle.NewTreeFromHashes(accHashes)
-	valTree := simplemerkle.NewTreeFromHashes(valHashes)
-	accRootHash := accTree.Root()
-	valRootHash := valTree.Root()
-
-	return *simplemerkle.HashMerkleBranches(&accRootHash, &valRootHash)
-}
-
 func (st *state) Close() {
 	st.lk.RLock()
 	defer st.lk.RUnlock()
@@ -509,7 +469,7 @@ func (st *state) CommitBlock(blk *block.Block, cert *certificate.BlockCertificat
 	}
 
 	// -----------------------------------
-	// Publishing the events to the zmq
+	// Publishing the events to the nano message.
 	st.publishEvents(height, blk)
 
 	return nil

state/state_test.go

@@ -561,29 +561,6 @@ func TestLoadState(t *testing.T) {
 	require.NoError(t, newState.CommitBlock(blk6, cert6))
 }
 
-func TestLoadStateAfterChangingGenesis(t *testing.T) {
-	td := setup(t)
-
-	_, err := LoadOrNewState(td.state.genDoc, td.state.valKeys,
-		td.state.store, txpool.MockingTxPool(), nil)
-	require.NoError(t, err)
-
-	pub, _ := td.RandBLSKeyPair()
-	val := validator.NewValidator(pub, 4)
-	newVals := append(td.state.genDoc.Validators(), val)
-
-	genDoc := genesis.MakeGenesis(
-		td.state.genDoc.GenesisTime(),
-		td.state.genDoc.Accounts(),
-		newVals,
-		td.state.genDoc.Params())
-
-	// Load last state info after modifying genesis
-	_, err = LoadOrNewState(genDoc, td.state.valKeys,
-		td.state.store, txpool.MockingTxPool(), nil)
-	require.Error(t, err)
-}
-
 func TestIsValidator(t *testing.T) {
 	td := setup(t)
 
@@ -643,9 +620,6 @@ func TestCommittedBlock(t *testing.T) {
 		assert.NoError(t, err)
 		assert.Nil(t, blockOne.PrevCertificate())
 		assert.Equal(t, hash.UndefHash, blockOne.Header().PrevBlockHash())
-
-		r := td.state.calculateGenesisStateRootFromGenesisDoc()
-		assert.Equal(t, blockOne.Header().StateRoot(), r)
 	})
 
 	t.Run("Last block", func(t *testing.T) {

store/block.go

@@ -11,7 +11,6 @@ import (
 	"github.com/pactus-project/pactus/types/block"
 	"github.com/pactus-project/pactus/util"
 	"github.com/pactus-project/pactus/util/encoding"
-	"github.com/pactus-project/pactus/util/logger"
 	"github.com/pactus-project/pactus/util/pairslice"
 	"github.com/syndtr/goleveldb/leveldb"
 )
@@ -47,15 +46,6 @@ func newBlockStore(db *leveldb.DB, sortitionCacheSize uint32, publicKeyCacheSize
 }
 
 func (bs *blockStore) saveBlock(batch *leveldb.Batch, height uint32, blk *block.Block) []blockRegion {
-	if height > 1 {
-		if !bs.hasBlock(height - 1) {
-			logger.Panic("previous block not found", "height", height)
-		}
-	}
-	if bs.hasBlock(height) {
-		logger.Panic("duplicated block", "height", height)
-	}
-
 	blockHash := blk.Hash()
 	regs := make([]blockRegion, blk.Transactions().Len())
 	w := bytes.NewBuffer(make([]byte, 0, blk.SerializeSize()+hash.HashSize))

store/block_test.go

@@ -13,16 +13,6 @@ func TestBlockStore(t *testing.T) {
 	lastCert := td.store.LastCertificate()
 	lastHeight := lastCert.Height()
 	nextBlk, nextCert := td.GenerateTestBlock(lastHeight + 1)
-	nextNextBlk, nextNextCert := td.GenerateTestBlock(lastHeight + 2)
-
-	t.Run("Missed block, Should panic ", func(t *testing.T) {
-		defer func() {
-			if r := recover(); r == nil {
-				t.Errorf("The code did not panic")
-			}
-		}()
-		td.store.SaveBlock(nextNextBlk, nextNextCert)
-	})
 
 	t.Run("Add block, don't batch write", func(t *testing.T) {
 		td.store.SaveBlock(nextBlk, nextCert)
@@ -46,15 +36,6 @@ func TestBlockStore(t *testing.T) {
 		assert.NoError(t, err)
 		assert.Equal(t, cert.Hash(), nextCert.Hash())
 	})
-
-	t.Run("Duplicated block, Should panic ", func(t *testing.T) {
-		defer func() {
-			if r := recover(); r == nil {
-				t.Errorf("The code did not panic")
-			}
-		}()
-		td.store.SaveBlock(nextBlk, nextCert)
-	})
 }
 
 func TestSortitionSeed(t *testing.T) {

store/config.go

@@ -9,7 +9,7 @@ import (
 
 type Config struct {
 	Path          string `toml:"path"`
-	RetentionDays uint   `toml:"retention_days"`
+	RetentionDays uint32 `toml:"retention_days"`
 
 	// Private configs
 	TxCacheSize        uint32                  `toml:"-"`
@@ -66,5 +66,5 @@ func (conf *Config) BasicCheck() error {
 }
 
 func (conf *Config) RetentionBlocks() uint32 {
-	return uint32(conf.RetentionDays * 8640)
+	return conf.RetentionDays * 8640
 }

store/interface.go

@@ -98,7 +98,7 @@ type Reader interface {
 	TotalValidators() int32
 	LastCertificate() *certificate.BlockCertificate
 	IsBanned(addr crypto.Address) bool
-	Prune(resultFunc func(pruned, skipped, pruningHeight uint32)) error
+	IsPruned() bool
 }
 
 type Store interface {
@@ -107,6 +107,7 @@ type Store interface {
 	UpdateAccount(addr crypto.Address, acc *account.Account)
 	UpdateValidator(val *validator.Validator)
 	SaveBlock(blk *block.Block, cert *certificate.BlockCertificate)
+	Prune(resultFunc func(pruned, skipped, pruningHeight uint32)) error
 	WriteBatch() error
 	Close()
 }

store/mock.go

@@ -293,3 +293,7 @@ func (*MockStore) IsBanned(_ crypto.Address) bool {
 func (*MockStore) Prune(_ func(pruned, skipped, pruningHeight uint32)) error {
 	return nil
 }
+
+func (*MockStore) IsPruned() bool {
+	return false
+}

store/store.go

@@ -73,6 +73,7 @@ type store struct {
 	txStore        *txStore
 	accountStore   *accountStore
 	validatorStore *validatorStore
+	isPruned       bool
 }
 
 func NewStore(conf *Config) (Store, error) {
@@ -93,21 +94,28 @@ func NewStore(conf *Config) (Store, error) {
 		txStore:        newTxStore(db, conf.TxCacheSize),
 		accountStore:   newAccountStore(db, conf.AccountCacheSize),
 		validatorStore: newValidatorStore(db),
+		isPruned:       false,
 	}
 
-	lc := s.LastCertificate()
+	lc := s.lastCertificate()
 	if lc == nil {
 		return s, nil
 	}
 
+	// Check if the node is pruned by checking genesis block.
+	blockOne, _ := s.block(1)
+	if blockOne == nil {
+		s.isPruned = true
+	}
+
 	currentHeight := lc.Height()
 	startHeight := uint32(1)
 	if currentHeight > conf.TxCacheSize {
 		startHeight = currentHeight - conf.TxCacheSize
 	}
 
 	for i := startHeight; i < currentHeight+1; i++ {
-		committedBlock, err := s.Block(i)
+		committedBlock, err := s.block(i)
 		if err != nil {
 			return nil, err
 		}
@@ -147,6 +155,22 @@ func (s *store) SaveBlock(blk *block.Block, cert *certificate.BlockCertificate)
 	s.txStore.saveTxs(s.batch, blk.Transactions(), regs)
 	s.txStore.pruneCache(height)
 
+	// Removing old block from prune node store.
+	if s.isPruned && height > s.config.RetentionBlocks() {
+		pruneHeight := height - s.config.RetentionBlocks()
+		deleted, err := s.pruneBlock(pruneHeight)
+		if err != nil {
+			panic(err)
+		}
+
+		if deleted {
+			// TODO: Let's use state logger in store[?].
+			logger.Debug("old block is pruned", "height", pruneHeight)
+		} else {
+			logger.Warn("unable to prune the old block", "height", pruneHeight, "error", err)
+		}
+	}
+
 	// Save last certificate: [version: 4 bytes]+[certificate: variant]
 	w := bytes.NewBuffer(make([]byte, 0, 4+cert.SerializeSize()))
 	err := encoding.WriteElements(w, lastStoreVersion)
@@ -387,6 +411,10 @@ func (s *store) IsBanned(addr crypto.Address) bool {
 	return s.config.BannedAddrs[addr]
 }
 
+func (s *store) IsPruned() bool {
+	return s.isPruned
+}
+
 func (s *store) Prune(resultFunc func(pruned, skipped, pruningHeight uint32)) error {
 	cert := s.lastCertificate()
 
@@ -425,17 +453,13 @@ func (s *store) Prune(resultFunc func(pruned, skipped, pruningHeight uint32)) er
 	return nil
 }
 
-func (s *store) pruneBlock(blockHeight uint32) (bool, error) { //nolint
+func (s *store) pruneBlock(blockHeight uint32) (bool, error) {
 	if !s.blockStore.hasBlock(blockHeight) {
 		return false, nil
 	}
 
-	cBlock, err := s.block(blockHeight)
-	if err != nil {
-		return false, err
-	}
-
-	blk, err := cBlock.ToBlock()
+	cBlock, _ := s.block(blockHeight)
+	blk, err := block.FromBytes(cBlock.Data)
 	if err != nil {
 		return false, err
 	}

tests/main_test.go

@@ -72,6 +72,7 @@ func TestMain(m *testing.M) {
 
 		tConfigs[i].TxPool.MinFeePAC = 0.000001
 		tConfigs[i].Store.Path = util.TempDirPath()
+		tConfigs[i].Store.RetentionDays = 10
 		tConfigs[i].Consensus.ChangeProposerTimeout = 4 * time.Second
 		tConfigs[i].Consensus.ChangeProposerDelta = 4 * time.Second
 		tConfigs[i].Consensus.QueryVoteTimeout = 4 * time.Second

types/block/block.go

@@ -134,6 +134,14 @@ func (b *Block) Hash() hash.Hash {
 	return h
 }
 
+func (b *Block) Height() uint32 {
+	if b.data.PrevCert == nil {
+		return 1
+	}
+
+	return b.PrevCertificate().Height() + 1
+}
+
 func (b *Block) String() string {
 	return fmt.Sprintf("{⌘ %v 👤 %v 💻 %v 📨 %d}",
 		b.Hash().ShortString(),

types/block/block_test.go

@@ -341,3 +341,16 @@ func TestMakeBlock(t *testing.T) {
 
 	assert.Equal(t, blk0.Hash(), blk1.Hash())
 }
+
+func TestBlockHeight(t *testing.T) {
+	ts := testsuite.NewTestSuite(t)
+
+	blk1, _ := ts.GenerateTestBlock(1, testsuite.BlockWithPrevCert(nil), testsuite.BlockWithPrevHash(hash.UndefHash))
+	blk2, _ := ts.GenerateTestBlock(2)
+
+	assert.NoError(t, blk1.BasicCheck())
+	assert.NoError(t, blk2.BasicCheck())
+
+	assert.Equal(t, uint32(1), blk1.Height())
+	assert.Equal(t, uint32(2), blk2.Height())
+}