fix(sync): make sure we use a latest state index as a start point (#760)

Co-authored-by: github-actions <[email protected]>
Co-authored-by: Omri <[email protected]>

CHANGELOG.md

@@ -3,11 +3,15 @@
 
 ### 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))
 * **da:** fixed da path seperator and encoding issue ([#731](https://github.com/dymensionxyz/dymint/issues/731)) ([3a3b219](https://github.com/dymensionxyz/dymint/commit/3a3b21932750fee7eaaa9c186f78e36e3e597746))
 * **DA:** use expo backoff in retries ([#739](https://github.com/dymensionxyz/dymint/issues/739)) ([848085f](https://github.com/dymensionxyz/dymint/commit/848085f70bcaae81fb80da3ab78c4d8b399e13b1))
+* **logging:** added reason for websocket closed debug msg ([#746](https://github.com/dymensionxyz/dymint/issues/746)) ([3aa7d80](https://github.com/dymensionxyz/dymint/commit/3aa7d80ace92b3b0f79e4f338f10bb94c96ab6dd))
+* **logs:** make logs more readable in a couple places, fix race cond ([#749](https://github.com/dymensionxyz/dymint/issues/749)) ([f05ef39](https://github.com/dymensionxyz/dymint/commit/f05ef3957b754c05fbc90aa39eabce80bbe65933))
 * **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))
 
 
 

block/retriever.go

@@ -4,6 +4,9 @@ import (
 	"context"
 	"fmt"
 	"sync/atomic"
+	"time"
+
+	"github.com/avast/retry-go/v4"
 
 	"code.cloudfoundry.org/go-diodes"
 	"github.com/dymensionxyz/dymint/da"
@@ -41,6 +44,29 @@ 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)
@@ -64,7 +90,7 @@ func (m *Manager) syncUntilTarget(syncTarget uint64) error {
 	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)
 	}

block/synctarget.go

@@ -23,6 +23,7 @@ func (m *Manager) SyncTargetLoop(ctx context.Context) {
 			return
 		case event := <-subscription.Out():
 			eventData := event.Data().(*settlement.EventDataNewBatchAccepted)
+
 			if eventData.EndHeight <= m.Store.Height() {
 				m.logger.Debug(
 					"syncTargetLoop: received new settlement batch accepted with batch end height <= current store height, skipping.",

settlement/base.go

@@ -101,6 +101,10 @@ func (b *BaseLayerClient) RetrieveBatch(stateIndex ...uint64) (*ResultRetrieveBa
 	return resultRetrieveBatch, nil
 }
 
+func (b *BaseLayerClient) GetHeightState(u uint64) (*ResultGetHeightState, error) {
+	return b.client.GetHeightState(b.config.RollappID, u)
+}
+
 // GetSequencersList returns the current list of sequencers from the settlement layer
 func (b *BaseLayerClient) GetSequencersList() []*types.Sequencer {
 	return b.sequencersList

settlement/dymension/dymension.go

@@ -6,6 +6,8 @@ import (
 	"strconv"
 	"time"
 
+	"github.com/dymensionxyz/dymint/gerr"
+
 	uevent "github.com/dymensionxyz/dymint/utils/event"
 
 	"google.golang.org/grpc/codes"
@@ -350,6 +352,39 @@ func (d *HubClient) GetBatchAtIndex(rollappID string, index uint64) (*settlement
 	return d.convertStateInfoToResultRetrieveBatch(&stateInfoResp.StateInfo)
 }
 
+func (d *HubClient) GetHeightState(rollappID string, h uint64) (*settlement.ResultGetHeightState, error) {
+	// TODO: dry out with GetBatchAtIndex
+	var stateInfoResp *rollapptypes.QueryGetStateInfoResponse
+	err := d.RunWithRetry(func() error {
+		var err error
+		stateInfoResp, err = d.rollappQueryClient.StateInfo(d.ctx,
+			&rollapptypes.QueryGetStateInfoRequest{RollappId: d.config.RollappID, Height: h})
+
+		if status.Code(err) == codes.NotFound {
+			return retry.Unrecoverable(gerr.ErrNotFound)
+		}
+
+		return err
+	})
+	if err != nil {
+		return nil, err
+	}
+	// not supposed to happen, but just in case
+	if stateInfoResp == nil {
+		return nil, settlement.ErrEmptyResponse
+	}
+	res, err := d.convertStateInfoToResultRetrieveBatch(&stateInfoResp.StateInfo)
+	if err != nil {
+		return nil, fmt.Errorf("convert state info to result retrieve batch: %w", err)
+	}
+	return &settlement.ResultGetHeightState{
+		BaseResult: res.BaseResult,
+		State: settlement.State{
+			StateIndex: res.BaseResult.StateIndex,
+		},
+	}, nil
+}
+
 // GetSequencers returns the bonded sequencers of the given rollapp.
 func (d *HubClient) GetSequencers(rollappID string) ([]*types.Sequencer, error) {
 	var res *sequencertypes.QueryGetSequencersByRollappByStatusResponse
@@ -428,10 +463,7 @@ func (d *HubClient) eventHandler() {
 			if err != nil {
 				panic(err)
 			}
-			err = d.pubsub.PublishWithEvents(d.ctx, eventData, map[string][]string{settlement.EventTypeKey: {d.eventMap[event.Query]}})
-			if err != nil {
-				panic(err)
-			}
+			uevent.MustPublish(d.ctx, d.pubsub, eventData, map[string][]string{settlement.EventTypeKey: {d.eventMap[event.Query]}})
 		}
 	}
 }

settlement/grpc/grpc.go

@@ -73,6 +73,11 @@ type HubGrpcClient struct {
 	refreshTime    int
 }
 
+func (c *HubGrpcClient) GetHeightState(rollappID string, index uint64) (*settlement.ResultGetHeightState, error) {
+	// TODO implement me
+	panic("implement me")
+}
+
 func newHubClient(config settlement.Config, pubsub *pubsub.Server, logger types.Logger) (*HubGrpcClient, error) {
 	ctx := context.Background()
 

settlement/local/local.go

@@ -72,6 +72,11 @@ type HubClient struct {
 	settlementKV   store.KVStore
 }
 
+func (c *HubClient) GetHeightState(rollappID string, index uint64) (*settlement.ResultGetHeightState, error) {
+	// TODO implement me
+	panic("implement me")
+}
+
 var _ settlement.HubClient = &HubClient{}
 
 func newHubClient(config settlement.Config, pubsub *pubsub.Server, logger types.Logger) (*HubClient, error) {

settlement/settlement.go

@@ -48,6 +48,15 @@ type ResultRetrieveBatch struct {
 	*Batch
 }
 
+type State struct {
+	StateIndex uint64
+}
+
+type ResultGetHeightState struct {
+	BaseResult // NOTE: the state index of this will not be populated
+	State
+}
+
 // Option is a function that sets a parameter on the settlement layer.
 type Option func(LayerI)
 
@@ -74,6 +83,8 @@ type LayerI interface {
 
 	// GetProposer returns the current proposer for this chain.
 	GetProposer() *types.Sequencer
+
+	GetHeightState(uint64) (*ResultGetHeightState, error)
 }
 
 // HubClient is a helper interface for a more granular interaction with the hub.
@@ -85,5 +96,6 @@ type HubClient interface {
 	PostBatch(batch *types.Batch, daClient da.Client, daResult *da.ResultSubmitBatch) error
 	GetLatestBatch(rollappID string) (*ResultRetrieveBatch, error)
 	GetBatchAtIndex(rollappID string, index uint64) (*ResultRetrieveBatch, error)
+	GetHeightState(rollappID string, index uint64) (*ResultGetHeightState, error)
 	GetSequencers(rollappID string) ([]*types.Sequencer, error)
 }