Fix/reset height pending txs (#24)

* delete pending txs & processed msgs

* send tx immediately even if node is syncing

* l1startheight starts from the given height

* search tx from the exact l1sequence

executor/db.go

@@ -19,6 +19,12 @@ func ResetHeights(db types.DB) error {
 		if err := node.DeleteSyncInfo(db); err != nil {
 			return err
 		}
+		if err := node.DeletePendingTxs(db); err != nil {
+			return err
+		}
+		if err := node.DeleteProcessedMsgs(db); err != nil {
+			return err
+		}
 		fmt.Printf("reset height to 0 for node %s\n", string(db.GetPrefix()))
 	}
 	return nil
@@ -35,6 +41,12 @@ func ResetHeight(db types.DB, nodeName string) error {
 	if err != nil {
 		return err
 	}
+	if err := node.DeletePendingTxs(db); err != nil {
+		return err
+	}
+	if err := node.DeleteProcessedMsgs(db); err != nil {
+		return err
+	}
 	fmt.Printf("reset height to 0 for node %s\n", string(nodeDB.GetPrefix()))
 	return nil
 }

executor/executor.go

@@ -236,14 +236,13 @@ func (ex *Executor) getStartHeights(ctx context.Context, bridgeId uint64) (l1Sta
 	if err != nil {
 		return 0, 0, 0, 0, err
 	}
-	if l1Sequence > 1 {
-		depositTxHeight, err := ex.host.QueryDepositTxHeight(ctx, bridgeId, l1Sequence-1)
-		if err != nil {
-			return 0, 0, 0, 0, err
-		}
-		if l1StartHeight > depositTxHeight {
-			l1StartHeight = depositTxHeight
-		}
+
+	depositTxHeight, err := ex.host.QueryDepositTxHeight(ctx, bridgeId, l1Sequence)
+	if err != nil {
+		return 0, 0, 0, 0, err
+	}
+	if l1StartHeight > depositTxHeight {
+		l1StartHeight = depositTxHeight
 	}
 
 	if l2StartHeight == 0 {
@@ -252,5 +251,8 @@ func (ex *Executor) getStartHeights(ctx context.Context, bridgeId uint64) (l1Sta
 	if ex.cfg.BatchStartHeight > 0 {
 		batchStartHeight = ex.cfg.BatchStartHeight - 1
 	}
+	if l1StartHeight > 0 {
+		l1StartHeight--
+	}
 	return l1StartHeight, l2StartHeight, startOutputIndex, batchStartHeight, err
 }

executor/host/handler.go

@@ -22,9 +22,6 @@ func (h *Host) endBlockHandler(_ context.Context, args nodetypes.EndBlockArgs) e
 	// collect more msgs if block height is not latest
 	blockHeight := args.Block.Header.Height
 	msgQueue := h.GetMsgQueue()
-	if blockHeight != args.LatestHeight && len(msgQueue) > 0 && len(msgQueue) <= 10 {
-		return nil
-	}
 
 	batchKVs := []types.RawKV{
 		h.Node().SyncInfoToRawKV(blockHeight),

node/db.go

@@ -2,6 +2,7 @@ package node
 
 import (
 	dbtypes "github.com/initia-labs/opinit-bots/db/types"
+	btypes "github.com/initia-labs/opinit-bots/node/broadcaster/types"
 	nodetypes "github.com/initia-labs/opinit-bots/node/types"
 	"github.com/initia-labs/opinit-bots/types"
 	"go.uber.org/zap"
@@ -54,3 +55,23 @@ func (n Node) DeleteSyncInfo() error {
 func DeleteSyncInfo(db types.DB) error {
 	return db.Delete(nodetypes.LastProcessedBlockHeightKey)
 }
+
+func DeleteProcessedMsgs(db types.DB) error {
+	return db.PrefixedIterate(btypes.ProcessedMsgsKey, func(key, _ []byte) (stop bool, err error) {
+		err = db.Delete(key)
+		if err != nil {
+			return stop, err
+		}
+		return false, nil
+	})
+}
+
+func DeletePendingTxs(db types.DB) error {
+	return db.PrefixedIterate(btypes.PendingTxsKey, func(key, _ []byte) (stop bool, err error) {
+		err = db.Delete(key)
+		if err != nil {
+			return stop, err
+		}
+		return false, nil
+	})
+}