Merge remote-tracking branch 'origin/main' into omritoptix/382-dymint…

…-hub-sync-issue

block/manager.go

@@ -9,9 +9,9 @@ import (
 
 	"code.cloudfoundry.org/go-diodes"
 
+	"cosmossdk.io/errors"
 	"github.com/avast/retry-go"
 	cryptocodec "github.com/cosmos/cosmos-sdk/crypto/codec"
-	"github.com/cosmos/cosmos-sdk/types/errors"
 	abciconv "github.com/dymensionxyz/dymint/conv/abci"
 	"github.com/dymensionxyz/dymint/node/events"
 	"github.com/dymensionxyz/dymint/p2p"
@@ -565,24 +565,30 @@ func (m *Manager) alignStoreWithApp(ctx context.Context, block *types.Block) (bo
 	// Validate incosistency in height wasn't caused by a crash and if so handle it.
 	proxyAppInfo, err := m.executor.GetAppInfo()
 	if err != nil {
-		m.logger.Error("Failed to get app info", "error", err)
-		return isRequired, err
-	}
-	if uint64(proxyAppInfo.LastBlockHeight) == block.Header.Height {
-		isRequired = true
-		m.logger.Info("Skipping block application and only updating store height and state hash", "height", block.Header.Height)
-		// update the state with the hash, last store height and last validators.
-		m.lastState.AppHash = *(*[32]byte)(proxyAppInfo.LastBlockAppHash)
-		m.lastState.LastStoreHeight = block.Header.Height
-		m.lastState.LastValidators = m.lastState.Validators.Copy()
-		_, err := m.store.UpdateState(m.lastState, nil)
-		if err != nil {
-			m.logger.Error("Failed to update state", "error", err)
-			return isRequired, err
-		}
-		m.store.SetHeight(block.Header.Height)
+		return isRequired, errors.Wrap(err, "failed to get app info")
+	}
+	if uint64(proxyAppInfo.LastBlockHeight) != block.Header.Height {
 		return isRequired, nil
 	}
+
+	isRequired = true
+	m.logger.Info("Skipping block application and only updating store height and state hash", "height", block.Header.Height)
+	// update the state with the hash, last store height and last validators.
+	m.lastState.AppHash = *(*[32]byte)(proxyAppInfo.LastBlockAppHash)
+	m.lastState.LastStoreHeight = block.Header.Height
+	m.lastState.LastValidators = m.lastState.Validators.Copy()
+
+	resp, err := m.store.LoadBlockResponses(block.Header.Height)
+	if err != nil {
+		return isRequired, errors.Wrap(err, "failed to load block responses")
+	}
+	copy(m.lastState.LastResultsHash[:], tmtypes.NewResults(resp.DeliverTxs).Hash())
+
+	_, err = m.store.UpdateState(m.lastState, nil)
+	if err != nil {
+		return isRequired, errors.Wrap(err, "failed to update state")
+	}
+	m.store.SetHeight(block.Header.Height)
 	return isRequired, nil
 }
 

da/celestia/celestia.go

@@ -21,12 +21,6 @@ import (
 	pb "github.com/dymensionxyz/dymint/types/pb/dymint"
 )
 
-const (
-	defaultTxPollingRetryDelay = 20 * time.Second
-	defaultSubmitRetryDelay    = 10 * time.Second
-	defaultTxPollingAttempts   = 5
-)
-
 type CNCClientI interface {
 	SubmitPFD(ctx context.Context, namespaceID [8]byte, blob []byte, fee int64, gasLimit uint64) (*cnc.TxResponse, error)
 	NamespacedShares(ctx context.Context, namespaceID [8]byte, height uint64) ([][]byte, error)
@@ -50,16 +44,6 @@ type DataAvailabilityLayerClient struct {
 var _ da.DataAvailabilityLayerClient = &DataAvailabilityLayerClient{}
 var _ da.BatchRetriever = &DataAvailabilityLayerClient{}
 
-// Config stores Celestia DALC configuration parameters.
-type Config struct {
-	BaseURL     string        `json:"base_url"`
-	AppNodeURL  string        `json:"app_node_url"`
-	Timeout     time.Duration `json:"timeout"`
-	Fee         int64         `json:"fee"`
-	GasLimit    uint64        `json:"gas_limit"`
-	NamespaceID [8]byte       `json:"namespace_id"`
-}
-
 // WithCNCClient sets CNC client.
 func WithCNCClient(client CNCClientI) da.Option {
 	return func(daLayerClient da.DataAvailabilityLayerClient) {
@@ -99,16 +83,20 @@ func WithSubmitRetryDelay(delay time.Duration) da.Option {
 func (c *DataAvailabilityLayerClient) Init(config []byte, pubsubServer *pubsub.Server, kvStore store.KVStore, logger log.Logger, options ...da.Option) error {
 	c.logger = logger
 
-	if len(config) > 0 {
-		err := json.Unmarshal(config, &c.config)
-		if err != nil {
-			return err
-		}
+	if len(config) <= 0 {
+		return errors.New("config is empty")
+	}
+	err := json.Unmarshal(config, &c.config)
+	if err != nil {
+		return err
+	}
+	err = (&c.config).InitNamespaceID()
+	if err != nil {
+		return err
 	}
 
 	c.pubsubServer = pubsubServer
 	// Set defaults
-	var err error
 	c.txPollingRetryDelay = defaultTxPollingRetryDelay
 	c.txPollingAttempts = defaultTxPollingAttempts
 	c.submitRetryDelay = defaultSubmitRetryDelay

da/celestia/config.go

@@ -0,0 +1,43 @@
+package celestia
+
+import (
+	"encoding/hex"
+	"time"
+)
+
+const (
+	defaultTxPollingRetryDelay = 20 * time.Second
+	defaultSubmitRetryDelay    = 10 * time.Second
+	defaultTxPollingAttempts   = 5
+)
+
+// Config stores Celestia DALC configuration parameters.
+type Config struct {
+	BaseURL        string        `json:"base_url"`
+	AppNodeURL     string        `json:"app_node_url"`
+	Timeout        time.Duration `json:"timeout"`
+	Fee            int64         `json:"fee"`
+	GasLimit       uint64        `json:"gas_limit"`
+	NamespaceIDStr string        `json:"namespace_id"`
+	NamespaceID    [8]byte       `json:"-"`
+}
+
+var CelestiaDefaultConfig = Config{
+	BaseURL:        "http://127.0.0.1:26659",
+	AppNodeURL:     "",
+	Timeout:        30 * time.Second,
+	Fee:            20000,
+	GasLimit:       20000000,
+	NamespaceIDStr: "000000000000ffff",
+	NamespaceID:    [8]byte{0, 0, 0, 0, 0, 0, 255, 255},
+}
+
+func (c *Config) InitNamespaceID() error {
+	// Decode NamespaceID from string to byte array
+	namespaceBytes, err := hex.DecodeString(c.NamespaceIDStr)
+	if err != nil {
+		return err
+	}
+	copy(c.NamespaceID[:], namespaceBytes)
+	return nil
+}

da/da_test.go

@@ -37,16 +37,26 @@ func TestLifecycle(t *testing.T) {
 			t.Skip("TODO")
 		}
 		t.Run(dalc, func(t *testing.T) {
-			doTestLifecycle(t, registry.GetClient(dalc))
+			doTestLifecycle(t, dalc)
 		})
 	}
 }
 
-func doTestLifecycle(t *testing.T, dalc da.DataAvailabilityLayerClient) {
+func doTestLifecycle(t *testing.T, daType string) {
+	var err error
 	require := require.New(t)
 	pubsubServer := pubsub.NewServer()
 	pubsubServer.Start()
-	err := dalc.Init([]byte{}, pubsubServer, nil, test.NewLogger(t))
+
+	dacfg := []byte{}
+	dalc := registry.GetClient(daType)
+
+	if daType == "celestia" {
+		dacfg, err = json.Marshal(celestia.CelestiaDefaultConfig)
+		require.NoError(err)
+	}
+
+	err = dalc.Init(dacfg, pubsubServer, nil, test.NewLogger(t))
 	require.NoError(err)
 
 	err = dalc.Start()

state/executor.go

@@ -175,6 +175,7 @@ func (e *BlockExecutor) Commit(ctx context.Context, state *types.State, block *t
 	}
 
 	copy(state.AppHash[:], appHash[:])
+	copy(state.LastResultsHash[:], tmtypes.NewResults(resp.DeliverTxs).Hash())
 
 	err = e.publishEvents(resp, block, *state)
 	if err != nil {
@@ -203,6 +204,7 @@ func (e *BlockExecutor) updateState(state types.State, block *types.Block, abciR
 	}
 
 	hash := block.Header.Hash()
+	//TODO: we can probably pass the state as a pointer and update it directly
 	s := types.State{
 		Version:         state.Version,
 		ChainID:         state.ChainID,
@@ -223,8 +225,10 @@ func (e *BlockExecutor) updateState(state types.State, block *types.Block, abciR
 		AppHash:         state.AppHash,
 		LastValidators:  state.LastValidators.Copy(),
 		LastStoreHeight: state.LastStoreHeight,
+
+		LastResultsHash: state.LastResultsHash,
+		BaseHeight:      state.BaseHeight,
 	}
-	copy(s.LastResultsHash[:], tmtypes.NewResults(abciResponses.DeliverTxs).Hash())
 
 	return s, nil
 }
@@ -276,7 +280,6 @@ func (e *BlockExecutor) validateBlock(state types.State, block *types.Block) err
 	if !bytes.Equal(block.Header.AppHash[:], state.AppHash[:]) {
 		return errors.New("AppHash mismatch")
 	}
-
 	if !bytes.Equal(block.Header.LastResultsHash[:], state.LastResultsHash[:]) {
 		return errors.New("LastResultsHash mismatch")
 	}

testutil/types.go

@@ -39,40 +39,76 @@ func createRandomHashes() [][32]byte {
 }
 
 // GenerateBlocks generates random blocks.
-func GenerateBlocks(startHeight uint64, num uint64, proposerKey crypto.PrivKey) ([]*types.Block, error) {
+func generateBlock(height uint64) *types.Block {
+	h := createRandomHashes()
+	block := &types.Block{
+		Header: types.Header{
+			Version: types.Version{
+				Block: BlockVersion,
+				App:   AppVersion,
+			},
+			NamespaceID:    [8]byte{0, 1, 2, 3, 4, 5, 6, 7},
+			Height:         height,
+			Time:           4567,
+			LastHeaderHash: h[0],
+			LastCommitHash: h[1],
+			DataHash:       h[2],
+			ConsensusHash:  h[3],
+			// AppHash:         h[4],
+			AppHash:         [32]byte{},
+			LastResultsHash: getEmptyLastResultsHash(),
+			ProposerAddress: []byte{4, 3, 2, 1},
+			AggregatorsHash: h[6],
+		},
+		Data: types.Data{
+			Txs:                    nil,
+			IntermediateStateRoots: types.IntermediateStateRoots{RawRootsList: [][]byte{{0x1}}},
+			Evidence:               types.EvidenceData{Evidence: nil},
+		},
+		LastCommit: types.Commit{
+			Height:     8,
+			HeaderHash: h[7],
+			Signatures: []types.Signature{},
+		},
+	}
+
+	return block
+}
+
+func GenerateBlocksWithTxs(startHeight uint64, num uint64, proposerKey crypto.PrivKey, nTxs int) ([]*types.Block, error) {
 	blocks := make([]*types.Block, num)
 	for i := uint64(0); i < num; i++ {
-		h := createRandomHashes()
-		block := &types.Block{
-			Header: types.Header{
-				Version: types.Version{
-					Block: BlockVersion,
-					App:   AppVersion,
-				},
-				NamespaceID:    [8]byte{0, 1, 2, 3, 4, 5, 6, 7},
-				Height:         i + startHeight,
-				Time:           4567,
-				LastHeaderHash: h[0],
-				LastCommitHash: h[1],
-				DataHash:       h[2],
-				ConsensusHash:  h[3],
-				// AppHash:         h[4],
-				AppHash:         [32]byte{},
-				LastResultsHash: getEmptyLastResultsHash(),
-				ProposerAddress: []byte{4, 3, 2, 1},
-				AggregatorsHash: h[6],
-			},
-			Data: types.Data{
-				Txs:                    nil,
-				IntermediateStateRoots: types.IntermediateStateRoots{RawRootsList: [][]byte{{0x1}}},
-				Evidence:               types.EvidenceData{Evidence: nil},
-			},
-			LastCommit: types.Commit{
-				Height:     8,
-				HeaderHash: h[7],
-				Signatures: []types.Signature{},
+
+		block := generateBlock(i + startHeight)
+
+		block.Data = types.Data{
+			Txs: make(types.Txs, nTxs),
+			IntermediateStateRoots: types.IntermediateStateRoots{
+				RawRootsList: make([][]byte, nTxs),
 			},
 		}
+
+		for i := 0; i < nTxs; i++ {
+			block.Data.Txs[i] = GetRandomTx()
+			block.Data.IntermediateStateRoots.RawRootsList[i] = GetRandomBytes(32)
+		}
+
+		signature, err := generateSignature(proposerKey, &block.Header)
+		if err != nil {
+			return nil, err
+		}
+		block.LastCommit.Signatures = []types.Signature{signature}
+		blocks[i] = block
+	}
+	return blocks, nil
+}
+
+// GenerateBlocks generates random blocks.
+func GenerateBlocks(startHeight uint64, num uint64, proposerKey crypto.PrivKey) ([]*types.Block, error) {
+	blocks := make([]*types.Block, num)
+	for i := uint64(0); i < num; i++ {
+		block := generateBlock(i + startHeight)
+
 		signature, err := generateSignature(proposerKey, &block.Header)
 		if err != nil {
 			return nil, err