reverted executer key

block/block.go

@@ -62,7 +62,7 @@ func (m *Manager) applyBlock(block *types.Block, commit *types.Commit, blockMeta
 	}
 
 	m.State = newState
-	dbBatch, err = m.Store.UpdateState(m.State, dbBatch)
+	dbBatch, err = m.Store.SaveState(m.State, dbBatch)
 	if err != nil {
 		dbBatch.Discard()
 		return fmt.Errorf("update state: %w", err)
@@ -79,15 +79,15 @@ func (m *Manager) applyBlock(block *types.Block, commit *types.Commit, blockMeta
 	}
 
 	// Commit block to app
-	appHash, retainHeight, err := m.Executor.Commit(&newState, block, responses)
+	appHash, retainHeight, err := m.Executor.Commit(newState, block, responses)
 	if err != nil {
 		return fmt.Errorf("commit block: %w", err)
 	}
 
 	// Update the state with the new app hash, last validators and store height from the commit.
 	// Every one of those, if happens before commit, prevents us from re-executing the block in case failed during commit.
-	newState.SetABCICommitResult(responses, appHash, block.Header.Height)
-	_, err = m.Store.UpdateState(newState, nil)
+	m.Executor.UpdateStateFromCommitResponse(&newState, responses, appHash, block.Header.Height)
+	_, err = m.Store.SaveState(newState, nil)
 	if err != nil {
 		return fmt.Errorf("final update state: %w", err)
 	}
@@ -100,12 +100,13 @@ func (m *Manager) applyBlock(block *types.Block, commit *types.Commit, blockMeta
 		} else {
 			m.logger.Debug("pruned blocks", "pruned", pruned, "retain_height", retainHeight)
 		}
-		newState.BaseHeight = m.State.Base()
-		//TODO: update state
+		newState.BaseHeight = m.State.BaseHeight
+		_, err = m.Store.SaveState(newState, nil)
+		if err != nil {
+			return fmt.Errorf("final update state: %w", err)
+		}
 	}
-
 	m.State = newState
-
 	return nil
 }
 

block/executor.go

@@ -13,8 +13,6 @@ import (
 	tmtypes "github.com/tendermint/tendermint/types"
 	"go.uber.org/multierr"
 
-	libp2pcrypto "github.com/libp2p/go-libp2p/core/crypto"
-
 	"github.com/dymensionxyz/dymint/mempool"
 	"github.com/dymensionxyz/dymint/types"
 )
@@ -35,12 +33,7 @@ type Executor struct {
 
 // NewExecutor creates new instance of BlockExecutor.
 // Proposer address and namespace ID will be used in all newly created blocks.
-func NewExecutor(proposerKey libp2pcrypto.PrivKey, namespaceID string, chainID string, mempool mempool.Mempool, proxyApp proxy.AppConns, eventBus *tmtypes.EventBus, logger types.Logger) (*Executor, error) {
-	proposerAddress, err := getAddress(proposerKey)
-	if err != nil {
-		return nil, err
-	}
-
+func NewExecutor(proposerAddress []byte, namespaceID string, chainID string, mempool mempool.Mempool, proxyApp proxy.AppConns, eventBus *tmtypes.EventBus, logger types.Logger) (*Executor, error) {
 	bytes, err := hex.DecodeString(namespaceID)
 	if err != nil {
 		return nil, err
@@ -141,13 +134,14 @@ func (e *Executor) CreateBlock(height uint64, lastCommit *types.Commit, lastHead
 }
 
 // Commit commits the block
-func (e *Executor) Commit(state *types.State, block *types.Block, resp *tmstate.ABCIResponses) ([]byte, int64, error) {
+func (e *Executor) Commit(state types.State, block *types.Block, resp *tmstate.ABCIResponses) ([]byte, int64, error) {
 	appHash, retainHeight, err := e.commit(state, block, resp.DeliverTxs)
 	if err != nil {
 		return nil, 0, err
 	}
 
-	err = e.publishEvents(resp, block, *state)
+	//FIXME: state is wrong here
+	err = e.publishEvents(resp, block, state)
 	if err != nil {
 		e.logger.Error("fire block events", "error", err)
 		return nil, 0, err
@@ -160,7 +154,7 @@ func (e *Executor) GetAppInfo() (*abci.ResponseInfo, error) {
 	return e.proxyAppQueryConn.InfoSync(abci.RequestInfo{})
 }
 
-func (e *Executor) commit(state *types.State, block *types.Block, deliverTxs []*abci.ResponseDeliverTx) ([]byte, int64, error) {
+func (e *Executor) commit(state types.State, block *types.Block, deliverTxs []*abci.ResponseDeliverTx) ([]byte, int64, error) {
 	e.mempool.Lock()
 	defer e.mempool.Unlock()
 

block/executor_test.go

@@ -185,10 +185,10 @@ func TestApplyBlock(t *testing.T) {
 	newState, err := executor.UpdateStateFromResponses(resp, state, block)
 	require.NoError(err)
 	require.NotNil(newState)
-	assert.Equal(1, newState.LastBlockHeight)
-	_, err = executor.Commit(&newState, block, resp)
+	assert.Equal(uint64(1), newState.LastBlockHeight)
+	appHash, _, err := executor.Commit(newState, block, resp)
 	require.NoError(err)
-	assert.Equal(mockAppHash, newState.AppHash)
+	assert.Equal(mockAppHash, appHash)
 	newState.LastStoreHeight = uint64(newState.LastBlockHeight)
 
 	// Create another block with multiple Tx from mempool
@@ -240,7 +240,7 @@ func TestApplyBlock(t *testing.T) {
 	require.NoError(err)
 	require.NotNil(newState)
 	assert.Equal(uint64(2), newState.LastBlockHeight)
-	_, err = executor.Commit(&newState, block, resp)
+	_, _, err = executor.Commit(newState, block, resp)
 	require.NoError(err)
 
 	// wait for at least 4 Tx events, for up to 3 second.

block/gossip.go

@@ -0,0 +1,76 @@
+package block
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/dymensionxyz/dymint/p2p"
+	"github.com/dymensionxyz/dymint/types"
+	"github.com/tendermint/tendermint/libs/pubsub"
+)
+
+// onNewGossippedBlock will take a block and apply it
+func (m *Manager) onNewGossipedBlock(event pubsub.Message) {
+	m.retrieverMutex.Lock() // needed to protect blockCache access
+	eventData := event.Data().(p2p.GossipedBlock)
+	block := eventData.Block
+	commit := eventData.Commit
+	m.logger.Debug("Received new block via gossip", "height", block.Header.Height, "n cachedBlocks", len(m.blockCache))
+
+	nextHeight := m.State.NextHeight()
+	if block.Header.Height >= nextHeight {
+		m.blockCache[block.Header.Height] = CachedBlock{
+			Block:  &block,
+			Commit: &commit,
+		}
+		m.logger.Debug("caching block", "block height", block.Header.Height, "store height", m.State.Height())
+	}
+	m.retrieverMutex.Unlock() // have to give this up as it's locked again in attempt apply, and we're not re-entrant
+	err := m.attemptApplyCachedBlocks()
+	if err != nil {
+		m.logger.Error("applying cached blocks", "err", err)
+	}
+}
+
+func (m *Manager) attemptApplyCachedBlocks() error {
+	m.retrieverMutex.Lock()
+	defer m.retrieverMutex.Unlock()
+
+	for {
+		expectedHeight := m.State.NextHeight()
+
+		cachedBlock, blockExists := m.blockCache[expectedHeight]
+		if !blockExists {
+			break
+		}
+		if err := m.validateBlock(cachedBlock.Block, cachedBlock.Commit); err != nil {
+			delete(m.blockCache, cachedBlock.Block.Header.Height)
+			/// TODO: can we take an action here such as dropping the peer / reducing their reputation?
+			return fmt.Errorf("block not valid at height %d, dropping it: err:%w", cachedBlock.Block.Header.Height, err)
+		}
+
+		err := m.applyBlock(cachedBlock.Block, cachedBlock.Commit, blockMetaData{source: gossipedBlock})
+		if err != nil {
+			return fmt.Errorf("apply cached block: expected height: %d: %w", expectedHeight, err)
+		}
+		m.logger.Debug("applied cached block", "height", expectedHeight)
+
+		delete(m.blockCache, cachedBlock.Block.Header.Height)
+	}
+
+	return nil
+}
+
+func (m *Manager) gossipBlock(ctx context.Context, block types.Block, commit types.Commit) error {
+	gossipedBlock := p2p.GossipedBlock{Block: block, Commit: commit}
+	gossipedBlockBytes, err := gossipedBlock.MarshalBinary()
+	if err != nil {
+		return fmt.Errorf("marshal binary: %w: %w", err, ErrNonRecoverable)
+	}
+	if err := m.p2pClient.GossipBlock(ctx, gossipedBlockBytes); err != nil {
+		// Although this boils down to publishing on a topic, we don't want to speculate too much on what
+		// could cause that to fail, so we assume recoverable.
+		return fmt.Errorf("p2p gossip block: %w: %w", err, ErrRecoverable)
+	}
+	return nil
+}

block/initchain.go

@@ -24,8 +24,8 @@ func (m *Manager) RunInitChain(ctx context.Context) error {
 
 	// update the state with only the consensus pubkey
 	m.Executor.UpdateStateAfterInitChain(&m.State, res, gensisValSet)
-	m.Executor.UpdateMempoolAfterInitChain(&m.State)
-	if _, err := m.Store.UpdateState(m.State, nil); err != nil {
+	m.Executor.UpdateMempoolAfterInitChain(m.State)
+	if _, err := m.Store.SaveState(m.State, nil); err != nil {
 		return err
 	}
 

block/manager.go

@@ -86,7 +86,12 @@ func NewManager(
 	p2pClient *p2p.Client,
 	logger types.Logger,
 ) (*Manager, error) {
-	exec, err := NewExecutor(proposerKey, conf.NamespaceID, genesis.ChainID, mempool, proxyApp, eventBus, logger)
+	proposerAddress, err := getAddress(proposerKey)
+	if err != nil {
+		return nil, err
+	}
+
+	exec, err := NewExecutor(proposerAddress, conf.NamespaceID, genesis.ChainID, mempool, proxyApp, eventBus, logger)
 	if err != nil {
 		return nil, fmt.Errorf("create block executor: %w", err)
 	}
@@ -170,8 +175,7 @@ func (m *Manager) syncBlockManager() error {
 	res, err := m.SLClient.RetrieveBatch()
 	if errors.Is(err, gerr.ErrNotFound) {
 		// The SL hasn't got any batches for this chain yet.
-		m.logger.Info("No batches for chain found in SL. Start writing first batch.")
-		//FIXME: set correct syncTarget
+		m.logger.Info("No batches for chain found in SL.")
 		m.SyncTarget.Store(uint64(m.Genesis.InitialHeight - 1))
 		return nil
 	}
@@ -196,18 +200,3 @@ func (m *Manager) UpdateSyncParams(endHeight uint64) {
 	m.logger.Info("Received new syncTarget", "syncTarget", endHeight)
 	m.SyncTarget.Store(endHeight)
 }
-
-// getInitialState tries to load lastState from Store, and if it's not available it reads GenesisDoc.
-func getInitialState(store store.Store, genesis *tmtypes.GenesisDoc, logger types.Logger) (s types.State, err error) {
-	s, err = store.LoadState()
-	if errors.Is(err, types.ErrNoStateFound) {
-		logger.Info("failed to find state in the store, creating new state from genesis")
-		s, err = types.NewFromGenesisDoc(genesis)
-	}
-
-	if err != nil {
-		return types.State{}, fmt.Errorf("get initial state: %w", err)
-	}
-
-	return s, nil
-}

block/manager_test.go

@@ -47,7 +47,7 @@ func TestInitialState(t *testing.T) {
 	// Init empty store and full store
 	emptyStore := store.New(store.NewDefaultInMemoryKVStore())
 	fullStore := store.New(store.NewDefaultInMemoryKVStore())
-	_, err = fullStore.UpdateState(sampleState, nil)
+	_, err = fullStore.SaveState(sampleState, nil)
 	require.NoError(t, err)
 
 	// Init p2p client

block/produce.go

@@ -19,7 +19,6 @@ import (
 func (m *Manager) ProduceBlockLoop(ctx context.Context) {
 	m.logger.Debug("Started produce loop")
 
-	// Main ticker for block production
 	ticker := time.NewTicker(m.Conf.BlockTime)
 	defer ticker.Stop()
 
@@ -90,19 +89,14 @@ func (m *Manager) ProduceAndGossipBlock(ctx context.Context, allowEmpty bool) (*
 
 func (m *Manager) produceBlock(allowEmpty bool) (*types.Block, *types.Commit, error) {
 	var (
-		lastCommit     *types.Commit
-		lastHeaderHash [32]byte
-		newHeight      uint64
 		err            error
+		lastHeaderHash [32]byte
+		lastCommit     = &types.Commit{}
+		newHeight      = m.State.NextHeight()
 	)
 
-	if m.State.IsGenesis() {
-		newHeight = uint64(m.State.InitialHeight)
-		lastCommit = &types.Commit{}
-		m.State.BaseHeight = newHeight
-	} else {
-		height := m.State.Height()
-		newHeight = height + 1
+	if !m.State.IsGenesis() {
+		height := newHeight - 1
 		lastCommit, err = m.Store.LoadCommit(height)
 		if err != nil {
 			return nil, nil, fmt.Errorf("load commit: height: %d: %w: %w", height, err, ErrNonRecoverable)
@@ -190,6 +184,7 @@ func (m *Manager) createTMSignature(block *types.Block, proposerAddress []byte,
 	}
 	v := vote.ToProto()
 	// convert libp2p key to tm key
+	//TODO: move to types
 	raw_key, _ := m.ProposerKey.Raw()
 	tmprivkey := tmed25519.PrivKey(raw_key)
 	tmprivkey.PubKey().Bytes()