Enhance error handle when signing with duplicate tx_id (Issue #60)

pkg/eventconsumer/event_consumer.go

@@ -336,6 +336,7 @@ func (ec *eventConsumer) consumeTxSigningEvent() error {
 		}
 
 		var session mpc.SigningSession
+		idempotentKey := composeSigningIdempotentKey(msg.TxID, natMsg)
 		switch msg.KeyType {
 		case types.KeyTypeSecp256k1:
 			session, err = ec.node.CreateSigningSession(
@@ -344,6 +345,7 @@ func (ec *eventConsumer) consumeTxSigningEvent() error {
 				msg.TxID,
 				msg.NetworkInternalCode,
 				ec.signingResultQueue,
+				idempotentKey,
 			)
 		case types.KeyTypeEd25519:
 			session, err = ec.node.CreateSigningSession(
@@ -352,6 +354,7 @@ func (ec *eventConsumer) consumeTxSigningEvent() error {
 				msg.TxID,
 				msg.NetworkInternalCode,
 				ec.signingResultQueue,
+				idempotentKey,
 			)
 
 		}
@@ -477,7 +480,7 @@ func (ec *eventConsumer) handleSigningSessionError(walletID, txID, networkIntern
 		return
 	}
 	err = ec.signingResultQueue.Enqueue(event.SigningResultCompleteTopic, signingResultBytes, &messaging.EnqueueOptions{
-		IdempotententKey: txID,
+		IdempotententKey: composeSigningIdempotentKey(txID, natMsg),
 	})
 	if err != nil {
 		logger.Error("Failed to enqueue signing result event", err,
@@ -809,3 +812,11 @@ func composeKeygenIdempotentKey(walletID string, natMsg *nats.Msg) string {
 	}
 	return fmt.Sprintf(mpc.TypeGenerateWalletResultFmt, uniqueKey)
 }
+
+func composeSigningIdempotentKey(txID string, natMsg *nats.Msg) string {
+	sid := natMsg.Header.Get("SessionID")
+	if sid != "" {
+		return fmt.Sprintf("%s:%s", txID, sid)
+	}
+	return txID
+}

pkg/mpc/ecdsa_signing_session.go

@@ -52,6 +52,7 @@ func newECDSASigningSession(
 	keyinfoStore keyinfo.Store,
 	resultQueue messaging.MessageQueue,
 	identityStore identity.Store,
+	idempotentKey string,
 ) *ecdsaSigningSession {
 	return &ecdsaSigningSession{
 		session: session{
@@ -81,6 +82,7 @@ func newECDSASigningSession(
 			getRoundFunc:  GetEcdsaMsgRound,
 			resultQueue:   resultQueue,
 			identityStore: identityStore,
+			idempotentKey: idempotentKey,
 		},
 		endCh:               make(chan *common.SignatureData),
 		txID:                txID,
@@ -178,7 +180,7 @@ func (s *ecdsaSigningSession) Sign(onSuccess func(data []byte)) {
 			}
 
 			err = s.resultQueue.Enqueue(event.SigningResultCompleteTopic, bytes, &messaging.EnqueueOptions{
-				IdempotententKey: s.txID,
+				IdempotententKey: s.idempotentKey,
 			})
 			if err != nil {
 				s.ErrCh <- errors.Wrap(err, "Failed to publish sign success message")

pkg/mpc/eddsa_signing_session.go

@@ -43,6 +43,7 @@ func newEDDSASigningSession(
 	keyinfoStore keyinfo.Store,
 	resultQueue messaging.MessageQueue,
 	identityStore identity.Store,
+	idempotentKey string,
 ) *eddsaSigningSession {
 	return &eddsaSigningSession{
 		session: session{
@@ -72,6 +73,7 @@ func newEDDSASigningSession(
 			getRoundFunc:  GetEddsaMsgRound,
 			resultQueue:   resultQueue,
 			identityStore: identityStore,
+			idempotentKey:  idempotentKey,
 		},
 		endCh:               make(chan *common.SignatureData),
 		txID:                txID,
@@ -167,7 +169,7 @@ func (s *eddsaSigningSession) Sign(onSuccess func(data []byte)) {
 			}
 
 			err = s.resultQueue.Enqueue(event.SigningResultCompleteTopic, bytes, &messaging.EnqueueOptions{
-				IdempotententKey: s.txID,
+				IdempotententKey: s.idempotentKey,
 			})
 			if err != nil {
 				s.ErrCh <- errors.Wrap(err, "Failed to publish sign success message")

pkg/mpc/node.go

@@ -165,6 +165,7 @@ func (p *Node) CreateSigningSession(
 	txID string,
 	networkInternalCode string,
 	resultQueue messaging.MessageQueue,
+	idempotentKey string,
 ) (SigningSession, error) {
 	version := p.getVersion(sessionType, walletID)
 	keyInfo, err := p.getKeyInfo(sessionType, walletID)
@@ -211,6 +212,7 @@ func (p *Node) CreateSigningSession(
 			p.keyinfoStore,
 			resultQueue,
 			p.identityStore,
+			idempotentKey,
 		), nil
 
 	case SessionTypeEDDSA:
@@ -228,6 +230,7 @@ func (p *Node) CreateSigningSession(
 			p.keyinfoStore,
 			resultQueue,
 			p.identityStore,
+			idempotentKey,
 		), nil
 	}
 

pkg/mpc/session.go

@@ -72,8 +72,9 @@ type session struct {
 	getRoundFunc  GetRoundFunc
 	mu            sync.Mutex
 	// After the session is done, the key will be stored pubkeyBytes
-	pubkeyBytes []byte
-	sessionType SessionType
+	pubkeyBytes   []byte
+	sessionType   SessionType
+	idempotentKey string
 }
 
 func (s *session) PartyID() *tss.PartyID {