[Persistence] TxIndexer to StateHash Bridge (Issue-#315)

consensus/CHANGELOG.md

@@ -7,7 +7,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [0.0.0.7] - 2022-11-01
+
+- Removed `apphash` and `txResults` from `consensusModule` structure
+- Modified lifecycle to `set` the proposal block within a `PersistenceContext`
+- Allow block and parts to be committed with the persistence context
+
 ## [0.0.0.6] - 2022-10-12
+
 - Stores transactions alongside blocks during `commit`
 - Added current block `[]TxResult` to the module
 

consensus/block.go

@@ -5,29 +5,11 @@ import (
 	"unsafe"
 
 	typesCons "github.com/pokt-network/pocket/consensus/types"
-	"github.com/pokt-network/pocket/shared/codec"
 )
 
 func (m *consensusModule) commitBlock(block *typesCons.Block) error {
 	m.nodeLog(typesCons.CommittingBlock(m.Height, len(block.Transactions)))
 
-	// Store the block in the KV store
-	codec := codec.GetCodec()
-	blockProtoBytes, err := codec.Marshal(block)
-	if err != nil {
-		return err
-	}
-
-	// IMPROVE(olshansky): temporary solution. `ApplyBlock` above applies the
-	// transactions to the postgres database, and this stores it in the KV store upon commitment.
-	// Instead of calling this directly, an alternative solution is to store the block metadata in
-	// the persistence context and have `CommitPersistenceContext` do this under the hood. However,
-	// additional `Block` metadata will need to be passed through and may change when we merkle the
-	// state hash.
-	if err := m.storeBlock(block, blockProtoBytes); err != nil {
-		return err
-	}
-
 	// Commit and release the context
 	if err := m.utilityContext.CommitPersistenceContext(); err != nil {
 		return err
@@ -36,31 +18,6 @@ func (m *consensusModule) commitBlock(block *typesCons.Block) error {
 	m.utilityContext.ReleaseContext()
 	m.utilityContext = nil
 
-	m.lastAppHash = block.BlockHeader.Hash
-
-	return nil
-}
-
-// IMPROVE(#285): ensure these persistence operations are atomic, we can't commit block without
-// committing the transactions and metadata at the same time
-func (m *consensusModule) storeBlock(block *typesCons.Block, blockProtoBytes []byte) error {
-	store := m.utilityContext.GetPersistenceContext()
-	// Store in KV Store
-	if err := store.StoreBlock(blockProtoBytes); err != nil {
-		return err
-	}
-	// Store transactions in Indexer
-	for _, txResult := range m.TxResults {
-		if err := store.StoreTransaction(txResult); err != nil {
-			return err
-		}
-	}
-
-	// Store in SQL Store
-	header := block.BlockHeader
-	if err := store.InsertBlock(uint64(header.Height), header.Hash, header.ProposerAddress, header.QuorumCertificate); err != nil {
-		return err
-	}
 	return nil
 }
 

consensus/consensus_tests/utils_test.go

@@ -359,22 +359,23 @@ func baseUtilityContextMock(t *testing.T) *modulesMock.MockUtilityContext {
 	ctrl := gomock.NewController(t)
 	utilityContextMock := modulesMock.NewMockUtilityContext(ctrl)
 	persistenceContextMock := modulesMock.NewMockPersistenceRWContext(ctrl)
+	persistenceContextMock.EXPECT().SetProposalBlock(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
+	persistenceContextMock.EXPECT().GetLastAppHash().Return("", nil).AnyTimes()
 
 	utilityContextMock.EXPECT().
-		GetProposalTransactions(gomock.Any(), maxTxBytes, gomock.AssignableToTypeOf(emptyByzValidators)).
-		Return(make([][]byte, 0), nil, nil).
+		CreateAndApplyBlock(gomock.Any(), maxTxBytes).
+		Return(appHash, make([][]byte, 0), nil).
 		AnyTimes()
 	utilityContextMock.EXPECT().
-		ApplyBlock(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
-		Return(appHash, nil, nil).
+		ApplyBlock().
+		Return(appHash, nil).
 		AnyTimes()
 	utilityContextMock.EXPECT().CommitPersistenceContext().Return(nil).AnyTimes()
 	utilityContextMock.EXPECT().ReleaseContext().Return().AnyTimes()
 	utilityContextMock.EXPECT().GetPersistenceContext().Return(persistenceContextMock).AnyTimes()
 
-	persistenceContextMock.EXPECT().StoreTransaction(gomock.Any()).Return(nil).AnyTimes()
-	persistenceContextMock.EXPECT().StoreBlock(gomock.Any()).Return(nil).AnyTimes()
-	persistenceContextMock.EXPECT().InsertBlock(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes().Return(nil)
+	persistenceContextMock.EXPECT().StoreTransactions().Return(nil).AnyTimes()
+	persistenceContextMock.EXPECT().StoreBlock().Return(nil).AnyTimes()
 
 	return utilityContextMock
 }

consensus/helpers.go

@@ -126,7 +126,6 @@ func (m *consensusModule) isOptimisticThresholdMet(n int) error {
 func (m *consensusModule) resetForNewHeight() {
 	m.Round = 0
 	m.Block = nil
-	m.TxResults = nil
 	m.highPrepareQC = nil
 	m.lockedQC = nil
 }

consensus/hotstuff_leader.go

@@ -2,7 +2,7 @@ package consensus
 
 import (
 	"encoding/hex"
-	"github.com/pokt-network/pocket/shared/modules"
+	"github.com/pokt-network/pocket/shared/codec"
 	"unsafe"
 
 	consensusTelemetry "github.com/pokt-network/pocket/consensus/telemetry"
@@ -54,27 +54,24 @@ func (handler *HotstuffLeaderMessageHandler) HandleNewRoundMessage(m *consensusM
 	// TODO: Add more unit tests for these checks...
 	if m.shouldPrepareNewBlock(highPrepareQC) {
 		// Leader prepares a new block if `highPrepareQC` is not applicable
-		block, txResults, err := m.prepareAndApplyBlock()
+		block, err := m.prepareAndApplyBlock()
 		if err != nil {
 			m.nodeLogError(typesCons.ErrPrepareBlock.Error(), err)
 			m.paceMaker.InterruptRound()
 			return
 		}
 		m.Block = block
-		m.TxResults = txResults
 	} else {
 		// DISCUSS: Do we need to call `validateProposal` here?
 		// Leader acts like a replica if `highPrepareQC` is not `nil`
 		// TODO(olshansky): Add test to make sure same block is not applied twice if round is interrrupted.
 		// been 'Applied'
-		txResults, err := m.applyBlock(highPrepareQC.Block)
-		if err != nil {
+		if err := m.applyBlock(highPrepareQC.Block); err != nil {
 			m.nodeLogError(typesCons.ErrApplyBlock.Error(), err)
 			m.paceMaker.InterruptRound()
 			return
 		}
 		m.Block = highPrepareQC.Block
-		m.TxResults = txResults
 	}
 
 	m.Step = Prepare
@@ -337,36 +334,33 @@ func (m *consensusModule) tempIndexHotstuffMessage(msg *typesCons.HotstuffMessag
 
 // This is a helper function intended to be called by a leader/validator during a view change
 // to prepare a new block that is applied to the new underlying context.
-func (m *consensusModule) prepareAndApplyBlock() (*typesCons.Block, []modules.TxResult, error) {
+func (m *consensusModule) prepareAndApplyBlock() (*typesCons.Block, error) {
 	if m.isReplica() {
-		return nil, nil, typesCons.ErrReplicaPrepareBlock
+		return nil, typesCons.ErrReplicaPrepareBlock
 	}
 
 	// TECHDEBT: Retrieve this from consensus consensus config
 	maxTxBytes := 90000
 
-	// TECHDEBT: Retrieve this from persistence
-	lastByzValidators := make([][]byte, 0)
-
 	// Reap the mempool for transactions to be applied in this block
-	txs, _, err := m.utilityContext.GetProposalTransactions(m.privateKey.Address(), maxTxBytes, lastByzValidators)
+	appHash, txs, err := m.utilityContext.CreateAndApplyProposalBlock(m.privateKey.Address(), maxTxBytes)
 	if err != nil {
-		return nil, nil, err
+		return nil, err
 	}
 
-	// OPTIMIZE: Determine if we can avoid the `ApplyBlock` call here
-	// Apply all the transactions in the block
-	appHash, txResults, err := m.utilityContext.ApplyBlock(int64(m.Height), m.privateKey.Address(), txs, lastByzValidators)
+	persistenceContext := m.utilityContext.GetPersistenceContext()
+
+	lastAppHash, err := persistenceContext.GetPrevAppHash()
 	if err != nil {
-		return nil, nil, err
+		return nil, err
 	}
 
 	// Construct the block
 	blockHeader := &typesCons.BlockHeader{
 		Height:            int64(m.Height),
 		Hash:              hex.EncodeToString(appHash),
 		NumTxs:            uint32(len(txs)),
-		LastBlockHash:     m.lastAppHash,
+		LastBlockHash:     lastAppHash, // IMRPROVE: this should be a block hash not the appHash
 		ProposerAddress:   m.privateKey.Address().Bytes(),
 		QuorumCertificate: []byte("HACK: Temporary placeholder"),
 	}
@@ -375,7 +369,18 @@ func (m *consensusModule) prepareAndApplyBlock() (*typesCons.Block, []modules.Tx
 		Transactions: txs,
 	}
 
-	return block, txResults, nil
+	cdc := codec.GetCodec()
+	blockProtoBz, err := cdc.Marshal(block)
+	if err != nil {
+		return nil, err
+	}
+
+	// Set the proposal block in the persistence context
+	if err = persistenceContext.SetProposalBlock(blockHeader.Hash, blockProtoBz, blockHeader.ProposerAddress, blockHeader.QuorumCertificate, block.Transactions); err != nil {
+		return nil, err
+	}
+
+	return block, nil
 }
 
 // Return true if this node, the leader, should prepare a new block

consensus/hotstuff_replica.go

@@ -3,11 +3,10 @@ package consensus
 import (
 	"encoding/hex"
 	"fmt"
-	"github.com/pokt-network/pocket/shared/modules"
-
 	consensusTelemetry "github.com/pokt-network/pocket/consensus/telemetry"
 	"github.com/pokt-network/pocket/consensus/types"
 	typesCons "github.com/pokt-network/pocket/consensus/types"
+	"github.com/pokt-network/pocket/shared/codec"
 )
 
 type HotstuffReplicaMessageHandler struct{}
@@ -61,14 +60,12 @@ func (handler *HotstuffReplicaMessageHandler) HandlePrepareMessage(m *consensusM
 	}
 
 	block := msg.GetBlock()
-	txResults, err := m.applyBlock(block)
-	if err != nil {
+	if err := m.applyBlock(block); err != nil {
 		m.nodeLogError(typesCons.ErrApplyBlock.Error(), err)
 		m.paceMaker.InterruptRound()
 		return
 	}
 	m.Block = block
-	m.TxResults = txResults
 	m.Step = PreCommit
 
 	prepareVoteMessage, err := CreateVoteMessage(m.Height, m.Round, Prepare, m.Block, m.privateKey)
@@ -228,22 +225,29 @@ func (m *consensusModule) validateProposal(msg *typesCons.HotstuffMessage) error
 }
 
 // This helper applies the block metadata to the utility & persistence layers
-func (m *consensusModule) applyBlock(block *typesCons.Block) ([]modules.TxResult, error) {
-	// TECHDEBT: Retrieve this from persistence
-	lastByzValidators := make([][]byte, 0)
+func (m *consensusModule) applyBlock(block *typesCons.Block) error {
+	blockProtoBz, err := codec.GetCodec().Marshal(block)
+	if err != nil {
+		return err
+	}
+	persistenceContext := m.utilityContext.GetPersistenceContext()
+	// Set the proposal block in the persistence context
+	if err = persistenceContext.SetProposalBlock(block.BlockHeader.Hash, blockProtoBz, block.BlockHeader.ProposerAddress, block.BlockHeader.QuorumCertificate, block.Transactions); err != nil {
+		return err
+	}
 
 	// Apply all the transactions in the block and get the appHash
-	appHash, txResults, err := m.utilityContext.ApplyBlock(int64(m.Height), block.BlockHeader.ProposerAddress, block.Transactions, lastByzValidators)
+	appHash, err := m.utilityContext.ApplyBlock()
 	if err != nil {
-		return txResults, err
+		return err
 	}
 
-	// CONSOLIDATE: Terminology of `blockHash`, `appHash` and `stateHash`
+	// CONSOLIDATE: Terminology of `appHash` and `stateHash`
 	if block.BlockHeader.Hash != hex.EncodeToString(appHash) {
-		return txResults, typesCons.ErrInvalidAppHash(block.BlockHeader.Hash, hex.EncodeToString(appHash))
+		return typesCons.ErrInvalidAppHash(block.BlockHeader.Hash, hex.EncodeToString(appHash))
 	}
 
-	return txResults, nil
+	return nil
 }
 
 func (m *consensusModule) validateQuorumCertificate(qc *typesCons.QuorumCertificate) error {

consensus/module.go

@@ -59,7 +59,6 @@ type consensusModule struct {
 	idToValAddrMap typesCons.IdToValAddrMap // TODO: This needs to be updated every time the ValMap is modified
 
 	// Consensus State
-	lastAppHash  string // TODO: Always retrieve this variable from the persistence module and simplify this struct
 	validatorMap typesCons.ValidatorMap
 
 	// Module Dependencies
@@ -136,7 +135,6 @@ func (*consensusModule) Create(runtimeMgr modules.RuntimeMgr) (modules.Module, e
 		valAddrToIdMap: valIdMap,
 		idToValAddrMap: idValMap,
 
-		lastAppHash:  "",
 		validatorMap: valMap,
 
 		utilityContext:    nil,
@@ -235,10 +233,6 @@ func (m *consensusModule) HandleMessage(message *anypb.Any) error {
 	return nil
 }
 
-func (m *consensusModule) AppHash() string {
-	return m.lastAppHash
-}
-
 func (m *consensusModule) CurrentHeight() uint64 {
 	return m.Height
 }
@@ -266,13 +260,7 @@ func (m *consensusModule) loadPersistedState() error {
 		return nil
 	}
 
-	appHash, err := persistenceContext.GetBlockHash(int64(latestHeight))
-	if err != nil {
-		return fmt.Errorf("error getting block hash for height %d even though it's in the database: %s", latestHeight, err)
-	}
-
 	m.Height = uint64(latestHeight) + 1 // +1 because the height of the consensus module is where it is actively participating in consensus
-	m.lastAppHash = string(appHash)
 
 	m.nodeLog(fmt.Sprintf("Starting node at height %d", latestHeight))
 	return nil

persistence/CHANGELOG.md

@@ -7,6 +7,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [0.0.0.7] - 2022-11-01
+
+- Ported over storing blocks and block components to the Persistence module from Consensus and Utility modules
+- Encapsulated `TxIndexer` logic to the persistence context only
+
 ## [0.0.0.6] - 2022-10-06
 
 - Don't ignore the exit code of `m.Run()` in the unit tests