feat: unlock pending btc proposals when doing retry

app/app.go

@@ -225,7 +225,7 @@ func Run() error {
 				eventHandlers = append(eventHandlers, hubEventHandlers.NewKeygenEventHandler(l, tssListener, coordinator, host, communication, keyshareStore, bridgeAddress, networkTopology.Threshold))
 				eventHandlers = append(eventHandlers, hubEventHandlers.NewFrostKeygenEventHandler(l, tssListener, coordinator, host, communication, frostKeyshareStore, frostAddress, networkTopology.Threshold))
 				eventHandlers = append(eventHandlers, hubEventHandlers.NewRefreshEventHandler(l, topologyProvider, topologyStore, tssListener, coordinator, host, communication, connectionGate, keyshareStore, bridgeAddress))
-				eventHandlers = append(eventHandlers, hubEventHandlers.NewRetryEventHandler(l, tssListener, depositHandler, bridgeAddress, *config.GeneralChainConfig.Id, config.BlockConfirmations, msgChan))
+				eventHandlers = append(eventHandlers, hubEventHandlers.NewRetryEventHandler(l, tssListener, depositHandler, propStore, bridgeAddress, *config.GeneralChainConfig.Id, config.BlockConfirmations, msgChan))
 				evmListener := listener.NewEVMListener(client, eventHandlers, blockstore, sygmaMetrics, *config.GeneralChainConfig.Id, config.BlockRetryInterval, config.BlockConfirmations, config.BlockInterval)
 				executor := executor.NewExecutor(host, communication, coordinator, bridgeContract, keyshareStore, exitLock, config.GasLimit.Uint64())
 

chains/btc/executor/executor.go

@@ -182,6 +182,7 @@ func (e *Executor) watchExecution(ctx context.Context, cancelExecution context.C
 					return err
 				}
 
+				e.storeProposalsStatus(proposals, store.ExecutedProp)
 				log.Info().Str("SessionID", sessionID).Msgf("Sent proposals execution with hash: %s", hash)
 			}
 		case <-timeout.C:

chains/evm/listener/eventHandlers/event-handler.go

@@ -13,6 +13,8 @@ import (
 	"github.com/rs/zerolog/log"
 
 	"github.com/ChainSafe/sygma-relayer/chains/evm/calls/consts"
+	"github.com/ChainSafe/sygma-relayer/relayer/transfer"
+	"github.com/ChainSafe/sygma-relayer/store"
 	"github.com/ethereum/go-ethereum/accounts/abi"
 	"github.com/sygmaprotocol/sygma-core/relayer/message"
 
@@ -38,10 +40,16 @@ type EventListener interface {
 	FetchDeposits(ctx context.Context, address common.Address, startBlock *big.Int, endBlock *big.Int) ([]*events.Deposit, error)
 }
 
+type PropStorer interface {
+	StorePropStatus(source, destination uint8, depositNonce uint64, status store.PropStatus) error
+	PropStatus(source, destination uint8, depositNonce uint64) (store.PropStatus, error)
+}
+
 type RetryEventHandler struct {
 	log                zerolog.Logger
 	eventListener      EventListener
 	depositHandler     DepositHandler
+	propStorer         PropStorer
 	bridgeAddress      common.Address
 	bridgeABI          abi.ABI
 	domainID           uint8
@@ -53,6 +61,7 @@ func NewRetryEventHandler(
 	logC zerolog.Context,
 	eventListener EventListener,
 	depositHandler DepositHandler,
+	propStorer PropStorer,
 	bridgeAddress common.Address,
 	domainID uint8,
 	blockConfirmations *big.Int,
@@ -63,6 +72,7 @@ func NewRetryEventHandler(
 		log:                logC.Logger(),
 		eventListener:      eventListener,
 		depositHandler:     depositHandler,
+		propStorer:         propStorer,
 		bridgeAddress:      bridgeAddress,
 		bridgeABI:          bridgeABI,
 		domainID:           domainID,
@@ -105,6 +115,15 @@ func (eh *RetryEventHandler) HandleEvents(
 					eh.log.Err(err).Str("messageID", msg.ID).Msgf("Failed handling deposit %+v", d)
 					continue
 				}
+				isExecuted, err := eh.isExecuted(msg)
+				if err != nil {
+					eh.log.Err(err).Str("messageID", msg.ID).Msgf("Failed checking if deposit executed %+v", d)
+					continue
+				}
+				if isExecuted {
+					eh.log.Debug().Str("messageID", msg.ID).Msgf("Deposit marked as executed %+v", d)
+					continue
+				}
 
 				eh.log.Info().Str("messageID", msg.ID).Msgf(
 					"Resolved retry message %+v in block range: %s-%s", msg, startBlock.String(), endBlock.String(),
@@ -121,6 +140,31 @@ func (eh *RetryEventHandler) HandleEvents(
 	return nil
 }
 
+func (eh *RetryEventHandler) isExecuted(msg *message.Message) (bool, error) {
+	var err error
+	propStatus, err := eh.propStorer.PropStatus(
+		msg.Source,
+		msg.Destination,
+		msg.Data.(transfer.TransferMessageData).DepositNonce)
+	if err != nil {
+		return true, err
+	}
+
+	if propStatus == store.ExecutedProp {
+		return true, nil
+	}
+
+	// change the status to failed if proposal is stuck to be able to retry it
+	if propStatus == store.PendingProp {
+		err = eh.propStorer.StorePropStatus(
+			msg.Source,
+			msg.Destination,
+			msg.Data.(transfer.TransferMessageData).DepositNonce,
+			store.FailedProp)
+	}
+	return false, err
+}
+
 type KeygenEventHandler struct {
 	log           zerolog.Logger
 	eventListener EventListener

chains/evm/listener/eventHandlers/event-handler_test.go

@@ -18,13 +18,15 @@ import (
 	"github.com/ChainSafe/sygma-relayer/chains/evm/listener/eventHandlers"
 	mock_listener "github.com/ChainSafe/sygma-relayer/chains/evm/listener/eventHandlers/mock"
 	"github.com/ChainSafe/sygma-relayer/relayer/transfer"
+	"github.com/ChainSafe/sygma-relayer/store"
 	"github.com/sygmaprotocol/sygma-core/relayer/message"
 )
 
 type RetryEventHandlerTestSuite struct {
 	suite.Suite
 	retryEventHandler  *eventHandlers.RetryEventHandler
 	mockDepositHandler *mock_listener.MockDepositHandler
+	mockPropStorer     *mock_listener.MockPropStorer
 	mockEventListener  *mock_listener.MockEventListener
 	domainID           uint8
 	msgChan            chan []*message.Message
@@ -39,8 +41,17 @@ func (s *RetryEventHandlerTestSuite) SetupTest() {
 	s.domainID = 1
 	s.mockEventListener = mock_listener.NewMockEventListener(ctrl)
 	s.mockDepositHandler = mock_listener.NewMockDepositHandler(ctrl)
+	s.mockPropStorer = mock_listener.NewMockPropStorer(ctrl)
 	s.msgChan = make(chan []*message.Message, 1)
-	s.retryEventHandler = eventHandlers.NewRetryEventHandler(log.With(), s.mockEventListener, s.mockDepositHandler, common.Address{}, s.domainID, big.NewInt(5), s.msgChan)
+	s.retryEventHandler = eventHandlers.NewRetryEventHandler(
+		log.With(),
+		s.mockEventListener,
+		s.mockDepositHandler,
+		s.mockPropStorer,
+		common.Address{},
+		s.domainID,
+		big.NewInt(5),
+		s.msgChan)
 }
 
 func (s *RetryEventHandlerTestSuite) Test_FetchDepositFails() {
@@ -79,6 +90,7 @@ func (s *RetryEventHandlerTestSuite) Test_FetchDepositFails_ExecutionContinues()
 			DepositNonce: 2,
 		},
 	}, nil)
+	s.mockPropStorer.EXPECT().PropStatus(gomock.Any(), gomock.Any(), gomock.Any()).Return(store.MissingProp, nil)
 
 	err := s.retryEventHandler.HandleEvents(big.NewInt(0), big.NewInt(5))
 	msgs := <-s.msgChan
@@ -132,6 +144,7 @@ func (s *RetryEventHandlerTestSuite) Test_HandleDepositFails_ExecutionContinues(
 	).Return(&message.Message{Data: transfer.TransferMessageData{
 		DepositNonce: 2,
 	}}, nil)
+	s.mockPropStorer.EXPECT().PropStatus(gomock.Any(), gomock.Any(), gomock.Any()).Return(store.MissingProp, nil)
 
 	err := s.retryEventHandler.HandleEvents(big.NewInt(0), big.NewInt(5))
 	msgs := <-s.msgChan
@@ -183,6 +196,7 @@ func (s *RetryEventHandlerTestSuite) Test_HandlingRetryPanics_ExecutionContinue(
 	).Return(&message.Message{Data: transfer.TransferMessageData{
 		DepositNonce: 2,
 	}}, nil)
+	s.mockPropStorer.EXPECT().PropStatus(gomock.Any(), gomock.Any(), gomock.Any()).Return(store.MissingProp, nil)
 
 	err := s.retryEventHandler.HandleEvents(big.NewInt(0), big.NewInt(5))
 	msgs := <-s.msgChan
@@ -235,6 +249,7 @@ func (s *RetryEventHandlerTestSuite) Test_MultipleDeposits() {
 	).Return(&message.Message{Data: transfer.TransferMessageData{
 		DepositNonce: 2,
 	}}, nil)
+	s.mockPropStorer.EXPECT().PropStatus(gomock.Any(), gomock.Any(), gomock.Any()).Return(store.MissingProp, nil).Times(2)
 
 	err := s.retryEventHandler.HandleEvents(big.NewInt(0), big.NewInt(5))
 	msgs := <-s.msgChan
@@ -247,6 +262,65 @@ func (s *RetryEventHandlerTestSuite) Test_MultipleDeposits() {
 	}}})
 }
 
+func (s *RetryEventHandlerTestSuite) Test_MultipleDeposits_ExecutedIgnored() {
+	d1 := events.Deposit{
+		DepositNonce:        1,
+		DestinationDomainID: 2,
+		ResourceID:          [32]byte{},
+		HandlerResponse:     []byte{},
+		Data:                []byte{},
+	}
+	d2 := events.Deposit{
+		DepositNonce:        2,
+		DestinationDomainID: 2,
+		ResourceID:          [32]byte{},
+		HandlerResponse:     []byte{},
+		Data:                []byte{},
+	}
+	s.mockEventListener.EXPECT().FetchRetryEvents(
+		gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(),
+	).Return([]events.RetryEvent{{TxHash: "event1"}}, nil)
+	s.mockEventListener.EXPECT().FetchRetryDepositEvents(events.RetryEvent{TxHash: "event1"}, gomock.Any(), big.NewInt(5)).Return([]events.Deposit{d1, d2}, nil)
+	msgID := fmt.Sprintf("retry-%d-%d-%d-%d", 1, 2, 0, 5)
+	s.mockDepositHandler.EXPECT().HandleDeposit(
+		s.domainID,
+		d1.DestinationDomainID,
+		d1.DepositNonce,
+		d1.ResourceID,
+		d1.Data,
+		d1.HandlerResponse,
+		msgID,
+	).Return(&message.Message{Data: transfer.TransferMessageData{
+		DepositNonce: 1,
+	}}, nil)
+	s.mockPropStorer.EXPECT().PropStatus(gomock.Any(), gomock.Any(), gomock.Any()).Return(store.ExecutedProp, nil)
+	s.mockDepositHandler.EXPECT().HandleDeposit(
+		s.domainID,
+		d2.DestinationDomainID,
+		d2.DepositNonce,
+		d2.ResourceID,
+		d2.Data,
+		d2.HandlerResponse,
+		msgID,
+	).Return(&message.Message{Data: transfer.TransferMessageData{
+		DepositNonce: 2,
+	}}, nil)
+	s.mockPropStorer.EXPECT().PropStatus(gomock.Any(), gomock.Any(), gomock.Any()).Return(store.PendingProp, nil)
+	s.mockPropStorer.EXPECT().StorePropStatus(gomock.Any(), gomock.Any(), gomock.Any(), store.FailedProp).Return(nil)
+
+	err := s.retryEventHandler.HandleEvents(big.NewInt(0), big.NewInt(5))
+	msgs := <-s.msgChan
+
+	s.Nil(err)
+	s.Equal(msgs, []*message.Message{
+		{
+			Data: transfer.TransferMessageData{
+				DepositNonce: 2,
+			},
+		},
+	})
+}
+
 type DepositHandlerTestSuite struct {
 	suite.Suite
 	depositEventHandler *eventHandlers.DepositEventHandler

example/app/app.go

@@ -197,7 +197,7 @@ func Run() error {
 				eventHandlers = append(eventHandlers, hubEventHandlers.NewKeygenEventHandler(l, tssListener, coordinator, host, communication, keyshareStore, bridgeAddress, networkTopology.Threshold))
 				eventHandlers = append(eventHandlers, hubEventHandlers.NewFrostKeygenEventHandler(l, tssListener, coordinator, host, communication, frostKeyshareStore, frostAddress, networkTopology.Threshold))
 				eventHandlers = append(eventHandlers, hubEventHandlers.NewRefreshEventHandler(l, nil, nil, tssListener, coordinator, host, communication, connectionGate, keyshareStore, bridgeAddress))
-				eventHandlers = append(eventHandlers, hubEventHandlers.NewRetryEventHandler(l, tssListener, depositHandler, bridgeAddress, *config.GeneralChainConfig.Id, config.BlockConfirmations, msgChan))
+				eventHandlers = append(eventHandlers, hubEventHandlers.NewRetryEventHandler(l, tssListener, depositHandler, propStore, bridgeAddress, *config.GeneralChainConfig.Id, config.BlockConfirmations, msgChan))
 				evmListener := listener.NewEVMListener(client, eventHandlers, blockstore, sygmaMetrics, *config.GeneralChainConfig.Id, config.BlockRetryInterval, config.BlockConfirmations, config.BlockInterval)
 				executor := executor.NewExecutor(host, communication, coordinator, bridgeContract, keyshareStore, exitLock, config.GasLimit.Uint64())