Improve mpcium resiliency

CHANGELOG.md

@@ -1,6 +1,13 @@
 # CHANGELOG
 
-This changelog is a work in progress and may contain notes for versions which have not actually been released. Check the [Releases](https://github.com/cryptoniumX/mpcium/releases) page to see full release notes and more information about the latest released versions.
+## v0.2.0 (2025-04-12)
+
+- Use onSuccess callback when sign session succeeds [View](https://github.com/cryptoniumX/mpcium/commit/9602d4d9bfe37c2d038856d3ed206bfecd2e8c93)
+- Fix bug signing doesn't work after all nodes are backup [View](https://github.com/cryptoniumX/mpcium/commit/a9192ca11581dd986bdd21728cbda4b78d75a753)
+- Handle duplicate message [View](https://github.com/cryptoniumX/mpcium/commit/e79f6e20fbe225e5aad8b0c9e70578356fce9573)
+- Update timeout consumer keep subscribe on time [View](https://github.com/cryptoniumX/mpcium/commit/52ee83c3ecc2bbb8c16a8227f4f00b72a57c8499)
+- Update signing timeout logic when not enough participants [View](https://github.com/cryptoniumX/mpcium/commit/e8ffa381f489a83e60dbcbf5262927e99eca2382)
+- Persit message, handle failure and timeout sign tx [View](https://github.com/cryptoniumX/mpcium/commit/400f26912ea6b31cbf511de93c1270776055c758)
 
 ## v0.1.7 (2024-10-05)
 

README.md

@@ -32,3 +32,7 @@ $ go run cmd/main.go --name=mpcium1
 $ go run cmd/main.go --name=mpcium2
 
 ```
+
+### Diagaram
+
+![Diagram](images/diagram.png)

cmd/main.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"context"
 	"flag"
 	"fmt"
 	"os"
@@ -9,6 +10,7 @@ import (
 
 	"github.com/cryptoniumX/mpcium/pkg/config"
 	"github.com/cryptoniumX/mpcium/pkg/constant"
+	"github.com/cryptoniumX/mpcium/pkg/event"
 	"github.com/cryptoniumX/mpcium/pkg/eventconsumer"
 	"github.com/cryptoniumX/mpcium/pkg/infra"
 	"github.com/cryptoniumX/mpcium/pkg/keyinfo"
@@ -54,16 +56,23 @@ func main() {
 	defer natsConn.Close()
 
 	pubsub := messaging.NewNATSPubSub(natsConn)
+	signingStream, err := messaging.NewJetStreamPubSub(natsConn, event.SigningPublisherStream, []string{
+		event.SigningRequestTopic,
+	})
+	if err != nil {
+		logger.Fatal("Failed to create JetStream PubSub", err)
+	}
+
 	directMessaging := messaging.NewNatsDirectMessaging(natsConn)
 	mqManager := messaging.NewNATsMessageQueueManager("mpc", []string{
 		"mpc.mpc_keygen_success.*",
-		"mpc.mpc_sign_success.*",
+		event.SigningResultTopic,
 	}, natsConn)
 
 	genKeySuccessQueue := mqManager.NewMessageQueue("mpc_keygen_success")
 	defer genKeySuccessQueue.Close()
-	singingSuccessQueue := mqManager.NewMessageQueue("mpc_sign_success")
-	defer singingSuccessQueue.Close()
+	singingResultQueue := mqManager.NewMessageQueue("signing_result")
+	defer singingResultQueue.Close()
 
 	logger.Info("Node is running", "peerID", nodeID, "name", *nodeName)
 
@@ -85,18 +94,36 @@ func main() {
 		mpcNode,
 		pubsub,
 		genKeySuccessQueue,
-		singingSuccessQueue,
+		singingResultQueue,
 	)
 	eventConsumer.Run()
 	defer eventConsumer.Close()
-	// Create a channel to receive signals
 
-	signals := make(chan os.Signal, 1)
-	signal.Notify(signals, syscall.SIGINT, syscall.SIGTERM)
-
-	// Block the execution until a signal is received
-	<-signals
+	timeoutConsumer := eventconsumer.NewTimeOutConsumer(
+		natsConn,
+		singingResultQueue,
+	)
 
+	timeoutConsumer.Run()
+	defer timeoutConsumer.Close()
+	signingConsumer := eventconsumer.NewSigningConsumer(natsConn, signingStream, pubsub)
+
+	// Make the node ready before starting the signing consumer
+	peerRegistry.Ready()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	// Setup signal handling to cancel context on termination signals.
+	go func() {
+		sigChan := make(chan os.Signal, 1)
+		signal.Notify(sigChan, os.Interrupt, syscall.SIGTERM)
+		<-sigChan
+		logger.Warn("Shutdown signal received, canceling context...")
+		cancel()
+	}()
+
+	if err := signingConsumer.Run(ctx); err != nil {
+		logger.Error("error running consumer:", err)
+	}
 }
 
 func NewConsulClient(addr string) *api.Client {

pkg/event/sign.go

@@ -0,0 +1,54 @@
+package event
+
+const (
+	SigningPublisherStream     = "mpc-signing"
+	SigningConsumerStream      = "mpc-signing-consumer"
+	SigningRequestTopic        = "mpc.signing_request.*"
+	SigningResultTopic         = "mpc.signing_result.*"
+	SigningResultCompleteTopic = "mpc.signing_result.complete"
+	MPCSigningEventTopic       = "mpc:sign"
+	SigningRequestEventTopic   = "mpc.signing_request.event"
+)
+
+type SigningResultType int
+
+const (
+	SigningResultTypeUnknown SigningResultType = iota
+	SigningResultTypeSuccess
+	SigningResultTypeError
+)
+
+type SigningResultEvent struct {
+	ResultType          SigningResultType `json:"result_type"`
+	ErrorReason         string            `json:"error_reason"`
+	IsTimeout           bool              `json:"is_timeout"`
+	NetworkInternalCode string            `json:"network_internal_code"`
+	WalletID            string            `json:"wallet_id"`
+	TxID                string            `json:"tx_id"`
+	R                   []byte            `json:"r"`
+	S                   []byte            `json:"s"`
+	SignatureRecovery   []byte            `json:"signature_recovery"`
+
+	// TODO: define two separate events for eddsa and ecdsa
+	Signature []byte `json:"signature"`
+}
+
+type SigningResultSuccessEvent struct {
+	NetworkInternalCode string `json:"network_internal_code"`
+	WalletID            string `json:"wallet_id"`
+	TxID                string `json:"tx_id"`
+	R                   []byte `json:"r"`
+	S                   []byte `json:"s"`
+	SignatureRecovery   []byte `json:"signature_recovery"`
+
+	// TODO: define two separate events for eddsa and ecdsa
+	Signature []byte `json:"signature"`
+}
+
+type SigningResultErrorEvent struct {
+	NetworkInternalCode string `json:"network_internal_code"`
+	WalletID            string `json:"wallet_id"`
+	TxID                string `json:"tx_id"`
+	ErrorReason         string `json:"error_reason"`
+	IsTimeout           bool   `json:"is_timeout"`
+}