Error handling for not enough majority nodes to sign

Author: anhthii

cmd/mpcium/main.go

@@ -194,7 +194,7 @@ func runNode(ctx context.Context, c *cli.Command) error {
 	timeoutConsumer.Run()
 	defer timeoutConsumer.Close()
 	keygenConsumer := eventconsumer.NewKeygenConsumer(natsConn, keygenBroker, pubsub, peerRegistry)
-	signingConsumer := eventconsumer.NewSigningConsumer(natsConn, signingBroker, pubsub, peerRegistry)
+	signingConsumer := eventconsumer.NewSigningConsumer(natsConn, signingBroker, pubsub, peerRegistry, singingResultQueue)
 
 	// Make the node ready before starting the signing consumer
 	if err := peerRegistry.Ready(); err != nil {

pkg/event/types.go

@@ -91,6 +91,7 @@ const (
 	// Context and cancellation errors
 	ErrorCodeContextCancelled ErrorCode = "ERROR_CONTEXT_CANCELLED"
 	ErrorCodeOperationAborted ErrorCode = "ERROR_OPERATION_ABORTED"
+	ErrorCodeNotMajority      ErrorCode = "ERROR_NOT_MAJORITY"
 )
 
 // GetErrorCodeFromError attempts to categorize a generic error into a specific error code

pkg/eventconsumer/sign_consumer.go

@@ -2,13 +2,15 @@ package eventconsumer
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"time"
 
 	"github.com/fystack/mpcium/pkg/event"
 	"github.com/fystack/mpcium/pkg/logger"
 	"github.com/fystack/mpcium/pkg/messaging"
 	"github.com/fystack/mpcium/pkg/mpc"
+	"github.com/fystack/mpcium/pkg/types"
 	"github.com/google/uuid"
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
@@ -34,25 +36,27 @@ type SigningConsumer interface {
 
 // signingConsumer implements SigningConsumer.
 type signingConsumer struct {
-	natsConn     *nats.Conn
-	pubsub       messaging.PubSub
-	jsBroker     messaging.MessageBroker
-	peerRegistry mpc.PeerRegistry
-	mpcThreshold int
+	natsConn           *nats.Conn
+	pubsub             messaging.PubSub
+	jsBroker           messaging.MessageBroker
+	peerRegistry       mpc.PeerRegistry
+	mpcThreshold       int
+	signingResultQueue messaging.MessageQueue
 
 	// jsSub holds the JetStream subscription, so it can be cleaned up during Close().
 	jsSub messaging.Subscription
 }
 
 // NewSigningConsumer returns a new instance of SigningConsumer.
-func NewSigningConsumer(natsConn *nats.Conn, jsBroker messaging.MessageBroker, pubsub messaging.PubSub, peerRegistry mpc.PeerRegistry) SigningConsumer {
+func NewSigningConsumer(natsConn *nats.Conn, jsBroker messaging.MessageBroker, pubsub messaging.PubSub, peerRegistry mpc.PeerRegistry, signingResultQueue messaging.MessageQueue) SigningConsumer {
 	mpcThreshold := viper.GetInt("mpc_threshold")
 	return &signingConsumer{
-		natsConn:     natsConn,
-		pubsub:       pubsub,
-		jsBroker:     jsBroker,
-		peerRegistry: peerRegistry,
-		mpcThreshold: mpcThreshold,
+		natsConn:           natsConn,
+		pubsub:             pubsub,
+		jsBroker:           jsBroker,
+		peerRegistry:       peerRegistry,
+		mpcThreshold:       mpcThreshold,
+		signingResultQueue: signingResultQueue,
 	}
 }
 
@@ -136,18 +140,25 @@ func (sc *signingConsumer) Run(ctx context.Context) error {
 // When signing completes, the session publishes the result to a queue and calls the onSuccess callback, which sends a reply to the inbox that the SigningConsumer is monitoring.
 // The reply signals completion, allowing the SigningConsumer to acknowledge the original message.
 func (sc *signingConsumer) handleSigningEvent(msg jetstream.Msg) {
-	// Check if we still have enough peers before processing the message
-	requiredPeers := int64(sc.mpcThreshold + 1)
-	readyPeers := sc.peerRegistry.GetReadyPeersCount()
-
-	if readyPeers < requiredPeers {
-		logger.Warn("SigningConsumer: Not enough peers to process signing request, rejecting message",
-			"ready", readyPeers,
-			"required", requiredPeers)
-		// Immediately return and let nats redeliver the message with backoff
+	// Parse the signing request message to extract transaction details
+	raw := msg.Data()
+	var signingMsg types.SignTxMessage
+	sessionID := msg.Headers().Get("SessionID")
+
+	err := json.Unmarshal(raw, &signingMsg)
+	if err != nil {
+		logger.Error("SigningConsumer: Failed to unmarshal signing message", err)
+		sc.handleSigningError(signingMsg, event.ErrorCodeUnmarshalFailure, err, sessionID)
+		_ = msg.Nak()
 		return
 	}
 
+	if !sc.peerRegistry.AreMajorityReady() {
+		requiredPeers := int64(sc.mpcThreshold + 1)
+		err := fmt.Errorf("not enough peers to process signing request: ready=%d, required=%d", sc.peerRegistry.GetReadyPeersCount(), requiredPeers)
+		sc.handleSigningError(signingMsg, event.ErrorCodeNotMajority, err, sessionID)
+		return
+	}
 	// Create a reply inbox to receive the signing event response.
 	replyInbox := nats.NewInbox()
 
@@ -199,6 +210,36 @@ func (sc *signingConsumer) handleSigningEvent(msg jetstream.Msg) {
 	_ = msg.Nak()
 }
 
+func (sc *signingConsumer) handleSigningError(signMsg types.SignTxMessage, errorCode event.ErrorCode, err error, sessionID string) {
+	signingResult := event.SigningResultEvent{
+		ResultType:          event.ResultTypeError,
+		ErrorCode:           errorCode,
+		NetworkInternalCode: signMsg.NetworkInternalCode,
+		WalletID:            signMsg.WalletID,
+		TxID:                signMsg.TxID,
+		ErrorReason:         err.Error(),
+	}
+
+	signingResultBytes, err := json.Marshal(signingResult)
+	if err != nil {
+		logger.Error("Failed to marshal signing result event", err,
+			"walletID", signMsg.WalletID,
+			"txID", signMsg.TxID,
+		)
+		return
+	}
+
+	err = sc.signingResultQueue.Enqueue(event.SigningResultCompleteTopic, signingResultBytes, &messaging.EnqueueOptions{
+		IdempotententKey: buildSigningIdempotentKey(signMsg.TxID, sessionID, mpc.TypeSigningResultFmt),
+	})
+	if err != nil {
+		logger.Error("Failed to enqueue signing result event", err,
+			"walletID", signMsg.WalletID,
+			"txID", signMsg.TxID,
+		)
+	}
+}
+
 // Close unsubscribes from the JetStream subject and cleans up resources.
 func (sc *signingConsumer) Close() error {
 	if sc.jsSub != nil {
@@ -210,3 +251,13 @@ func (sc *signingConsumer) Close() error {
 	}
 	return nil
 }
+
+func buildSigningIdempotentKey(baseID string, sessionID string, formatTemplate string) string {
+	var uniqueKey string
+	if sessionID != "" {
+		uniqueKey = fmt.Sprintf("%s:%s", baseID, sessionID)
+	} else {
+		uniqueKey = baseID
+	}
+	return fmt.Sprintf(formatTemplate, uniqueKey)
+}

pkg/mpc/registry.go

@@ -14,6 +14,7 @@ import (
 	"github.com/fystack/mpcium/pkg/messaging"
 	"github.com/hashicorp/consul/api"
 	"github.com/samber/lo"
+	"github.com/spf13/viper"
 )
 
 const (
@@ -23,6 +24,7 @@ const (
 type PeerRegistry interface {
 	Ready() error
 	ArePeersReady() bool
+	AreMajorityReady() bool
 	WatchPeersReady()
 	// Resign is called by the node when it is going to shutdown
 	Resign() error
@@ -49,6 +51,7 @@ type registry struct {
 	pubSub        messaging.PubSub
 	identityStore identity.Store
 	ecdhSession   ECDHSession
+	mpcThreshold  int
 
 	onPeerConnected    func(peerID string)
 	onPeerDisconnected func(peerID string)
@@ -64,6 +67,10 @@ func NewRegistry(
 	identityStore identity.Store,
 ) *registry {
 	ecdhSession := NewECDHSession(nodeID, peerNodeIDs, pubSub, identityStore)
+	mpcThreshold := viper.GetInt("mpc_threshold")
+	if mpcThreshold <= 2 {
+		logger.Fatal("mpc_threshold must be greater than 2", nil)
+	}
 
 	return &registry{
 		consulKV:      consulKV,
@@ -75,6 +82,7 @@ func NewRegistry(
 		pubSub:        pubSub,
 		identityStore: identityStore,
 		ecdhSession:   ecdhSession,
+		mpcThreshold:  mpcThreshold,
 	}
 }
 
@@ -286,6 +294,15 @@ func (r *registry) ArePeersReady() bool {
 	return r.ready && r.isECDHReady()
 }
 
+// AreMajorityReady checks if a majority of peers are ready.
+// Returns true only if:
+//  1. The number of ready peers (including self) is greater than mpcThreshold+1
+//  2. Symmetric keys are fully established among all ready peers (excluding self).
+func (r *registry) AreMajorityReady() bool {
+	readyCount := r.GetReadyPeersCount()
+	return int(readyCount) >= r.mpcThreshold+1 && r.isECDHReady()
+}
+
 func (r *registry) GetTotalPeersCount() int64 {
 	var self int64 = 1
 	return int64(len(r.peerNodeIDs)) + self
@@ -332,8 +349,8 @@ func (r *registry) GetReadyPeersCountExcludeSelf() int64 {
 }
 
 func (r *registry) isECDHReady() bool {
-	requiredKeyCount := int(r.GetReadyPeersCount()) - 1
-	return r.identityStore.CheckSymmetricKeyComplete(requiredKeyCount)
+	requiredKeyCount := r.GetReadyPeersCountExcludeSelf()
+	return r.identityStore.CheckSymmetricKeyComplete(int(requiredKeyCount))
 }
 
 func (r *registry) composeHealthCheckTopic(nodeID string) string {