fix(bug): sync from da and p2p when starting a node

CHANGELOG.md

@@ -1,7 +1,7 @@
-# [](https://github.com/dymensionxyz/dymint/compare/v1.1.0-rc02...v) (2024-05-01)
+# [](https://github.com/dymensionxyz/dymint/compare/v1.1.0-rc02...v) (2024-05-02)
 
 
 ### Bug Fixes
 
 * **celestia test:** fix race in test ([#755](https://github.com/dymensionxyz/dymint/issues/755)) ([0b36781](https://github.com/dymensionxyz/dymint/commit/0b367818bf6aa8da4a4fd8e4e5c78223b60b44e0))
 * **celestia:** impl retry on submit ([#748](https://github.com/dymensionxyz/dymint/issues/748)) ([61630eb](https://github.com/dymensionxyz/dymint/commit/61630eb458197abe2440a81426210000dff25d40))
@@ -12,17 +12,18 @@
 * **p2p:** validate block before applying and not before caching in p2p gossiping ([#723](https://github.com/dymensionxyz/dymint/issues/723)) ([98371b5](https://github.com/dymensionxyz/dymint/commit/98371b5220613e70f3274fab5593e02ba532f7db))
 * **produce loop:** handle unauthenticated error in settlement layer ([#726](https://github.com/dymensionxyz/dymint/issues/726)) ([33e78d1](https://github.com/dymensionxyz/dymint/commit/33e78d116b5f14b91b8b3bda2b6cbfee9040e2d3))
 * **rpc:** nil panic in rpc/json/handler.go WriteError ([#750](https://github.com/dymensionxyz/dymint/issues/750)) ([e09709b](https://github.com/dymensionxyz/dymint/commit/e09709b428a33da002defb9f13178fa19b81a69b))
+* **sync:** make sure we use a latest state index as a start point ([#760](https://github.com/dymensionxyz/dymint/issues/760)) ([43e2d96](https://github.com/dymensionxyz/dymint/commit/43e2d965f2b505751f8e5260549e909c976141ee))
 
 
 
 # [1.1.0-rc02](https://github.com/dymensionxyz/dymint/compare/v1.1.0-rc01...v1.1.0-rc02) (2024-04-26)
 
 
 
 # [1.1.0-rc01](https://github.com/dymensionxyz/dymint/compare/v1.0.1-alpha...v1.1.0-rc01) (2024-04-25)
 
 
 ### Bug Fixes
 
 * **block production:** apply block before gossiping ([#695](https://github.com/dymensionxyz/dymint/issues/695)) ([5c496b4](https://github.com/dymensionxyz/dymint/commit/5c496b453e98bbcc67feb6df3a2d4ad340586816))
 * **block:** Only register `nodeHealthStatusHandler` for sequencer ([#683](https://github.com/dymensionxyz/dymint/issues/683)) ([da2ff94](https://github.com/dymensionxyz/dymint/commit/da2ff94bcdd064109da703fa885609846a94e180))

block/manager.go

@@ -166,6 +166,9 @@
 			return err
 		}
 	}
+	if !isAggregator {
+		go uevent.MustSubscribe(ctx, m.Pubsub, "applyBlockLoop", p2p.EventQueryNewNewGossipedBlock, m.onNewGossipedBlock, m.logger)
+	}
 
 	err := m.syncBlockManager()
 	if err != nil {
@@ -178,7 +181,6 @@
 		go m.ProduceBlockLoop(ctx)
 		go m.SubmitLoop(ctx)
 	} else {
-		go uevent.MustSubscribe(ctx, m.Pubsub, "applyBlockLoop", p2p.EventQueryNewNewGossipedBlock, m.onNewGossipedBlock, m.logger, 100)
 		go m.RetrieveLoop(ctx)
 		go m.SyncTargetLoop(ctx)
 	}
@@ -236,9 +238,7 @@
 // TODO: move to gossip.go
 // onNewGossippedBlock will take a block and apply it
 func (m *Manager) onNewGossipedBlock(event pubsub.Message) {
-	m.retrieverMutex.Lock() // needed to protect blockCache access
-
-	m.logger.Debug("Received new block via gossip", "n cachedBlocks", len(m.blockCache))
+	m.logger.Debug("Received new block event", "eventData", event.Data(), "cachedBlocks", len(m.blockCache))
 	eventData := event.Data().(p2p.GossipedBlock)
 	block := eventData.Block
 	commit := eventData.Commit
@@ -252,8 +252,6 @@
 		m.logger.Debug("caching block", "block height", block.Header.Height, "store height", m.Store.Height())
 	}
 
-	m.retrieverMutex.Unlock() // have to give this up as it's locked again in attempt apply, and we're not re-entrant
-
 	if block.Header.Height == nextHeight {
 		err := m.attemptApplyCachedBlocks()
 		if err != nil {

block/retriever.go

@@ -3,7 +3,6 @@ package block
 import (
 	"context"
 	"fmt"
-	"sync/atomic"
 	"time"
 
 	"github.com/avast/retry-go/v4"
@@ -44,33 +43,33 @@ func (m *Manager) syncUntilTarget(syncTarget uint64) error {
 		return nil
 	}
 
-	var stateIndex uint64
-	h := m.Store.Height()
-	err := retry.Do(
-		func() error {
-			res, err := m.SLClient.GetHeightState(h)
-			if err != nil {
-				m.logger.Debug("sl client get height state", "error", err)
-				return err
-			}
-			stateIndex = res.State.StateIndex
-			return nil
-		},
-		retry.Attempts(0),
-		retry.Delay(500*time.Millisecond),
-		retry.LastErrorOnly(true),
-		retry.DelayType(retry.FixedDelay),
-	)
-	if err != nil {
-		return fmt.Errorf("get height state: %w", err)
-	}
-	m.updateStateIndex(stateIndex - 1)
-	m.logger.Debug("Sync until target: updated state index pre loop", "stateIndex", stateIndex, "height", h, "syncTarget", syncTarget)
-
 	for currentHeight < syncTarget {
-		currStateIdx := atomic.LoadUint64(&m.LastState.SLStateIndex) + 1
-		m.logger.Info("Syncing until target", "height", currentHeight, "state_index", currStateIdx, "syncTarget", syncTarget)
-		settlementBatch, err := m.SLClient.RetrieveBatch(currStateIdx)
+		var stateIndex uint64
+		h := m.Store.Height() + 1
+
+		m.logger.Debug("Sync until target: updated state index pre loop", "height", h, "syncTarget", syncTarget)
+
+		err := retry.Do(
+			func() error {
+				res, err := m.SLClient.GetHeightState(h)
+				if err != nil {
+					m.logger.Debug("sl client get height state", "error", err)
+					return err
+				}
+				stateIndex = res.State.StateIndex
+				return nil
+			},
+			retry.Attempts(0),
+			retry.Delay(500*time.Millisecond),
+			retry.LastErrorOnly(true),
+			retry.DelayType(retry.FixedDelay),
+		)
+		if err != nil {
+			return fmt.Errorf("get height state: %w", err)
+		}
+
+		m.logger.Info("Retrieving batch", "state_index", stateIndex)
+		settlementBatch, err := m.SLClient.RetrieveBatch(stateIndex)
 		if err != nil {
 			return err
 		}
@@ -82,32 +81,18 @@ func (m *Manager) syncUntilTarget(syncTarget uint64) error {
 
 		currentHeight = m.Store.Height()
 
-		err = m.updateStateIndex(settlementBatch.StateIndex)
-		if err != nil {
-			return err
-		}
 	}
 	m.logger.Info("Synced", "current height", currentHeight, "syncTarget", syncTarget)
 
 	// check for cached blocks
-	err = m.attemptApplyCachedBlocks()
+	err := m.attemptApplyCachedBlocks()
 	if err != nil {
 		m.logger.Error("applying previous cached blocks", "err", err)
 	}
 
 	return nil
 }
 
-func (m *Manager) updateStateIndex(stateIndex uint64) error {
-	atomic.StoreUint64(&m.LastState.SLStateIndex, stateIndex)
-	_, err := m.Store.UpdateState(m.LastState, nil)
-	if err != nil {
-		m.logger.Error("update state", "error", err)
-		return err
-	}
-	return nil
-}
-
 func (m *Manager) ProcessNextDABatch(daMetaData *da.DASubmitMetaData) error {
 	m.logger.Debug("trying to retrieve batch from DA", "daHeight", daMetaData.Height)
 	batchResp := m.fetchBatch(daMetaData)

da/celestia/celestia.go

@@ -2,6 +2,7 @@ package celestia
 
 import (
 	"context"
+	"encoding/hex"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -319,6 +320,7 @@ func (c *DataAvailabilityLayerClient) retrieveBatches(daMetaData *da.DASubmitMet
 	ctx, cancel := context.WithTimeout(c.ctx, c.config.Timeout)
 	defer cancel()
 
+	c.logger.Debug("Celestia DA getting blob", "height", daMetaData.Height, "namespace", hex.EncodeToString(daMetaData.Namespace), "commitment", hex.EncodeToString(daMetaData.Commitment))
 	var batches []*types.Batch
 	blob, err := c.rpc.Get(ctx, daMetaData.Height, daMetaData.Namespace, daMetaData.Commitment)
 	if err != nil {
@@ -345,6 +347,9 @@ func (c *DataAvailabilityLayerClient) retrieveBatches(daMetaData *da.DASubmitMet
 	if err != nil {
 		c.logger.Error("unmarshal block", "daHeight", daMetaData.Height, "error", err)
 	}
+
+	c.logger.Debug("Celestia DA get blob successful", "DA height", daMetaData.Height, "lastBlockHeight", batch.EndHeight)
+
 	parsedBatch := new(types.Batch)
 	err = parsedBatch.FromProto(&batch)
 	if err != nil {

da/celestia/config.go

@@ -12,7 +12,7 @@ import (
 )
 
 const (
-	defaultRpcRetryDelay            = 30 * time.Second
+	defaultRpcRetryDelay            = 1 * time.Second
 	defaultRpcCheckAttempts         = 10
 	namespaceVersion                = 0
 	defaultGasPrices                = 0.1
@@ -42,7 +42,7 @@ type Config struct {
 var CelestiaDefaultConfig = Config{
 	BaseURL:        "http://127.0.0.1:26658",
 	AppNodeURL:     "",
-	Timeout:        30 * time.Second,
+	Timeout:        5 * time.Second,
 	Fee:            0,
 	GasLimit:       20000000,
 	GasPrices:      defaultGasPrices,

utils/event/funcs.go

@@ -20,9 +20,8 @@ func MustSubscribe(
 	eventQuery pubsub.Query,
 	callback func(event pubsub.Message),
 	logger types.Logger,
-	outCapacity ...int,
 ) {
-	subscription, err := pubsubServer.Subscribe(ctx, clientID, eventQuery, outCapacity...)
+	subscription, err := pubsubServer.SubscribeUnbuffered(ctx, clientID, eventQuery)
 	if err != nil {
 		logger.Error("subscribe to events")
 		panic(err)