feat(da): RetrieveBatchesV2 method

.mockery.yaml

@@ -10,10 +10,10 @@ packages:
   github.com/dymensionxyz/dymint/settlement/dymension:
     interfaces:
       CosmosClient:
-  github.com/dymensionxyz/dymension/v3/x/sequencer/types:
+  github.com/dymensionxyz/dymint/third_party/dymension/sequencer/types:
     interfaces:
       QueryClient:
-  github.com/dymensionxyz/dymension/v3/x/rollapp/types:
+  github.com/dymensionxyz/dymint/third_party/dymension/rollapp/types:
     interfaces:
       QueryClient:
   github.com/tendermint/tendermint/abci/types:
@@ -32,5 +32,10 @@ packages:
   github.com/dymensionxyz/dymint/da:
     interfaces:
      DataAvailabilityLayerClient:
+  github.com/dymensionxyz/dymint/p2p:
+    interfaces:
+     GetProposerI:
+
+
 
 

block/block.go

@@ -14,88 +14,110 @@ import (
 // As the entire process can't be atomic we need to make sure the following condition apply before
 // - block height is the expected block height on the store (height + 1).
 // - block height is the expected block height on the app (last block height + 1).
-func (m *Manager) applyBlock(block *types.Block, commit *types.Commit, blockMetaData blockMetaData) error {
+func (m *Manager) applyBlock(block *types.Block, commit *types.Commit, blockMetaData types.BlockMetaData) error {
+	var retainHeight int64
+
 	// TODO: add switch case to have defined behavior for each case.
 	// validate block height
 	if block.Header.Height != m.State.NextHeight() {
 		return types.ErrInvalidBlockHeight
 	}
 
-	m.logger.Debug("Applying block", "height", block.Header.Height, "source", blockMetaData.source)
+	types.SetLastAppliedBlockSource(blockMetaData.Source.String())
+
+	m.logger.Debug("Applying block", "height", block.Header.Height, "source", blockMetaData.Source.String())
 
 	// Check if the app's last block height is the same as the currently produced block height
 	isBlockAlreadyApplied, err := m.isHeightAlreadyApplied(block.Header.Height)
 	if err != nil {
 		return fmt.Errorf("check if block is already applied: %w", err)
 	}
-	// In case the following true, it means we crashed after the commit and before updating the store height.
-	// In that case we'll want to align the store with the app state and continue to the next block.
+	// In case the following true, it means we crashed after the app commit but before updating the state
+	// In that case we'll want to align the state with the app commit result, as if the block was applied.
 	if isBlockAlreadyApplied {
-		// In this case, where the app was committed, but the state wasn't updated
-		// it will update the state from appInfo, saved responses and validators.
 		err := m.UpdateStateFromApp()
 		if err != nil {
 			return fmt.Errorf("update state from app: %w", err)
 		}
-		m.logger.Debug("Aligned with app state required. Skipping to next block", "height", block.Header.Height)
-		return nil
-	}
-	// Start applying the block assuming no inconsistency was found.
-	_, err = m.Store.SaveBlock(block, commit, nil)
-	if err != nil {
-		return fmt.Errorf("save block: %w", err)
-	}
+		m.logger.Info("updated state from app commit", "height", block.Header.Height)
+	} else {
+		var appHash []byte
+		// Start applying the block assuming no inconsistency was found.
+		_, err = m.Store.SaveBlock(block, commit, nil)
+		if err != nil {
+			return fmt.Errorf("save block: %w", err)
+		}
 
-	responses, err := m.Executor.ExecuteBlock(m.State, block)
-	if err != nil {
-		return fmt.Errorf("execute block: %w", err)
-	}
+		err := m.saveP2PBlockToBlockSync(block, commit)
+		if err != nil {
+			m.logger.Error("save block blocksync", "err", err)
+		}
 
-	dbBatch := m.Store.NewBatch()
-	dbBatch, err = m.Store.SaveBlockResponses(block.Header.Height, responses, dbBatch)
-	if err != nil {
-		dbBatch.Discard()
-		return fmt.Errorf("save block responses: %w", err)
-	}
+		responses, err := m.Executor.ExecuteBlock(m.State, block)
+		if err != nil {
+			return fmt.Errorf("execute block: %w", err)
+		}
+
+		_, err = m.Store.SaveBlockResponses(block.Header.Height, responses, nil)
+		if err != nil {
+			return fmt.Errorf("save block responses: %w", err)
+		}
 
-	// Get the validator changes from the app
-	validators := m.State.NextValidators.Copy() // TODO: this will be changed when supporting multiple sequencers from the hub
+		// Commit block to app
+		appHash, retainHeight, err = m.Executor.Commit(m.State, block, responses)
+		if err != nil {
+			return fmt.Errorf("commit block: %w", err)
+		}
 
-	dbBatch, err = m.Store.SaveValidators(block.Header.Height, validators, dbBatch)
-	if err != nil {
-		dbBatch.Discard()
-		return fmt.Errorf("save validators: %w", err)
+		// Update the state with the new app hash, 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.
+		m.Executor.UpdateStateAfterCommit(m.State, responses, appHash, block.Header.Height)
 	}
 
-	err = dbBatch.Commit()
+	// check if the proposer needs to be changed
+	switchRole := m.Executor.UpdateProposerFromBlock(m.State, block)
+
+	// save sequencers to store to be queried over RPC
+	batch := m.Store.NewBatch()
+	batch, err = m.Store.SaveSequencers(block.Header.Height, &m.State.Sequencers, batch)
 	if err != nil {
-		return fmt.Errorf("commit batch to disk: %w", err)
+		return fmt.Errorf("save sequencers: %w", err)
 	}
 
-	// Commit block to app
-	appHash, retainHeight, err := m.Executor.Commit(m.State, block, responses)
+	batch, err = m.Store.SaveState(m.State, batch)
 	if err != nil {
-		return fmt.Errorf("commit block: %w", err)
+		return fmt.Errorf("update state: %w", err)
 	}
 
-	// If failed here, after the app committed, but before the state is updated, we'll update the state on
-	// UpdateStateFromApp using the saved responses and validators.
-
-	// 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.
-	m.Executor.UpdateStateAfterCommit(m.State, responses, appHash, block.Header.Height, validators)
-	_, err = m.Store.SaveState(m.State, nil)
+	err = batch.Commit()
 	if err != nil {
-		return fmt.Errorf("update state: %w", err)
+		return fmt.Errorf("commit state: %w", err)
 	}
 
+	types.RollappHeightGauge.Set(float64(block.Header.Height))
+
 	// Prune old heights, if requested by ABCI app.
 	if 0 < retainHeight {
-		err = m.pruneBlocks(uint64(retainHeight))
+		err = m.PruneBlocks(uint64(retainHeight))
 		if err != nil {
 			m.logger.Error("prune blocks", "retain_height", retainHeight, "err", err)
 		}
 	}
+
+	m.blockCache.Delete(block.Header.Height)
+
+	if switchRole {
+		// TODO: graceful role change (https://github.com/dymensionxyz/dymint/issues/1008)
+		m.logger.Info("Node changing to proposer role")
+		panic("sequencer is no longer the proposer")
+	}
+
+	// validate whether configuration params and rollapp consensus params keep in line, after rollapp params are updated from the responses received in the block execution
+	err = m.ValidateConfigWithRollappParams()
+	if err != nil {
+		return err
+	}
+
 	return nil
 }
 
@@ -120,32 +142,28 @@ func (m *Manager) attemptApplyCachedBlocks() error {
 	for {
 		expectedHeight := m.State.NextHeight()
 
-		cachedBlock, blockExists := m.blockCache[expectedHeight]
+		cachedBlock, blockExists := m.blockCache.Get(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?
+		if err := m.validateBlockBeforeApply(cachedBlock.Block, cachedBlock.Commit); err != nil {
+			m.blockCache.Delete(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})
+		err := m.applyBlock(cachedBlock.Block, cachedBlock.Commit, types.BlockMetaData{Source: cachedBlock.Source})
 		if err != nil {
 			return fmt.Errorf("apply cached block: expected height: %d: %w", expectedHeight, err)
 		}
 		m.logger.Info("Block applied", "height", expectedHeight)
 
-		delete(m.blockCache, cachedBlock.Block.Header.Height)
 	}
 
 	return nil
 }
 
-func (m *Manager) validateBlock(block *types.Block, commit *types.Commit) error {
-	// Currently we're assuming proposer is never nil as it's a pre-condition for
-	// dymint to start
-	proposer := m.SLClient.GetProposer()
-
-	return types.ValidateProposedTransition(m.State, block, commit, proposer)
+// This function validates the block and commit against the state before applying it.
+func (m *Manager) validateBlockBeforeApply(block *types.Block, commit *types.Commit) error {
+	return types.ValidateProposedTransition(m.State, block, commit, m.GetProposerPubKey())
 }

block/block_cache.go

@@ -0,0 +1,34 @@
+package block
+
+import (
+	"github.com/dymensionxyz/dymint/types"
+)
+
+type Cache struct {
+	// concurrency managed by Manager.retrieverMu mutex
+	cache map[uint64]types.CachedBlock
+}
+
+func (m *Cache) Add(h uint64, b *types.Block, c *types.Commit, source types.BlockSource) {
+	m.cache[h] = types.CachedBlock{Block: b, Commit: c, Source: source}
+	types.BlockCacheSizeGauge.Set(float64(m.Size()))
+}
+
+func (m *Cache) Delete(h uint64) {
+	delete(m.cache, h)
+	types.BlockCacheSizeGauge.Set(float64(m.Size()))
+}
+
+func (m *Cache) Get(h uint64) (types.CachedBlock, bool) {
+	ret, found := m.cache[h]
+	return ret, found
+}
+
+func (m *Cache) Has(h uint64) bool {
+	_, found := m.Get(h)
+	return found
+}
+
+func (m *Cache) Size() int {
+	return len(m.cache)
+}

block/errors.go

@@ -5,5 +5,4 @@ import "errors"
 var (
 	ErrNonRecoverable = errors.New("non recoverable")
 	ErrRecoverable    = errors.New("recoverable")
-	ErrWrongDA        = errors.New("wrong DA")
 )